Files
ModelForge/backend/tests/test_invariants_m16.py

761 lines
24 KiB
Python

"""M16 invariant tests.
An invariant check that never fires is worse than none: it reports safety it did not verify. Each
check therefore has a negative case that constructs the violation and proves the check detects it.
"""
from __future__ import annotations
import uuid
from datetime import UTC, datetime, timedelta
import pytest
from sqlalchemy import create_engine
from sqlalchemy import text as sa_text
from sqlalchemy.engine import Engine
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session
from modelforge_api.domain.audit import AUDIT_HASH_FORMAT_V1
from modelforge_api.persistence.models import (
Accelerator,
ArtifactLocation,
ArtifactSet,
AuditEvent,
BackupSet,
Base,
Capability,
CapabilityContract,
CapabilityDeployment,
ComputeNode,
EmbeddingSpace,
LifecycleApprovalRequest,
LifecycleOperation,
LifecyclePolicyRevision,
LifecyclePromotionPlan,
LifecycleSubject,
MigrationCutoverOperation,
Model,
ModelArtifact,
ModelRevision,
NodeCredential,
RecoveryPolicyRevision,
ResidencyAllocation,
RestorePlan,
RuntimeProfile,
ServiceClient,
ServiceCredential,
ServingGpuLease,
ServingJob,
StorageRoot,
UpstreamSnapshot,
)
from modelforge_api.services.audit import AuditWriter
from modelforge_api.services.invariants import (
CHECKS,
InvariantStatus,
check_invariants,
invariant_keys,
summarise,
)
@pytest.fixture
def session() -> Session:
engine = create_engine("sqlite+pysqlite:///:memory:")
Base.metadata.create_all(engine)
with Session(engine) as value:
yield value
def _now() -> datetime:
return datetime.now(UTC)
def drop_guard(session: Session, index_name: str) -> None:
"""Remove one protective index on the in-memory test database.
Several invariants are also enforced by a partial unique index, which is the right primary
defence. The invariant exists as defence in depth for the case where that guard is bypassed —
a hand-edited database, a bad migration, a restore from an older schema — so the negative test
has to remove the guard to reach the code path it is meant to cover.
"""
session.commit()
session.execute(sa_text(f"DROP INDEX IF EXISTS {index_name}"))
session.commit()
def violated(session: Session, key: str) -> bool:
report = check_invariants(session)
result = next(item for item in report.results if item.key == key)
return result.status is InvariantStatus.VIOLATED
# --------------------------------------------------------------------- framework
def test_an_empty_platform_violates_nothing(session: Session) -> None:
report = check_invariants(session)
assert report.checked == 16
assert report.violated == 0
assert report.holding == 16
assert report.ok is True
assert summarise(report)["violations"] == []
def test_every_declared_invariant_key_is_actually_checked(session: Session) -> None:
report = check_invariants(session)
assert {item.key for item in report.results} == set(invariant_keys())
assert len(CHECKS) == len(invariant_keys())
assert len({item.key for item in report.results}) == len(report.results)
def test_invariants_never_mutate_state(session: Session) -> None:
"""A safety check that writes is a safety check that can cause the incident it looks for."""
session.add(ComputeNode(key="gpu_node", hostname="gpu_node"))
session.commit()
before = session.query(ComputeNode).count()
check_invariants(session)
check_invariants(session)
assert session.query(ComputeNode).count() == before
assert not session.dirty and not session.new and not session.deleted
# --------------------------------------------------------------------- fixtures
def contract(session: Session, key: str = "rag.embedding") -> CapabilityContract:
capability = Capability(key=key, description=key)
session.add(capability)
session.flush()
record = CapabilityContract(
capability_id=capability.id,
version=1,
input_schema={},
output_schema={},
contract={},
upgrade_class="behavioral",
)
session.add(record)
session.flush()
return record
def deployment(
session: Session,
contract_id: uuid.UUID,
*,
production: bool = True,
status: str = "stable",
artifact_set_id: uuid.UUID | None = None,
embedding_space_id: uuid.UUID | None = None,
approval: uuid.UUID | None = None,
) -> CapabilityDeployment:
record = CapabilityDeployment(
capability_contract_id=contract_id,
deployment_candidate_id=uuid.uuid4(),
production=production,
status=status,
channel="stable",
artifact_set_id=artifact_set_id or uuid.uuid4(),
runtime_profile_id=uuid.uuid4(),
compute_node_id=uuid.uuid4(),
accelerator_id=uuid.uuid4(),
embedding_space_id=embedding_space_id,
production_approval_id=approval or uuid.uuid4(),
health_status="ready_on_demand",
config_fingerprint=uuid.uuid4().hex * 2,
provenance={},
)
session.add(record)
session.flush()
return record
# --------------------------------------------------------------------- negative cases
def test_the_schema_refuses_a_second_stable_production_deployment(session: Session) -> None:
record = contract(session)
deployment(session, record.id)
session.commit()
assert not violated(session, "single_production_stable")
with pytest.raises(IntegrityError):
deployment(session, record.id)
session.rollback()
def test_a_second_stable_production_deployment_is_detected_if_the_guard_is_bypassed(
session: Session,
) -> None:
record = contract(session)
deployment(session, record.id)
session.commit()
drop_guard(session, "uq_capability_deployment_one_production_stable")
deployment(session, record.id)
session.commit()
assert violated(session, "single_production_stable")
def test_two_stable_artifact_sets_for_one_contract_are_detected(session: Session) -> None:
record = contract(session)
first = deployment(session, record.id)
session.commit()
assert not violated(session, "stable_identity_is_singular")
drop_guard(session, "uq_capability_deployment_one_production_stable")
deployment(session, record.id, artifact_set_id=uuid.uuid4())
session.commit()
assert violated(session, "stable_identity_is_singular")
assert first.artifact_set_id is not None
def test_two_enabled_nodes_sharing_hardware_are_detected(session: Session) -> None:
session.add(ComputeNode(key="node-a", hostname="gpu_node", hardware_fingerprint="a" * 64))
session.commit()
assert not violated(session, "single_node_identity")
session.add(ComputeNode(key="node-b", hostname="gpu_node", hardware_fingerprint="a" * 64))
session.commit()
assert violated(session, "single_node_identity")
def test_a_disabled_duplicate_node_is_not_a_violation(session: Session) -> None:
"""Superseding a node by disabling it is the documented recovery, not a violation."""
session.add(ComputeNode(key="node-a", hostname="gpu_node", hardware_fingerprint="a" * 64))
session.add(
ComputeNode(
key="node-b", hostname="gpu_node", hardware_fingerprint="a" * 64, enabled=False
)
)
session.commit()
assert not violated(session, "single_node_identity")
def test_the_schema_refuses_a_second_active_node_credential(session: Session) -> None:
node = ComputeNode(key="gpu_node", hostname="gpu_node")
session.add(node)
session.flush()
session.add(NodeCredential(compute_node_id=node.id, secret_hash="a" * 64))
session.commit()
assert not violated(session, "single_active_node_credential")
session.add(NodeCredential(compute_node_id=node.id, secret_hash="b" * 64))
with pytest.raises(IntegrityError):
session.flush()
session.rollback()
def test_a_second_active_node_credential_is_detected_if_the_guard_is_bypassed(
session: Session,
) -> None:
node = ComputeNode(key="gpu_node", hostname="gpu_node")
session.add(node)
session.flush()
session.add(NodeCredential(compute_node_id=node.id, secret_hash="a" * 64))
session.commit()
drop_guard(session, "uq_active_node_credential")
session.add(NodeCredential(compute_node_id=node.id, secret_hash="b" * 64))
session.commit()
assert violated(session, "single_active_node_credential")
def test_an_expired_but_active_gpu_lease_is_detected(session: Session) -> None:
node = ComputeNode(key="gpu_node", hostname="gpu_node")
session.add(node)
session.flush()
accelerator = Accelerator(
compute_node_id=node.id, device_index=0, device_uuid="GPU-1", name="RTX"
)
session.add(accelerator)
session.flush()
lease = ServingGpuLease(
accelerator_id=accelerator.id,
capability_deployment_id=uuid.uuid4(),
request_id=uuid.uuid4(),
reserved_vram_bytes=1024,
priority="production",
state="active",
owner="test",
lease_type="request",
expires_at=_now() + timedelta(minutes=5),
generation=1,
)
session.add(lease)
session.commit()
assert not violated(session, "no_stale_gpu_lease")
lease.expires_at = _now() - timedelta(minutes=5)
session.commit()
assert violated(session, "no_stale_gpu_lease")
def test_a_released_expired_lease_is_not_a_violation(session: Session) -> None:
node = ComputeNode(key="gpu_node", hostname="gpu_node")
session.add(node)
session.flush()
accelerator = Accelerator(
compute_node_id=node.id, device_index=0, device_uuid="GPU-1", name="RTX"
)
session.add(accelerator)
session.flush()
session.add(
ServingGpuLease(
accelerator_id=accelerator.id,
capability_deployment_id=uuid.uuid4(),
request_id=uuid.uuid4(),
reserved_vram_bytes=1024,
priority="production",
state="released",
owner="test",
lease_type="request",
expires_at=_now() - timedelta(hours=2),
released_at=_now() - timedelta(hours=2),
generation=1,
)
)
session.commit()
assert not violated(session, "no_stale_gpu_lease")
def test_two_embedding_spaces_in_stable_production_are_detected(session: Session) -> None:
record = contract(session)
first = EmbeddingSpace(
capability_contract_id=record.id,
artifact_set_id=uuid.uuid4(),
runtime_profile_id=uuid.uuid4(),
identity_digest="a" * 64,
dimension=1024,
normalized=True,
migration_class="requires_reindex",
identity_facts={},
immutable_at=_now(),
)
second = EmbeddingSpace(
capability_contract_id=record.id,
artifact_set_id=uuid.uuid4(),
runtime_profile_id=uuid.uuid4(),
identity_digest="b" * 64,
dimension=1024,
normalized=True,
migration_class="requires_reindex",
identity_facts={},
immutable_at=_now(),
)
session.add_all([first, second])
session.flush()
shared = uuid.uuid4()
deployment(session, record.id, embedding_space_id=first.id, artifact_set_id=shared)
session.commit()
assert not violated(session, "no_mixed_embedding_space")
drop_guard(session, "uq_capability_deployment_one_production_stable")
deployment(session, record.id, embedding_space_id=second.id, artifact_set_id=shared)
session.commit()
assert violated(session, "no_mixed_embedding_space")
def lifecycle_operation(session: Session, *, stage: str, approver: str = "approver") -> LifecycleOperation:
subject = LifecycleSubject(
target_type="LAB_REHEARSAL", target_ref="test", environment="LAB", state="LAB_READY"
)
session.add(subject)
session.flush()
policy = LifecyclePolicyRevision(
key="test",
revision=1,
scope="LAB",
requirements={},
fingerprint=uuid.uuid4().hex,
created_by="test",
)
session.add(policy)
session.flush()
approval = LifecycleApprovalRequest(
policy_revision_id=policy.id,
subject_id=subject.id,
target_type="LAB_REHEARSAL",
target_ref="test",
environment="LAB",
requested_transition="LAB_STABLE",
evidence_snapshot={},
evidence_fingerprint="c" * 64,
status="APPROVED",
requested_by="test",
reason="invariant regression fixture",
)
session.add(approval)
session.flush()
plan = LifecyclePromotionPlan(
approval_request_id=approval.id,
subject_id=subject.id,
current_state="LAB_READY",
desired_state="LAB_STABLE",
migration_class="behavioral",
rollback_target_ref="test",
project_consumers=[],
affected_identities={},
impact_analysis={},
canary_strategy={},
drain_strategy={},
health_gates={},
automatic_abort_conditions=[],
plan_fingerprint=uuid.uuid4().hex,
status="APPROVED",
created_by="test",
immutable_at=_now(),
)
session.add(plan)
session.flush()
operation = LifecycleOperation(
promotion_plan_id=plan.id,
stage=stage,
idempotency_key=uuid.uuid4().hex,
expected_subject_version=1,
requester="test",
approver=approver,
executor="test",
)
session.add(operation)
session.flush()
return operation
def test_a_lifecycle_commit_without_an_approver_is_detected(session: Session) -> None:
operation = lifecycle_operation(session, stage="COMMITTED")
session.commit()
assert not violated(session, "lifecycle_commit_has_evidence")
operation.approver = ""
session.commit()
assert violated(session, "lifecycle_commit_has_evidence")
def test_a_cutover_commit_without_external_truth_is_detected(session: Session) -> None:
cutover = MigrationCutoverOperation(
migration_plan_id=uuid.uuid4(),
stage="COMMITTED",
idempotency_key=uuid.uuid4().hex,
generation=1,
expected_plan_version=1,
source_before="source",
target_after="target",
external_state_fingerprint="d" * 64,
)
session.add(cutover)
session.commit()
assert not violated(session, "cutover_commit_has_validation")
cutover.external_state_fingerprint = ""
session.commit()
assert violated(session, "cutover_commit_has_validation")
def test_a_restore_plan_against_an_unverified_backup_is_detected(session: Session) -> None:
policy = RecoveryPolicyRevision(
key="control-plane.database",
revision=1,
name="db",
asset_class="AUTHORITATIVE",
backup_method="POSTGRES_LOGICAL_CUSTOM",
retention_days=30,
minimum_verified_backups=1,
restore_verification="FULL_RESTORE",
rationale="test recovery policy",
fingerprint=uuid.uuid4().hex,
created_by="test",
)
session.add(policy)
session.flush()
backup = BackupSet(
backup_id="test-backup",
state="VERIFIED",
policy_revision_id=policy.id,
modelforge_version="0.1.0",
destination_root="/data/backups",
environment_fingerprint={},
database_identity={},
included_asset_classes=[],
excluded_asset_classes=[],
verification_details={},
reason="test",
created_by="test",
)
session.add(backup)
session.flush()
plan = RestorePlan(
backup_set_id=backup.id,
mode="VALIDATION",
target_environment="ISOLATED",
target_label="test",
database_destination="postgresql+psycopg://u:p@h:5432/d",
artifact_strategy="NONE",
secret_strategy="ROTATE", # noqa: S106 - a recovery strategy name, not a secret
node_strategy="NONE",
preflight={},
validation_requirements={},
fingerprint=uuid.uuid4().hex,
reason="test",
created_by="test",
)
session.add(plan)
session.commit()
assert not violated(session, "restore_requires_verified_backup")
backup.state = "CREATED"
session.commit()
assert violated(session, "restore_requires_verified_backup")
def test_a_credential_used_after_revocation_is_detected(session: Session) -> None:
node = ComputeNode(key="gpu_node", hostname="gpu_node")
session.add(node)
session.flush()
credential = NodeCredential(
compute_node_id=node.id,
secret_hash="a" * 64,
revoked_at=_now() - timedelta(minutes=10),
last_used_at=_now() - timedelta(minutes=20),
)
session.add(credential)
session.commit()
assert not violated(session, "revoked_credentials_stay_revoked")
credential.last_used_at = _now()
session.commit()
assert violated(session, "revoked_credentials_stay_revoked")
def test_a_service_credential_used_after_revocation_is_detected(session: Session) -> None:
client = ServiceClient(
name="test-client",
allowed_capabilities=["rag.embedding@1"],
requests_per_minute=60,
max_concurrent_requests=2,
workload_priority="production",
integration_environment="LAB",
purpose="test",
)
session.add(client)
session.flush()
credential = ServiceCredential(
service_client_id=client.id,
secret_hash="a" * 64,
secret_prefix="mfsvc_", # noqa: S106 - a non-secret credential prefix
revoked_at=_now() - timedelta(minutes=10),
last_used_at=_now(),
)
session.add(credential)
session.commit()
assert violated(session, "revoked_credentials_stay_revoked")
def test_a_capability_client_claiming_operator_scope_is_detected(session: Session) -> None:
client = ServiceClient(
name="test-client",
allowed_capabilities=["rag.embedding@1"],
requests_per_minute=60,
max_concurrent_requests=2,
workload_priority="production",
integration_environment="LAB",
purpose="test",
)
session.add(client)
session.commit()
assert not violated(session, "capability_clients_are_not_operators")
client.allowed_capabilities = ["rag.embedding@1", "admin.recovery"]
session.commit()
assert violated(session, "capability_clients_are_not_operators")
@pytest.mark.parametrize("scope", ["admin.lifecycle", "operator", "node.publish", "recovery"])
def test_every_forbidden_client_scope_is_rejected(session: Session, scope: str) -> None:
client = ServiceClient(
name="test-client",
allowed_capabilities=[scope],
requests_per_minute=60,
max_concurrent_requests=2,
workload_priority="production",
integration_environment="LAB",
purpose="test",
)
session.add(client)
session.commit()
assert violated(session, "capability_clients_are_not_operators")
def test_an_unsafe_artifact_in_a_verified_location_is_detected(session: Session) -> None:
model = Model(
key="m", display_name="M", upstream_provider="huggingface", upstream_source="org/model"
)
session.add(model)
session.flush()
revision = ModelRevision(
model_id=model.id, upstream_revision="main", resolved_commit_sha="a" * 40
)
session.add(revision)
session.flush()
artifact = ModelArtifact(
revision_id=revision.id,
filename="model.safetensors",
artifact_type="weights",
serialization_format="safetensors",
sha256="b" * 64,
size_bytes=1024,
security_status="verified",
)
session.add(artifact)
node = ComputeNode(key="gpu_node", hostname="gpu_node")
session.add(node)
session.flush()
root = StorageRoot(compute_node_id=node.id, name="root", path="/mnt/models")
session.add(root)
session.flush()
location = ArtifactLocation(
artifact_id=artifact.id,
storage_root_id=root.id,
relative_path="model.safetensors",
status="verified",
)
session.add(location)
session.commit()
assert not violated(session, "no_unsafe_artifact_promoted")
artifact.security_status = "blocked"
session.commit()
assert violated(session, "no_unsafe_artifact_promoted")
def test_an_orphaned_serving_job_is_detected(session: Session) -> None:
node = ComputeNode(key="gpu_node", hostname="gpu_node")
session.add(node)
session.flush()
job = ServingJob(
capability_deployment_id=uuid.uuid4(),
compute_node_id=node.id,
operation="invoke",
status="leased",
priority="production",
idempotency_key=uuid.uuid4().hex,
lease_expires_at=_now() + timedelta(minutes=2),
)
session.add(job)
session.commit()
assert not violated(session, "no_orphan_serving_work")
job.lease_expires_at = _now() - timedelta(minutes=2)
session.commit()
assert violated(session, "no_orphan_serving_work")
def test_a_residency_allocation_on_a_disabled_node_is_detected(session: Session) -> None:
node = ComputeNode(key="gpu_node", hostname="gpu_node")
session.add(node)
session.flush()
allocation = ResidencyAllocation(
capability_deployment_id=uuid.uuid4(),
compute_node_id=node.id,
accelerator_id=uuid.uuid4(),
state="resident",
health="healthy",
generation=1,
)
session.add(allocation)
session.commit()
assert not violated(session, "no_orphan_serving_work")
node.enabled = False
session.commit()
assert violated(session, "no_orphan_serving_work")
def test_a_production_deployment_without_an_approval_is_detected(session: Session) -> None:
record = contract(session)
item = deployment(session, record.id)
session.commit()
assert not violated(session, "no_hidden_auto_promotion")
item.production_approval_id = None
session.commit()
assert violated(session, "no_hidden_auto_promotion")
def test_a_duplicated_audit_sequence_is_detected(session: Session) -> None:
writer = AuditWriter(session, "test", "test")
writer.write("TEST", "test", None, {})
writer.write("TEST", "test", None, {})
session.commit()
assert not violated(session, "audit_chain_intact")
drop_guard(session, "ix_audit_events_sequence")
session.commit()
engine = session.get_bind()
assert isinstance(engine, Engine)
with engine.begin() as connection:
connection.execute(
AuditEvent.__table__.insert().values(
correlation_id=str(uuid.uuid4()),
actor_type="test",
actor_id="test",
action="TEST",
resource_type="test",
outcome="success",
details={},
event_hash=uuid.uuid4().hex + uuid.uuid4().hex,
hash_format=AUDIT_HASH_FORMAT_V1,
sequence=2,
)
)
session.commit()
assert violated(session, "audit_chain_intact")
def test_a_resurrected_decommissioned_node_is_detected(session: Session) -> None:
session.add(
ComputeNode(
key="decommissioned-invariant-test",
hostname="disposable-invariant-test",
enabled=True,
status="active",
liveness_state="online",
decommissioned_at=_now(),
production_eligible=True,
inventory={"unexpected": "current state"},
)
)
session.commit()
assert violated(session, "decommissioned_nodes_are_terminal")
def test_the_report_summarises_only_violations(session: Session) -> None:
record = contract(session)
deployment(session, record.id)
session.commit()
drop_guard(session, "uq_capability_deployment_one_production_stable")
deployment(session, record.id)
session.commit()
report = check_invariants(session)
assert report.ok is False
payload = summarise(report)
assert payload["violated"] >= 1
keys = {item["key"] for item in payload["violations"]}
assert "single_production_stable" in keys
assert all(item["examples"] for item in payload["violations"])
def test_unused_fixtures_are_referenced() -> None:
"""Keep the imported fixtures honest; unused imports would drift into false coverage."""
assert ArtifactSet is not None
assert UpstreamSnapshot is not None
assert RuntimeProfile is not None