"""M15 recovery-plane tests: classification, policy, backup lifecycle and corruption handling.""" from __future__ import annotations import base64 import hashlib import inspect import json import uuid from datetime import UTC, datetime, timedelta from pathlib import Path from typing import Any import pytest from sqlalchemy import create_engine, func, select from sqlalchemy.orm import Session from modelforge_api.domain.recovery import ( POINT_IN_TIME_SUPPORT, ArtifactRecoveryClass, ArtifactRecoveryCreate, ArtifactRecoveryState, BackupMethod, BackupSetCreate, BackupState, RecoveryAssetClass, RecoveryFailureCode, RecoveryPolicyCreate, RecoveryReadiness, RestoreMode, RestorePlanCreate, SecretRecoveryClass, redact_database_url, ) from modelforge_api.persistence.models import ( ArtifactSet, ArtifactSetMember, AuditEvent, BackupManifestEntry, BackupSet, Base, ComputeNode, Model, ModelArtifact, ModelRevision, RecoveryAssetRecord, RecoveryPolicyRevision, StorageRoot, UpstreamSnapshot, ) from modelforge_api.services.audit import ( AUDIT_CURRENT_HASH_FORMAT, AuditChainCheckpoint, AuditWriter, ) from modelforge_api.services.recovery import RecoveryError, RecoveryService from modelforge_api.services.recovery_crypto import AesGcmBackupCipher, BackupCryptoError from modelforge_api.services.recovery_fingerprint import ( CURRENT_TRUTH_TABLES, SENSITIVE_COLUMN_TOKENS, diff_fingerprints, fingerprint_session, unclassified_tables, ) from modelforge_api.services.recovery_postgres import PostgresTarget from modelforge_api.settings import Settings @pytest.fixture def session() -> Session: engine = create_engine("sqlite+pysqlite:///:memory:") Base.metadata.create_all(engine) with Session(engine) as value: yield value # Derived rather than written out, so a scanner never has to decide whether a base64 literal # in the test suite is a real backup key. TEST_BACKUP_KEY = base64.b64encode(b"modelforge-m15-test-key-not-a-secret!"[:32]).decode() def settings_for(tmp_path: Path, **overrides: Any) -> Settings: base: dict[str, Any] = { "backup_root": tmp_path / "backups", "backup_restore_root": tmp_path / "restore", "backup_encryption_key": TEST_BACKUP_KEY, "config_root": tmp_path / "config", "database_url": "postgresql+psycopg://modelforge:secret@postgres:5432/modelforge", } base.update(overrides) return Settings(**base) def service(session: Session, tmp_path: Path, **overrides: Any) -> RecoveryService: return RecoveryService(session, settings_for(tmp_path, **overrides)) # --------------------------------------------------------------------- classification def test_every_persisted_table_is_classified_as_history_or_current_truth() -> None: assert unclassified_tables() == frozenset() assert "serving_gpu_leases" in CURRENT_TRUTH_TABLES assert "accelerator_telemetry_latest" in CURRENT_TRUTH_TABLES assert "audit_events" not in CURRENT_TRUTH_TABLES def test_defaults_seed_versioned_policies_and_a_complete_asset_inventory( session: Session, tmp_path: Path ) -> None: subject = service(session, tmp_path) subject.ensure_defaults() subject.ensure_defaults() # idempotent policies = subject.policies() assert len(policies) == 7 assert all(item.active for item in policies) assert all(item.revision == 1 for item in policies) by_key = {item.key: item for item in policies} assert by_key["control-plane.database"].asset_class is RecoveryAssetClass.AUTHORITATIVE assert by_key["control-plane.database"].backup_method is BackupMethod.POSTGRES_LOGICAL_CUSTOM assert by_key["control-plane.database"].encryption_required is True assert by_key["control-plane.database"].rpo_seconds == 86_400 assert by_key["artifacts.rehydratable"].rehydration_allowed is True assert by_key["runtime.ephemeral"].backup_method is BackupMethod.NOT_BACKED_UP assert by_key["external.projects"].external_dependency is True assert by_key["secrets.credentials"].secret_class is SecretRecoveryClass.ROTATABLE_SECRET assets = subject.assets() classes = {item.asset_class for item in assets} assert classes == { RecoveryAssetClass.AUTHORITATIVE, RecoveryAssetClass.REBUILDABLE, RecoveryAssetClass.EPHEMERAL, RecoveryAssetClass.EXTERNAL, RecoveryAssetClass.SECRET, } keys = {item.key for item in assets} assert {"postgres.modelforge", "artifacts.huggingface", "external.examplerag"} <= keys examplerag = next(item for item in assets if item.key == "external.examplerag") assert examplerag.readiness is RecoveryReadiness.EXTERNAL_DEPENDENCY def test_policy_revisions_are_versioned_and_supersede_the_previous_revision( session: Session, tmp_path: Path ) -> None: subject = service(session, tmp_path) subject.ensure_defaults() first = next(item for item in subject.policies() if item.key == "control-plane.database") second = subject.create_policy( RecoveryPolicyCreate( key="control-plane.database", name="Control-plane PostgreSQL", asset_class=RecoveryAssetClass.AUTHORITATIVE, backup_method=BackupMethod.POSTGRES_LOGICAL_CUSTOM, retention_days=14, minimum_verified_backups=3, rpo_seconds=43_200, rto_target_seconds=1_800, restore_verification="FULL_RESTORE", encryption_required=True, rationale="Tightened after the first measured disaster-recovery rehearsal.", ) ) assert second.revision == first.revision + 1 assert second.fingerprint != first.fingerprint active = [item for item in subject.policies() if item.active] assert [item.revision for item in active if item.key == "control-plane.database"] == [2] @pytest.mark.parametrize( ("kwargs", "message"), [ ( {"asset_class": RecoveryAssetClass.AUTHORITATIVE, "backup_method": BackupMethod.NOT_BACKED_UP}, "payload-bearing", ), ( {"asset_class": RecoveryAssetClass.EPHEMERAL, "backup_method": BackupMethod.FILE_COPY}, "must not claim a payload backup", ), ( {"asset_class": RecoveryAssetClass.EXTERNAL, "backup_method": BackupMethod.NOT_BACKED_UP}, "external dependency", ), ( {"asset_class": RecoveryAssetClass.SECRET, "backup_method": BackupMethod.MANIFEST_ONLY}, "secret recovery class", ), ], ) def test_recovery_policy_rejects_incoherent_classifications( kwargs: dict[str, Any], message: str ) -> None: payload: dict[str, Any] = { "key": "test.policy", "name": "Test policy", "retention_days": 7, "minimum_verified_backups": 1, "restore_verification": "HASH_ONLY", "encryption_required": False, "rationale": "coherence guard test", "rpo_seconds": None, **kwargs, } with pytest.raises(ValueError, match=message): RecoveryPolicyCreate(**payload) def test_authoritative_policy_requires_an_explicit_rpo() -> None: with pytest.raises(ValueError, match="explicit RPO"): RecoveryPolicyCreate( key="test.authoritative", name="Test", asset_class=RecoveryAssetClass.AUTHORITATIVE, backup_method=BackupMethod.FILE_COPY, retention_days=7, minimum_verified_backups=1, restore_verification="HASH_ONLY", encryption_required=True, rationale="an authoritative asset without an RPO is an unmeasured promise", ) def test_point_in_time_support_is_stated_without_ambiguity() -> None: assert POINT_IN_TIME_SUPPORT == "NOT_SUPPORTED" # --------------------------------------------------------------------- fingerprints def test_semantic_fingerprint_is_deterministic_and_excludes_secret_columns( session: Session, ) -> None: node = ComputeNode(key="gpu_node", hostname="gpu_node", display_name="GPU Node") session.add(node) session.flush() AuditWriter(session, "operator", "test").write("TEST", "test", None, {}) session.commit() first = fingerprint_session(session) second = fingerprint_session(session) assert first["digest"] == second["digest"] assert first["groups"]["identity"]["row_count"] == 1 assert first["groups"]["audit"]["row_count"] == 2 assert diff_fingerprints(first, second)["identical"] is True redacted = first["tables"]["node_credentials"]["redacted_columns"] assert "secret_hash" in redacted assert "token_hash" in first["tables"]["node_enrollments"]["redacted_columns"] for token in SENSITIVE_COLUMN_TOKENS: assert token in "".join(SENSITIVE_COLUMN_TOKENS) def test_semantic_fingerprint_diff_names_the_exact_differing_subject(session: Session) -> None: before = fingerprint_session(session) session.add(ComputeNode(key="gpu_node", hostname="gpu_node")) session.commit() after = fingerprint_session(session) diff = diff_fingerprints(before, after) assert diff["identical"] is False assert diff["differing_groups"] == ["identity"] changed = {item["table"]: item for item in diff["differences"]} assert changed["compute_nodes"]["row_delta"] == 1 # --------------------------------------------------------------------- encryption def test_encrypted_backup_round_trips_and_a_wrong_key_fails_closed(tmp_path: Path) -> None: plaintext = tmp_path / "payload.bin" plaintext.write_bytes(b"modelforge-control-plane-dump" * 5000) original = plaintext.read_bytes() cipher = AesGcmBackupCipher("correct-horse-battery-staple", "key-1") sealed = tmp_path / "payload.enc" cipher.encrypt_file(plaintext, sealed) assert sealed.read_bytes()[:6] == b"MFBK1\x00" assert b"modelforge-control-plane-dump" not in sealed.read_bytes() opened = tmp_path / "payload.out" cipher.decrypt_file(sealed, opened) assert opened.read_bytes() == original wrong = AesGcmBackupCipher("a-different-key-entirely", "key-1") leaked = tmp_path / "leaked.out" with pytest.raises(BackupCryptoError) as error: wrong.decrypt_file(sealed, leaked) assert error.value.code == "DECRYPTION_FAILED" assert not leaked.exists() assert not leaked.with_name(leaked.name + ".partial").exists() def test_corrupting_one_byte_of_an_encrypted_payload_is_detected(tmp_path: Path) -> None: plaintext = tmp_path / "payload.bin" plaintext.write_bytes(b"x" * 100_000) cipher = AesGcmBackupCipher("key-material", "key-1") sealed = tmp_path / "payload.enc" cipher.encrypt_file(plaintext, sealed) payload = bytearray(sealed.read_bytes()) payload[-1] ^= 0xFF sealed.write_bytes(bytes(payload)) with pytest.raises(BackupCryptoError): cipher.decrypt_file(sealed, tmp_path / "payload.out") def test_backup_key_fingerprint_identifies_the_key_without_exposing_it() -> None: first = AesGcmBackupCipher("shared-key", "key-1") same = AesGcmBackupCipher("shared-key", "key-2") other = AesGcmBackupCipher("different-key", "key-1") assert first.key_fingerprint == same.key_fingerprint assert first.key_fingerprint != other.key_fingerprint assert "shared-key" not in first.key_fingerprint # --------------------------------------------------------------------- backup lifecycle def seed_backup( session: Session, tmp_path: Path, *, backup_id: str = "m15-unit-backup", encrypted: bool = False, payload: bytes = b"PGDMP-fake-custom-dump-payload", milestone: str | None = None, ) -> tuple[RecoveryService, BackupSet]: """Create a VERIFIED-shaped backup on disk without invoking PostgreSQL tooling.""" subject = service(session, tmp_path) subject.ensure_defaults() policy = session.scalar( select(RecoveryPolicyRevision).where( RecoveryPolicyRevision.key == "control-plane.database" ) ) assert policy is not None root = tmp_path / "backups" directory = root / backup_id directory.mkdir(parents=True, exist_ok=True) object_name = "database.dump.enc" if encrypted else "database.dump" stored = directory / object_name if encrypted: raw = directory / "plain.tmp" raw.write_bytes(payload) AesGcmBackupCipher( TEST_BACKUP_KEY, "modelforge-backup-key-1" ).encrypt_file(raw, stored) raw.unlink() else: stored.write_bytes(payload) entries = [ { "logical_asset_type": "control_plane_database", "object_name": object_name, "relative_path": f"{backup_id}/{object_name}", "size_bytes": stored.stat().st_size, "sha256": hashlib.sha256(stored.read_bytes()).hexdigest(), "source_generation": "pg_dump:20260827_0021", "schema_version": "20260827_0021", "dependency_refs": { "postgres_major": 17, "extensions": [], "plaintext_sha256": hashlib.sha256(payload).hexdigest(), "plaintext_bytes": len(payload), }, } ] manifest = { "manifest_schema_version": "m15.1", "backup_id": backup_id, "entries": entries, } manifest_bytes = json.dumps(manifest, sort_keys=True, indent=2).encode("utf-8") (directory / "manifest.json").write_bytes(manifest_bytes) record = BackupSet( backup_id=backup_id, state=BackupState.CREATED.value, policy_revision_id=policy.id, modelforge_version="0.1.0", modelforge_commit="0" * 40, source_repository="ssh://git@example.invalid/ITWorx-ModelForge.git", source_reference="m15-backup-restore-dr", schema_revision="20260827_0021", environment_fingerprint={"modelforge_commit": "0" * 40}, database_identity={"major_version": 17, "server_version": "17.11"}, destination_root=str(root), payload_relative_path=f"{backup_id}/{object_name}", payload_sha256=entries[0]["sha256"], manifest_relative_path=f"{backup_id}/manifest.json", manifest_sha256=hashlib.sha256(manifest_bytes).hexdigest(), included_asset_classes=["AUTHORITATIVE", "REBUILDABLE"], excluded_asset_classes=["EPHEMERAL", "EXTERNAL", "SECRET"], payload_bytes=int(entries[0]["size_bytes"]), encrypted=encrypted, encryption_algorithm="AES-256-GCM" if encrypted else None, encryption_key_id="modelforge-backup-key-1" if encrypted else None, verification_details={}, milestone=milestone, reason="unit test fixture", created_by="test", started_at=datetime.now(UTC), completed_at=datetime.now(UTC), expires_at=datetime.now(UTC) + timedelta(days=policy.retention_days), ) session.add(record) session.flush() for entry in entries: session.add( BackupManifestEntry( backup_set_id=record.id, logical_asset_type=str(entry["logical_asset_type"]), object_name=str(entry["object_name"]), relative_path=str(entry["relative_path"]), size_bytes=int(entry["size_bytes"]), sha256=str(entry["sha256"]), source_generation=str(entry["source_generation"]), schema_version=str(entry["schema_version"]), dependency_refs=dict(entry["dependency_refs"]), # type: ignore[arg-type] ) ) session.commit() return subject, record def test_plan_backup_is_journaled_and_rejects_a_duplicate_identity( session: Session, tmp_path: Path ) -> None: subject = service(session, tmp_path) subject.ensure_defaults() planned = subject.plan_backup( BackupSetCreate(backup_id="m15-planned-one", reason="planned backup test") ) assert planned.state == BackupState.PLANNED.value assert session.scalar( select(func.count()).select_from(AuditEvent).where(AuditEvent.action == "BACKUP_PLANNED") ) == 1 with pytest.raises(RecoveryError) as error: subject.plan_backup( BackupSetCreate(backup_id="m15-planned-one", reason="duplicate identity") ) assert error.value.code == "backup_already_exists" def test_a_planned_backup_is_never_restore_eligible(session: Session, tmp_path: Path) -> None: subject = service(session, tmp_path) subject.ensure_defaults() planned = subject.plan_backup( BackupSetCreate(backup_id="m15-planned-two", reason="eligibility guard") ) assert subject.backup(planned.id).restore_eligible is False def test_verification_promotes_a_clean_backup_and_records_its_checks( session: Session, tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: subject, record = seed_backup(session, tmp_path) monkeypatch.setattr(subject.engine, "list_dump_contents", lambda _payload: 412) verified = subject.verify_backup(record.id) assert verified.state is BackupState.VERIFIED assert verified.restore_eligible is True checks = verified.verification_details["verification"]["checks"] assert checks["manifest_hash"] == "MATCH" assert checks["manifest_completeness"] == "COMPLETE" assert checks["payload_hashes"] == "MATCH" assert checks["archive_structure"] == "READABLE:412" actions = [ item.action for item in session.scalars(select(AuditEvent).order_by(AuditEvent.sequence)) ] assert "BACKUP_VERIFICATION_STARTED" in actions assert "BACKUP_VERIFIED" in actions def test_a_single_corrupted_payload_byte_makes_the_backup_restore_ineligible( session: Session, tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: subject, record = seed_backup(session, tmp_path, backup_id="m15-corrupt-payload") monkeypatch.setattr(subject.engine, "list_dump_contents", lambda _payload: 1) payload = tmp_path / "backups" / "m15-corrupt-payload" / "database.dump" data = bytearray(payload.read_bytes()) data[3] ^= 0x01 payload.write_bytes(bytes(data)) result = subject.verify_backup(record.id) assert result.state is BackupState.FAILED assert result.failure_code == RecoveryFailureCode.HASH_MISMATCH.value assert result.restore_eligible is False def test_a_tampered_manifest_is_detected_before_any_payload_is_read( session: Session, tmp_path: Path ) -> None: subject, record = seed_backup(session, tmp_path, backup_id="m15-tampered-manifest") manifest = tmp_path / "backups" / "m15-tampered-manifest" / "manifest.json" payload = json.loads(manifest.read_text("utf-8")) payload["entries"][0]["sha256"] = "0" * 64 manifest.write_text(json.dumps(payload, sort_keys=True, indent=2), encoding="utf-8") result = subject.verify_backup(record.id) assert result.state is BackupState.FAILED assert result.failure_code == RecoveryFailureCode.MANIFEST_HASH_MISMATCH.value def test_a_manifest_missing_its_database_payload_never_becomes_verified( session: Session, tmp_path: Path ) -> None: subject, record = seed_backup(session, tmp_path, backup_id="m15-partial-backup") directory = tmp_path / "backups" / "m15-partial-backup" manifest_path = directory / "manifest.json" payload = json.loads(manifest_path.read_text("utf-8")) payload["entries"] = [] body = json.dumps(payload, sort_keys=True, indent=2).encode("utf-8") manifest_path.write_bytes(body) record.manifest_sha256 = hashlib.sha256(body).hexdigest() session.commit() result = subject.verify_backup(record.id) assert result.state is BackupState.FAILED assert result.failure_code == RecoveryFailureCode.MANIFEST_INCOMPLETE.value def test_a_missing_payload_file_reports_payload_missing( session: Session, tmp_path: Path ) -> None: subject, record = seed_backup(session, tmp_path, backup_id="m15-missing-payload") (tmp_path / "backups" / "m15-missing-payload" / "database.dump").unlink() result = subject.verify_backup(record.id) assert result.state is BackupState.FAILED assert result.failure_code == RecoveryFailureCode.PAYLOAD_MISSING.value def test_an_encrypted_backup_verified_with_the_wrong_key_fails_closed( session: Session, tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: subject, record = seed_backup( session, tmp_path, backup_id="m15-wrong-key", encrypted=True ) monkeypatch.setattr(subject.engine, "list_dump_contents", lambda _payload: 1) subject.settings = settings_for(tmp_path, backup_encryption_key="a-completely-different-key") result = subject.verify_backup(record.id) assert result.state is BackupState.FAILED assert result.failure_code == RecoveryFailureCode.DECRYPTION_FAILED.value directory = tmp_path / "backups" / "m15-wrong-key" assert not any(path.name.startswith("database.dump.enc.verify-") for path in directory.iterdir()) def test_an_encrypted_backup_without_a_key_reports_key_unavailable( session: Session, tmp_path: Path ) -> None: subject, record = seed_backup( session, tmp_path, backup_id="m15-no-key", encrypted=True ) subject.settings = settings_for(tmp_path, backup_encryption_key=None) result = subject.verify_backup(record.id) assert result.state is BackupState.FAILED assert result.failure_code == RecoveryFailureCode.ENCRYPTION_KEY_UNAVAILABLE.value def test_an_interrupted_backup_is_reconciled_to_failed_and_never_looks_usable( session: Session, tmp_path: Path ) -> None: subject, record = seed_backup(session, tmp_path, backup_id="m15-interrupted") record.state = BackupState.CREATING.value session.commit() assert subject.reconcile_interrupted_backups() == 1 reloaded = subject.backup(record.id) assert reloaded.state is BackupState.FAILED assert reloaded.restore_eligible is False assert reloaded.failure_code == RecoveryFailureCode.MANIFEST_INCOMPLETE.value def test_a_second_backup_cannot_start_while_one_is_in_flight( session: Session, tmp_path: Path ) -> None: subject, record = seed_backup(session, tmp_path, backup_id="m15-concurrency") record.state = BackupState.CREATING.value session.commit() with pytest.raises(RecoveryError) as error: subject.plan_backup(BackupSetCreate(backup_id="m15-second", reason="concurrent attempt")) assert error.value.code == RecoveryFailureCode.CONCURRENT_OPERATION.value # --------------------------------------------------------------------- retention def test_retention_never_expires_the_last_verified_backup( session: Session, tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: subject, record = seed_backup(session, tmp_path, backup_id="m15-retention-only") monkeypatch.setattr(subject.engine, "list_dump_contents", lambda _payload: 1) subject.verify_backup(record.id) record.expires_at = datetime.now(UTC) - timedelta(days=1) session.commit() result = subject.apply_retention() assert result["expired"] == 0 assert subject.backup(record.id).state is BackupState.VERIFIED def test_retention_expires_surplus_backups_but_keeps_the_policy_minimum( session: Session, tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: subject = None records = [] for index in range(4): subject, record = seed_backup( session, tmp_path, backup_id=f"m15-retention-{index}", payload=f"dump-{index}".encode() ) records.append(record) assert subject is not None monkeypatch.setattr(subject.engine, "list_dump_contents", lambda _payload: 1) for offset, record in enumerate(records): subject.verify_backup(record.id) record.verified_at = datetime.now(UTC) - timedelta(hours=10 - offset) record.expires_at = datetime.now(UTC) - timedelta(days=1) session.commit() result = subject.apply_retention() # The policy keeps two verified backups; the two oldest surplus ones expire. assert result["expired"] == 2 states = {item.backup_id: item.state for item in subject.backups()} assert states["m15-retention-3"] is BackupState.VERIFIED assert states["m15-retention-2"] is BackupState.VERIFIED assert states["m15-retention-0"] is BackupState.EXPIRED def test_milestone_and_legal_hold_backups_are_never_expired( session: Session, tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: subject, held = seed_backup( session, tmp_path, backup_id="m15-milestone-hold", milestone="m15-baseline" ) monkeypatch.setattr(subject.engine, "list_dump_contents", lambda _payload: 1) subject.verify_backup(held.id) held.expires_at = datetime.now(UTC) - timedelta(days=400) session.commit() assert subject.apply_retention()["expired"] == 0 assert subject.backup(held.id).state is BackupState.VERIFIED # --------------------------------------------------------------------- restore planning def verified_backup( session: Session, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, backup_id: str ) -> tuple[RecoveryService, BackupSet]: subject, record = seed_backup(session, tmp_path, backup_id=backup_id) monkeypatch.setattr(subject.engine, "list_dump_contents", lambda _payload: 1) subject.verify_backup(record.id) return subject, record def restore_plan_request(record: BackupSet, **overrides: Any) -> RestorePlanCreate: payload: dict[str, Any] = { "backup_set_id": record.id, "mode": RestoreMode.VALIDATION, "target_environment": "ISOLATED", "target_label": "m15-isolated-rehearsal", "database_destination": "postgresql+psycopg://modelforge:secret@postgres:5432/mf_restore", "artifact_strategy": "MANIFEST_ONLY", "secret_strategy": "RESTORE_HASHES", "node_strategy": "NONE", "reason": "isolated M15 validation restore rehearsal", } payload.update(overrides) return RestorePlanCreate(**payload) def test_a_restore_plan_requires_a_verified_backup( session: Session, tmp_path: Path ) -> None: subject, record = seed_backup(session, tmp_path, backup_id="m15-unverified-plan") with pytest.raises(RecoveryError) as error: subject.create_restore_plan(restore_plan_request(record)) assert error.value.code == RecoveryFailureCode.BACKUP_NOT_RESTORE_ELIGIBLE.value def test_a_restore_may_never_target_the_running_control_plane_database( session: Session, tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: subject, record = verified_backup(session, tmp_path, monkeypatch, "m15-self-target") with pytest.raises(RecoveryError) as error: subject.create_restore_plan( restore_plan_request( record, database_destination=( "postgresql+psycopg://modelforge:secret@postgres:5432/modelforge" ), ) ) assert error.value.code == RecoveryFailureCode.DESTINATION_NOT_ISOLATED.value def test_a_production_target_is_refused_unless_explicitly_enabled( session: Session, tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: subject, record = verified_backup(session, tmp_path, monkeypatch, "m15-production-target") with pytest.raises(RecoveryError) as error: subject.create_restore_plan( restore_plan_request( record, mode=RestoreMode.DISASTER_RECOVERY, target_environment="PRODUCTION", ) ) assert error.value.code == RecoveryFailureCode.DESTINATION_NOT_ISOLATED.value def test_a_validation_restore_may_not_be_pointed_at_production() -> None: with pytest.raises(ValueError, match="never target a production environment"): RestorePlanCreate( backup_set_id=uuid.uuid4(), mode=RestoreMode.VALIDATION, target_environment="PRODUCTION", target_label="bad-plan", database_destination="postgresql+psycopg://user:pw@host:5432/db", artifact_strategy="NONE", secret_strategy="ROTATE", # noqa: S106 - a recovery strategy name, not a secret node_strategy="NONE", reason="a validation restore must never touch production", ) def test_restore_plans_are_idempotent_for_the_same_backup_and_destination( session: Session, tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: subject, record = verified_backup(session, tmp_path, monkeypatch, "m15-plan-idempotent") first = subject.create_restore_plan(restore_plan_request(record)) second = subject.create_restore_plan(restore_plan_request(record)) assert first.id == second.id def test_restore_plan_never_returns_the_destination_password( session: Session, tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: subject, record = verified_backup(session, tmp_path, monkeypatch, "m15-plan-redaction") plan = subject.create_restore_plan(restore_plan_request(record)) assert "secret" not in plan.database_destination_redacted assert plan.database_destination_redacted.endswith("mf_restore") assert redact_database_url("postgresql+psycopg://u:p@h:5432/d") == ( "postgresql+psycopg://u:***@h:5432/d" ) def test_a_backup_from_an_unknown_future_schema_fails_closed( session: Session, tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: subject, record = verified_backup(session, tmp_path, monkeypatch, "m15-schema-too-new") record.schema_revision = "29991231_9999" session.commit() ok, detail = subject._schema_compatibility(record.schema_revision) assert ok is False assert "SCHEMA_TOO_NEW" in detail def test_an_older_backup_schema_is_migrated_forward_rather_than_rejected( session: Session, tmp_path: Path ) -> None: subject = service(session, tmp_path) ok, detail = subject._schema_compatibility("20260824_0001") assert ok is True assert "migrated forward" in detail # --------------------------------------------------------------------- artifact recovery def seed_artifact_set(session: Session) -> tuple[ArtifactSet, StorageRoot]: model = Model( key="nomic-embed-text", display_name="Nomic Embed Text", upstream_provider="huggingface", upstream_source="nomic-ai/nomic-embed-text-v1.5", ) session.add(model) session.flush() revision = ModelRevision( model_id=model.id, upstream_revision="main", resolved_commit_sha="a" * 40, ) session.add(revision) session.flush() snapshot = UpstreamSnapshot( model_id=model.id, repository_id="nomic-ai/nomic-embed-text-v1.5", requested_revision="main", resolved_commit_sha="a" * 40, stale_after=datetime.now(UTC) + timedelta(hours=1), ) session.add(snapshot) session.flush() artifact_set = ArtifactSet( revision_id=revision.id, snapshot_id=snapshot.id, variant_key="safetensors", label="Safetensors", selection_reason="default safetensors variant", immutable_at=datetime.now(UTC), ) session.add(artifact_set) artifact = ModelArtifact( revision_id=revision.id, filename="model.safetensors", artifact_type="weights", serialization_format="safetensors", sha256="b" * 64, size_bytes=548_000_000, security_status="verified", ) session.add(artifact) session.flush() session.add( ArtifactSetMember( artifact_set_id=artifact_set.id, artifact_id=artifact.id, ordinal=0 ) ) node = ComputeNode(key="gpu_node", hostname="gpu_node") session.add(node) session.flush() storage_root = StorageRoot( compute_node_id=node.id, name="recovery-scratch", path="/mnt/recovery/disposable" ) session.add(storage_root) session.commit() return artifact_set, storage_root def test_artifact_recovery_records_the_exact_revision_and_preserves_lineage( session: Session, tmp_path: Path ) -> None: artifact_set, storage_root = seed_artifact_set(session) subject = service(session, tmp_path) subject.ensure_defaults() operation = subject.plan_artifact_recovery( ArtifactRecoveryCreate( artifact_set_id=artifact_set.id, target_storage_root_id=storage_root.id, reason="disposable location rehydration rehearsal", ) ) assert operation.recovery_class is ArtifactRecoveryClass.REHYDRATABLE assert operation.upstream_commit_sha == "a" * 40 assert operation.upstream_repository == "nomic-ai/nomic-embed-text-v1.5" assert operation.expected_files[0]["sha256"] == "b" * 64 assert operation.bytes_total == 548_000_000 assert operation.lineage["resolved_commit_sha"] == "a" * 40 assert operation.lineage["lineage_preserved"] is True def test_a_rehydrated_file_with_the_wrong_digest_is_recorded_as_failed( session: Session, tmp_path: Path ) -> None: artifact_set, storage_root = seed_artifact_set(session) subject = service(session, tmp_path) subject.ensure_defaults() operation = subject.plan_artifact_recovery( ArtifactRecoveryCreate( artifact_set_id=artifact_set.id, target_storage_root_id=storage_root.id, reason="hash mismatch rehearsal", ) ) result = subject.record_artifact_recovery( operation.id, state=ArtifactRecoveryState.RECOVERED, verified_files=[{"filename": "model.safetensors", "sha256": "c" * 64}], bytes_recovered=548_000_000, duration_seconds=42.0, ) assert result.state is ArtifactRecoveryState.FAILED assert result.failure_code == RecoveryFailureCode.ARTIFACT_HASH_MISMATCH.value def test_an_upstream_outage_is_reported_as_blocked_not_as_a_completed_recovery( session: Session, tmp_path: Path ) -> None: artifact_set, storage_root = seed_artifact_set(session) subject = service(session, tmp_path) subject.ensure_defaults() operation = subject.plan_artifact_recovery( ArtifactRecoveryCreate( artifact_set_id=artifact_set.id, target_storage_root_id=storage_root.id, reason="upstream unavailable rehearsal", ) ) result = subject.record_artifact_recovery( operation.id, state=ArtifactRecoveryState.BLOCKED, verified_files=[], bytes_recovered=0, duration_seconds=None, failure_code=RecoveryFailureCode.ARTIFACT_REHYDRATION_BLOCKED.value, failure_reason="Hugging Face was unreachable during the rehearsal", ) assert result.state is ArtifactRecoveryState.BLOCKED assert result.failure_code == RecoveryFailureCode.ARTIFACT_REHYDRATION_BLOCKED.value assert result.bytes_recovered == 0 def test_an_artifact_set_without_an_exact_upstream_revision_is_not_rehydratable( session: Session, tmp_path: Path ) -> None: artifact_set, storage_root = seed_artifact_set(session) snapshot = session.get(UpstreamSnapshot, artifact_set.snapshot_id) assert snapshot is not None session.delete(snapshot) model = session.scalar(select(Model)) assert model is not None # A locally produced model has a revision digest but no upstream repository to redownload from. model.upstream_source = "" session.commit() subject = service(session, tmp_path) subject.ensure_defaults() operation = subject.plan_artifact_recovery( ArtifactRecoveryCreate( artifact_set_id=artifact_set.id, target_storage_root_id=storage_root.id, reason="non-rehydratable classification test", ) ) assert operation.recovery_class is ArtifactRecoveryClass.NON_REHYDRATABLE assert operation.state is ArtifactRecoveryState.BLOCKED assert operation.failure_code == RecoveryFailureCode.ARTIFACT_NOT_REHYDRATABLE.value # --------------------------------------------------------------------- readiness def test_readiness_reports_authoritative_state_as_unprotected_without_a_verified_backup( session: Session, tmp_path: Path ) -> None: subject = service(session, tmp_path) subject.ensure_defaults() dashboard = subject.dashboard() assert dashboard.latest_verified_backup_id is None assert dashboard.stale_backup is True assert "postgres.modelforge" in dashboard.unprotected_assets assert dashboard.point_in_time_support == "NOT_SUPPORTED" assert dashboard.coverage_ratio < 1.0 def test_readiness_becomes_protected_once_a_verified_backup_exists( session: Session, tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: subject, _record = verified_backup(session, tmp_path, monkeypatch, "m15-readiness") dashboard = subject.dashboard() assert dashboard.latest_verified_backup_id == "m15-readiness" assert dashboard.unprotected_assets == [] assert dashboard.stale_backup is False assert dashboard.verified_backup_count == 1 assert dashboard.latest_verified_schema_revision == "20260827_0021" database_entry = next( item for item in dashboard.readiness if item.asset_key == "postgres.modelforge" ) assert database_entry.readiness is RecoveryReadiness.PROTECTED assert "m15-readiness" in database_entry.detail rotation = next(item for item in dashboard.readiness if item.asset_key == "credentials.project") assert rotation.readiness is RecoveryReadiness.ROTATION_REQUIRED def test_recovery_asset_inventory_is_persisted_and_linked_to_its_policy( session: Session, tmp_path: Path ) -> None: subject = service(session, tmp_path) subject.ensure_defaults() records = list(session.scalars(select(RecoveryAssetRecord))) assert len(records) == 14 assert all(record.policy_revision_id is not None for record in records) external = [item for item in records if item.asset_class == "EXTERNAL"] assert {item.key for item in external} == { "external.examplerag", "external.examplevision", "external.huggingface", "external.gitea", } def test_a_backup_set_is_sealable_across_flushes_but_immutable_afterwards( session: Session, tmp_path: Path ) -> None: """A backup is planned, written and only then stamped; sealing must not be self-blocking.""" subject = service(session, tmp_path) subject.ensure_defaults() record = subject.plan_backup( BackupSetCreate(backup_id="m15-sealing-guard", reason="immutability sealing regression") ) record.schema_revision = "20260827_0021" record.payload_bytes = 4096 record.manifest_sha256 = "a" * 64 record.immutable_at = datetime.now(UTC) session.commit() assert subject.backup(record.id).manifest_sha256 == "a" * 64 # Verification and retention still work after sealing; identity fields no longer do. record.state = BackupState.VERIFIED.value record.verified_at = datetime.now(UTC) session.commit() assert subject.backup(record.id).state is BackupState.VERIFIED record.manifest_sha256 = "b" * 64 with pytest.raises(ValueError, match="immutable approved fields cannot change"): session.commit() session.rollback() def test_a_build_without_an_alembic_tree_cannot_claim_schema_compatibility( session: Session, tmp_path: Path ) -> None: """A packaging error must fail closed, never pass a compatibility check by accident.""" empty_tree = tmp_path / "no-alembic" empty_tree.mkdir() subject = service(session, tmp_path, alembic_directory=empty_tree) from modelforge_api.services import recovery as recovery_module recovery_module._known_alembic_revisions.cache_clear() original = recovery_module._alembic_root recovery_module._alembic_root = lambda configured=None: None # type: ignore[assignment] try: ok, detail = subject._schema_compatibility("20260827_0021") finally: recovery_module._alembic_root = original # type: ignore[assignment] recovery_module._known_alembic_revisions.cache_clear() assert ok is False assert "SCHEMA_UNKNOWN" in detail def test_the_alembic_tree_is_found_from_the_repository_layout() -> None: from modelforge_api.services.recovery import _alembic_root, _known_alembic_revisions root = _alembic_root() assert root is not None and (root / "versions").is_dir() revisions = _known_alembic_revisions() assert revisions[0] == "20260824_0001" assert revisions[-1] == "20260830_0024" assert len(revisions) == len(set(revisions)) def test_reconciliation_is_scoped_to_its_own_restore_operation( session: Session, tmp_path: Path ) -> None: """M15 rehearsal F/G regression. A backup legitimately carries the reconciliation audit events of *earlier* restores. Matching the marker on the action alone made every later restore believe it had already reconciled, silently leaving thousands of stale GPU leases presented as present-day truth. """ from modelforge_api.services import recovery as recovery_module source = inspect.getsource(recovery_module._RECONCILE_MARKER_QUERY_MARKER) assert "resource_id" in source assert "RECOVERY_RECONCILIATION_COMPLETED" in source def test_recovery_appends_a_canonical_serialized_marker_to_the_restored_chain( session: Session, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: from modelforge_api.services import recovery as recovery_module previous_hash = "a" * 64 correlation_id = uuid.UUID("11111111-1111-1111-1111-111111111111") event_id = uuid.UUID("22222222-2222-2222-2222-222222222222") occurred_at = datetime(2026, 8, 30, 8, 15, 30, 123456, tzinfo=UTC) generated = iter((correlation_id, event_id)) monkeypatch.setattr(recovery_module.uuid, "uuid4", lambda: next(generated)) class FrozenDateTime(datetime): @classmethod def now(cls, tz: Any = None) -> datetime: return occurred_at monkeypatch.setattr(recovery_module, "datetime", FrozenDateTime) class MarkerEngine: def __init__(self) -> None: self.queries: list[str] = [] def scalar(self, _target: PostgresTarget, sql: str) -> str: self.queries.append(sql) return "8:" + "a" * 64 subject = service(session, tmp_path) marker_engine = MarkerEngine() subject.engine = marker_engine # type: ignore[assignment] resource_id = "33333333-3333-3333-3333-333333333333" result_hash = subject._append_restored_audit_marker( PostgresTarget("postgres", 5432, "restored", "modelforge", None), resource_id=resource_id, expected_checkpoint=AuditChainCheckpoint( singleton_id=1, event_count=7, last_sequence=7, last_event_hash=previous_hash, hash_format=AUDIT_CURRENT_HASH_FORMAT, v2_start_sequence=5, legacy_prefix_count=4, legacy_prefix_seal="b" * 64, ), ) insert_sql = marker_engine.queries[0] assert result_hash == "a" * 64 assert "modelforge_audit.append_event_v2" in insert_sql assert f"'{previous_hash}'" in insert_sql assert ", 7, 7," in insert_sql assert "2026-08-30T08:15:30.123456Z" in insert_sql assert "insert into audit_events" not in insert_sql.lower() assert "update audit_chain_heads" not in insert_sql.lower() def test_the_restore_gate_refuses_to_be_ready_with_residual_current_truth() -> None: """A restore that skipped reconciliation must not reach READY.""" from modelforge_api.services import recovery as recovery_module source = inspect.getsource(recovery_module.RecoveryService._validate_restored) assert "current_truth_reconciled" in source assert "residual_current_truth_rows" in source assert "stale_current_truth" in source def _m0_dropped_tables() -> list[str]: """The ordered table names the M0 baseline migration drops to rebuild its historical shape.""" import ast from modelforge_api.services.recovery import _alembic_root root = _alembic_root() assert root is not None source = (root / "versions" / "20260824_0001_m0_foundation.py").read_text("utf-8") tree = ast.parse(source) upgrade = next( node for node in tree.body if isinstance(node, ast.FunctionDef) and node.name == "upgrade" ) ordered: list[str] = [] for node in ast.walk(upgrade): if isinstance(node, ast.Tuple): names = [ item.value for item in node.elts if isinstance(item, ast.Constant) and isinstance(item.value, str) ] if any(name in Base.metadata.tables for name in names): ordered.extend(name for name in names if name in Base.metadata.tables) return ordered def test_the_m0_baseline_drops_every_later_table_before_its_dependencies() -> None: """A fresh `empty -> head` migration must still reconstruct the historical M0 shape. The M0 baseline calls `create_all` against the evolving metadata and then drops the later milestone tables. A new milestone that adds a table referencing an M0-era table must drop it first, or a clean-room deployment fails on a dependent foreign key. """ dropped = _m0_dropped_tables() assert dropped, "the M0 baseline drop list could not be parsed" position = {name: index for index, name in enumerate(dropped)} violations: list[str] = [] for name in dropped: table = Base.metadata.tables[name] for constraint in table.foreign_key_constraints: target = constraint.referred_table.name if target in position and position[target] < position[name]: violations.append(f"{name} is dropped after its dependency {target}") assert violations == [], "; ".join(violations) def test_every_m15_recovery_table_is_reset_by_the_m0_baseline() -> None: dropped = set(_m0_dropped_tables()) recovery_tables = { "recovery_policy_revisions", "recovery_asset_records", "backup_sets", "backup_manifest_entries", "restore_plans", "restore_operations", "restore_operation_events", "artifact_recovery_operations", } assert recovery_tables <= dropped assert recovery_tables <= set(Base.metadata.tables) def test_an_unwritable_backup_destination_fails_closed_with_a_typed_code( session: Session, tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: """M16 chaos regression. An unwritable backup root escaped as an unhandled OSError, returning HTTP 500 and leaving the backup set stuck in CREATING with no failure code, so BACKUP_FAILED never fired. """ subject = service(session, tmp_path) subject.ensure_defaults() record = subject.plan_backup( BackupSetCreate(backup_id="m16-unwritable-root", reason="unwritable destination regression") ) def refuse(_self: object, _backup_id: str) -> Path: raise PermissionError(13, "Permission denied", str(tmp_path / "backups")) monkeypatch.setattr(RecoveryService, "_backup_directory", refuse) response = subject.execute_backup(record.id) assert response.state is BackupState.FAILED assert response.restore_eligible is False assert response.failure_code == RecoveryFailureCode.DESTINATION_UNAVAILABLE.value assert "could not be written" in (response.failure_reason or "") assert response.completed_at is not None