Add DB+.env auto backup with download and upload restore in Settings.
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>
This commit is contained in:
@@ -0,0 +1,75 @@
|
||||
"""备份 / 恢复单元测试。"""
|
||||
|
||||
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()
|
||||
Reference in New Issue
Block a user