"""Model-only independent reader bounds and no-payload projection tests.""" import copy import json import sys import unittest from pathlib import Path from unittest.mock import patch sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts")) import cadence_observer_readback as lane # noqa: E402 SINCE = "2026-10-15T10:00:00Z" UNTIL = "2026-10-15T11:00:00Z" class ReadbackBounds(unittest.TestCase): def record(self): return dict( event_id="test-1", source="qonto-assistant", tenant="tenant:friendly:binky", action="audit-core.heartbeat", accepted_at="2026-10-15T10:30:00Z", observed_at="2026-10-15T10:29:59Z", details={ "data": { "stream_instance_id": "instance", "stream_sequence": 1, "class": "audit.deny", "sensitive-request": "must-never-leave", } }, ) def sanitize(self, record): return lane.sanitize_record( record, "qonto-assistant", "tenant:friendly:binky", "instance", "test-1", SINCE, UNTIL, ) def test_payload_is_never_returned(self): result = self.sanitize(self.record()) self.assertNotIn("must-never-leave", json.dumps(result)) self.assertEqual(result["covered_event_classes"], ["audit.deny"]) self.assertTrue(result["independent_custody_readback"]) def test_wrong_source_tenant_instance_and_id_are_refused(self): for key in ("source", "tenant", "event_id", "stream_instance_id"): record = self.record() if key == "stream_instance_id": record["details"]["data"][key] = "other" else: record[key] = "other" with self.subTest(key=key), self.assertRaises(lane.LaneError): self.sanitize(record) def test_outside_acceptance_window_is_refused(self): record = self.record() record["accepted_at"] = "2026-10-15T12:00:00Z" with self.assertRaisesRegex(lane.LaneError, "acceptance_window"): self.sanitize(record) def test_exact_until_is_excluded_like_receiver_sql(self): record = self.record() record["accepted_at"] = UNTIL with self.assertRaisesRegex(lane.LaneError, "acceptance_window"): self.sanitize(record) def test_paths_are_bounded_to_exact_approved_coordinates(self): result = lane.paths( "qonto-assistant", "tenant:friendly:binky", "instance", SINCE, UNTIL, ["test-1"], ) self.assertEqual(result[-1], "/v1/events/test-1") with self.assertRaisesRegex(lane.LaneError, "exact_source_tenant"): lane.paths( "qonto-assistant", "tenant:other", "instance", SINCE, UNTIL, ["test-1"] ) with self.assertRaisesRegex(lane.LaneError, "invalid_event_id"): lane.paths( "qonto-assistant", "tenant:friendly:binky", "instance", SINCE, UNTIL, ["../secret"], ) def test_reader_cannot_open_arbitrary_paths(self): with self.assertRaisesRegex(lane.LaneError, "exact_packet"): lane.read_http( "http://127.0.0.1:1", "fixture-only", "/v1/events/other", ["/v1/events/test-1"], ) def test_readback_discards_unrelated_findings_and_checks_exact_reconciliation(self): query = dict( source="qonto-assistant", tenant="tenant:friendly:binky", stream_instance_id="instance", since=SINCE, until=UNTIL, counts=[], ) findings = { "stream_findings": [ dict(source="unrelated", **{"class": "sensitive"}), dict( source="qonto-assistant", kind="missing_heartbeat", **{"class": "audit.deny"}, private="no-output", ), ] } with patch.object( lane, "read_http", side_effect=[query, findings, self.record()] ): result = lane.readback( "http://127.0.0.1:1", "fixture", "qonto-assistant", "tenant:friendly:binky", "instance", SINCE, UNTIL, ["test-1"], ) self.assertEqual(len(result["stream_findings"]), 1) self.assertNotIn("no-output", json.dumps(result)) wrong = copy.deepcopy(query) wrong["tenant"] = "other" with patch.object(lane, "read_http", return_value=wrong): with self.assertRaisesRegex(lane.LaneError, "scope_window_mismatch"): lane.readback( "http://127.0.0.1:1", "fixture", "qonto-assistant", "tenant:friendly:binky", "instance", SINCE, UNTIL, ["test-1"], ) class ApprovalReadback(unittest.TestCase): def test_approval_omits_instance_filter_without_inventing_one(self): paths = lane.paths( "approval-engine", "tenant:platform", None, SINCE, UNTIL, ["approval-1"] ) self.assertNotIn("stream_instance_id", paths[0]) with self.assertRaisesRegex(lane.LaneError, "no_declared_instance"): lane.paths( "approval-engine", "tenant:platform", "invented", SINCE, UNTIL, ["approval-1"], ) with self.assertRaisesRegex(lane.LaneError, "exact_instance_required"): lane.paths( "qonto-assistant", "tenant:friendly:binky", None, SINCE, UNTIL, ["q-1"] ) def test_actual_approval_record_needs_no_instance_sequence(self): record = dict( event_id="approval-1", source="approval-engine", tenant="tenant:platform", action="approval.issuance", accepted_at="2026-10-15T10:30:00Z", observed_at="2026-10-15T10:29:59Z", details={"data": {"transition": "issuance"}}, ) result = lane.sanitize_record( record, "approval-engine", "tenant:platform", None, "approval-1", SINCE, UNTIL, ) self.assertIsNone(result["stream_instance_id"]) self.assertIsNone(result["stream_sequence"]) self.assertEqual(result["type"], "approval.issuance") def test_reconciliation_compares_semantic_window_times(self): response = dict( source="approval-engine", tenant="tenant:platform", since="2026-10-15T10:00:00+00:00", until="2026-10-15T11:00:00+00:00", counts=[], ) record = dict( event_id="approval-1", source="approval-engine", tenant="tenant:platform", action="audit-core.heartbeat", accepted_at="2026-10-15T10:30:00Z", observed_at="2026-10-15T10:29:59Z", details={"data": {"class": "revocation"}}, ) with patch.object( lane, "read_http", side_effect=[response, {"stream_findings": []}, record] ): result = lane.readback( "http://127.0.0.1:1", "fixture-only", "approval-engine", "tenant:platform", None, SINCE, UNTIL, ["approval-1"], ) self.assertEqual(result["events"][0]["covered_event_classes"], ["revocation"]) self.assertIn("no process-instance completeness claim", result["scope"]) if __name__ == "__main__": unittest.main()