coulomb/runbook-tutorials
5
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Engine v0.2: flow graph, datapoint states, gates, modes, overrides, deviations (RBT-WP-0007)
All checks were successful
CI Smoke / host-smoke (push) Successful in 1s
Details
CI Smoke / container-smoke (push) Successful in 4s
Details
Work Records / validate (push) Successful in 12s
Details

Browse source 
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>

Assistant: claude-code
Assistant-Model: sonnet
Assistant-Process: 295952@bnt-lap001
Assistant-Session: e93f64ad-516c-46eb-9666-aad8d300c477
This commit is contained in:
tegwick 2026-09-29 11:34:07 +02:00
parent 3fc279a6ed
commit 60d983769f
13 changed files with 1017 additions and 22 deletions
Download patch file Download diff file Expand all files Collapse all files

316
tests/test_run2.py Normal file
View file

@ -0,0 +1,316 @@
import copy
import subprocess
import uuid
from datetime import datetime, timedelta, timezone
from pathlib import Path
import pytest
import yaml
from rtut import pack as P, run as R1, run2 as R
ROOT = Path(__file__).resolve().parent.parent
SRC = yaml.safe_load((ROOT / "runbooks/showcase-lighthouse/pack.v0.2.yaml").read_text())
PHRASE = "I understand this cannot be undone"
def mkpack(tmp_path, mutate=None):
d = copy.deepcopy(SRC)
if mutate:
mutate(d)
f = tmp_path / "runbooks" / "lh" / "pack.yaml"
f.parent.mkdir(parents=True, exist_ok=True)
f.write_text(yaml.safe_dump(d, sort_keys=False))
pk = P.load(f)
assert pk.ok, pk.errors
return pk
_SCRATCH: list[str] = []
@pytest.fixture(autouse=True)
def _cleanup_scratch():
yield
import shutil
while _SCRATCH:
shutil.rmtree(_SCRATCH.pop(), ignore_errors=True)
def new(tmp_path, mutate=None, **kw):
pk = mkpack(tmp_path, mutate)
logdir = f"/tmp/rtut-{uuid.uuid4().hex[:10]}"
_SCRATCH.append(logdir)
vals = {"logdir": logdir, **kw.pop("values", {})}
return R.create(pk, "tester", vals, **kw)
def do_step(r, sid, alt=None, phrase=None, manual=True, run_cmd=True, evidence=None):
"""What an operator does: start, run the command, check, mark done."""
R.start(r, sid, alt=alt, phrase=phrase)
cmd = R.command_for(r, sid)
if cmd and run_cmd:
subprocess.run(cmd, shell=True, check=True, capture_output=True)
R.verify(r, sid, confirm=True, manual_ok=manual)
R.complete(r, sid, evidence)
def pass_gate(r, gid):
for i, ck in enumerate(R.flow(r).gates[gid]["checks"]):
R.gate_check(r, gid, i, confirm=True, manual_ok=True)
# ---- datapoints -----------------------------------------------------------------------------------
def test_create_sets_datapoint_states(tmp_path):
r = new(tmp_path)
dp = r.data["datapoints"]
assert dp["keeper_name"]["state"] == "set" and dp["keeper_name"]["value"] == "Ada"
assert dp["lamp_lit"]["state"] == "nil"
assert r.data["params"]["weather"] == "fog"
def test_invalid_datapoint_is_err_and_not_stored(tmp_path):
r = new(tmp_path)
assert R.set_datapoint(r, "weather", "hail") == "err"
assert "value" not in r.data["datapoints"]["weather"] and "weather" not in r.data["params"]
assert R.set_datapoint(r, "weather", "storm") == "set"
assert R.set_datapoint(r, "logdir", "/etc") == "err"
assert R.set_datapoint(r, "motto", 'x"; rm -rf /') == "err"
with pytest.raises(R.RunError):
R.create(mkpack(tmp_path), "t", {"weather": "hail"})
# ---- readiness, recommendation, parallelism, alternatives ---------------------------------------------
def test_recommended_next_is_first_ready_in_canonical_order_and_others_are_parallel(tmp_path):
r = new(tmp_path)
na = R.next_actions(r)
assert na["recommended"] == "read-the-sign"
assert na["ready"] == ["read-the-sign", "light-lamp", "read-the-sky"] # parallel, canonical order
blocked = {b["id"]: b["unmet"] for b in na["blocked"]}
assert "lamp-check" in blocked and "warn" in blocked and "close-log" in blocked
R.start(r, "light-lamp") # any ready item may be taken
assert R.next_actions(r)["recommended"] == "light-lamp" # a running item comes first
def test_alternatives_recommended_by_datapoint_and_explicit_choice(tmp_path):
for weather, alt in (("fog", "sound-horn"), ("clear", "wave-flags"), ("storm", "shelter")):
r = new(tmp_path, values={"weather": weather})
assert R.recommended_alt(r, "warn") == alt
r = new(tmp_path)
for s in ("read-the-sign", "light-lamp", "read-the-sky"):
do_step(r, s)
pass_gate(r, "lamp-check")
with pytest.raises(R.RunError, match="choose one of"):
R.start(r, "warn", alt="dance")
R.start(r, "warn", alt="wave-flags") # operator may override the recommendation
assert R.command_for(r, "warn").startswith("echo") and "signals with flags" in R.command_for(r, "warn")
assert r.data["steps"]["warn"]["alternative"] == "wave-flags"
def test_datapoint_ok_via_confirmed_by_and_gate_flow(tmp_path):
r = new(tmp_path)
assert R.unmet(r, "lamp-check") # stage steps not done yet
do_step(r, "light-lamp")
dp = r.data["datapoints"]["lamp_lit"]
assert dp["state"] == "ok" and dp["how"] # produced, then confirmed by the step's own check
for s in ("read-the-sign", "read-the-sky"):
do_step(r, s)
assert R.unmet(r, "lamp-check") == []
R.gate_check(r, "lamp-check", 0, manual_ok=True)
assert r.data["gates"]["lamp-check"]["state"] == "open" # not all checks evaluated yet
R.gate_check(r, "lamp-check", 1) # datapoints_ok is automatic
assert r.data["gates"]["lamp-check"]["state"] == "passed"
assert "warn" in R.next_actions(r)["ready"]
def test_failed_step_marks_produced_datapoint_err_and_retry_clears_it(tmp_path):
r = new(tmp_path)
R.start(r, "light-lamp"); R.fail(r, "light-lamp", "no oil")
assert r.data["datapoints"]["lamp_lit"]["state"] == "err"
R.retry(r, "light-lamp")
assert r.data["datapoints"]["lamp_lit"]["state"] == "nil"
# ---- overrides, waivers, deviations ------------------------------------------------------------------
def test_override_needs_reason_and_acknowledged_risk_and_is_recorded(tmp_path):
r = new(tmp_path)
with pytest.raises(R.RunError, match="Risk:"):
R.start(r, "warn", alt="sound-horn") # gate not passed: names unmet + risk
with pytest.raises(R.RunError, match="acknowledge"):
R.start(r, "warn", alt="sound-horn", override_reason="need to see the horn text")
R.start(r, "warn", alt="sound-horn", override_reason="need to see the horn text", acknowledge_risk=True)
dev = r.data["deviations"][0]
assert dev["kind"] == "override" and dev["item"] == "warn" and dev["mode"] == "manual"
assert any("lamp-check" in u for u in dev["unmet"]) and dev["risk_acknowledged"] and dev["reason"]
assert "not complete" in dev["unmet"][0]
def test_waiver_only_where_declared_with_reason_and_recorded(tmp_path):
r = new(tmp_path)
for s in ("read-the-sign", "light-lamp", "read-the-sky"):
do_step(r, s)
R.gate_check(r, "lamp-check", 0, manual_ok=False)
R.gate_check(r, "lamp-check", 1)
assert r.data["gates"]["lamp-check"]["state"] == "failed"
with pytest.raises(R.RunError, match="reason"):
R.waive(r, "lamp-check", " ")
R.waive(r, "lamp-check", "harbour master accepts the risk")
assert r.data["gates"]["lamp-check"]["state"] == "waived"
assert r.data["deviations"][-1]["kind"] == "waiver" and "check 0 not passed" in r.data["deviations"][-1]["unmet"]
assert "warn" in R.next_actions(r)["ready"] # a waived gate unblocks dependents
with pytest.raises(R.RunError, match="not waivable"):
R.waive(r, "night-on-record", "try")
def test_deviations_block_exercise_but_not_finalize(tmp_path):
r = new(tmp_path)
R.start(r, "warn", alt="sound-horn", override_reason="probe", acknowledge_risk=True)
r.data["steps"]["warn"]["state"] = "failed"
assert r.data["deviations"]
with pytest.raises(R1.RunError, match="deviations"):
for k in r.data["steps"]:
r.data["steps"][k]["state"] = "done"
R1.exercise_update(r)
# ---- confirmation styles ------------------------------------------------------------------------------
def test_irreversible_step_needs_the_typed_phrase(tmp_path):
r = new(tmp_path)
assert R.confirmation_style(r, "read-the-sign") == "confirm" and R.confirmation_style(r, "close-log") == "phrase"
for s in ("read-the-sign", "light-lamp", "read-the-sky"):
do_step(r, s)
pass_gate(r, "lamp-check")
do_step(r, "warn")
do_step(r, "write-logbook")
with pytest.raises(R.RunError, match="confirmation phrase"):
R.start(r, "close-log")
with pytest.raises(R.RunError, match="confirmation phrase"):
R.start(r, "close-log", phrase="yes")
R.start(r, "close-log", phrase=PHRASE)
assert r.data["steps"]["close-log"]["authorized_by"] == "tester"
def test_preauthorization_only_for_irreversible_steps(tmp_path):
with pytest.raises(R.RunError, match="only irreversible"):
new(tmp_path, mode="autopilot", preauthorize=["light-lamp"])
# ---- autopilot ------------------------------------------------------------------------------------------------
def drive(r, ticks=60, now=None, operator=None):
log = []
for _ in range(ticks):
out = R.autopilot_tick(r, now)
log.append(out)
if out["action"] in ("finished", "stopped", "awaiting-operator", "waiting"):
break
if out["action"] == "started" and operator:
operator(out["item"]) # the operator runs the step's command while autopilot confirms
return log
def test_autopilot_confirms_when_preconditions_are_met_and_waits_for_the_operator_on_manual_only(tmp_path):
r = new(tmp_path, mode="autopilot")
def operator(sid):
cmd = R.command_for(r, sid)
if cmd:
subprocess.run(cmd, shell=True, check=True, capture_output=True)
log = drive(r, operator=operator)
assert log[-1] == {"action": "awaiting-operator", "item": "close-log", "detail": "step is manual-only"}
done = [x["item"] for x in log if x["action"] == "completed"]
assert done[:3] == ["read-the-sign", "light-lamp", "read-the-sky"] and "lamp-check" in done and "write-logbook" in done
assert r.data["steps"]["warn"]["alternative"] == "sound-horn" # recommended by the fog datapoint
assert r.data["gates"]["lamp-check"]["state"] == "passed"
assert r.data["steps"]["read-the-sign"]["verification"].get("auto_confirmed")
assert r.data["deviations"] == []
R.start(r, "close-log", phrase=PHRASE) # the operator takes over the last step
subprocess.run(R.command_for(r, "close-log"), shell=True, check=True, capture_output=True)
R.verify(r, "close-log", confirm=True); R.complete(r, "close-log")
assert R.autopilot_tick(r)["action"] == "completed" # gate night-on-record
assert R.autopilot_tick(r)["action"] == "finished"
R1.finalize(r, "succeeded")
assert r.data["outcome"] == "succeeded" and not r.data["deviations"]
def test_autopilot_timeout_stop(tmp_path):
def m(d):
d["timeout"] = {"seconds": 30, "on_timeout": "stop"}
r = new(tmp_path, m, mode="autopilot")
log = drive(r) # write-logbook's file check cannot pass: nobody ran its command
assert log[-1]["action"] == "waiting" and log[-1]["item"] == "write-logbook"
late = datetime.now(timezone.utc) + timedelta(seconds=31)
out = R.autopilot_tick(r, late)
assert out["action"] == "stopped" and "timed out after 30s" in out["detail"]
assert R.autopilot_tick(r)["action"] == "stopped" # stays stopped
assert r.data["deviations"] == [] # stopping is not a deviation
def test_autopilot_push_through_records_deviation_and_moves_on(tmp_path):
def m(d):
d["timeout"] = {"seconds": 30, "on_timeout": "push-through"}
r = new(tmp_path, m, mode="autopilot")
drive(r)
late = datetime.now(timezone.utc) + timedelta(seconds=31)
out = R.autopilot_tick(r, late)
assert out["action"] == "push-through" and out["item"] == "write-logbook"
assert r.data["steps"]["write-logbook"]["state"] == "done" and r.data["steps"]["write-logbook"]["pushed_through"]
dev = r.data["deviations"][0]
assert dev["kind"] == "push-through" and dev["mode"] == "autopilot" and dev["risk_acknowledged"] is False
nxt = R.autopilot_tick(r, late) # continues to the manual-only step
assert nxt["item"] == "close-log" and nxt["action"] == "awaiting-operator"
with pytest.raises(R1.RunError, match="deviations"):
for k in r.data["steps"]:
r.data["steps"][k]["state"] = "done"
R1.exercise_update(r)
def test_autopilot_gate_push_through_and_stop(tmp_path):
def m(d):
d["stages"][0]["gates"][0]["checks"].append({"done_when": "always fails", "command": "false", "expect": "exit-0"})
d["stages"][0]["gates"][0]["timeout"] = {"seconds": 10, "on_timeout": "push-through"}
r = new(tmp_path, m, mode="autopilot")
drive(r)
assert r.data["gates"]["lamp-check"]["state"] == "failed" # all checks evaluated, one failing
out = R.autopilot_tick(r, datetime.now(timezone.utc) + timedelta(seconds=11))
assert out["action"] == "push-through" and r.data["gates"]["lamp-check"]["state"] == "waived"
assert r.data["deviations"][0]["item"] == "lamp-check" and r.data["deviations"][0]["kind"] == "push-through"
def test_autopilot_never_confirms_irreversible_steps_unless_preauthorized(tmp_path):
def m(d):
for st in d["stages"]:
for s in st.get("steps", []):
s.pop("autopilot", None) # allow automation of close-log
r = new(tmp_path, m, mode="autopilot")
drive(r, operator=lambda sid: (subprocess.run(R.command_for(r, sid), shell=True, check=True, capture_output=True) if R.command_for(r, sid) else None))
last = R.autopilot_tick(r)
assert last["action"] == "awaiting-operator" and "not pre-authorized" in last["detail"]
r2 = new(tmp_path, m, mode="autopilot", preauthorize=["close-log"])
drive(r2, operator=lambda sid: (subprocess.run(R.command_for(r2, sid), shell=True, check=True, capture_output=True) if R.command_for(r2, sid) else None))
assert r2.data["steps"]["close-log"]["state"] in ("running", "done")
assert r2.data["steps"]["close-log"]["authorized_by"] == "tester"
def test_manual_mode_rejects_autopilot_and_autopilot_rejects_overrides(tmp_path):
with pytest.raises(R.RunError, match="autopilot run"):
R.autopilot_tick(new(tmp_path))
r = new(tmp_path, mode="autopilot")
with pytest.raises(R.RunError, match="manual mode"):
R.start(r, "warn", alt="sound-horn", override_reason="x", acknowledge_risk=True)
def test_finished_v02_run_is_read_only_and_scores_count_it(tmp_path):
r = new(tmp_path)
R1.finalize(r, "abandoned")
with pytest.raises(R.RunError, match="finished"):
R.start(r, "read-the-sign")
with pytest.raises(R.RunError, match="finished"):
R.set_datapoint(r, "weather", "clear")
assert R1.scorecard(r.pack)["abandoned"] == 1
assert R1.load(r.pack, r.data["run_id"]).data["spec"] == "runbook-run/v0.2" # reloads through the shared loader