audit-core/tests/test_cadence_registration.py

"""Guarded registration never changes custody or unrelated registry state."""
import copy
import hashlib
import json
from pathlib import Path

import pytest

from audit_core.cadence_registration import prepare, strict_json


def fixture():
    frame = {"frame_seconds": 3600}
    entries = [{"source_id": n, "event_class": n, "form": "heartbeat-or-reconciliation",
                "heartbeat": {"event_class": "audit-core.heartbeat", "interval_seconds": 3600,
                              "assertion": "nothing-to-report", "missing": "finding",
                              "extensions": {"approval-engine": {"asserted_class": n,
                                  "requested_observer_max_gap_seconds": 7200,
                                  "effective_deadline_seconds": 10800}}}}
               for n in ("issuance", "use", "supersession", "revocation")]
    entries.append({"source_id": "heartbeat", "event_class": "heartbeat",
                    "form": "heartbeat-or-reconciliation", "reconciliation": {
                        "compare_local": "local", "compare_observed": "receiver", "divergence": "finding"}})
    doc = {"schema_version": "0.1", "declaration_id": "test", "source": "approval-engine",
           "sources": entries, "extensions": {"netkingdom": {"coordination_frame": frame}}}
    raw = json.dumps(doc).encode()
    proposal = {"sender": "approval-engine", "source": "approval-engine", "tenant": "tenant:platform",
                "may_read": False, "preserve_existing_credentials_and_other_registry_fields": True,
                "coordination_frame": frame, "emission_cadence_sha256": hashlib.sha256(raw).hexdigest(),
                "heartbeat_classes": dict.fromkeys(("issuance", "use", "supersession", "revocation"),7200),
                "receiver_grace_factor": 1.5, "effective_deadline_seconds": 10800}
    rows = [{"name": "sibling", "sources": ["sibling"], "tokens": ["sibling-fixture"]},
            {"name": "approval-engine", "sources": ["approval-engine"], "tenants": ["tenant:platform"],
             "tokens": ["current-fixture", "overlap-fixture"], "may_read": False,
             "evidence_kind": "load-bearing", "secret_policy": "redact", "owner_property": {"keep": True}}]
    return {"senders.json": json.dumps(rows), "unrelated": "preserve"}, raw, proposal


def test_preserves_every_non_metadata_field_and_inputs_and_is_idempotent():
    body, raw, proposal = fixture()
    original = copy.deepcopy(body)
    updated, receipt = prepare(body, raw, proposal)
    before, after = json.loads(body["senders.json"]), json.loads(updated["senders.json"])
    assert body == original and updated["unrelated"] == "preserve" and before[0] == after[0]
    assert {k:v for k,v in after[1].items() if k not in ("heartbeat_classes", "emission_cadence")} == before[1]
    assert "fixture" not in json.dumps(receipt)
    assert prepare(updated, raw, proposal)[0] == updated
    assert prepare(updated, raw, proposal)[1]["changed"] is False


@pytest.mark.parametrize("mutation", ["digest", "read", "tenant", "class", "credentials", "duplicate", "absent"])
def test_fail_closed(mutation):
    body, raw, proposal = fixture()
    rows = json.loads(body["senders.json"])
    if mutation == "digest": proposal["emission_cadence_sha256"] = "bad"
    if mutation == "read": rows[1]["may_read"] = True
    if mutation == "tenant": rows[1]["tenants"] = ["*"]
    if mutation == "class": rows[1]["heartbeat_classes"] = {"must-not-drop": 3600}
    if mutation == "credentials": rows[1]["tokens"] = []
    if mutation == "duplicate": rows.append(copy.deepcopy(rows[1]))
    if mutation == "absent": rows.pop()
    body["senders.json"] = json.dumps(rows)
    with pytest.raises(ValueError): prepare(body, raw, proposal)


def test_duplicate_json_refused():
    with pytest.raises(ValueError): strict_json('{"name":1,"name":2}')


def test_real_owner_proposal_when_checkout_available():
    root = Path(__file__).resolve().parents[2]
    proposal = root / "net-kingdom/generated/coordination-frame/approval-engine-receiver.json"
    declaration = root / "approval-engine/cadence.yaml"
    if not proposal.exists() or not declaration.exists(): pytest.skip("owner checkouts unavailable")
    body, _, _ = fixture()
    _, receipt = prepare(body, declaration.read_bytes(), strict_json(proposal.read_bytes()))
    assert receipt["effective_deadline_seconds"] == 10800


def test_qonto_proposal_retains_mixed_classification():
    from audit_core.cadence_registration import qonto_metadata
    body, raw, proposal = fixture()
    doc = strict_json(raw)
    doc["source"] = "qonto-assistant"
    doc["stream_id"] = "qonto-assistant.audit"
    kinds = {"audit.allow": "attributive", "audit.deny": "load-bearing", "audit.heartbeat": "load-bearing"}
    doc["sources"] = [{"source_id": name, "event_class": name, "form": "heartbeat-or-reconciliation",
                       "extensions": {"netkingdom": {"evidence_class": kind}},
                       "heartbeat": {"event_class": "audit.heartbeat", "interval": "PT3600S",
                                     "assertion": "nothing-to-report", "missing": "finding",
                                     "extensions": {"qonto-assistant": {"assertion_scope": "no-deny-transitions-in-window"}}}}
                      for name, kind in kinds.items()]
    entry = qonto_metadata(doc, "a" * 64)
    assert "tokens" not in entry and entry["may_read"] is False
    assert entry["heartbeat_classes"] == {"audit.deny": 7200}
    assert entry["emission_cadence"] == doc
    doc["sources"][0]["extensions"]["netkingdom"]["evidence_class"] = "load-bearing"
    with pytest.raises(ValueError, match="classification"): qonto_metadata(doc, "a" * 64)