fc5: backup + rollback services (pg_dump + tar.zst, manifest JSON, find-by-tag)

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-22 08:12:17 -04:00
parent da9c19b2dc
commit e1b057f494
4 changed files with 215 additions and 0 deletions
@@ -0,0 +1,64 @@
+"""FC-5: backup + rollback service tests.
+
+Uses tmp_path to avoid touching real /images/_backups/. Subprocess
+calls (pg_dump, tar) are monkeypatched — the real shell-out is exercised
+in operator-side smoke testing on the homelab, not in unit tests.
+"""
+import pytest
+
+from backend.app.services.migrators import backup as backup_mod
+
+pytestmark = pytest.mark.integration
+
+
+@pytest.mark.asyncio
+async def test_create_backup_writes_sql_and_tar(monkeypatch, tmp_path):
+    calls = []
+
+    def fake_run(cmd, **kwargs):
+        calls.append(cmd)
+        # Touch the expected output file so existence checks pass downstream.
+        if "pg_dump" in cmd[0]:
+            outpath = tmp_path / "_backups" / f"fc_{kwargs['_test_ts']}.sql"
+            outpath.write_text("-- fake pg_dump output")
+        else:
+            outpath = tmp_path / "_backups" / f"fc_{kwargs['_test_ts']}.tar.zst"
+            outpath.write_bytes(b"\x28\xb5\x2f\xfd")  # zstd magic
+        from types import SimpleNamespace
+        return SimpleNamespace(returncode=0, stdout=b"", stderr=b"")
+
+    monkeypatch.setattr(backup_mod, "_run_subprocess", fake_run)
+
+    result = backup_mod.create_backup(
+        db_url="postgresql://x", images_root=tmp_path, tag="pre_migration",
+    )
+    assert "sql_path" in result
+    assert "tar_path" in result
+    assert result["tag"] == "pre_migration"
+    assert any("pg_dump" in c[0] for c in calls)
+    assert any(c[0] == "tar" for c in calls)
+
+
+@pytest.mark.asyncio
+async def test_find_latest_backup_by_tag(monkeypatch, tmp_path):
+    # Create two manifests with different tags.
+    backups_dir = tmp_path / "_backups"
+    backups_dir.mkdir()
+    import json
+    (backups_dir / "fc_20260101T000000Z.json").write_text(json.dumps({
+        "backup_id": "20260101T000000Z", "tag": "manual",
+        "created_at": "2026-01-01T00:00:00+00:00",
+        "sql_path": "/x/a.sql", "tar_path": "/x/a.tar.zst",
+    }))
+    (backups_dir / "fc_20260202T000000Z.json").write_text(json.dumps({
+        "backup_id": "20260202T000000Z", "tag": "pre_migration",
+        "created_at": "2026-02-02T00:00:00+00:00",
+        "sql_path": "/x/b.sql", "tar_path": "/x/b.tar.zst",
+    }))
+
+    found = backup_mod.find_latest_backup(tmp_path, tag="pre_migration")
+    assert found is not None
+    assert found["backup_id"] == "20260202T000000Z"
+
+    missing = backup_mod.find_latest_backup(tmp_path, tag="nonexistent")
+    assert missing is None