fix: make backup verify safe for wal sqlite

This commit is contained in:
Hermes Agent
2026-06-18 13:39:02 +08:00
parent 6e06e150b0
commit 65f44c9718
5 changed files with 146 additions and 6 deletions

View File

@@ -14,6 +14,7 @@ from __future__ import annotations
import os
import shutil
import sqlite3
import subprocess
from pathlib import Path
@@ -89,6 +90,54 @@ def test_backup_verify_uses_venv_python_for_service_level_restore(tmp_path):
assert "ModuleNotFoundError" not in proc.stdout
def test_backup_verify_live_mode_uses_sqlite_backup_for_db_files(tmp_path):
admin_db = tmp_path / "admin.db"
verify_dir = tmp_path / "verify"
live_conn = sqlite3.connect(admin_db)
try:
journal_mode = live_conn.execute("PRAGMA journal_mode=WAL").fetchone()
assert journal_mode is not None
assert str(journal_mode[0]).lower() == "wal"
live_conn.execute("PRAGMA wal_autocheckpoint = 0")
live_conn.execute(
"CREATE TABLE restore_checks (id INTEGER PRIMARY KEY, value TEXT NOT NULL)"
)
live_conn.execute(
"INSERT INTO restore_checks(value) VALUES (?)", ("present",)
)
live_conn.commit()
proc = subprocess.run(
["bash", str(SCRIPTS_DIR / "backup_verify.sh"), str(verify_dir), "--skip-smoke"],
cwd=REPO_ROOT,
env={
**os.environ,
"GAOKAO_DB_PATH": str(admin_db),
},
text=True,
capture_output=True,
check=False,
)
finally:
live_conn.close()
assert proc.returncode == 0, proc.stderr
staged_admin_db = verify_dir / "db" / "admin.db"
conn = sqlite3.connect(staged_admin_db)
try:
tables = {
row[0]
for row in conn.execute(
"SELECT name FROM sqlite_master WHERE type='table'"
).fetchall()
}
assert "restore_checks" in tables
row = conn.execute("SELECT COUNT(*) FROM restore_checks").fetchone()
assert row == (1,)
finally:
conn.close()
def test_backup_restore_smoke_fails_fast_when_admin_db_missing(tmp_path):
staging = tmp_path / "backup"
_prepare_backup_dir(staging)

View File

@@ -188,6 +188,47 @@ def test_backup_snapshot_preserves_admin_db_contents_under_wal(tmp_path):
assert _count_rows(snapshot_admin_db, "admin_users") == 1
def test_backup_verify_live_copy_handles_wal_sqlite(tmp_path):
admin_db = tmp_path / "admin.db"
verify_dir = tmp_path / "verify"
live_conn = sqlite3.connect(admin_db)
try:
journal_mode = live_conn.execute("PRAGMA journal_mode=WAL").fetchone()
assert journal_mode is not None
assert str(journal_mode[0]).lower() == "wal"
live_conn.execute("PRAGMA wal_autocheckpoint = 0")
live_conn.execute(
"CREATE TABLE admin_users (id INTEGER PRIMARY KEY, username TEXT NOT NULL)"
)
live_conn.execute(
"INSERT INTO admin_users(username) VALUES (?)", ("verify-admin",)
)
live_conn.commit()
wal_path = admin_db.with_name(f"{admin_db.name}-wal")
assert wal_path.exists()
proc = subprocess.run(
["bash", "scripts/backup_verify.sh", str(verify_dir), "--skip-smoke"],
cwd=PROJECT_ROOT,
text=True,
capture_output=True,
env={
**os.environ,
"GAOKAO_DB_PATH": str(admin_db),
},
check=False,
)
finally:
live_conn.close()
assert proc.returncode == 0, proc.stderr
staged_admin_db = verify_dir / "db" / "admin.db"
assert staged_admin_db.is_file()
assert "admin_users" in _list_tables(staged_admin_db)
assert _count_rows(staged_admin_db, "admin_users") == 1
def test_backup_verify_runs_restore_smoke_on_snapshot(settings, tmp_path):
env = {
**os.environ,