py/tests/test_saga.py
"""The saga producer: wire fidelity, capture semantics, and a live round-trip.
Three layers, cheapest first:
1. **Wire drift** — `facett_demo/proto/saga.proto` is a hand-kept SUBSET of
`edda/crates/nornir-wire/proto/nornir.proto`. A subset that drifts is worse than
no subset: it fails at runtime on a field number nobody re-checked. This test
re-reads the source-of-truth proto and compares field-for-field.
2. **Capture** — seq monotonicity, `state_json` passthrough vs encode, build stamp,
kind validation. No server needed.
3. **Round-trip** — record → `RecordEvents` → `ReadSession` against a live
nornir-server. SKIPPED (loudly) when no server answers; never silently green.
"""
from __future__ import annotations
import json
import re
import uuid
import os
from pathlib import Path
import pytest
from facett_demo.saga import KINDS, SagaRecorder, SagaUnavailable
import facett_demo
HERE = Path(__file__).resolve().parent
SUBSET = Path(facett_demo.__file__).resolve().parent / "proto" / "saga.proto"
def _locate_nornir_proto() -> Path:
"""Where `nornir.proto` is, in a checkout OR in the container.
The line this replaces was `HERE.parents[2] / "edda" / ...`, evaluated at
IMPORT time. That is correct on the host and raises `IndexError: 2` inside the
container, where the package is mounted at `/app` and there is no third parent
— so this module could not even be COLLECTED there, and `cargo xtask test-demo`
quietly ran one test file instead of the suite. A guard that cannot be imported
on the road that matters is not a guard.
`NORNIR_PROTO` is the explicit door (the container passes it, pointing at the
read-only mount); the relative candidates stay for a normal checkout. A missing
proto still FAILS rather than skips — that is the whole point of the file.
"""
env = os.environ.get("NORNIR_PROTO")
if env:
return Path(env)
rel = Path("edda") / "crates" / "nornir-wire" / "proto" / "nornir.proto"
for parent in HERE.parents:
cand = parent / rel
if cand.is_file():
return cand
# Nothing found: return the conventional location so the failure NAMES a path
# the reader can look at, rather than a None nobody can act on.
return HERE.parents[min(2, len(HERE.parents) - 1)] / rel
# The source of truth lives in the sibling `edda` checkout.
TRUTH = _locate_nornir_proto()
# What the subset claims to mirror.
MIRRORED_MESSAGES = [
"SagaEventPb",
"RecordSagaEventsRequest",
"SagaSessionQuery",
"SagaEventList",
"SubmitCountResponse",
"JsonResponse",
]
def _strip_comments(text: str) -> str:
return re.sub(r"//[^\n]*", "", text)
def _message_fields(proto: str, name: str) -> list[tuple[str, str, str]]:
"""[(type, field_name, number)] for `message <name>`, comments stripped."""
m = re.search(rf"\bmessage\s+{re.escape(name)}\s*\{{(.*?)\}}", _strip_comments(proto), re.S)
if m is None:
return []
body = m.group(1)
out = []
for fm in re.finditer(r"(?:(repeated)\s+)?(\w[\w.]*)\s+(\w+)\s*=\s*(\d+)\s*;", body):
rep, ftype, fname, num = fm.groups()
out.append((f"repeated {ftype}" if rep else ftype, fname, num))
return out
def _service_rpcs(proto: str, name: str) -> dict[str, tuple[str, str]]:
m = re.search(rf"\bservice\s+{re.escape(name)}\s*\{{(.*?)\}}", _strip_comments(proto), re.S)
if m is None:
return {}
return {
rpc: (req, resp)
for rpc, req, resp in re.findall(
r"rpc\s+(\w+)\s*\(\s*([\w.]+)\s*\)\s*returns\s*\(\s*([\w.]+)\s*\)", m.group(1)
)
}
@pytest.fixture(scope="module")
def protos() -> tuple[str, str]:
# A drift guard that SKIPS itself when the other side is missing is not a
# guard — it is a green tick. (That is exactly how the korp copy next door
# got three RPCs behind unnoticed.) Fail loudly, and say how to point it.
import os
env = os.environ.get("NORNIR_PROTO")
truth = Path(env) if env else TRUTH
if not truth.is_file():
pytest.fail(
f"nornir-wire's proto was not found at {truth} — this guard must not be "
"skipped. Check out the sibling `edda` repo, or point NORNIR_PROTO at "
"crates/nornir-wire/proto/nornir.proto."
)
return SUBSET.read_text(), truth.read_text()
def test_subset_declares_the_same_package(protos):
subset, truth = protos
assert re.search(r"^package\s+nornir\.v1;", subset, re.M), "subset must be in `nornir.v1`"
assert re.search(r"^package\s+nornir\.v1;", truth, re.M)
@pytest.mark.parametrize("msg", MIRRORED_MESSAGES)
def test_message_fields_match_nornir_wire(protos, msg):
subset, truth = protos
mine, theirs = _message_fields(subset, msg), _message_fields(truth, msg)
assert theirs, f"{msg} not found in the source-of-truth proto {TRUTH}"
assert mine == theirs, (
f"`{msg}` drifted from nornir-wire.\n subset: {mine}\n truth : {theirs}\n"
f"Re-copy it from {TRUTH} and regenerate the stubs."
)
def test_saga_service_rpcs_match_nornir_wire(protos):
subset, truth = protos
mine, theirs = _service_rpcs(subset, "Saga"), _service_rpcs(truth, "Saga")
assert theirs, "service Saga not found in nornir-wire"
assert mine == theirs, f"Saga RPCs drifted.\n subset: {mine}\n truth : {theirs}"
def test_generated_stubs_are_in_step_with_the_proto():
"""The checked-in gencode must expose every field the .proto declares."""
from facett_demo.proto import saga_pb2 as pb
declared = {f for _, f, _ in _message_fields(SUBSET.read_text(), "SagaEventPb")}
generated = set(pb.SagaEventPb.DESCRIPTOR.fields_by_name)
assert declared == generated, (
f"stubs are stale — regenerate: missing={declared - generated}, extra={generated - declared}"
)
# ── capture semantics (no server) ──────────────────────────────────────────────
def _offline() -> SagaRecorder:
"""A recorder pointed at a dead port: capture works, shipping does not."""
return SagaRecorder(app="test-capture", target="127.0.0.1:1", workspace="nordisk")
def test_seq_is_monotonic_from_one():
r = _offline()
for i in range(3):
r.record("log", detail=str(i))
assert [e["seq"] for e in r._buf] == [1, 2, 3]
assert r.pending == 3
def test_state_json_string_passes_through_but_objects_are_encoded():
r = _offline()
r.mounted("table", '{"rows":3}')
r.record("state", component="plot", state={"series": 1})
assert r._buf[0]["state_json"] == '{"rows":3}' # verbatim, not double-encoded
assert json.loads(r._buf[1]["state_json"]) == {"series": 1}
def test_every_event_carries_the_build_stamp():
r = _offline()
r.record("log", detail="x")
e = r._buf[0]
assert e["app"] == "test-capture" and e["app_version"] and e["git_sha"]
def test_unknown_kind_is_refused_at_capture():
r = _offline()
with pytest.raises(ValueError):
r.record("mount") # not one of nav|click|log|error|state
assert set(KINDS) == {"nav", "click", "log", "error", "state"}
def test_strict_recorder_raises_when_the_sink_is_dead():
r = SagaRecorder(app="test-strict", target="127.0.0.1:1", workspace="nordisk", strict=True)
r.record("log", detail="x")
with pytest.raises(SagaUnavailable):
r.flush()
def test_non_strict_recorder_keeps_the_buffer_when_shipping_fails():
r = _offline()
r.record("log", detail="x")
assert r.flush() == 0
assert r.pending == 1, "a failed flush must not lose the tape"
assert r.last_error, "a failed flush must be reported, not swallowed"
# ── live round-trip ────────────────────────────────────────────────────────────
@pytest.fixture(scope="module")
def live() -> SagaRecorder:
r = SagaRecorder(app="facett-pytest", session_id=f"facett-pytest-{uuid.uuid4().hex[:10]}")
r.record("log", action="probe", detail="reachability")
if r.flush() != 1:
pytest.skip(f"no saga sink at {r.target} [{r.workspace}]: {r.last_error}")
return r
def test_recorded_events_read_back_in_order(live):
live.mounted("table", '{"columns":["id","name"],"rows":2}', detail="assert=pass")
live.refused("console", "no data seam")
assert live.flush() == 2
tape = live.read_session()
assert len(tape) == 3, f"expected probe+state+error, got {[e['kind'] for e in tape]}"
assert [e["seq"] for e in tape] == [1, 2, 3]
state = next(e for e in tape if e["kind"] == "state")
assert state["component"] == "table"
assert json.loads(state["state_json"])["rows"] == 2, "state_json must survive the round-trip"
err = next(e for e in tape if e["kind"] == "error")
assert err["action"] == "refuse" and "no data seam" in err["detail"]
def test_session_appears_in_the_server_roll_up(live):
sessions = live.list_sessions()
assert sessions, "ListSessions returned nothing after a successful record"
assert any(str(s.get("session_id", "")) == live.session_id for s in sessions)