53 lines
2.2 KiB
Python
53 lines
2.2 KiB
Python
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"
|