303 lines
12 KiB
Python
303 lines
12 KiB
Python
"""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")
|