File size: 3,263 Bytes
7c6ffa6 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 | from __future__ import annotations
import os
import sys
from datetime import datetime, timedelta, timezone
from pathlib import Path
import pytest
SCRIPT_DIR = Path(__file__).resolve().parents[1] / "scripts"
if str(SCRIPT_DIR) not in sys.path:
sys.path.insert(0, str(SCRIPT_DIR))
import database_backup # noqa: E402
import database_restore # noqa: E402
def test_parse_postgres_url_supports_sqlalchemy_driver_and_sslmode() -> None:
target = database_backup.parse_postgres_url(
"postgresql+psycopg://student:p%40ss@db.example.com:6543/docdoe?sslmode=require"
)
assert target.host == "db.example.com"
assert target.port == 6543
assert target.database == "docdoe"
assert target.username == "student"
assert target.password == "p@ss"
assert target.sslmode == "require"
@pytest.mark.parametrize("database_url", ["", "sqlite:///local.db", "postgresql://localhost"])
def test_parse_postgres_url_rejects_unsafe_or_incomplete_targets(database_url: str) -> None:
with pytest.raises(ValueError):
database_backup.parse_postgres_url(database_url)
def test_process_environment_keeps_password_out_of_command_data() -> None:
target = database_backup.parse_postgres_url(
"postgresql://student:super-secret@db.example.com/docdoe"
)
env = target.process_environment()
assert env["PGPASSWORD"] == "super-secret"
assert env["PGDATABASE"] == "docdoe"
def test_prune_expired_backups_only_removes_owned_backup_pattern(tmp_path: Path) -> None:
now = datetime(2026, 7, 16, tzinfo=timezone.utc)
old = tmp_path / "docdoe-20260101T000000Z.dump"
old_manifest = old.with_suffix(".manifest.json")
recent = tmp_path / "docdoe-20260715T000000Z.dump"
unrelated = tmp_path / "customer-export.dump"
for path in (old, old_manifest, recent, unrelated):
path.write_bytes(b"safe-test-data")
old_time = (now - timedelta(days=30)).timestamp()
recent_time = (now - timedelta(days=1)).timestamp()
os.utime(old, (old_time, old_time))
os.utime(old_manifest, (old_time, old_time))
os.utime(recent, (recent_time, recent_time))
removed = database_backup.prune_expired_backups(tmp_path, 14, now)
assert set(removed) == {old, old_manifest}
assert recent.exists()
assert unrelated.exists()
def test_backup_dry_run_does_not_write_or_print_password(tmp_path: Path, capsys) -> None:
result = database_backup.create_backup(
database_url="postgresql://student:super-secret@db.example.com/docdoe",
output_dir=tmp_path,
retention_days=14,
verify=True,
dry_run=True,
now=datetime(2026, 7, 16, tzinfo=timezone.utc),
)
assert result is None
assert list(tmp_path.iterdir()) == []
output = capsys.readouterr().out
assert "super-secret" not in output
assert "db.example.com:5432/docdoe" in output
def test_restore_manifest_detects_tampering(tmp_path: Path) -> None:
backup = tmp_path / "docdoe-20260716T000000Z.dump"
backup.write_bytes(b"original")
backup.with_suffix(".manifest.json").write_text(
'{"sha256":"not-the-real-checksum"}', encoding="utf-8"
)
with pytest.raises(ValueError, match="checksum"):
database_restore.verify_manifest(backup)
|