kings-guard/tests/test_cadence_declaration.py

from pathlib import Path

import pytest

from kings_guard.cadence import load_qonto_assistant_source_cadence

ROOT = Path(__file__).resolve().parents[1]
DRAFT = ROOT / "specs" / "EmissionCadenceDeclaration.md"
QONTO_SOURCE_DECLARATION = Path("/home/worsch/qonto-assistant/specs/audit-emission-cadence.yaml")


def test_emission_cadence_draft_exists_for_taxonomy_handover() -> None:
    text = DRAFT.read_text(encoding="utf-8")
    assert DRAFT.is_file()
    assert "owner: Taxonomy" in text
    assert "drafter: kings-guard" in text
    assert "qonto-assistant" in text
    assert "GH-WP-0002-T04" in text
    assert "approval-engine/cadence.yaml" in text
    assert "expected rate" in text.lower() or "expected-rate" in text
    assert "heartbeat" in text.lower()
    assert "reconciliation" in text.lower()
    assert "alongside the security genome" in text.lower() or "alongside the genome" in text.lower()
    assert "security_genome" in text


def test_qonto_source_declaration_is_consumed_separately_from_the_taxonomy_draft() -> None:
    cadence = load_qonto_assistant_source_cadence()

    assert cadence.status == "source-declared"
    assert cadence.owner == "qonto-assistant"
    assert cadence.reference_instance == "QONTO-WP-0005"
    assert cadence.forms() == {"heartbeat-or-reconciliation"}
    assert cadence.rates == ()
    assert cadence.heartbeats[0].interval.total_seconds() == 86400


@pytest.mark.skipif(
    not QONTO_SOURCE_DECLARATION.is_file(),
    reason="qonto-assistant source declaration checkout is unavailable",
)
def test_qonto_source_cadence_fixture_matches_the_source_owned_declaration() -> None:
    yaml = pytest.importorskip("yaml")
    document = yaml.safe_load(QONTO_SOURCE_DECLARATION.read_text(encoding="utf-8"))
    source = document.get("audit_emission_cadence", document)
    cadence = load_qonto_assistant_source_cadence()

    assert cadence.source_system == source["source"]
    if "sources" in source:
        deny = next(item for item in source["sources"] if item["event_class"] == "audit.deny")
        heartbeat = deny["heartbeat"]
        assert cadence.heartbeats[0].event_class == heartbeat["event_class"]
        from kings_guard.cadence import parse_interval

        interval = heartbeat.get("interval_seconds", heartbeat.get("interval"))
        assert cadence.heartbeats[0].interval == parse_interval(interval)
        assert deny["form"] in cadence.forms()
        assert deny["extensions"]["netkingdom"]["evidence_class"] == "load-bearing"
    else:
        # Historical owner envelope is checked only as provenance during migration.
        # The runtime reader accepts the canonical namespace exclusively.
        assert cadence.heartbeats[0].event_class == source["heartbeat"]["event_class"]
        assert (
            cadence.heartbeats[0].interval.total_seconds()
            == source["heartbeat"]["default_interval_seconds"]
        )
        assert source["event_classes"]["audit.deny"]["form"] in cadence.forms()
        assert source["event_classes"]["audit.allow"]["completeness_claimed"] is False
    assert cadence.rates == ()


@pytest.mark.parametrize(
    "filename",
    [
        "qonto_assistant_cadence.json",
        "qonto_assistant_source_cadence.json",
    ],
)
def test_cadence_fixtures_conform_to_owner_schema(filename):
    import json

    import jsonschema
    import yaml

    schema_path = Path(
        "/home/worsch/info-tech-canon/infospace/schemas/emission-cadence.schema.yaml"
    )
    if not schema_path.is_file():
        pytest.skip("canonical InfoTechCanon schema checkout is unavailable")
    schema = yaml.safe_load(schema_path.read_text())
    payload = json.loads((ROOT / "src/kings_guard/fixtures" / filename).read_text())
    jsonschema.Draft202012Validator(schema).validate(payload)


@pytest.mark.parametrize(("value", "seconds"), [("P1DT2H30M5S", 95405), ("P1D", 86400)])
def test_canonical_compound_durations_are_consumed(value, seconds):
    from kings_guard.cadence import parse_interval

    assert parse_interval(value).total_seconds() == seconds


def _canonical_cadence_document():
    import json

    return json.loads(
        (ROOT / "src/kings_guard/fixtures/qonto_assistant_source_cadence.json").read_text()
    )


def test_canonical_profile_namespace_preserves_source_classification():
    from kings_guard.cadence import emission_cadence_from_dict
    from kings_guard.contracts import EvidenceClass

    payload = _canonical_cadence_document()
    cadence = emission_cadence_from_dict(payload)
    assert cadence.heartbeats[0].evidence_class is EvidenceClass.LOAD_BEARING
    payload["sources"][0]["extensions"]["netkingdom"]["evidence_class"] = "attributive"
    assert (
        emission_cadence_from_dict(payload).heartbeats[0].evidence_class
        is EvidenceClass.ATTRIBUTIVE
    )


@pytest.mark.parametrize("retain_canonical", [False, True])
def test_legacy_profile_namespace_requires_explicit_owner_migration(retain_canonical):
    from kings_guard.cadence import emission_cadence_from_dict

    payload = _canonical_cadence_document()
    extensions = payload["sources"][0]["extensions"]
    extensions["net-kingdom"] = dict(extensions["netkingdom"])
    if not retain_canonical:
        del extensions["netkingdom"]
    with pytest.raises(ValueError, match="legacy extensions.net-kingdom must migrate"):
        emission_cadence_from_dict(payload)


@pytest.mark.parametrize("profile", [None, [], "load-bearing"])
def test_missing_or_malformed_profile_does_not_infer_classification(profile):
    from kings_guard.cadence import emission_cadence_from_dict

    payload = _canonical_cadence_document()
    payload["sources"][0]["extensions"]["netkingdom"] = profile
    with pytest.raises(ValueError, match="requires source-owned extensions.netkingdom"):
        emission_cadence_from_dict(payload)


@pytest.mark.parametrize("classification", [None, True, "unknown"])
def test_invalid_evidence_classification_is_refused(classification):
    from kings_guard.cadence import emission_cadence_from_dict

    payload = _canonical_cadence_document()
    payload["sources"][0]["extensions"]["netkingdom"] = {"evidence_class": classification}
    with pytest.raises(ValueError, match="evidence_class must be source-declared"):
        emission_cadence_from_dict(payload)


def test_malformed_entry_extensions_are_diagnosed():
    from kings_guard.cadence import emission_cadence_from_dict

    payload = _canonical_cadence_document()
    payload["sources"][0]["extensions"] = []
    with pytest.raises(ValueError, match="extensions must be an object"):
        emission_cadence_from_dict(payload)