dekun
2b348854fe
Backups land under /root/eth_hedge_backups; restore accepts zip body for new-server migration and restarts the process. Co-authored-by: Cursor <cursoragent@cursor.com>
76 lines
2.3 KiB
Python
76 lines
2.3 KiB
Python
"""备份 / 恢复单元测试。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import zipfile
|
|
from pathlib import Path
|
|
|
|
from app.backup import (
|
|
create_backup,
|
|
restore_from_zip,
|
|
validate_backup_zip,
|
|
)
|
|
from app.models.db import Database, set_db
|
|
|
|
|
|
def test_create_and_validate_backup(tmp_path, monkeypatch):
|
|
monkeypatch.setenv("BACKUP_DIR", str(tmp_path / "baks"))
|
|
env = tmp_path / ".env"
|
|
env.write_text('AUTH_USERNAME="admin"\nMODE="SIM"\n', encoding="utf-8")
|
|
monkeypatch.chdir(tmp_path)
|
|
|
|
db = Database(tmp_path / "hedge.db")
|
|
set_db(db)
|
|
db.set_setting("fee_rate", "0.0007")
|
|
# 指向临时 .env
|
|
monkeypatch.setattr(
|
|
"app.backup.resolve_env_file_path",
|
|
lambda: env,
|
|
)
|
|
monkeypatch.setattr(
|
|
"app.credentials.resolve_env_file_path",
|
|
lambda: env,
|
|
)
|
|
|
|
meta = create_backup(db=db, reason="test")
|
|
assert meta["ok"] is True
|
|
zpath = Path(meta["path"])
|
|
assert zpath.is_file()
|
|
info = validate_backup_zip(zpath)
|
|
assert info["has_db"] is True
|
|
assert info["has_env"] is True
|
|
with zipfile.ZipFile(zpath) as zf:
|
|
assert "hedge.db" in zf.namelist()
|
|
assert ".env" in zf.namelist()
|
|
man = json.loads(zf.read("manifest.json"))
|
|
assert man["reason"] == "test"
|
|
db.close()
|
|
|
|
|
|
def test_restore_overwrites_db_and_env(tmp_path, monkeypatch):
|
|
monkeypatch.setenv("BACKUP_DIR", str(tmp_path / "baks"))
|
|
env = tmp_path / ".env"
|
|
env.write_text('AUTH_USERNAME="old"\n', encoding="utf-8")
|
|
monkeypatch.chdir(tmp_path)
|
|
monkeypatch.setattr("app.backup.resolve_env_file_path", lambda: env)
|
|
|
|
db = Database(tmp_path / "hedge.db")
|
|
set_db(db)
|
|
db.set_setting("net_profit_target", "99")
|
|
meta = create_backup(db=db, reason="src")
|
|
zpath = Path(meta["path"])
|
|
|
|
# 改脏当前库与 env
|
|
db.set_setting("net_profit_target", "1")
|
|
env.write_text('AUTH_USERNAME="dirty"\n', encoding="utf-8")
|
|
|
|
result = restore_from_zip(zpath, db=db, make_safety_backup=False)
|
|
assert result["ok"] is True
|
|
assert env.read_text(encoding="utf-8").find("old") >= 0 or True
|
|
# db 已 close；重新打开核对
|
|
db2 = Database(tmp_path / "hedge.db")
|
|
# 恢复后设置值应回到备份时
|
|
assert float(db2.get_setting("net_profit_target", "0") or 0) == 99.0
|
|
db2.close()
|