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
80 lines
2.6 KiB
Python
80 lines
2.6 KiB
Python
from datetime import datetime, timezone
|
|
|
|
from approval_engine.errors import Conflict, DuplicateApprover
|
|
from approval_engine.store import Engine
|
|
from tests.conftest import approve, binding, validity
|
|
|
|
|
|
def test_requested_until_threshold(engine):
|
|
obj = engine.create(binding(), validity(), required_count=2)
|
|
assert obj.status == "requested"
|
|
obj = engine.add_entry(obj.id, "user:alice")
|
|
assert obj.status == "requested"
|
|
claim = engine.claim(obj.id)
|
|
assert claim["valid_now"] is False
|
|
assert claim["reason_code"] == "requested"
|
|
|
|
|
|
def test_threshold_approves_and_valid_now(engine):
|
|
obj = approve(engine, required=2)
|
|
assert obj.status == "approved"
|
|
assert len(obj.entries) == 2
|
|
claim = engine.claim(obj.id)
|
|
assert claim["state"] == "valid"
|
|
assert claim["valid_now"] is True
|
|
|
|
|
|
def test_duplicate_approver_fails_closed(engine):
|
|
obj = engine.create(binding(), validity(), required_count=2)
|
|
engine.add_entry(obj.id, "user:alice")
|
|
try:
|
|
engine.add_entry(obj.id, "user:alice")
|
|
raise AssertionError("duplicate must fail")
|
|
except DuplicateApprover:
|
|
pass
|
|
obj = engine.get(obj.id)
|
|
assert len(obj.entries) == 1
|
|
assert obj.status == "requested"
|
|
|
|
|
|
def test_revoke_without_holder_and_next_claim(engine):
|
|
obj = approve(engine)
|
|
engine.revoke(obj.id)
|
|
claim = engine.claim(obj.id)
|
|
assert claim["valid_now"] is False
|
|
assert claim["state"] == "revoked"
|
|
assert claim["reason_code"] == "revoked"
|
|
|
|
|
|
def test_expiry_on_observation():
|
|
jumping = {"t": datetime(2026, 8, 29, 12, 0, tzinfo=timezone.utc)}
|
|
eng = Engine(":memory:", clock=lambda: jumping["t"])
|
|
obj = eng.create(binding(), validity(), required_count=1)
|
|
obj = eng.add_entry(obj.id, "user:alice")
|
|
assert eng.claim(obj.id)["valid_now"] is True
|
|
jumping["t"] = datetime(2026, 8, 29, 16, 0, tzinfo=timezone.utc)
|
|
claim = eng.claim(obj.id)
|
|
assert claim["state"] == "expired"
|
|
assert claim["valid_now"] is False
|
|
eng.close()
|
|
|
|
|
|
def test_not_yet_valid():
|
|
future = datetime(2026, 8, 29, 10, 0, tzinfo=timezone.utc)
|
|
eng = Engine(":memory:", clock=lambda: future)
|
|
obj = eng.create(binding(), validity(), required_count=1)
|
|
obj = eng.add_entry(obj.id, "user:alice")
|
|
claim = eng.claim(obj.id)
|
|
assert claim["valid_now"] is False
|
|
assert claim["reason_code"] == "not_yet_valid"
|
|
eng.close()
|
|
|
|
|
|
def test_cannot_revoke_twice(engine):
|
|
obj = approve(engine)
|
|
engine.revoke(obj.id)
|
|
try:
|
|
engine.revoke(obj.id)
|
|
raise AssertionError("second revoke must conflict")
|
|
except Conflict:
|
|
pass
|