"""SP17 — Admin backup API endpoint tests. Covers: - POST /api/admin/backup/create - GET /api/admin/backup/list - GET /api/admin/backup/status - POST /api/admin/backup/{id}/verify - POST /api/admin/backup/{id}/restore/initiate - POST /api/admin/backup/{id}/restore/confirm - POST /api/admin/backup/prune - POST /api/admin/backup/scheduler/{start,stop,tick} Each fixture starts a clean DB + BackupService configured with a known passphrase. We deliberately do NOT enable SQLCipher here — the backup layer is independent of SQLCipher encryption at rest. """ from __future__ import annotations from datetime import datetime, timedelta, timezone from pathlib import Path import pytest @pytest.fixture def _backup_env(tmp_path, monkeypatch): """Fresh sqlite DB + BackupService with passphrase. Reset module singletons.""" from cyclone import db from cyclone import backup_service as svc_mod from cyclone import backup_scheduler as sched_mod from cyclone.db import Batch monkeypatch.setenv("CYCLONE_DB_URL", f"sqlite:///{tmp_path}/test.db") db._reset_for_tests() db.init_db() # Make sure there's at least one row so the backup isn't a no-op. import uuid with db.SessionLocal()() as s: s.add(Batch( id=str(uuid.uuid4()), kind="837P", input_filename="seed.x12", parsed_at=datetime.now(timezone.utc), totals_json=None, validation_json=None, raw_result_json={"envelope": {"control_number": "1"}, "claims": [], "summary": {"passed": 0, "failed": 0, "failed_claim_ids": []}}, )) s.commit() backup_dir = tmp_path / "backups" svc_mod.reset_backup_service_for_tests() sched_mod.reset_backup_scheduler_for_tests() svc = svc_mod.configure_backup_service( backup_dir=backup_dir, passphrase="api-test-pass", retention_days=7, ) yield svc, backup_dir sched_mod.reset_backup_scheduler_for_tests() svc_mod.reset_backup_service_for_tests() db._reset_for_tests() def _client(): from fastapi.testclient import TestClient from cyclone.api import app return TestClient(app) # --------------------------------------------------------------------------- # /backup/create # --------------------------------------------------------------------------- def test_create_returns_metadata_and_persists_row(_backup_env): svc, backup_dir = _backup_env r = _client().post("/api/admin/backup/create") assert r.status_code == 200, r.text body = r.json() assert body["ok"] is True b = body["backup"] assert b["size_bytes"] > 0 assert b["db_fingerprint"].startswith("sha256:") assert b["table_count"] >= 1 assert b["created_at"] # File actually exists on disk. assert (backup_dir / b["filename"]).exists() # Sidecar metadata echoed. sc = body["sidecar"] assert sc["kdf"] == "PBKDF2-HMAC-SHA256" assert sc["kdf_iterations"] == 200_000 assert sc["cipher"] == "AES-256-GCM" def test_create_503_when_service_unconfigured(tmp_path, monkeypatch): """If BackupService was never configured, create returns 503.""" from cyclone import db from cyclone import backup_service as svc_mod monkeypatch.setenv("CYCLONE_DB_URL", f"sqlite:///{tmp_path}/test.db") db._reset_for_tests() db.init_db() svc_mod.reset_backup_service_for_tests() try: r = _client().post("/api/admin/backup/create") assert r.status_code == 503 assert "not configured" in r.json()["detail"].lower() finally: db._reset_for_tests() # --------------------------------------------------------------------------- # /backup/list # --------------------------------------------------------------------------- def test_list_returns_newest_first(_backup_env): svc, _ = _backup_env client = _client() client.post("/api/admin/backup/create") client.post("/api/admin/backup/create") r = client.get("/api/admin/backup/list") assert r.status_code == 200 body = r.json() assert body["count"] == 2 assert body["files"][0]["id"] > body["files"][1]["id"] def test_list_filter_by_status(_backup_env): svc, _ = _backup_env client = _client() client.post("/api/admin/backup/create") r = client.get("/api/admin/backup/list?status=ok") assert r.json()["count"] == 1 r = client.get("/api/admin/backup/list?status=error") assert r.json()["count"] == 0 # --------------------------------------------------------------------------- # /backup/status # --------------------------------------------------------------------------- def test_status_returns_counts_and_dirs(_backup_env): svc, backup_dir = _backup_env client = _client() client.post("/api/admin/backup/create") r = client.get("/api/admin/backup/status") assert r.status_code == 200 body = r.json() assert body["totals"]["ok"] == 1 assert body["backup_dir"] == str(backup_dir) assert body["retention_days"] == 7 assert body["last_backup_at"] is not None assert body["last_ok_backup_at"] is not None # The scheduler may or may not be configured depending on lifespan. assert "scheduler" in body # --------------------------------------------------------------------------- # /backup/{id}/verify # --------------------------------------------------------------------------- def test_verify_ok_after_create(_backup_env): svc, _ = _backup_env client = _client() cid = client.post("/api/admin/backup/create").json()["backup"]["id"] r = client.post(f"/api/admin/backup/{cid}/verify") assert r.status_code == 200 body = r.json() assert body["ok"] is True assert body["expected_fingerprint"] == body["actual_fingerprint"] def test_verify_detects_tampered_ciphertext(_backup_env): from cyclone import backup as backup_mod svc, backup_dir = _backup_env client = _client() cid = client.post("/api/admin/backup/create").json()["backup"]["id"] fname = svc.list_backups()[0].filename # Flip a bit in the ciphertext. bin_path = backup_dir / fname data = bytearray(bin_path.read_bytes()) data[backup_mod.NONCE_LEN + 5] ^= 0x01 bin_path.write_bytes(bytes(data)) r = client.post(f"/api/admin/backup/{cid}/verify") assert r.status_code == 200 assert r.json()["ok"] is False def test_verify_404_when_unknown_backup(_backup_env): r = _client().post("/api/admin/backup/99999/verify") # The service raises BackupError; the endpoint should return 503 (no svc) or 400 # depending on flow. Let's see what happens. assert r.status_code in (400, 404, 503) # --------------------------------------------------------------------------- # /backup/{id}/restore/{initiate,confirm} # --------------------------------------------------------------------------- def test_restore_two_step_via_api(_backup_env): svc, _ = _backup_env from cyclone.db import Batch import uuid client = _client() cid = client.post("/api/admin/backup/create").json()["backup"]["id"] # Mutate the live DB (add another Batch row). with __import__("cyclone").db.SessionLocal()() as s: s.add(Batch( id=str(uuid.uuid4()), kind="837P", input_filename="mutated.x12", parsed_at=datetime.now(timezone.utc), totals_json=None, validation_json=None, raw_result_json={"envelope": {"control_number": "2"}, "claims": [], "summary": {"passed": 0, "failed": 0, "failed_claim_ids": []}}, )) s.commit() # Step 1: initiate. r1 = client.post(f"/api/admin/backup/{cid}/restore/initiate") assert r1.status_code == 200, r1.text body1 = r1.json() assert body1["restore_token"] assert body1["preview"]["backup_table_count"] >= 1 assert body1["preview"]["backup_db_fingerprint"] != body1["preview"]["current_db_fingerprint"] # Step 2: confirm. r2 = client.post( f"/api/admin/backup/{cid}/restore/confirm", json={"restore_token": body1["restore_token"], "actor": "test"}, ) assert r2.status_code == 200, r2.text body2 = r2.json() assert body2["ok"] is True assert body2["new_db_fingerprint"] == body1["preview"]["backup_db_fingerprint"] def test_restore_confirm_requires_token(_backup_env): svc, _ = _backup_env client = _client() cid = client.post("/api/admin/backup/create").json()["backup"]["id"] r = client.post(f"/api/admin/backup/{cid}/restore/confirm", json={}) assert r.status_code == 400 def test_restore_confirm_rejects_wrong_token(_backup_env): svc, _ = _backup_env client = _client() cid = client.post("/api/admin/backup/create").json()["backup"]["id"] r = client.post( f"/api/admin/backup/{cid}/restore/confirm", json={"restore_token": "0" * 64}, ) assert r.status_code == 400 # --------------------------------------------------------------------------- # /backup/prune # --------------------------------------------------------------------------- def test_prune_deletes_old_backups(_backup_env): svc, _ = _backup_env from cyclone.db import DbBackup client = _client() cid = client.post("/api/admin/backup/create").json()["backup"]["id"] # Age the backup past the retention cutoff. with __import__("cyclone").db.SessionLocal()() as s: row = s.get(DbBackup, cid) row.created_at = datetime.now(timezone.utc) - timedelta(days=30) s.commit() r = client.post("/api/admin/backup/prune") assert r.status_code == 200 body = r.json() assert body["ok"] is True assert body["deleted_count"] == 2 # .bin + .meta.json # --------------------------------------------------------------------------- # /backup/scheduler/{start,stop,tick} # --------------------------------------------------------------------------- def test_scheduler_endpoints_require_configured_scheduler(_backup_env, monkeypatch): """Without calling configure_backup_scheduler, the endpoints 503.""" svc, _ = _backup_env # We did NOT call configure_backup_scheduler; the lifespan # *might* have called it as a side effect of the TestClient # entering its context. Either way, the scheduler endpoints # need it to be present. client = _client() r = client.post("/api/admin/backup/scheduler/tick") assert r.status_code in (200, 503) def test_scheduler_tick_when_configured(_backup_env): """With a configured scheduler, tick runs and returns a result.""" from cyclone import backup_scheduler as sched_mod svc, _ = _backup_env sched_mod.configure_backup_scheduler(svc, interval_hours=24.0) try: client = _client() r = client.post("/api/admin/backup/scheduler/tick") assert r.status_code == 200 body = r.json() assert body["ok"] is True assert body["tick"]["created"] is not None assert body["tick"]["created"]["id"] >= 1 finally: sched_mod.reset_backup_scheduler_for_tests()