From 8a5cbc1c29dd683a820e131937e9b76d6abcb69e Mon Sep 17 00:00:00 2001 From: tegwick Date: Tue, 29 Sep 2026 11:15:58 +0200 Subject: [PATCH] Run outcomes, score, explicit exercise recording, and step captures (RBT-WP-0004-T13) Co-Authored-By: Claude Sonnet 5.5 Assistant: claude-code Assistant-Model: sonnet Assistant-Process: 295952@bnt-lap001 Assistant-Session: e93f64ad-516c-46eb-9666-aad8d300c477 --- docs/run-spec.md | 36 +++++++ rtut/__main__.py | 20 +++- rtut/run.py | 118 ++++++++++++++++++++++ rtut/ui.py | 88 +++++++++++++++- tests/test_score.py | 104 +++++++++++++++++++ tests/test_ui.py | 66 +++++++++++- workplans/RBT-WP-0004-local-runbook-ui.md | 17 ++++ 7 files changed, 444 insertions(+), 5 deletions(-) create mode 100644 tests/test_score.py diff --git a/docs/run-spec.md b/docs/run-spec.md index 2a98618..6eb1f66 100644 --- a/docs/run-spec.md +++ b/docs/run-spec.md @@ -24,3 +24,39 @@ One run of one native pack, stored as JSON in diff; `--apply` writes it, and only for a run in which every step is done. CLI: `rtut run new|list|status|cmd|start|verify|done|fail|retry|skip|finish ...` + +## Outcome, score and exercise status + +A run is **open** until finalized. `rtut run finalize` (or the UI's *Close this run*) records +an `outcome`: + +| Outcome | Rule | +| --- | --- | +| `succeeded` | every step is `done` (skipped steps do not count) | +| `failed` | at least one step has failed; otherwise abandon | +| `abandoned` | any time. An open run untouched for 7 days also counts as abandoned in the score | + +A finished run is read-only, except for feedback. The **score** of a runbook is derived from +its run records: tried, succeeded, failed, abandoned, open, plus a tally of feedback ratings +(`rtut score `; shown in the UI). Run records live in `.local/` of the machine that ran +them, so the score is per machine until run histories are shared. + +Finalizing never edits the pack by itself. For a `succeeded` run the UI offers **Record pack +as exercised**, which shows what will be written, and on click sets +`exercise_status: exercised by ` in `pack.yaml` (`rtut run finalize ... --outcome +succeeded --apply` does the same). The change is an ordinary edit in the owning repository, to +be reviewed and committed there. + +## Feedback + +`feedback` entries (`rating` good|neutral|poor, `comment`, `at`) are added after a run; history +is kept and the latest counts. `rtut feedback ` lists them. + +## Captures + +A step may carry optional `captures`: the command actually run, and the output that came +back, as documentation or for inspection when the engine cannot reach the system itself. Limits: +20 000 characters of output, 2 000 of command. Obvious secret markers are refused; the engine cannot +detect every secret, so the operator stays responsible. For `output-contains` steps the output can +be **pasted for verification** (`source: pasted-for-verification`), which checks the `contains` text +without running anything. Exit-code checks cannot be verified from pasted output. diff --git a/rtut/__main__.py b/rtut/__main__.py index 24b1c96..5ee1a5f 100644 --- a/rtut/__main__.py +++ b/rtut/__main__.py @@ -45,6 +45,13 @@ def _run(pk, a) -> int: R.retry(r, a.step); return 0 if a.action == "skip": R.skip(r, a.step, a.note); return 0 + if a.action == "finalize": + R.finalize(r, a.outcome or "", a.note) + if r.data["outcome"] == "succeeded" and a.apply: + print("recorded:", R.apply_exercise(r)) + print(r.data["outcome"]); return 0 + if a.action == "capture": + R.add_capture(r, a.step, sys.stdin.read(), a.command); print("captured"); return 0 if a.action == "feedback": R.add_feedback(r, a.rating or "", a.note); print("recorded"); return 0 old, new = R.exercise_update(r) @@ -65,7 +72,7 @@ def main(argv: list[str] | None = None) -> int: sp = sub.add_parser(name) sp.add_argument("root", nargs="?", default=".") sp = sub.add_parser("run", help="record a run of a native pack") - sp.add_argument("action", choices=["new", "list", "status", "cmd", "start", "verify", "done", "fail", "retry", "skip", "finish", "feedback"]) + sp.add_argument("action", choices=["new", "list", "status", "cmd", "start", "verify", "done", "fail", "retry", "skip", "finish", "feedback", "finalize", "capture"]) sp.add_argument("manifest") sp.add_argument("run_id", nargs="?") sp.add_argument("step", nargs="?") @@ -73,6 +80,8 @@ def main(argv: list[str] | None = None) -> int: sp.add_argument("--param", "-p", action="append", default=[], help="name=value") sp.add_argument("--evidence", "-e", action="append", default=[], help="name=value (non-secret)") sp.add_argument("--note", default="") + sp.add_argument("--outcome", choices=["succeeded", "failed", "abandoned"], help="finalize: how the run ended") + sp.add_argument("--command", default="", help="capture: the command you actually ran; output is read from stdin") sp.add_argument("--rating", choices=["good", "neutral", "poor"], help="feedback: how the runbook felt") sp.add_argument("--yes", action="store_true", help="confirm running a verify command") sp.add_argument("--pass", dest="passed", action="store_true", help="manual check: operator confirms it passed") @@ -80,6 +89,8 @@ def main(argv: list[str] | None = None) -> int: sp = sub.add_parser("ui", help="serve the local UI on 127.0.0.1") sp.add_argument("roots", nargs="*", default=["."]) sp.add_argument("--port", type=int, default=8877) + sp = sub.add_parser("score", help="how often each runbook was tried and how it went") + sp.add_argument("root", nargs="?", default=".") sp = sub.add_parser("feedback", help="list the latest feedback of every run of a pack") sp.add_argument("manifest") sp = sub.add_parser("launch"); sp.add_argument("manifest") @@ -98,6 +109,13 @@ def main(argv: list[str] | None = None) -> int: except KeyboardInterrupt: pass return 0 + if a.cmd == "score": + from . import run as R + for p in _packs(a.root): + if p.ok and p.data.get("engine") == "native" and p.spec == "runbook-pack/v0.1": + c = R.scorecard(p) + print(f"{p.id}\ttried {c['tried']}\tsucceeded {c['succeeded']}\tfailed {c['failed']}\tabandoned {c['abandoned']}\topen {c['open']}\t[{p.data['exercise_status']}]") + return 0 if a.cmd in ("list", "validate"): packs = _packs(a.root) if not packs: diff --git a/rtut/run.py b/rtut/run.py index b0cb19c..78321d6 100644 --- a/rtut/run.py +++ b/rtut/run.py @@ -63,6 +63,12 @@ class Run: return all(v["state"] in ("done", "skipped") for v in self.data["steps"].values()) + +def _writable(run: Run) -> None: + if run.data.get("outcome"): + raise RunError(f"run is finished (outcome: {run.data['outcome']}); start a new run") + + def create(pack: Pack, operator: str, params: dict[str, str] | None = None) -> Run: if not pack.ok or pack.data.get("engine") != "native": raise RunError("only valid native packs can be run") @@ -125,6 +131,7 @@ def command_for(run: Run, step_id: str) -> str | None: def start(run: Run, step_id: str) -> None: + _writable(run) run.step_def(step_id) _order(run, step_id) st = run.state(step_id) @@ -140,6 +147,7 @@ def verify(run: Run, step_id: str, confirm: bool = False, manual_ok: bool | None Stored: exit code, duration, output size, and whether 'contains' matched. Output text is never stored. """ + _writable(run) v = run.step_def(step_id)["verify"] st = run.state(step_id) if st["state"] != "running": @@ -174,6 +182,7 @@ def verify(run: Run, step_id: str, confirm: bool = False, manual_ok: bool | None def complete(run: Run, step_id: str, evidence: dict[str, str] | None = None, note: str = "") -> None: + _writable(run) sd = run.step_def(step_id) st = run.state(step_id) if st["state"] != "running": @@ -192,6 +201,7 @@ def complete(run: Run, step_id: str, evidence: dict[str, str] | None = None, not def fail(run: Run, step_id: str, note: str) -> None: + _writable(run) st = run.state(step_id) if st["state"] != "running": raise RunError(f"step {step_id} is not running") @@ -200,6 +210,7 @@ def fail(run: Run, step_id: str, note: str) -> None: def retry(run: Run, step_id: str) -> None: + _writable(run) st = run.state(step_id) if st["state"] != "failed": raise RunError(f"step {step_id} has not failed") @@ -209,6 +220,7 @@ def retry(run: Run, step_id: str) -> None: def skip(run: Run, step_id: str, reason: str) -> None: + _writable(run) sd = run.step_def(step_id) if sd.get("risk", "read-only") != "read-only": raise RunError("only read-only steps may be skipped") @@ -262,3 +274,109 @@ def all_feedback(pack: Pack) -> list[dict]: if fb: out.append({"run_id": rid, "operator": r.data["operator"], **fb}) return sorted(out, key=lambda x: x["at"], reverse=True) + + +# ---- captures: optional documentation of what was run and what came back -------------------- + +CAPTURE_MAX_OUTPUT = 20000 +CAPTURE_MAX_COMMAND = 2000 + + +def add_capture(run: Run, step_id: str, output: str, command: str = "", source: str = "operator") -> dict: + """Keep the command actually run and its output as documentation or for inspection. + + The engine cannot detect every secret. It refuses obvious markers and oversized text; + the operator remains responsible for what is pasted. + """ + _writable(run) + run.step_def(step_id) + st = run.state(step_id) + if st["state"] == "todo": + raise RunError("start the step before capturing its output") + if len(output) > CAPTURE_MAX_OUTPUT or len(command) > CAPTURE_MAX_COMMAND: + raise RunError(f"capture too large (output max {CAPTURE_MAX_OUTPUT}, command max {CAPTURE_MAX_COMMAND} characters)") + if not output.strip() and not command.strip(): + raise RunError("nothing to capture") + entry = {"command": _clean(command.strip(), "captured command"), "output": _clean(output.rstrip(), "captured output"), + "source": source, "at": _now()} + st.setdefault("captures", []).append(entry) + run.save() + return entry + + +def verify_pasted(run: Run, step_id: str, output: str, command: str = "") -> dict: + """Check an output-contains step against pasted output, for systems the engine cannot reach.""" + _writable(run) + v = run.step_def(step_id)["verify"] + st = run.state(step_id) + if st["state"] != "running": + raise RunError(f"step {step_id} is not running") + if v.get("expect") != "output-contains": + raise RunError("only output-contains checks can be verified from pasted output") + add_capture(run, step_id, output, command, source="pasted-for-verification") + needle = render(v["contains"], run.data["params"], run.pack) + result = {"kind": "output-contains", "source": "pasted", "passed": needle in output, + "output_bytes": len(output.encode()), "at": _now()} + st["verification"] = result + run.save() + return result + + +# ---- outcome and score ------------------------------------------------------------------------ + +OUTCOMES = ("succeeded", "failed", "abandoned") +STALE_DAYS = 7 + + +def finalize(run: Run, outcome: str, note: str = "") -> None: + """Close a run. 'succeeded' needs every step done; 'failed' needs a failed step.""" + if run.data.get("outcome"): + raise RunError(f"run already finished ({run.data['outcome']})") + if outcome not in OUTCOMES: + raise RunError(f"outcome must be one of {', '.join(OUTCOMES)}") + states = [v["state"] for v in run.data["steps"].values()] + if outcome == "succeeded" and any(x != "done" for x in states): + raise RunError("a run succeeds only when every step is done (skipped steps do not count)") + if outcome == "failed" and "failed" not in states: + raise RunError("a run can be finished as failed only when a step has failed; otherwise abandon it") + run.data["outcome"] = outcome + run.data["finished"] = _now() + if note.strip(): + run.data["outcome_note"] = _clean(note.strip(), "note") + run.save() + + +def effective_outcome(run: Run, now: datetime | None = None, stale_days: int = STALE_DAYS) -> str: + """finished outcome, or 'open', or 'abandoned' for an open run untouched for stale_days.""" + if run.data.get("outcome"): + return run.data["outcome"] + now = now or datetime.now(timezone.utc) + last = datetime.fromisoformat(run.data.get("updated") or run.data["created"]) + return "abandoned" if (now - last).days >= stale_days else "open" + + +def scorecard(pack: Pack, now: datetime | None = None, stale_days: int = STALE_DAYS) -> dict: + """How often a runbook was tried and how it went, derived from its run records.""" + card = {"tried": 0, "succeeded": 0, "failed": 0, "abandoned": 0, "open": 0, + "feedback": {"good": 0, "neutral": 0, "poor": 0}} + for rid in list_runs(pack): + r = load(pack, rid) + card["tried"] += 1 + card[effective_outcome(r, now, stale_days)] += 1 + fb = latest_feedback(r) + if fb: + card["feedback"][fb["rating"]] += 1 + return card + + +def apply_exercise(run: Run, today: date | None = None) -> str: + """Write 'exercised by ' into pack.yaml for a succeeded run. Returns the new status.""" + if run.data.get("outcome") != "succeeded": + raise RunError("only a run finished as succeeded can exercise the pack") + _, new = exercise_update(run, today) + run.pack.path.write_text(new) + status = STATUS_LINE.search(new).group(0).split(": ", 1)[1] + run.pack.data["exercise_status"] = status + run.data["exercise_recorded"] = status + run.save() + return status diff --git a/rtut/ui.py b/rtut/ui.py index 408abf2..6378e4e 100644 --- a/rtut/ui.py +++ b/rtut/ui.py @@ -34,7 +34,7 @@ def page(title: str, body: str) -> bytes: "const f=b.closest('form'),n=f.querySelector('[name=note]'),w=f.querySelector('[data-warn]');" "if(n&&!n.value.trim()&&!b.dataset.warned){ev.preventDefault();b.dataset.warned='1';w.hidden=false;n.focus();}}));" # keep the operator on the current step after every action - "const c=document.getElementById(location.hash==='#feedback'?'feedback':'current');if(c)c.scrollIntoView({block:'center'});" + "const c=document.getElementById(location.hash==='#feedback'?'feedback':location.hash==='#close'?'close':'current');if(c)c.scrollIntoView({block:'center'});" "").encode() @@ -45,6 +45,11 @@ def e(s) -> str: SMILEYS = (("good", "馃檪", "Went well"), ("neutral", "馃槓", "So-so"), ("poor", "馃檨", "Poor")) +def score_line(c: dict) -> str: + return (f"tried {c['tried']} 路 succeeded {c['succeeded']} 路 failed {c['failed']} 路 " + f"abandoned {c['abandoned']} 路 open {c['open']}") + + class App: def __init__(self, roots: list[Path]): self.token = secrets.token_urlsafe(16) @@ -67,7 +72,8 @@ class App: f"start with: python -m rtut launch {e(p.path)}") else: rows.append(f"
  • {e(d['title'])} " - f"{e(d['exercise_status'])}
    • ") + f"{e(d['exercise_status'])} " + f"{e(score_line(R.scorecard(p)))}") return page("Runbooks", f"

    Runbooks

      {''.join(rows) or '
    • No packs found.
      • '}
    ") def pack_page(self, i: int) -> bytes: @@ -81,11 +87,71 @@ class App: + f"
    {e(x.get('help', ''))} Never enter secrets.

    " for x in d.get("parameters", [])) return page(d["title"], f"

    {e(d['title'])}

    {e(d['exercise_status'])}

    " - f"

    {e(d.get('outcome', ''))}

    Prerequisites

      {prereq}
    " + f"

    {e(d.get('outcome', ''))}

    {e(score_line(R.scorecard(p)))}

    " + f"{self.runs_list(i, p)}

    Prerequisites

      {prereq}
    " f"
    {self.hidden()}{fields}" "

    " "
    ") + def runs_list(self, i: int, p) -> str: + rows = [] + for rid in reversed(R.list_runs(p)): + r = R.load(p, rid) + rows.append(f"
  • {e(rid)} 路 {e(r.data['operator'])} 路 " + f"{e(R.effective_outcome(r))}
    • ") + return f"

    Runs

      {''.join(rows)}
    " if rows else "" + + def capture_html(self, url: str, s: dict, st: dict, state: str, finished: bool) -> str: + saved = "" + for c in st.get("captures", []): + head = f"$ {c['command']}\n" if c.get("command") else "" + saved += f"
    {e(head + c['output'][:4000])}

    captured {e(c['at'][:19])} ({e(c['source'])})

    " + if finished or state == "todo": + return f"
    Captured output{saved}
    " if saved else "" + paste = "" + if state == "running" and s["verify"].get("expect") == "output-contains": + paste = f"" + return (f"
    Capture command and output (optional)" + f"{' - ' + str(len(st.get('captures', []))) + ' saved' if st.get('captures') else ''}{saved}" + f"
    {self.hidden()}" + "

    " + "

    " + f"

    {paste}

    " + "

    Kept in the run record as documentation. Never paste secrets: obvious markers are " + "refused, but not everything can be detected. Use the check button when the command ran on a system " + "this runbook cannot reach.

    ") + + def close_run_form(self, i: int, rid: str, r, p) -> str: + outcome = r.data.get("outcome") + if outcome: + note = f" - {e(r.data['outcome_note'])}" if r.data.get("outcome_note") else "" + box = f"

    Run finished: {e(outcome)}{note}

    " + if outcome == "succeeded": + if r.data.get("exercise_recorded"): + box += f"

    Pack recorded as {e(r.data['exercise_recorded'])}. Commit the change in the owning repository.

    " + else: + old, new = R.exercise_update(r) + line = new.split("exercise_status:", 1)[1].split("\n", 1)[0].strip() + box += (f"

    Current pack status: {e(p.data['exercise_status'])}. " + f"Recording this run would set it to {e(line)} in {e(p.path)} " + "(a reviewable edit; commit it in the owning repository).

    " + f"
    {self.hidden()}" + "
    ") + return f"

    Close this run

    {box}
    " + states = [v["state"] for v in r.data["steps"].values()] + btns = [] + if r.complete and all(x == "done" for x in states): + btns.append("") + if "failed" in states: + btns.append("") + btns.append("") + return (f"

    Close this run

    " + "

    Finishing records the outcome in the pack's score. A finished run is read-only. " + "Runs left open for a week count as abandoned.

    " + f"
    {self.hidden()}" + "

    " + f"

    {' '.join(btns)}

    ") + def feedback_form(self, i: int, rid: str, r) -> str: fb = R.latest_feedback(r) thanks = "" @@ -111,6 +177,7 @@ class App: out.append(f"

    {e(msg)}

    ") blocked = False waiting_for = "" + finished = bool(r.data.get("outcome")) total = len(d["steps"]) done_n = sum(1 for x in d["steps"] if r.state(x["id"])["state"] in ("done", "skipped")) out.append(f"

    {done_n} of {total} steps done

    " @@ -163,6 +230,8 @@ class App: act = f"

    Failed: {e(st.get('note') or '(no comment)')}

    {self.hidden()}
    " elif state == "done" and st.get("note"): act = f"

    Comment: {e(st['note'])}

    " + if finished and state != "done": + act = "" # a finished run is read-only if state not in ("done", "skipped") and not blocked: blocked = True waiting_for = f"step {n}, {s['title']}" @@ -173,12 +242,14 @@ class App: f"{n}. {e(s['title'])} " f"[{e(s['owner'])}] {e(s.get('risk', 'read-only'))}" f"{state}{block}" + f"{self.capture_html(url, s, st, state, finished)}" f"

    Done when: {e(s['verify']['done_when'])}

    {rb}{act}") if r.complete: out.append("

    All steps done. To record this run as exercising the pack, review and apply: " f"python -m rtut run finish {e(p.path)} {e(rid)} --apply

    ") if r.complete: out.append(self.feedback_form(i, rid, r)) + out.append(self.close_run_form(i, rid, r, p)) out.append("

    Threat checks

      " + "".join(f"
    • {e(t)}
      • " for t in d["threat_checks"]) + "
    ") return page(d["title"], "".join(out)) @@ -231,6 +302,15 @@ def make_handler(app: App, port_holder: list): i, rid = int(parts[1]), parts[2] R.add_feedback(R.load(app.packs[i], rid), form.get("rating", ""), form.get("comment", "")) return self._redirect(f"/run/{i}/{rid}#feedback") + if len(parts) == 4 and parts[0] == "run" and parts[3] in ("finalize", "exercise"): + i, rid = int(parts[1]), parts[2] + r = R.load(app.packs[i], rid) + if parts[3] == "finalize": + R.finalize(r, form.get("outcome", ""), form.get("note", "")) + else: + R.apply_exercise(r) + app.packs[i] = P.load(app.packs[i].path) # pick up the new exercise status + return self._redirect(f"/run/{i}/{rid}#close") i, rid, sid, action = int(parts[1]), parts[2], parts[3], parts[4] p = app.packs[i]; r = R.load(p, rid) if action == "start": R.start(r, sid) @@ -239,6 +319,8 @@ def make_handler(app: App, port_holder: list): elif action == "done": R.complete(r, sid, {k[2:]: v for k, v in form.items() if k.startswith("e_")}, form.get("note", "")) elif action == "fail": R.fail(r, sid, form.get("note", "")) + elif action == "capture": R.add_capture(r, sid, form.get("output", ""), form.get("command", "")) + elif action == "verify-pasted": R.verify_pasted(r, sid, form.get("output", ""), form.get("command", "")) elif action == "retry": R.retry(r, sid) return self._redirect(f"/run/{i}/{rid}#current") except (R.RunError, KeyError, ValueError, IndexError) as ex: diff --git a/tests/test_score.py b/tests/test_score.py new file mode 100644 index 0000000..a62c7df --- /dev/null +++ b/tests/test_score.py @@ -0,0 +1,104 @@ +import shutil +import subprocess +from datetime import datetime, timedelta, timezone +from pathlib import Path + +import pytest + +from rtut import pack as P, run as R + +ROOT = Path(__file__).resolve().parent.parent + + +@pytest.fixture +def pk(tmp_path): + d = tmp_path / "runbooks" / "show"; d.mkdir(parents=True) + shutil.copy(ROOT / "runbooks/showcase-lighthouse/pack.yaml", d / "pack.yaml") + return P.load(d / "pack.yaml") + + +def play_all(r): + for s in r.pack.data["steps"]: + sid = s["id"]; R.start(r, sid) + cmd = R.command_for(r, sid) + if cmd: + subprocess.run(cmd, shell=True, check=True, capture_output=True) + R.verify(r, sid, confirm=True, manual_ok=True if s["verify"]["expect"] == "manual" else None) + R.complete(r, sid, {n: "1" for n in s.get("evidence", [])}) + + +def test_finalize_rules(pk): + r = R.create(pk, "bernd") + with pytest.raises(R.RunError): + R.finalize(r, "succeeded") # nothing done yet + with pytest.raises(R.RunError): + R.finalize(r, "failed") # nothing failed: abandon instead + with pytest.raises(R.RunError): + R.finalize(r, "great") + R.start(r, "read-the-sign"); R.fail(r, "read-the-sign", "could not read it") + R.finalize(r, "failed", "gave up on step 1") + assert r.data["outcome"] == "failed" and r.data["outcome_note"] + with pytest.raises(R.RunError): + R.finalize(r, "abandoned") # already finished + + +def test_finished_run_is_read_only_but_takes_feedback(pk): + r = R.create(pk, "bernd"); R.finalize(r, "abandoned") + for call in (lambda: R.start(r, "read-the-sign"), lambda: R.add_capture(r, "read-the-sign", "x")): + with pytest.raises(R.RunError, match="finished"): + call() + R.add_feedback(r, "poor", "gave up: unclear") # feedback stays possible + + +def test_scorecard_counts_every_outcome_and_stale_open_runs(pk): + ok = R.create(pk, "a"); play_all(ok); R.finalize(ok, "succeeded"); R.add_feedback(ok, "good") + bad = R.create(pk, "b"); R.start(bad, "read-the-sign"); R.fail(bad, "read-the-sign", "x"); R.finalize(bad, "failed") + gone = R.create(pk, "c"); R.finalize(gone, "abandoned") + open_ = R.create(pk, "d") + stale = R.create(pk, "e") + stale.data["updated"] = (datetime.now(timezone.utc) - timedelta(days=9)).isoformat() + stale.path.write_text(__import__("json").dumps(stale.data)) + c = R.scorecard(pk) + assert (c["tried"], c["succeeded"], c["failed"], c["abandoned"], c["open"]) == (5, 1, 1, 2, 1) + assert c["feedback"]["good"] == 1 + + +def test_finalizing_success_can_record_exercise_status(pk): + r = R.create(pk, "bernd"); play_all(r) + with pytest.raises(R.RunError): + R.apply_exercise(r) # not finalized yet + R.finalize(r, "succeeded") + status = R.apply_exercise(r) + assert status.startswith("exercised ") and status.endswith(" by bernd") + reloaded = P.load(pk.path) + assert reloaded.ok and reloaded.data["exercise_status"] == status + + +def test_capture_documentation_and_limits(pk): + r = R.create(pk, "bernd") + with pytest.raises(R.RunError, match="start the step"): + R.add_capture(r, "read-the-sign", "out") + R.start(r, "read-the-sign") + R.add_capture(r, "read-the-sign", "Tower of stone", command="printf poem") + assert R.load(pk, r.data["run_id"]).state("read-the-sign")["captures"][0]["output"] == "Tower of stone" + with pytest.raises(R.RunError, match="too large"): + R.add_capture(r, "read-the-sign", "x" * 20001) + with pytest.raises(R.RunError, match="secret-looking"): + R.add_capture(r, "read-the-sign", "hvs.ABCDEFGHIJKLMN") + with pytest.raises(R.RunError, match="nothing"): + R.add_capture(r, "read-the-sign", " ") + + +def test_verify_against_pasted_output(pk): + r = R.create(pk, "bernd", {"keeper": "Ada"}) + R.start(r, "read-the-sign") + with pytest.raises(R.RunError, match="output-contains"): + R.verify_pasted(r, "read-the-sign", "x") # manual step: not applicable + R.complete # (step 2 is output-contains) + R.verify(r, "read-the-sign", manual_ok=True); R.complete(r, "read-the-sign") + R.start(r, "meet-the-keeper") + assert not R.verify_pasted(r, "meet-the-keeper", "nothing useful", "echo ...")["passed"] + ok = R.verify_pasted(r, "meet-the-keeper", "Ada climbs the stairs. Motto on the wall - x") + assert ok["passed"] and ok["source"] == "pasted" + caps = R.load(pk, r.data["run_id"]).state("meet-the-keeper")["captures"] + assert len(caps) == 2 and caps[0]["source"] == "pasted-for-verification" diff --git a/tests/test_ui.py b/tests/test_ui.py index 2a32975..4ead875 100644 --- a/tests/test_ui.py +++ b/tests/test_ui.py @@ -177,7 +177,9 @@ def test_done_steps_fold_to_the_title_line_and_others_stay_open(srv): _, _, body = req(srv, "GET", base) assert "
    " in body # done: no 'open' attribute, so folded assert "1. Read the poem" in body.split("
    ")[1].split("")[0] # title line stays in the summary - assert "Comment: fine" in body.split("
    ")[1].split("
    ")[0] # the rest is inside the fold + rest = body.split("
    ")[1] + step_end = min(x for x in (rest.find("