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
58 lines
2 KiB
Python
58 lines
2 KiB
Python
import json
|
|
from pathlib import Path
|
|
|
|
from approval_engine.binding import binding_digest
|
|
from tests.conftest import approve, binding
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
def test_schema_forbids_decision_keys():
|
|
schema = json.loads((ROOT / "schemas/approval_claim.schema.json").read_text())
|
|
assert "not" in schema
|
|
kinds = [item["required"][0] for item in schema["not"]["anyOf"]]
|
|
assert set(kinds) >= {"effect", "decision", "allow", "deny"}
|
|
|
|
|
|
def test_claim_has_issuer_digest_freshness(engine):
|
|
obj = approve(engine)
|
|
claim = engine.claim(obj.id)
|
|
assert claim["kind"] == "approval-claim"
|
|
assert claim["issuer"] == "approval-engine"
|
|
assert claim["approval_id"] == obj.id
|
|
assert claim["valid_now"] is True
|
|
assert claim["reason_code"] == "ok"
|
|
assert claim["binding"]["digest"] == binding_digest(binding())
|
|
assert claim["freshness"]["ttl_seconds"] == 30
|
|
assert claim["freshness"]["not_after"] == "2026-08-29T12:00:30+00:00"
|
|
assert "effect" not in claim
|
|
assert "decision" not in claim
|
|
assert "yields_to" in claim
|
|
|
|
|
|
def test_wrong_target_changes_digest():
|
|
a = binding()
|
|
b = binding(target={"id": "other-lane", "stage": "prod"})
|
|
assert binding_digest(a) != binding_digest(b)
|
|
|
|
|
|
def test_wrong_action_changes_digest():
|
|
assert binding_digest(binding()) != binding_digest(binding(action="secrets.kv.read"))
|
|
|
|
|
|
def test_pdp_digest_is_recorded_not_recomputed(engine):
|
|
pdp = "sha256:" + "ab" * 32
|
|
obj = approve(engine, pdp_digest=pdp)
|
|
claim = engine.claim(obj.id)
|
|
assert claim["binding"]["pdp_digest"] == pdp
|
|
assert claim["binding"]["digest"] == binding_digest(binding())
|
|
assert claim["binding"]["digest"] != pdp
|
|
|
|
|
|
def test_examples_are_claim_shaped():
|
|
for name in ("claim.valid.json", "claim.revoked.json"):
|
|
data = json.loads((ROOT / "examples" / name).read_text())
|
|
assert data["kind"] == "approval-claim"
|
|
assert data["issuer"] == "approval-engine"
|
|
for forbidden in ("effect", "decision", "allow", "deny"):
|
|
assert forbidden not in data
|