"""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