Add lighthouse showcase pack (v0.1 runnable, v0.2 draft); fix manual-check cycle and blocked-step feedback (RBT-WP-0004-T05)
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:
parent
f9fb9008bf
commit
b47c948a7e
8 changed files with 387 additions and 12 deletions
69
tests/test_showcase.py
Normal file
69
tests/test_showcase.py
Normal file
|
|
@ -0,0 +1,69 @@
|
|||
from pathlib import Path
|
||||
import shutil
|
||||
import subprocess
|
||||
|
||||
import pytest
|
||||
import yaml
|
||||
|
||||
from rtut import pack as P, run as R
|
||||
|
||||
ROOT = Path(__file__).resolve().parent.parent
|
||||
SRC = ROOT / "runbooks/showcase-lighthouse"
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def pk(tmp_path):
|
||||
d = tmp_path / "runbooks" / "showcase"; d.mkdir(parents=True)
|
||||
shutil.copy(SRC / "pack.yaml", d / "pack.yaml")
|
||||
return P.load(d / "pack.yaml")
|
||||
|
||||
|
||||
def test_showcase_is_valid():
|
||||
assert P.load(SRC / "pack.yaml").ok
|
||||
|
||||
|
||||
def test_v02_draft_is_wellformed_and_self_consistent():
|
||||
d = yaml.safe_load((SRC / "pack.v0.2.draft.yaml").read_text())
|
||||
assert d["spec"] == "runbook-pack/v0.2"
|
||||
actors = {a["id"] for a in d["actors"]}
|
||||
dps = {x["id"] for x in d["datapoints"]}
|
||||
steps = {}
|
||||
for st in d["stages"]:
|
||||
for s in st["steps"]:
|
||||
steps[s["id"]] = s
|
||||
assert s["actor"] in actors
|
||||
for u in s.get("uses", []):
|
||||
assert u in dps, u
|
||||
gates = {g["id"] for st in d["stages"] for g in st.get("gates", [])}
|
||||
stages = {st["id"] for st in d["stages"]}
|
||||
for st in d["stages"]:
|
||||
for r in st.get("requires", []):
|
||||
assert r.removeprefix("gate:") in gates | stages
|
||||
assert any(s.get("reversible") is False for s in steps.values())
|
||||
assert any("one_of" in s for s in steps.values())
|
||||
assert not any(k.startswith("secret") for x in d["datapoints"] for k in x)
|
||||
|
||||
|
||||
def test_scripted_playthrough(pk):
|
||||
r = R.create(pk, "tester", {"keeper": "Ada", "weather": "storm", "motto": "Light on"})
|
||||
for s in pk.data["steps"]:
|
||||
sid = s["id"]
|
||||
R.start(r, sid)
|
||||
cmd = R.command_for(r, sid) # the operator runs the command...
|
||||
if cmd:
|
||||
subprocess.run(cmd, shell=True, check=True, capture_output=True)
|
||||
v = s["verify"] # ...then the engine verifies the result
|
||||
res = R.verify(r, sid, confirm=True, manual_ok=True if v["expect"] == "manual" else None)
|
||||
assert res["passed"], sid
|
||||
R.complete(r, sid, {n: "1" for n in s.get("evidence", [])})
|
||||
assert r.complete
|
||||
assert not Path(r.data["params"]["logdir"]).exists()
|
||||
old, new = R.exercise_update(r)
|
||||
assert "exercised" in new
|
||||
|
||||
|
||||
def test_bad_logdir_rejected(pk):
|
||||
with pytest.raises(R.RunError):
|
||||
R.create(pk, "tester", {"logdir": "/etc"})
|
||||
with pytest.raises(R.RunError):
|
||||
R.create(pk, "tester", {"motto": 'x"; rm -rf /'})
|
||||
Loading…
Add table
Add a link
Reference in a new issue