From 86ad9b80e943c5a2816cacb1ad30d6c97fb5eb02 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Sun, 24 May 2026 23:00:40 -0400 Subject: [PATCH] fc3h: backup_service unit tests (subprocess monkeypatched) Co-Authored-By: Claude Opus 4.7 (1M context) --- tests/test_backup_service.py | 197 +++++++++++++++++++++++++++++++++++ 1 file changed, 197 insertions(+) create mode 100644 tests/test_backup_service.py diff --git a/tests/test_backup_service.py b/tests/test_backup_service.py new file mode 100644 index 0000000..7d39120 --- /dev/null +++ b/tests/test_backup_service.py @@ -0,0 +1,197 @@ +"""FC-3h: backup_service unit tests. + +Subprocess calls (pg_dump, tar, psql) are monkeypatched so tests run +without external binaries. The real subprocess behavior is exercised +implicitly via the Celery task tests in test_tasks_backup.py. +""" +import json +from pathlib import Path + +import pytest + +from backend.app.services import backup_service + +pytestmark = pytest.mark.integration + + +@pytest.fixture +def fake_subprocess(monkeypatch): + """Replace subprocess.run with a fake that writes a sentinel to + the target path (for pg_dump's -f, for tar's -cf). Captures all + calls in a list.""" + calls = [] + + class _FakeProc: + returncode = 0 + stdout = b"" + stderr = b"" + + def _fake_run(cmd, **kwargs): + calls.append(list(cmd)) + if cmd[0] == "pg_dump": + i = cmd.index("-f") + Path(cmd[i + 1]).write_bytes(b"-- fake pg_dump\n") + elif cmd[0] == "tar" and "-cf" in cmd: + i = cmd.index("-cf") + Path(cmd[i + 1]).write_bytes(b"fake tar payload") + return _FakeProc() + + monkeypatch.setattr("subprocess.run", _fake_run) + return calls + + +# --- backup_db ------------------------------------------------------- + + +def test_backup_db_writes_sql_and_manifest(tmp_path, fake_subprocess): + result = backup_service.backup_db( + db_url="postgresql://test@x/db", + images_root=tmp_path, + tag="pre-cutover", + triggered_by="manual", + ) + sql = Path(result["sql_path"]) + manifest = Path(result["manifest_path"]) + assert sql.is_file() and sql.suffix == ".sql" + assert manifest.is_file() and manifest.suffix == ".json" + assert result["kind"] == "db" + assert result["tar_path"] is None + assert result["size_bytes"] == len(b"-- fake pg_dump\n") + + parsed = json.loads(manifest.read_text()) + assert parsed["kind"] == "db" + assert parsed["tag"] == "pre-cutover" + assert parsed["triggered_by"] == "manual" + assert parsed["artifact_path"] == str(sql) + + +def test_backup_db_strips_sqlalchemy_psycopg_driver(tmp_path, fake_subprocess): + backup_service.backup_db( + db_url="postgresql+psycopg://u:p@h/d", images_root=tmp_path, + ) + cmd = fake_subprocess[0] + assert cmd[0] == "pg_dump" + assert cmd[-1].startswith("postgresql://") + assert "+psycopg" not in cmd[-1] + + +def test_backup_db_strips_asyncpg_driver(tmp_path, fake_subprocess): + backup_service.backup_db( + db_url="postgresql+asyncpg://u:p@h/d", images_root=tmp_path, + ) + assert "+asyncpg" not in fake_subprocess[0][-1] + + +def test_backup_db_default_tag_is_none(tmp_path, fake_subprocess): + result = backup_service.backup_db( + db_url="postgresql://u@h/d", images_root=tmp_path, + ) + parsed = json.loads(Path(result["manifest_path"]).read_text()) + assert parsed["tag"] is None + + +# --- backup_images --------------------------------------------------- + + +def test_backup_images_writes_tar_and_manifest(tmp_path, fake_subprocess): + result = backup_service.backup_images( + images_root=tmp_path, tag="monthly", triggered_by="manual", + ) + tar = Path(result["tar_path"]) + assert tar.is_file() and tar.name.endswith(".tar.zst") + assert result["sql_path"] is None + assert result["size_bytes"] == len(b"fake tar payload") + + +def test_backup_images_excludes_backups_and_quarantine(tmp_path, fake_subprocess): + backup_service.backup_images(images_root=tmp_path) + cmd = fake_subprocess[0] + excludes = [arg for arg in cmd if arg.startswith("--exclude=")] + assert any("_backups" in e for e in excludes) + assert any("_quarantine" in e for e in excludes) + + +# --- restore_db ------------------------------------------------------ + + +def test_restore_db_drops_schema_then_loads(tmp_path, fake_subprocess): + sql_path = tmp_path / "fake.sql" + sql_path.write_text("SELECT 1;") + backup_service.restore_db( + db_url="postgresql://u@h/d", sql_path=sql_path, + ) + # Two psql calls: one with -c (DROP SCHEMA), one with -f (load). + assert len(fake_subprocess) == 2 + assert "-c" in fake_subprocess[0] + assert "DROP SCHEMA IF EXISTS public CASCADE" in fake_subprocess[0][-1] + assert "-f" in fake_subprocess[1] + assert str(sql_path) in fake_subprocess[1] + + +# --- restore_images -------------------------------------------------- + + +def test_restore_images_untar_to_parent(tmp_path, fake_subprocess): + tar_path = tmp_path / "fake.tar.zst" + tar_path.write_bytes(b"fake") + backup_service.restore_images(images_root=tmp_path, tar_path=tar_path) + cmd = fake_subprocess[0] + assert cmd[:3] == ["tar", "--zstd", "-xf"] + assert str(tar_path) in cmd + assert "-C" in cmd + + +# --- unlink ---------------------------------------------------------- + + +def test_unlink_removes_present_files_and_reports(tmp_path): + sql = tmp_path / "x.sql"; sql.write_bytes(b"x") + tar = tmp_path / "x.tar.zst"; tar.write_bytes(b"x") + manifest = tmp_path / "x.json"; manifest.write_text("{}") + result = backup_service.unlink_artifact_files( + sql_path=str(sql), tar_path=str(tar), manifest_path=str(manifest), + ) + assert result == {"sql": True, "tar": True, "manifest": True} + assert not sql.exists() and not tar.exists() and not manifest.exists() + + +def test_unlink_missing_files_returns_true(tmp_path): + """missing_ok semantics: a non-existent file isn't an error.""" + result = backup_service.unlink_artifact_files( + sql_path=str(tmp_path / "nope.sql"), + tar_path=None, manifest_path=None, + ) + assert result == {"sql": True} + + +def test_unlink_skips_none_paths(): + """A None path is skipped — not added to the result dict.""" + result = backup_service.unlink_artifact_files( + sql_path=None, tar_path=None, manifest_path=None, + ) + assert result == {} + + +# --- helpers --------------------------------------------------------- + + +def test_libpq_url_strips_each_known_driver(): + assert backup_service._libpq_url( + "postgresql+psycopg://u@h/d" + ) == "postgresql://u@h/d" + assert backup_service._libpq_url( + "postgresql+asyncpg://u@h/d" + ) == "postgresql://u@h/d" + assert backup_service._libpq_url( + "postgresql+psycopg2://u@h/d" + ) == "postgresql://u@h/d" + # Plain URL passes through unchanged. + assert backup_service._libpq_url( + "postgresql://u@h/d" + ) == "postgresql://u@h/d" + + +def test_backups_dir_created_on_first_use(tmp_path): + d = backup_service._backups_dir(tmp_path) + assert d.is_dir() + assert d.name == "_backups"