98 lines
2.7 KiB
Python
98 lines
2.7 KiB
Python
|
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
|