kings-guard/tests/test_acceptance.py

"""All inputs here are test fixtures, never live operational receipts."""

from copy import deepcopy

import pytest

from kings_guard.acceptance import assess_capture


def capture():
    instance = "test-instance"
    events = []
    for seq, cls, stamp, reason, count in (
        (1, "audit.heartbeat", "2026-10-15T10:00:00Z", "startup", 0),
        (2, "audit.deny", "2026-10-15T10:01:00Z", None, 1),
        (3, "audit.heartbeat", "2026-10-15T11:00:00Z", "periodic", 1),
    ):
        event = dict(
            stream_id="qonto-assistant.audit",
            stream_instance_id=instance,
            stream_sequence=seq,
            event_class=cls,
            timestamp=stamp,
        )
        if reason:
            event.update(
                reason=reason,
                assertion="nothing-to-report" if not count else "transitions-reported",
                source_transition_counts={"audit.allow": 0, "audit.deny": count},
            )
        events.append(event)
    snapshot = dict(
        stream_id="qonto-assistant.audit",
        stream_instance_id=instance,
        last_stream_sequence=3,
        snapshot_at="2026-10-15T11:00:02Z",
        source_transition_counts={"audit.allow": 0, "audit.deny": 1},
    )
    receiver = dict(
        source="qonto-assistant",
        tenant="test-tenant",
        stream_instance_id=instance,
        since="2026-10-15T09:59:00Z",
        until="2026-10-15T11:00:03Z",
        counts=[{"class": "audit.deny", "count": 1}, {"class": "audit-core.heartbeat", "count": 2}],
    )
    receipts = [
        dict(
            event_id=f"test-{e['stream_sequence']}",
            status_code=202,
            accepted_at=e["timestamp"],
            stream_instance_id=instance,
            stream_sequence=e["stream_sequence"],
            type="audit-core.heartbeat" if e["reason"] else e["event_class"],
            source="qonto-assistant",
            tenant="test-tenant",
            covered_event_classes=["audit.deny"] if e["reason"] else [],
        )
        for e in [dict(e, reason=e.get("reason")) for e in events]
    ]
    return events, snapshot, receiver, receipts


def assess(parts):
    return assess_capture(
        *parts,
        source="qonto-assistant",
        tenant="test-tenant",
        frame_seconds=3600,
        now="2026-10-15T11:00:04Z",
    )


def test_complete_fixture_capture_passes_only_its_bounded_evidence():
    result = assess(capture())
    assert result["result"] == "pass"
    assert result["effective_deadline_seconds"] == 10800
    assert "supplied deployed capture only" in result["scope"]


@pytest.mark.parametrize(
    "kind",
    [
        "gap",
        "instance",
        "divergence",
        "missing_receipt",
        "wrong_class",
        "duplicate_200",
        "lag",
        "short_periodic",
        "source_binding",
        "tenant_binding",
    ],
)
def test_refuses_broken_correlation(kind):
    events, snapshot, receiver, receipts = deepcopy(capture())
    if kind == "gap":
        events[1]["stream_sequence"] = 4
    elif kind == "instance":
        receiver["stream_instance_id"] = "other"
    elif kind == "divergence":
        receiver["counts"][0]["count"] = 2
    elif kind == "missing_receipt":
        receipts.pop()
    elif kind == "wrong_class":
        receipts[0]["covered_event_classes"] = ["audit.allow"]
    elif kind == "duplicate_200":
        receipts[0]["status_code"] = 200
    elif kind == "lag":
        receipts[0]["accepted_at"] = "2026-10-15T11:00:01Z"
    elif kind == "short_periodic":
        events[2]["timestamp"] = "2026-10-15T10:30:00Z"
    elif kind == "source_binding":
        receipts[0]["source"] = "other"
    elif kind == "tenant_binding":
        receipts[0]["tenant"] = "other"
    assert assess((events, snapshot, receiver, receipts))["result"] == "fail"


def test_timezone_missing_is_invalid():
    parts = capture()
    parts[0][0]["timestamp"] = "2026-10-15T10:00:00"
    with pytest.raises(ValueError, match="timezone"):
        assess(parts)


def test_duplicate_receipt_sequence_is_invalid():
    parts = capture()
    parts[3].append(parts[3][0])
    with pytest.raises(ValueError, match="unique"):
        assess(parts)


def test_class_silence_does_not_accept_fresh_unrelated_class():
    from kings_guard.acceptance import assess_class_silence

    result = assess_class_silence(
        source="approval-engine",
        sender="approval-engine",
        registered_wait_seconds={"issuance": 7200, "revocation": 7200},
        last_seen={"issuance": "2026-10-15T11:00:00Z", "revocation": "2026-10-15T07:00:00Z"},
        findings=[],
        now="2026-10-15T11:00:01Z",
    )
    assert result["result"] == "fail"
    assert result["expected_missing_classes"] == ["revocation"]


def test_class_silence_matches_receiver_findings_per_class():
    from kings_guard.acceptance import assess_class_silence

    result = assess_class_silence(
        source="approval-engine",
        sender="approval-engine",
        registered_wait_seconds={"issuance": 7200, "revocation": 7200},
        last_seen={"issuance": "2026-10-15T11:00:00Z", "revocation": None},
        findings=[
            dict(
                source="approval-engine",
                sender="approval-engine",
                kind="no_heartbeat_since_registration",
                **{"class": "revocation"},
            )
        ],
        now="2026-10-15T11:00:01Z",
    )
    assert result["result"] == "pass"


def test_class_silence_refuses_absent_observation_instead_of_inventing_never_seen():
    from kings_guard.acceptance import assess_class_silence

    with pytest.raises(ValueError, match="last_seen"):
        assess_class_silence(
            source="a",
            sender="a",
            registered_wait_seconds={"deny": 7200},
            last_seen={},
            findings=[],
            now="2026-10-15T11:00:00Z",
        )