1243 lines
47 KiB
Python
1243 lines
47 KiB
Python
"""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
|