rebuild(stage-14): deliver migration and recovery tooling

2026-07-30 08:29:37 +08:00
parent 9eb7b548f3
commit fa7a8dde06
17 changed files with 1397 additions and 3 deletions
@@ -0,0 +1,52 @@
+from __future__ import annotations
+
+import sqlite3
+import zipfile
+
+import pytest
+
+from tools.backup import create_backup
+from tools.restore import restore_backup
+
+
+def test_backup_restore_round_trip_and_rejects_tampering(tmp_path) -> None:
+    database = tmp_path / "source.db"
+    with sqlite3.connect(database) as connection:
+        connection.execute("CREATE TABLE sample (value TEXT)")
+        connection.execute("INSERT INTO sample VALUES ('preserved')")
+    skills = tmp_path / "skills"
+    skills.mkdir()
+    skills.joinpath("private.md").write_text("private", encoding="utf-8")
+    environment = tmp_path / "source.env"
+    environment.write_text("APP_ENCRYPTION_KEY=hidden\n", encoding="utf-8")
+    archive = tmp_path / "xiaobai-2026-07-30.zip"
+    create_backup(database, archive, private_skills=skills, environment_file=environment)
+
+    restored = tmp_path / "restored.db"
+    restored_skills = tmp_path / "restored-skills"
+    restore_backup(archive, restored, private_skills=restored_skills)
+    with sqlite3.connect(restored) as connection:
+        assert connection.execute("SELECT value FROM sample").fetchone()[0] == "preserved"
+    assert restored_skills.joinpath("private.md").read_text(encoding="utf-8") == "private"
+
+    tampered = tmp_path / "tampered.zip"
+    with zipfile.ZipFile(archive) as source, zipfile.ZipFile(tampered, "w") as destination:
+        for item in source.infolist():
+            content = source.read(item.filename)
+            replacement = b"broken" if item.filename == "database.sqlite3" else content
+            destination.writestr(item, replacement)
+    with pytest.raises(ValueError, match="checksum mismatch"):
+        restore_backup(tampered, tmp_path / "rejected.db")
+
+
+def test_restore_refuses_implicit_overwrite(tmp_path) -> None:
+    database = tmp_path / "source.db"
+    with sqlite3.connect(database) as connection:
+        connection.execute("CREATE TABLE sample (value TEXT)")
+    archive = tmp_path / "backup.zip"
+    create_backup(database, archive)
+    destination = tmp_path / "existing.db"
+    destination.write_bytes(b"keep")
+    with pytest.raises(FileExistsError):
+        restore_backup(archive, destination)
+    assert destination.read_bytes() == b"keep"