tegwick
a786b061da
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a13eab-1e6a-7b33-9c37-372e90ada3a2
46 lines
2.2 KiB
Python
46 lines
2.2 KiB
Python
import json
|
|
from pathlib import Path
|
|
|
|
from approval_engine.audit import audit_envelope
|
|
from approval_engine.coordination import FRAME_SECONDS, RECEIVER_WAIT_SECONDS
|
|
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"] == FRAME_SECONDS
|
|
assert beat["extensions"]["approval-engine"]["requested_observer_max_gap_seconds"] == RECEIVER_WAIT_SECONDS
|
|
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
|
|
|
|
|
|
def test_default_runtime_heartbeat_matches_declared_coordination_frame():
|
|
from approval_engine.cli import _parser
|
|
|
|
args = _parser().parse_args(["serve"])
|
|
assert args.heartbeat_seconds == FRAME_SECONDS
|
|
assert _parser().parse_args(["serve", "--heartbeat-seconds", "600"]).heartbeat_seconds == 600
|