kairee
5c2ab7b187
Implement P1: seven-metric view snapshots, relation/stat fan snapshots, failure placeholders, resume by daily completion markers, Cookie login warnings, daily SQLite backups, Shanghai daily scheduling, and CLI snapshot/auth-check commands. Real read-only run captured 795/795 videos successfully over 1124s with a verified daily backup.
79 lines
2.6 KiB
Python
79 lines
2.6 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""每日备份与保留期测试: uv run python tests/test_maintenance.py"""
|
|
|
|
import os
|
|
import sqlite3
|
|
import sys
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
sys.stdout.reconfigure(encoding="utf-8")
|
|
|
|
from reaction_board.database import Database # noqa: E402
|
|
from reaction_board.services.maintenance import ( # noqa: E402
|
|
backup_daily,
|
|
daily_backup_path,
|
|
prune_daily_backups,
|
|
)
|
|
|
|
|
|
def test_daily_backup_and_restore():
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
db = Database(Path(tmp) / "app.db")
|
|
db.initialize()
|
|
try:
|
|
db.write(lambda c: c.execute("INSERT INTO kv(k, v) VALUES ('x', '42')"))
|
|
target = backup_daily(db, Path(tmp) / "backup", "2026-10-31")
|
|
assert target == daily_backup_path(Path(tmp) / "backup", "2026-10-31")
|
|
assert target.name == "app-20261031.db"
|
|
assert target.exists()
|
|
|
|
check = sqlite3.connect(target)
|
|
try:
|
|
assert check.execute("SELECT v FROM kv WHERE k='x'").fetchone()[0] == "42"
|
|
finally:
|
|
check.close()
|
|
|
|
# 覆盖同日备份也安全
|
|
db.write(lambda c: c.execute("UPDATE kv SET v='43' WHERE k='x'"))
|
|
backup_daily(db, Path(tmp) / "backup", "2026-10-31")
|
|
check = sqlite3.connect(target)
|
|
try:
|
|
assert check.execute("SELECT v FROM kv WHERE k='x'").fetchone()[0] == "43"
|
|
finally:
|
|
check.close()
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def test_prune_daily_backups_only():
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
backup_dir = Path(tmp)
|
|
old_daily = backup_dir / "app-20260101.db"
|
|
fresh_daily = backup_dir / "app-20261031.db"
|
|
old_migration = backup_dir / "app-pre-v1-20260101.db"
|
|
other = backup_dir / "keep.txt"
|
|
for path in (old_daily, fresh_daily, old_migration, other):
|
|
path.write_bytes(b"x")
|
|
now = 1_800_000_000
|
|
old_time = now - 20 * 86400
|
|
for path in (old_daily, old_migration):
|
|
os.utime(path, (old_time, old_time))
|
|
os.utime(fresh_daily, (now, now))
|
|
removed = prune_daily_backups(backup_dir, keep_days=14, now=now)
|
|
assert removed == [old_daily]
|
|
assert fresh_daily.exists()
|
|
assert old_migration.exists() # 迁移备份不属于日备份清理范围
|
|
assert other.exists()
|
|
|
|
|
|
def main():
|
|
test_daily_backup_and_restore()
|
|
test_prune_daily_backups_only()
|
|
print("OK: 每日备份与保留期测试全部通过")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|