tegwick
31e9963933
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a06e89-93a2-7aa2-82b3-ce5ccd2682e6
83 lines
3 KiB
Python
83 lines
3 KiB
Python
from copy import deepcopy
|
|
from dataclasses import asdict
|
|
|
|
import pytest
|
|
|
|
from kings_guard.adapters.secrets_engine import snapshot_from_engine
|
|
from kings_guard.contracts import StreamCompleteness
|
|
|
|
NOW = "2026-09-05T12:00:00Z"
|
|
PAYLOAD = {
|
|
"surface": "secret-use-evidence", "as_of": NOW, "completeness_claimed": False,
|
|
"cadence": {"form": "heartbeat", "interval": "1d"},
|
|
"lanes": [{"as_of": NOW, "catalog_id": "lane-1", "stage": "test", "kind": "kv"}],
|
|
}
|
|
BINDINGS = {"lane-1": ("tenant-1", "subject-1")}
|
|
|
|
|
|
def admit(payload, now=NOW, bindings=BINDINGS):
|
|
return snapshot_from_engine(payload, lane_bindings=bindings, now=now)
|
|
|
|
|
|
def test_catalog_only_snapshot_does_not_invent_evidence_or_completeness():
|
|
result = admit(PAYLOAD)
|
|
assert result.lanes[0].ready is None
|
|
assert result.lanes[0].revocation_succeeded is None
|
|
assert result.lanes[0].evidence_kind is None
|
|
assert result.completeness is StreamCompleteness.UNKNOWN
|
|
assert result.heartbeat_interval.total_seconds() == 86400
|
|
|
|
|
|
def test_evidence_preserves_false_and_drops_authorization_and_sensitive_fields():
|
|
payload = deepcopy(PAYLOAD)
|
|
payload["lanes"][0].update(ready=False, revocation_succeeded=False,
|
|
decision_id="dec-1", session_handle="session-1",
|
|
secret="never-retained", lifecycle_operation="revoke")
|
|
result = admit(payload)
|
|
assert result.lanes[0].ready is False
|
|
assert result.lanes[0].revocation_succeeded is False
|
|
assert result.lanes[0].lifecycle_operation == "revoke"
|
|
fields = asdict(result.lanes[0])
|
|
assert {"decision_id", "session_handle", "secret"}.isdisjoint(fields)
|
|
assert result.completeness is StreamCompleteness.UNKNOWN
|
|
|
|
|
|
@pytest.mark.parametrize(("now", "findings"), [
|
|
("2026-09-06T12:00:00Z", ()),
|
|
("2026-09-06T12:00:01Z", ("snapshot:stale",)),
|
|
("2026-09-05T11:59:59Z", ("snapshot:future_timestamp",)),
|
|
])
|
|
def test_freshness_does_not_upgrade_completeness(now, findings):
|
|
result = admit(PAYLOAD, now)
|
|
assert result.findings == findings
|
|
assert result.completeness is StreamCompleteness.UNKNOWN
|
|
|
|
|
|
def test_unbound_or_duplicate_lanes_cannot_cross_scope():
|
|
with pytest.raises(ValueError, match="unbound"):
|
|
admit(PAYLOAD, bindings={})
|
|
payload = deepcopy(PAYLOAD)
|
|
payload["lanes"].append(payload["lanes"][0])
|
|
with pytest.raises(ValueError, match="duplicate"):
|
|
admit(payload)
|
|
|
|
|
|
@pytest.mark.parametrize("change", [
|
|
{"completeness_claimed": True}, {"completeness_claimed": None},
|
|
{"cadence": {"form": "heartbeat", "interval": "0s"}},
|
|
{"surface": "authorization"}, {"lanes": None},
|
|
])
|
|
def test_invalid_envelope_is_rejected(change):
|
|
with pytest.raises(ValueError):
|
|
admit({**PAYLOAD, **change})
|
|
|
|
|
|
@pytest.mark.parametrize("change", [
|
|
{"ready": "false"}, {"revocation_attempted": 0},
|
|
{"as_of": "2026-09-04T12:00:00Z"}, {"evidence_kind": "guessed"},
|
|
])
|
|
def test_invalid_lane_is_rejected(change):
|
|
payload = deepcopy(PAYLOAD)
|
|
payload["lanes"][0].update(change)
|
|
with pytest.raises(ValueError):
|
|
admit(payload)
|