coulomb/runbook-tutorials
5
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
runbook-tutorials/tests/test_run2.py
tegwick 6d3414d8b7 Retire exercise_status: maturity is the only notion; add CLI tests (RBT-WP-0009-T01) 
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
2026-09-29 21:28:56 +02:00

464 lines
23 KiB
Python
Raw Permalink Blame History

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-v2/pack.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)
def counts_towards_maturity(r) -> bool:
"""Would this run be clean maturity evidence? (succeeded, no deviation, with an environment)"""
from rtut import maturity as M
r.data["outcome"] = "succeeded"
r.data["context"] = {"project": "p", "environment": "rehearsal"}
return M._evidence(r.pack, r.data.get("pack_digest"), [r])["clean_runs"] == 1
# ---- 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_keep_a_run_out_of_maturity(tmp_path):
r = new(tmp_path)
R.start(r, "warn", alt="sound-horn", override_reason="probe", acknowledge_risk=True)
assert r.data["deviations"] and not counts_towards_maturity(r)
assert counts_towards_maturity(new(tmp_path)) # the same without a deviation would count
# ---- 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="auto-confirm", preauthorize=["light-lamp"])
# ---- auto-confirm ------------------------------------------------------------------------------------------------
def drive(r, ticks=60, now=None, operator=None):
log = []
for _ in range(ticks):
out = R.auto_confirm_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 auto-confirm confirms
return log
def test_auto_confirm_confirms_when_preconditions_are_met_and_waits_for_the_operator_on_manual_only(tmp_path):
r = new(tmp_path, mode="auto-confirm")
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.auto_confirm_tick(r)["action"] == "completed" # gate night-on-record
assert R.auto_confirm_tick(r)["action"] == "finished"
R1.finalize(r, "succeeded")
assert r.data["outcome"] == "succeeded" and not r.data["deviations"]
def test_auto_confirm_timeout_stop(tmp_path):
def m(d):
d["timeout"] = {"seconds": 30, "on_timeout": "stop"}
r = new(tmp_path, m, mode="auto-confirm")
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.auto_confirm_tick(r, late)
assert out["action"] == "stopped" and "timed out after 30s" in out["detail"]
assert R.auto_confirm_tick(r)["action"] == "stopped" # stays stopped
assert r.data["deviations"] == [] # stopping is not a deviation
def test_auto_confirm_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="auto-confirm")
drive(r)
late = datetime.now(timezone.utc) + timedelta(seconds=31)
out = R.auto_confirm_tick(r, late)
assert out["action"] == "push-through" and out["item"] == "write-logbook"
assert r.data["steps"]["write-logbook"]["state"] == "pushed" # its own state: not done, not failed
dev = r.data["deviations"][0]
assert dev["kind"] == "push-through" and dev["mode"] == "auto-confirm" and dev["risk_acknowledged"] is False
nxt = R.auto_confirm_tick(r, late) # continues to the manual-only step
assert nxt["item"] == "close-log" and nxt["action"] == "awaiting-operator"
assert not counts_towards_maturity(r) # a pushed-through run is not evidence
def test_auto_confirm_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="auto-confirm")
drive(r)
assert r.data["gates"]["lamp-check"]["state"] == "failed" # all checks evaluated, one failing
out = R.auto_confirm_tick(r, datetime.now(timezone.utc) + timedelta(seconds=11))
assert out["action"] == "push-through" and r.data["gates"]["lamp-check"]["state"] == "pushed" # not "waived": nobody approved it
assert r.data["deviations"][0]["item"] == "lamp-check" and r.data["deviations"][0]["kind"] == "push-through"
def test_auto_confirm_never_confirms_irreversible_steps_unless_preauthorized(tmp_path):
def m(d):
for st in d["stages"]:
for s in st.get("steps", []):
s.pop("auto_confirm", None) # allow automation of close-log
r = new(tmp_path, m, mode="auto-confirm")
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.auto_confirm_tick(r)
assert last["action"] == "awaiting-operator" and "not pre-authorized" in last["detail"]
r2 = new(tmp_path, m, mode="auto-confirm", 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_auto_confirm_and_auto_confirm_rejects_overrides(tmp_path):
with pytest.raises(R.RunError, match="auto-confirm run"):
R.auto_confirm_tick(new(tmp_path))
r = new(tmp_path, mode="auto-confirm")
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
def test_pushed_items_are_a_distinct_state_that_lets_the_run_continue_but_never_succeed(tmp_path):
def m(d):
d["timeout"] = {"seconds": 5, "on_timeout": "push-through"}
light = next(s for st in d["stages"] for s in st.get("steps", []) if s["id"] == "light-lamp")
light["verify"] = {"done_when": "never true here", "command": "false", "expect": "exit-0"}
d["stages"][0]["gates"][0]["timeout"] = {"seconds": 5, "on_timeout": "push-through"} # the gate's own timeout wins
r = new(tmp_path, m, mode="auto-confirm")
late = datetime.now(timezone.utc) + timedelta(seconds=6)
seen = []
for _ in range(40):
out = R.auto_confirm_tick(r, late)
seen.append(out["action"])
if out["action"] in ("stopped", "awaiting-operator", "finished"):
break
late = late + timedelta(seconds=6)
assert r.data["steps"]["light-lamp"]["state"] == "pushed"
dp = r.data["datapoints"]["lamp_lit"]
assert dp["state"] == "pushed" and "value" not in dp and "error" not in dp # not set, not err
assert "lamp_lit" not in r.data["params"] # never usable as a value
assert R.is_complete(r, "light-lamp") # dependents may proceed
assert r.data["gates"]["lamp-check"]["state"] == "pushed" # datapoints_ok needs 'ok', so the gate was pushed too
kinds = [d["kind"] for d in r.data["deviations"]]
assert kinds.count("push-through") >= 2 and "override" not in kinds
with pytest.raises(R1.RunError, match="succeeds only when every step is done"):
R1.finalize(r, "succeeded")
R1.finalize(r, "failed", "pushed through to collect problems") # a clean-completion failure, finishable as failed
assert r.data["outcome"] == "failed"
# ---- retry policy, mode switch, views ------------------------------------------------------------------------------
def test_retry_of_a_non_idempotent_step_needs_the_partial_application_confirmation(tmp_path):
r = new(tmp_path)
for s in ("read-the-sign", "light-lamp", "read-the-sky"):
do_step(r, s)
pass_gate(r, "lamp-check")
do_step(r, "warn")
R.start(r, "write-logbook"); R.fail(r, "write-logbook", "disk full halfway") # non-idempotent: appends
with pytest.raises(R.RunError, match="did not partly apply"):
R.retry(r, "write-logbook")
R.retry(r, "write-logbook", acknowledge_partial=True)
assert r.data["steps"]["write-logbook"]["retries"][0]["acknowledged_partial"] is True
R.start(r, "read-the-sign") if False else None
r2 = new(tmp_path)
R.start(r2, "read-the-sign"); R.fail(r2, "read-the-sign", "x") # idempotent: free retry
R.retry(r2, "read-the-sign")
assert "retries" not in r2.data["steps"]["read-the-sign"]
def test_undeclared_idempotency_also_needs_the_confirmation(tmp_path):
def m(d):
step(d, "read-the-sign").pop("idempotent")
r = new(tmp_path, m)
R.start(r, "read-the-sign"); R.fail(r, "read-the-sign", "x")
with pytest.raises(R.RunError, match="not declared idempotent"):
R.retry(r, "read-the-sign")
def step(d, sid):
return next(s for st in d["stages"] for s in st.get("steps", []) if s["id"] == sid)
def test_switch_mode_records_history_and_clears_a_stop(tmp_path):
r = new(tmp_path, mode="auto-confirm")
R._stop(r, "x", "why")
assert r.data["auto_confirm"]["stopped"]
R.switch_mode(r, "manual")
assert r.data["mode"] == "manual" and r.data["mode_history"][0] == {**r.data["mode_history"][0], "from": "auto-confirm", "to": "manual"}
with pytest.raises(R.RunError, match="already in manual"):
R.switch_mode(r, "manual")
R.switch_mode(r, "auto-confirm")
assert "stopped" not in r.data["auto_confirm"] and len(r.data["mode_history"]) == 2
def test_overview_counts_progress_pushed_and_datapoint_states(tmp_path):
r = new(tmp_path)
ov = R.overview(r)
assert ov["progress"] == {"done": 0, "pushed": 0, "total": 8} and [s["id"] for s in ov["stages"]] == ["dusk", "night", "dawn"]
assert ov["datapoints"]["set"] == 5 and ov["datapoints"]["nil"] == 1
do_step(r, "light-lamp")
ov = R.overview(r)
assert ov["progress"]["done"] == 1 and ov["stages"][0]["steps_done"] == 1 and ov["datapoints"]["ok"] == 1
assert ov["stages"][1]["unmet"] and not ov["stages"][0]["complete"]
def test_datapoint_rows_group_and_filter_by_dimension(tmp_path):
r = new(tmp_path)
assert "actor" in R.dimension_keys(r) and "kind" in R.dimension_keys(r)
rows = R.datapoint_rows(r)
assert list(rows) == [""] and len(rows[""]) == 6
by_actor = R.datapoint_rows(r, group="actor")
assert set(by_actor) == {"keeper", "(none)"} and {x["id"] for x in by_actor["keeper"]} >= {"keeper_name", "lamp_fuel", "motto", "lamp_lit"}
only = R.datapoint_rows(r, flt="kind:identity")
assert {x["id"] for x in only[""]} == {"keeper_name", "motto"}
lamp = next(x for x in rows[""] if x["id"] == "lamp_lit")
assert lamp["editable"] is False and lamp["produced_by"] == "light-lamp"
def test_artefact_rows_resolve_location_from_a_datapoint(tmp_path):
r = new(tmp_path, values={"logdir": "/tmp/rtut-art-one"})
a = R.artefact_rows(r)[0]
assert a["id"] == "logbook-file" and a["location"] == "/tmp/rtut-art-one/logbook.txt" and a["produced_state"] == "todo"
def test_captures_and_pasted_verification_work_for_v02_runs(tmp_path):
r = new(tmp_path)
R.start(r, "read-the-sky")
R1.add_capture(r, "read-the-sky", "Weather tonight - fog.", "echo ...")
res = R1.verify_pasted(r, "read-the-sky", "Weather tonight - fog.")
assert res["passed"] and r.data["steps"]["read-the-sky"]["captures"][0]["output"].startswith("Weather")
# ---- changing datapoints after steps used them ------------------------------------------------------------------------
def test_changing_an_unused_datapoint_is_free_and_a_used_one_is_a_recorded_deviation(tmp_path):
r = new(tmp_path, values={"weather": "fog"})
assert R.set_datapoint(r, "weather", "clear") == "set" and r.data["deviations"] == [] # nobody used it yet
do_step(r, "read-the-sky") # uses weather
assert R.datapoint_users(r, "weather") == ["read-the-sky"]
with pytest.raises(R.RunError, match="already used by read-the-sky"):
R.set_datapoint(r, "weather", "storm")
with pytest.raises(R.RunError, match="acknowledge"):
R.set_datapoint(r, "weather", "storm", override_reason="the forecast changed")
assert r.data["params"]["weather"] == "clear" # refused: nothing changed
R.set_datapoint(r, "weather", "storm", override_reason="the forecast changed", acknowledge_risk=True)
assert r.data["params"]["weather"] == "storm"
dev = r.data["deviations"][0]
assert dev["kind"] == "changed-after-use" and dev["item"] == "weather" and "read-the-sky" in dev["unmet"][0] and "'clear'" in dev["unmet"][0]
assert dev["reason"] == "the forecast changed" and dev["risk_acknowledged"] is True
assert R.set_datapoint(r, "weather", "storm") == "set" and len(r.data["deviations"]) == 1 # the same value is not a change
def test_used_datapoints_cannot_be_changed_in_auto_confirm_mode(tmp_path):
r = new(tmp_path, mode="auto-confirm")
R.auto_confirm_tick(r); R.auto_confirm_tick(r)
r.data["steps"]["read-the-sky"]["state"] = "done"
with pytest.raises(R.RunError, match="take over"):
R.set_datapoint(r, "weather", "clear", override_reason="x", acknowledge_risk=True)
def test_a_run_that_changed_a_used_value_never_counts_towards_maturity(tmp_path):
r = new(tmp_path)
do_step(r, "read-the-sky")
R.set_datapoint(r, "weather", "clear", override_reason="changed my mind", acknowledge_risk=True)
assert r.data["deviations"] and not counts_towards_maturity(r)
Reference in a new issue View git blame Copy permalink