"""M15 recovery API contract tests: operator isolation, redaction and typed error envelopes.""" from __future__ import annotations import uuid from pathlib import Path from typing import Any import pytest from fastapi.testclient import TestClient from sqlalchemy import create_engine from sqlalchemy.orm import Session from sqlalchemy.pool import StaticPool from modelforge_api.api.routes.recovery import get_service from modelforge_api.db import get_session from modelforge_api.main import app from modelforge_api.persistence.models import Base from modelforge_api.services.recovery import RecoveryService from modelforge_api.settings import Settings, get_settings from tests.test_recovery_m15 import seed_backup, settings_for client = TestClient(app) ADMIN = {"X-ModelForge-Admin-Token": "m15-operator-token"} RECOVERY_ROUTES = ( ("GET", "/api/v1/admin/recovery/dashboard"), ("GET", "/api/v1/admin/recovery/policies"), ("GET", "/api/v1/admin/recovery/assets"), ("GET", "/api/v1/admin/recovery/capacity"), ("GET", "/api/v1/admin/recovery/backups"), ("GET", "/api/v1/admin/recovery/restore-plans"), ("GET", "/api/v1/admin/recovery/restore-operations"), ("GET", "/api/v1/admin/recovery/artifact-recoveries"), ("GET", "/api/v1/admin/recovery/fingerprint"), ("POST", "/api/v1/admin/recovery/retention/run"), ) @pytest.fixture def recovery(tmp_path: Path) -> Any: engine = create_engine( "sqlite+pysqlite:///:memory:", connect_args={"check_same_thread": False}, poolclass=StaticPool, ) Base.metadata.create_all(engine) settings = settings_for(tmp_path, operator_api_key="m15-operator-token") with Session(engine) as session: service = RecoveryService(session, settings) service.ensure_defaults() app.dependency_overrides[get_session] = lambda: session app.dependency_overrides[get_settings] = lambda: settings app.dependency_overrides[get_service] = lambda: service try: yield session, service, settings finally: app.dependency_overrides.clear() @pytest.mark.parametrize(("method", "path"), RECOVERY_ROUTES) def test_every_recovery_route_requires_an_operator_credential( recovery: Any, method: str, path: str ) -> None: response = client.request(method, path) assert response.status_code == 401 assert response.json()["error"]["code"] == "http_401" @pytest.mark.parametrize(("method", "path"), RECOVERY_ROUTES) def test_a_wrong_operator_credential_is_refused(recovery: Any, method: str, path: str) -> None: response = client.request(method, path, headers={"X-ModelForge-Admin-Token": "wrong"}) assert response.status_code == 401 def test_recovery_routes_are_unavailable_when_no_operator_key_is_configured( tmp_path: Path, ) -> None: engine = create_engine( "sqlite+pysqlite:///:memory:", connect_args={"check_same_thread": False}, poolclass=StaticPool, ) Base.metadata.create_all(engine) settings = settings_for(tmp_path, operator_api_key=None) with Session(engine) as session: app.dependency_overrides[get_session] = lambda: session app.dependency_overrides[get_settings] = lambda: settings try: response = client.get("/api/v1/admin/recovery/dashboard", headers=ADMIN) assert response.status_code == 503 finally: app.dependency_overrides.clear() def test_dashboard_reports_measured_state_and_never_invents_a_backup(recovery: Any) -> None: payload = client.get("/api/v1/admin/recovery/dashboard", headers=ADMIN).json() assert payload["point_in_time_support"] == "NOT_SUPPORTED" assert payload["latest_verified_backup_id"] is None assert payload["latest_verified_backup_age_seconds"] is None assert payload["observed_restore_seconds"] is None assert payload["observed_rpo_seconds"] is None assert payload["stale_backup"] is True assert "postgres.modelforge" in payload["unprotected_assets"] assert payload["verified_backup_count"] == 0 def test_policies_and_assets_expose_the_full_classification(recovery: Any) -> None: policies = client.get("/api/v1/admin/recovery/policies", headers=ADMIN).json() assert {item["key"] for item in policies} >= { "control-plane.database", "artifacts.rehydratable", "runtime.ephemeral", "external.projects", "secrets.credentials", } assets = client.get("/api/v1/admin/recovery/assets", headers=ADMIN).json() assert {item["asset_class"] for item in assets} == { "AUTHORITATIVE", "REBUILDABLE", "EPHEMERAL", "EXTERNAL", "SECRET", } external = [item for item in assets if item["asset_class"] == "EXTERNAL"] assert all(item["readiness"] == "EXTERNAL_DEPENDENCY" for item in external) def test_capacity_estimate_is_reported_before_any_backup_is_written(recovery: Any) -> None: payload = client.get("/api/v1/admin/recovery/capacity", headers=ADMIN).json() assert payload["bytes_to_copy"] >= 0 assert payload["sufficient"] is True assert isinstance(payload["detail"], str) def test_the_fingerprint_endpoint_returns_digests_without_secret_columns(recovery: Any) -> None: payload = client.get("/api/v1/admin/recovery/fingerprint", headers=ADMIN).json() assert payload["version"] == "m15.2" assert len(payload["digest"]) == 64 assert "audit" in payload["groups"] assert "secret_hash" in payload["tables"]["node_credentials"]["redacted_columns"] assert "serving_gpu_leases" in payload["excluded_current_truth_tables"] def test_a_backup_set_response_never_exposes_a_destination_password( recovery: Any, tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: session, service, _settings = recovery _subject, record = seed_backup(session, tmp_path, backup_id="m15-api-backup") monkeypatch.setattr(service.engine, "list_dump_contents", lambda _payload: 7) listed = client.get("/api/v1/admin/recovery/backups", headers=ADMIN).json() assert [item["backup_id"] for item in listed] == ["m15-api-backup"] assert listed[0]["restore_eligible"] is False verified = client.post( f"/api/v1/admin/recovery/backups/{record.id}/verify", headers=ADMIN ).json() assert verified["state"] == "VERIFIED" assert verified["restore_eligible"] is True assert "secret" not in verified["database_identity"].get("connection", "") assert verified["entries"][0]["logical_asset_type"] == "control_plane_database" def test_creating_a_restore_plan_against_the_live_database_is_refused( recovery: Any, tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: session, service, _settings = recovery _subject, record = seed_backup(session, tmp_path, backup_id="m15-api-selftarget") monkeypatch.setattr(service.engine, "list_dump_contents", lambda _payload: 1) service.verify_backup(record.id) response = client.post( "/api/v1/admin/recovery/restore-plans", headers=ADMIN, json={ "backup_set_id": str(record.id), "mode": "VALIDATION", "target_environment": "ISOLATED", "target_label": "self-target", "database_destination": ( "postgresql+psycopg://modelforge:secret@postgres:5432/modelforge" ), "artifact_strategy": "MANIFEST_ONLY", "secret_strategy": "RESTORE_HASHES", "node_strategy": "NONE", "reason": "attempting to restore over the running control plane", }, ) assert response.status_code == 409 assert response.json()["error"]["code"] == "DESTINATION_NOT_ISOLATED" def test_a_restore_plan_is_returned_with_a_redacted_destination( recovery: Any, tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: session, service, _settings = recovery _subject, record = seed_backup(session, tmp_path, backup_id="m15-api-plan") monkeypatch.setattr(service.engine, "list_dump_contents", lambda _payload: 1) service.verify_backup(record.id) created = client.post( "/api/v1/admin/recovery/restore-plans", headers=ADMIN, json={ "backup_set_id": str(record.id), "mode": "VALIDATION", "target_environment": "ISOLATED", "target_label": "m15-isolated", "database_destination": ( "postgresql+psycopg://modelforge:secret@postgres:5432/mf_restore" ), "artifact_strategy": "MANIFEST_ONLY", "secret_strategy": "RESTORE_HASHES", "node_strategy": "NONE", "reason": "isolated validation restore rehearsal for the M15 API contract", }, ) assert created.status_code == 201 payload = created.json() assert payload["state"] == "DRAFT" assert "secret" not in payload["database_destination_redacted"] assert "database_destination" not in payload fetched = client.get( f"/api/v1/admin/recovery/restore-plans/{payload['id']}", headers=ADMIN ).json() assert fetched["backup_id"] == "m15-api-plan" def test_starting_a_restore_without_a_passing_preflight_is_refused( recovery: Any, tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: session, service, _settings = recovery _subject, record = seed_backup(session, tmp_path, backup_id="m15-api-preflight") monkeypatch.setattr(service.engine, "list_dump_contents", lambda _payload: 1) service.verify_backup(record.id) plan = client.post( "/api/v1/admin/recovery/restore-plans", headers=ADMIN, json={ "backup_set_id": str(record.id), "mode": "VALIDATION", "target_environment": "ISOLATED", "target_label": "m15-isolated", "database_destination": ( "postgresql+psycopg://modelforge:secret@postgres:5432/mf_restore" ), "artifact_strategy": "MANIFEST_ONLY", "secret_strategy": "RESTORE_HASHES", "node_strategy": "NONE", "reason": "a restore must not start before its preflight has passed", }, ).json() response = client.post( f"/api/v1/admin/recovery/restore-plans/{plan['id']}/start", headers=ADMIN, json={"actor": "operator", "reason": "starting before preflight"}, ) assert response.status_code == 409 assert response.json()["error"]["code"] == "restore_preflight_required" def test_unknown_recovery_resources_return_typed_not_found_envelopes(recovery: Any) -> None: missing = uuid.uuid4() for path, code in ( (f"/api/v1/admin/recovery/backups/{missing}", "backup_not_found"), (f"/api/v1/admin/recovery/restore-plans/{missing}", "restore_plan_not_found"), ( f"/api/v1/admin/recovery/restore-operations/{missing}", "restore_operation_not_found", ), ): response = client.get(path, headers=ADMIN) assert response.status_code == 404 assert response.json()["error"]["code"] == code def test_the_recovery_contract_is_published_in_the_openapi_document() -> None: schema = app.openapi() recovery_paths = { path for path in schema["paths"] if path.startswith("/api/v1/admin/recovery") } assert len(recovery_paths) >= 15 assert "/api/v1/admin/recovery/backups" in recovery_paths assert "/api/v1/admin/recovery/restore-plans/{plan_id}/preflight" in recovery_paths components = schema["components"]["schemas"] assert "BackupSetResponse" in components assert "RestorePlanResponse" in components assert "RecoveryDashboard" in components # The restore destination DSN is never part of a response contract. assert "database_destination" not in components["RestorePlanResponse"]["properties"] assert "database_destination_redacted" in components["RestorePlanResponse"]["properties"] def test_recovery_settings_default_to_safe_values() -> None: settings = Settings(_env_file=None) # type: ignore[call-arg] assert settings.restore_allow_production_target is False assert settings.recovery_reconciliation_enabled is False assert settings.backup_encryption_key is None assert str(settings.backup_root).replace("\\", "/").endswith("/data/backups")