94 lines
3.1 KiB
Python
94 lines
3.1 KiB
Python
"""自动备份/恢复演练单元测试（临时目录，不碰真实数据）。"""
|
|
import os
|
|
import sqlite3
|
|
import zipfile
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
|
|
from app import backup as bk
|
|
|
|
|
|
@pytest.fixture()
|
|
def isolated_backup(monkeypatch, tmp_path):
|
|
"""隔离 DATA_DIR/KB_DIR/BACKUP_DIR 到临时目录，并造一个含业务表的库。"""
|
|
data_dir = tmp_path / "data"
|
|
data_dir.mkdir()
|
|
db_path = data_dir / "workspace.db"
|
|
with sqlite3.connect(str(db_path)) as conn:
|
|
conn.executescript("""
|
|
CREATE TABLE campaign (id INTEGER PRIMARY KEY, name TEXT);
|
|
CREATE TABLE agent (id INTEGER PRIMARY KEY);
|
|
CREATE TABLE appuser (id INTEGER PRIMARY KEY);
|
|
CREATE TABLE other (x INTEGER);
|
|
INSERT INTO campaign (name) VALUES ('任务一'), ('任务二');
|
|
""")
|
|
kb_dir = data_dir / "kb"
|
|
kb_dir.mkdir()
|
|
(kb_dir / "note.txt").write_text("知识内容", encoding="utf-8")
|
|
monkeypatch.setattr(bk, "DATA_DIR", data_dir)
|
|
monkeypatch.setattr(bk, "KB_DIR", kb_dir)
|
|
monkeypatch.setattr(bk, "BACKUP_DIR", data_dir / "backups")
|
|
return data_dir
|
|
|
|
|
|
def test_create_backup_returns_zip(isolated_backup):
|
|
result = bk.create_backup("test")
|
|
assert result is not None and result.exists()
|
|
assert result.name.startswith("backup-test-")
|
|
with zipfile.ZipFile(result) as zf:
|
|
assert "workspace.db" in zf.namelist()
|
|
assert "kb/note.txt" in zf.namelist()
|
|
|
|
|
|
def test_backup_skipped_when_no_db(tmp_path):
|
|
empty = tmp_path / "empty"
|
|
empty.mkdir()
|
|
(empty / "kb").mkdir()
|
|
with patch.object(bk, "DATA_DIR", empty), patch.object(bk, "KB_DIR", empty / "kb"), \
|
|
patch.object(bk, "BACKUP_DIR", empty / "backups"):
|
|
assert bk.create_backup("x") is None
|
|
|
|
|
|
def test_verify_backup_checks_tables(isolated_backup):
|
|
result = bk.create_backup("test")
|
|
ok, detail = bk.verify_backup(result)
|
|
assert ok, detail
|
|
assert "campaign 2 行" in detail
|
|
|
|
|
|
def test_verify_backup_rejects_corrupt(isolated_backup):
|
|
bk.BACKUP_DIR.mkdir(parents=True, exist_ok=True)
|
|
bad = bk.BACKUP_DIR / "backup-bad.zip"
|
|
bad.write_bytes(b"not a zip")
|
|
ok, _detail = bk.verify_backup(bad)
|
|
assert not ok
|
|
|
|
|
|
def test_prune_keeps_recent(isolated_backup, monkeypatch):
|
|
monkeypatch.setattr(bk, "KEEP_COUNT", 3)
|
|
bk.BACKUP_DIR.mkdir(parents=True, exist_ok=True)
|
|
for i in range(5):
|
|
target = bk.BACKUP_DIR / f"backup-auto-2026010{i}-000000.zip"
|
|
target.write_bytes(b"placeholder")
|
|
os.utime(target, (1700000000 + i * 100, 1700000000 + i * 100))
|
|
removed = bk._prune_old_backups()
|
|
assert removed == 2
|
|
remaining = list(bk.BACKUP_DIR.glob("backup-*.zip"))
|
|
assert len(remaining) == 3
|
|
|
|
|
|
def test_restore_drill_latest(isolated_backup):
|
|
bk.create_backup("drill")
|
|
report = bk.verify_restore_latest()
|
|
assert report["ok"], report["detail"]
|
|
assert "解压演练通过" in report["detail"]
|
|
|
|
|
|
def test_backup_status_summary(isolated_backup):
|
|
bk.create_backup("status")
|
|
status = bk.backup_status()
|
|
assert status["count"] == 1
|
|
assert status["latest"].startswith("backup-status-")
|
|
assert status["latest_size_kb"] > 0
|