tests/test_database_backup.py

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)