tegwick
931e0b3249
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a1029d-2c93-7500-a0e0-e3e089668b29
37 lines
1.8 KiB
Python
37 lines
1.8 KiB
Python
import json
|
|
from pathlib import Path
|
|
|
|
from approval_engine.audit import audit_envelope
|
|
from approval_engine.store import EVENT_CLASSES, HEARTBEAT_CLASSES
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
def test_layer_yaml_cadence_declared():
|
|
text = (ROOT / "layer.yaml").read_text()
|
|
assert "cadence_status: declared" in text
|
|
assert "cadence_form: heartbeat-or-reconciliation" in text
|
|
assert "cadence: cadence.yaml" in text
|
|
|
|
|
|
def test_cadence_inventory_matches_durable_emitter(engine):
|
|
declaration = json.loads((ROOT / "cadence.yaml").read_text())
|
|
inventory = declaration["extensions"]["approval-engine"]["inventory"]
|
|
assert set(inventory["load_bearing"]) == set(EVENT_CLASSES)
|
|
assert set(inventory["rare_load_bearing"]) == set(HEARTBEAT_CLASSES)
|
|
entries = {entry["event_class"]: entry for entry in declaration["sources"]}
|
|
assert set(entries) == set(EVENT_CLASSES)
|
|
engine.emit_heartbeat()
|
|
observed = {audit_envelope(row["payload"])["data"]["class"]:
|
|
audit_envelope(row["payload"]) for row in engine.undrained()}
|
|
assert set(observed) == set(HEARTBEAT_CLASSES)
|
|
for cls in HEARTBEAT_CLASSES:
|
|
beat = entries[cls]["heartbeat"]
|
|
assert beat["event_class"] == observed[cls]["type"] == "audit-core.heartbeat"
|
|
assert beat["assertion"] == observed[cls]["data"]["assertion"]
|
|
assert beat["extensions"]["approval-engine"]["asserted_class"] == cls
|
|
assert beat["interval_seconds"] == 86400
|
|
assert beat["extensions"]["approval-engine"]["requested_observer_max_gap_seconds"] == 172800
|
|
assert entries[cls]["extensions"]["netkingdom"]["rate_monitoring"] == "forbidden"
|
|
assert entries[cls]["reconciliation"]["undrained_local"] == "lag-not-divergence"
|
|
assert "heartbeat" not in entries["heartbeat"] # transport control has no recursive heartbeat
|