DB backup script and alembic auto-backup hook #6
1 changed files with 97 additions and 0 deletions
97
tests/test_backup_script.py
Normal file
97
tests/test_backup_script.py
Normal file
|
|
@ -0,0 +1,97 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import shutil
|
||||
import sqlite3
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parents[1]
|
||||
SCRIPT = REPO_ROOT / "scripts" / "backup-db.sh"
|
||||
|
||||
|
||||
def _run(args: list[str], env_overrides: dict[str, str], cwd: Path):
|
||||
env = os.environ.copy()
|
||||
env.update(env_overrides)
|
||||
return subprocess.run(
|
||||
[str(SCRIPT), *args],
|
||||
check=False,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
cwd=str(cwd),
|
||||
env=env,
|
||||
)
|
||||
|
||||
|
||||
def test_backup_exit_zero_when_db_missing(tmp_path):
|
||||
fake_db = tmp_path / "ghost.db"
|
||||
result = _run(
|
||||
[],
|
||||
{"QUARTERMASTER_DB_URL": f"sqlite:///{fake_db}"},
|
||||
tmp_path,
|
||||
)
|
||||
assert result.returncode == 0, result.stderr
|
||||
assert "nothing to back up" in result.stdout
|
||||
|
||||
|
||||
def test_backup_creates_file(tmp_path):
|
||||
# seed a real sqlite DB
|
||||
db_path = tmp_path / "qm.db"
|
||||
conn = sqlite3.connect(str(db_path))
|
||||
conn.execute("CREATE TABLE t (x INTEGER)")
|
||||
conn.execute("INSERT INTO t VALUES (1), (2), (3)")
|
||||
conn.commit()
|
||||
conn.close()
|
||||
|
||||
result = _run(
|
||||
["unit-test"],
|
||||
{"QUARTERMASTER_DB_URL": f"sqlite:///{db_path}"},
|
||||
tmp_path,
|
||||
)
|
||||
assert result.returncode == 0, result.stderr
|
||||
backup_dir = tmp_path / "backups"
|
||||
assert backup_dir.is_dir()
|
||||
backups = list(backup_dir.glob("quartermaster-*-unit-test.db"))
|
||||
assert len(backups) == 1
|
||||
# restored backup reads the same rows
|
||||
restored = sqlite3.connect(str(backups[0]))
|
||||
rows = list(restored.execute("SELECT x FROM t ORDER BY x"))
|
||||
restored.close()
|
||||
assert rows == [(1,), (2,), (3,)]
|
||||
|
||||
|
||||
def test_backup_rejects_non_sqlite_url(tmp_path):
|
||||
result = _run(
|
||||
[],
|
||||
{"QUARTERMASTER_DB_URL": "postgres://example"},
|
||||
tmp_path,
|
||||
)
|
||||
assert result.returncode == 1
|
||||
assert "not a sqlite URL" in result.stderr
|
||||
|
||||
|
||||
def test_backup_slug_defaults_to_manual(tmp_path):
|
||||
db_path = tmp_path / "qm.db"
|
||||
sqlite3.connect(str(db_path)).close()
|
||||
result = _run(
|
||||
[],
|
||||
{"QUARTERMASTER_DB_URL": f"sqlite:///{db_path}"},
|
||||
tmp_path,
|
||||
)
|
||||
assert result.returncode == 0, result.stderr
|
||||
backups = list((tmp_path / "backups").glob("*manual*.db"))
|
||||
assert len(backups) == 1
|
||||
|
||||
|
||||
def test_backup_slug_sanitizes_reason(tmp_path):
|
||||
db_path = tmp_path / "qm.db"
|
||||
sqlite3.connect(str(db_path)).close()
|
||||
result = _run(
|
||||
["alembic upgrade!! head"],
|
||||
{"QUARTERMASTER_DB_URL": f"sqlite:///{db_path}"},
|
||||
tmp_path,
|
||||
)
|
||||
assert result.returncode == 0, result.stderr
|
||||
backups = list((tmp_path / "backups").glob("*.db"))
|
||||
assert len(backups) == 1
|
||||
assert "alembic-upgrade-head" in backups[0].name
|
||||
Loading…
Reference in a new issue