"""AUDIT-WP-0009 T04/T06/T07 — heartbeats, reconciliation, findings.""" import io import json from datetime import datetime, timedelta, timezone import pytest from audit_core.ingestion import IngestionApplication from audit_core.senders import SenderIdentity, SenderRegistry from audit_core.sqlite_backend import SQLiteAuditBackend from audit_core.stream_findings import ( HEARTBEAT_ACTION, MISSING_HEARTBEAT, NEVER_HEARTBEAT, evaluate, ) def invoke(app, path, *, token="approval", method="GET", payload=None, key=None): raw = json.dumps(payload).encode() if payload is not None else b"" environ = { "PATH_INFO": path.split("?")[0], "QUERY_STRING": path.split("?")[1] if "?" in path else "", "REQUEST_METHOD": method, "CONTENT_LENGTH": str(len(raw)), "wsgi.input": io.BytesIO(raw), "HTTP_AUTHORIZATION": f"Bearer {token}", } if key: environ["HTTP_IDEMPOTENCY_KEY"] = key result = {} out = b"".join(app(environ, lambda status, headers: result.update(status=status))) return result["status"], (json.loads(out) if out else {}) SENDER = SenderIdentity( name="approval-engine", tokens=("approval",), sources=frozenset({"approval-engine"}), tenants=frozenset({"tenant:platform"}), may_write=True, may_read=False, evidence_kind="load-bearing", heartbeat_classes=(("approval.revocation", 86400),), ) OPERATOR = SenderIdentity( name="operator", tokens=("operator",), sources=frozenset({"approval-engine"}), tenants=frozenset({"*"}), may_write=False, may_read=True, ) @pytest.fixture def app(tmp_path): backend = SQLiteAuditBackend(str(tmp_path / "s.db")) return IngestionApplication(backend, SenderRegistry([SENDER, OPERATOR])) def emit(app, event_id, action, *, data=None, tenant="tenant:platform"): payload = { "id": event_id, "type": action, "source": "approval-engine", "subject": "approval:1", "tenant": tenant, "correlation_id": "c-1", "occurred_at": "2026-09-10T00:00:00+00:00", "data": data or {"k": "v"}, } return invoke(app, "/v1/events", method="POST", payload=payload, key=event_id) # --- T04: heartbeats ------------------------------------------------------- def test_a_heartbeat_is_an_ordinary_event_and_is_chained(app): """No special table. A heartbeat outside the chain could be back-dated.""" status, _ = emit( app, "hb-1", HEARTBEAT_ACTION, data={"class": "approval.revocation", "assertion": "nothing-to-report"}, ) assert status.startswith("202") status, body = invoke(app, "/v1/integrity", token="operator") assert status.startswith("200") assert body["intact"] is True assert body["events"] == 1 def test_a_declared_class_with_no_heartbeat_ever_is_its_own_finding(): """The case a naive 'compare against last seen' silently drops.""" findings = evaluate([SENDER], {}) assert [f.kind for f in findings] == [NEVER_HEARTBEAT] assert findings[0].event_class == "approval.revocation" assert findings[0].source == "approval-engine" def test_a_fresh_heartbeat_produces_no_finding(): now = datetime(2026, 9, 10, 12, tzinfo=timezone.utc) last = (now - timedelta(hours=2)).isoformat() assert evaluate([SENDER], {("approval-engine", "approval.revocation"): last}, now=now) == [] def test_an_overdue_heartbeat_is_a_finding_with_its_lateness(): now = datetime(2026, 9, 10, 12, tzinfo=timezone.utc) # 86400s interval, 1.5 grace -> overdue past 36h. last = (now - timedelta(hours=40)).isoformat() findings = evaluate([SENDER], {("approval-engine", "approval.revocation"): last}, now=now) assert [f.kind for f in findings] == [MISSING_HEARTBEAT] assert findings[0].overdue_seconds == int(timedelta(hours=4).total_seconds()) def test_grace_widens_the_window_but_never_removes_the_finding(): now = datetime(2026, 9, 10, 12, tzinfo=timezone.utc) last = (now - timedelta(days=30)).isoformat() findings = evaluate( [SENDER], {("approval-engine", "approval.revocation"): last}, now=now, grace_factor=100.0, ) # 86400 * 100 is ~100 days, so this one is inside the window... assert findings == [] # ...and no grace factor makes a year-old heartbeat acceptable. older = (now - timedelta(days=400)).isoformat() assert evaluate( [SENDER], {("approval-engine", "approval.revocation"): older}, now=now, grace_factor=100.0, ) def test_a_wildcard_source_is_not_held_to_a_heartbeat(): """No determinate set of streams to expect one from. Refused, not guessed.""" wild = SenderIdentity( name="w", tokens=("t",), sources=frozenset({"*"}), heartbeat_classes=(("anything", 60),), ) assert evaluate([wild], {}) == [] def test_every_finding_says_absence_is_not_proof_of_suppression(): """The surface invites over-reading; the bound travels on each finding.""" finding = evaluate([SENDER], {})[0].as_dict() assert "not proof of suppression" in finding["means"].lower() # --- T07: the findings surface --------------------------------------------- def test_stream_findings_needs_read_and_full_tenant_scope(app): assert invoke(app, "/v1/stream-findings")[0].startswith("403") status, body = invoke(app, "/v1/stream-findings", token="operator") assert status.startswith("200") assert body["stream_findings"][0]["kind"] == NEVER_HEARTBEAT def test_a_delivered_heartbeat_clears_the_finding(app): emit( app, "hb-2", HEARTBEAT_ACTION, data={"class": "approval.revocation", "assertion": "nothing-to-report"}, ) _, body = invoke(app, "/v1/stream-findings", token="operator") # Stored today against a 24h interval: no longer a finding. assert body["stream_findings"] == [] # --- T06: reconciliation --------------------------------------------------- def test_a_writer_may_count_its_own_events_without_may_read(app): """The §9.6 obligation would be undischargeable otherwise.""" assert SENDER.may_read is False emit(app, "e-1", "approval.issuance") emit(app, "e-2", "approval.issuance") emit(app, "e-3", "approval.revocation") status, body = invoke( app, "/v1/reconciliation?source=approval-engine&tenant=tenant:platform" "&since=2026-09-01T00:00:00%2B00:00&until=2027-01-01T00:00:00%2B00:00", ) assert status.startswith("200") assert body["counts"] == [ {"class": "approval.issuance", "count": 2}, {"class": "approval.revocation", "count": 1}, ] def test_reconciliation_returns_counts_and_never_payloads(app): emit(app, "e-9", "approval.issuance", data={"secret_shaped": "x", "binding": "b"}) _, body = invoke( app, "/v1/reconciliation?source=approval-engine&tenant=tenant:platform" "&since=2026-09-01T00:00:00%2B00:00&until=2027-01-01T00:00:00%2B00:00", ) assert set(body) == {"source", "tenant", "since", "until", "counts", "means"} assert "binding" not in json.dumps(body) def test_a_source_may_not_count_another_source(app): status, body = invoke( app, "/v1/reconciliation?source=user-engine&tenant=tenant:platform" "&since=2026-09-01T00:00:00%2B00:00&until=2027-01-01T00:00:00%2B00:00", ) # 403, not an empty count: a zero would read as "we hold none of yours", # which is a false answer to a question about completeness. assert status.startswith("403") assert body["error"] == "source_not_allowed" def test_reconciliation_requires_an_explicit_bounded_window(app): status, body = invoke(app, "/v1/reconciliation?source=approval-engine&tenant=tenant:platform") assert status.startswith("400") assert body["error"] == "since_and_until_required" def test_a_scoped_credential_must_name_its_tenant(app): status, body = invoke( app, "/v1/reconciliation?source=approval-engine" "&since=2026-09-01T00:00:00%2B00:00&until=2027-01-01T00:00:00%2B00:00", ) assert status.startswith("400") assert body["error"] == "tenant_required" def test_a_scoped_credential_may_not_count_another_tenant(app): status, body = invoke( app, "/v1/reconciliation?source=approval-engine&tenant=tenant:coulomb" "&since=2026-09-01T00:00:00%2B00:00&until=2027-01-01T00:00:00%2B00:00", ) assert status.startswith("403") assert body["error"] == "tenant_not_allowed" def test_the_count_response_states_what_it_does_not_prove(app): _, body = invoke( app, "/v1/reconciliation?source=approval-engine&tenant=tenant:platform" "&since=2026-09-01T00:00:00%2B00:00&until=2027-01-01T00:00:00%2B00:00", ) means = body["means"].lower() assert "proves neither completeness" in means # --- the overlay asymmetry ------------------------------------------------- def test_the_overlay_may_shorten_a_heartbeat_interval_but_not_lengthen_it(tmp_path): scope = tmp_path / "scope.json" base = [{ "name": "approval-engine", "tokens": ["t"], "sources": ["approval-engine"], "heartbeat_classes": {"approval.revocation": 3600}, }] scope.write_text(json.dumps( [{"name": "approval-engine", "heartbeat_classes": {"approval.revocation": 600}}] )) registry = SenderRegistry.from_env({ "AUDIT_CORE_SENDERS": json.dumps(base), "AUDIT_CORE_SENDERS_SCOPE_PATH": str(scope), }) assert dict(registry.identities[0].heartbeat_classes)["approval.revocation"] == 600 scope.write_text(json.dumps( [{"name": "approval-engine", "heartbeat_classes": {"approval.revocation": 99999}}] )) with pytest.raises(ValueError, match="lengthen"): SenderRegistry.from_env({ "AUDIT_CORE_SENDERS": json.dumps(base), "AUDIT_CORE_SENDERS_SCOPE_PATH": str(scope), }) scope.write_text(json.dumps([{"name": "approval-engine", "heartbeat_classes": {}}])) with pytest.raises(ValueError, match="remove"): SenderRegistry.from_env({ "AUDIT_CORE_SENDERS": json.dumps(base), "AUDIT_CORE_SENDERS_SCOPE_PATH": str(scope), })