coulomb/railiance-platform
5
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Record native receiver custody and prepare actual Approval lifecycle readback
All checks were successful
CI Smoke / host-smoke (push) Successful in 0s
Details
CI Smoke / container-smoke (push) Successful in 1s
Details

Browse source 
Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a13eab-1e6a-7b33-9c37-372e90ada3a2
This commit is contained in:
codex 2026-10-15 21:40:48 +02:00
parent ce5344da69
commit 6e24b09eaa
8 changed files with 372 additions and 40 deletions
Download patch file Download diff file Expand all files Collapse all files

77
tests/test_cadence_observer_readback.py
View file

@ -156,5 +156,82 @@ class ReadbackBounds(unittest.TestCase):
)
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()