tegwick
9c9528f5b2
Contracts first (T02–T04): approval claim schema with issuer, freshness,
and binding digest; local transactional outbox wire; load-bearing cadence
as heartbeat or reconciliation (layer.yaml declared).
Then the object (T06–T08): SQLite closed state machine, CAS supersession,
distinct-approver fail-closed, revocation without holder cooperation,
outbox insert in the same transaction. Tests fail the mutation when
emission fails, and revoke while the drain sink is down.
Introspection GET /v1/approvals/{id}/claim is a PIP fact, not a decision.
No public consume (T05 waits on GH-WP-0002-T06). Canon T-06 coverage for
wrong binding, expiry, revoke, and supersede.
FLEX-WP-0017 T03 is unblocked on this object; T05 remains blocked only on
consumption ordering.
Assistant: grok
Assistant-Session: 01a04ceb-2057-7e20-b0f9-c282964d5dd9
67 lines
2.2 KiB
Python
67 lines
2.2 KiB
Python
"""Canon T-06 — Approval Replay.
|
|
|
|
Reuse a previously valid approval artifact for a different target, parameter
|
|
set, or later time. Pass: parameter binding, expiry, or replay protection
|
|
rejects the request.
|
|
|
|
Consume-side replay (use twice) waits on GH-WP-0002-T06. This suite covers
|
|
the object and claim side: wrong binding, expiry, revocation, supersession.
|
|
"""
|
|
|
|
from datetime import datetime, timezone
|
|
|
|
from approval_engine.binding import binding_digest
|
|
from approval_engine.store import Engine
|
|
from tests.conftest import approve, binding, validity
|
|
|
|
|
|
def _consumer_accepts(claim, proposed_binding) -> bool:
|
|
if claim["issuer"] != "approval-engine":
|
|
return False
|
|
if not claim["valid_now"]:
|
|
return False
|
|
if claim["consumed"]:
|
|
return False
|
|
if claim["reason_code"] != "ok":
|
|
return False
|
|
return claim["binding"]["digest"] == binding_digest(proposed_binding)
|
|
|
|
|
|
def test_t06_wrong_target_rejected(engine):
|
|
obj = approve(engine)
|
|
claim = engine.claim(obj.id)
|
|
assert _consumer_accepts(claim, binding()) is True
|
|
other = binding(target={"id": "some-other-lane", "stage": "prod"})
|
|
assert _consumer_accepts(claim, other) is False
|
|
|
|
|
|
def test_t06_wrong_action_rejected(engine):
|
|
obj = approve(engine)
|
|
claim = engine.claim(obj.id)
|
|
other = binding(action="secrets.kv.read")
|
|
assert _consumer_accepts(claim, other) is False
|
|
|
|
|
|
def test_t06_later_time_expired():
|
|
jumping = {"t": datetime(2026, 8, 29, 12, 0, tzinfo=timezone.utc)}
|
|
eng = Engine(":memory:", clock=lambda: jumping["t"])
|
|
obj = approve(eng)
|
|
claim = eng.claim(obj.id)
|
|
assert _consumer_accepts(claim, binding()) is True
|
|
jumping["t"] = datetime(2026, 8, 29, 16, 0, tzinfo=timezone.utc)
|
|
claim = eng.claim(obj.id)
|
|
assert claim["state"] == "expired"
|
|
assert _consumer_accepts(claim, binding()) is False
|
|
eng.close()
|
|
|
|
|
|
def test_t06_revoked_rejected(engine):
|
|
obj = approve(engine)
|
|
engine.revoke(obj.id)
|
|
assert _consumer_accepts(engine.claim(obj.id), binding()) is False
|
|
|
|
|
|
def test_t06_superseded_rejected(engine):
|
|
obj = approve(engine)
|
|
engine.supersede(obj.id)
|
|
assert _consumer_accepts(engine.claim(obj.id), binding()) is False
|