dekun
89 lines
2.6 KiB
Python
89 lines
2.6 KiB
Python
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from packages.db.backup import (
|
|
DISPLAY_LIMIT,
|
|
backup_db,
|
|
is_valid_backup_name,
|
|
list_backups,
|
|
prune_backups,
|
|
resolve_backup_file,
|
|
restore_db,
|
|
should_auto_backup,
|
|
)
|
|
|
|
|
|
def _make_db(path: Path, value: str = "hello") -> None:
|
|
conn = __import__("sqlite3").connect(str(path))
|
|
conn.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, v TEXT)")
|
|
conn.execute("INSERT INTO t(v) VALUES (?)", (value,))
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
|
|
def test_backup_roundtrip(tmp_path: Path):
|
|
db = tmp_path / "live.db"
|
|
backup_dir = tmp_path / "backups"
|
|
_make_db(db)
|
|
|
|
dest = backup_db(db, backup_dir, tz="UTC")
|
|
assert dest.is_file()
|
|
assert is_valid_backup_name(dest.name)
|
|
|
|
conn = __import__("sqlite3").connect(str(db))
|
|
conn.execute("DELETE FROM t")
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
restore_db(dest, db)
|
|
conn = __import__("sqlite3").connect(str(db))
|
|
row = conn.execute("SELECT v FROM t").fetchone()
|
|
conn.close()
|
|
assert row[0] == "hello"
|
|
|
|
items = list_backups(backup_dir)
|
|
assert len(items) == 1
|
|
assert items[0]["name"] == dest.name
|
|
assert resolve_backup_file(backup_dir, dest.name) == dest.resolve()
|
|
|
|
|
|
def test_list_limit_and_prune(tmp_path: Path):
|
|
backup_dir = tmp_path / "backups"
|
|
backup_dir.mkdir()
|
|
for i in range(7):
|
|
p = backup_dir / f"market_intel_2026080{i+1}_120000.db"
|
|
p.write_bytes(b"x" * (i + 1))
|
|
# mtime 递增，便于 prune 保留最新
|
|
import os
|
|
|
|
os.utime(p, (1_700_000_000 + i, 1_700_000_000 + i))
|
|
|
|
all_items = list_backups(backup_dir)
|
|
assert len(all_items) == 7
|
|
limited = list_backups(backup_dir, limit=DISPLAY_LIMIT)
|
|
assert len(limited) == 5
|
|
assert limited[0]["name"] == "market_intel_20260807_120000.db"
|
|
|
|
removed = prune_backups(backup_dir, keep=3)
|
|
assert removed == 4
|
|
left = list_backups(backup_dir)
|
|
assert len(left) == 3
|
|
assert left[0]["name"] == "market_intel_20260807_120000.db"
|
|
|
|
|
|
def test_should_auto_backup(tmp_path: Path):
|
|
assert should_auto_backup(tmp_path, enabled=False, interval_hours=24) is False
|
|
assert should_auto_backup(tmp_path, enabled=True, interval_hours=24) is True
|
|
|
|
db = tmp_path / "live.db"
|
|
_make_db(db)
|
|
backup_db(db, tmp_path, tz="UTC")
|
|
assert should_auto_backup(tmp_path, enabled=True, interval_hours=24) is False
|
|
assert should_auto_backup(tmp_path, enabled=True, interval_hours=0.000001) is True
|
|
|
|
|
|
def test_invalid_backup_name(tmp_path: Path):
|
|
with pytest.raises(ValueError):
|
|
resolve_backup_file(tmp_path, "../etc/passwd")
|