761 lines
24 KiB
Python
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
|