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
286 lines
14 KiB
Python
286 lines
14 KiB
Python
import http.client
|
|
import re
|
|
import shutil
|
|
import threading
|
|
from pathlib import Path
|
|
from urllib.parse import urlencode
|
|
|
|
import pytest
|
|
|
|
from rtut import ui
|
|
|
|
ROOT = Path(__file__).resolve().parent.parent
|
|
|
|
|
|
@pytest.fixture
|
|
def srv(tmp_path):
|
|
d = tmp_path / "runbooks" / "hello"; d.mkdir(parents=True)
|
|
shutil.copy(ROOT / "runbooks/example-hello/pack.yaml", d / "pack.yaml")
|
|
s = ui.serve([tmp_path], 0)
|
|
threading.Thread(target=s.serve_forever, daemon=True).start()
|
|
yield s
|
|
s.shutdown()
|
|
|
|
|
|
def req(s, method, path, form=None, host="127.0.0.1"):
|
|
c = http.client.HTTPConnection("127.0.0.1", s.server_address[1])
|
|
body = urlencode(form) if form is not None else None
|
|
c.request(method, path, body, {"Host": host, "Content-Type": "application/x-www-form-urlencoded"})
|
|
r = c.getresponse()
|
|
return r.status, r.getheader("Location"), r.read().decode()
|
|
|
|
|
|
def test_index_lists_pack(srv):
|
|
st, _, body = req(srv, "GET", "/")
|
|
assert st == 200 and "Example - check a host answers" in body
|
|
|
|
|
|
def test_foreign_host_and_missing_token_refused(srv):
|
|
assert req(srv, "GET", "/", host="evil.example")[0] == 403
|
|
assert req(srv, "POST", "/run/new/0", {"operator": "bernd"})[0] == 403
|
|
|
|
|
|
def test_walk_a_run(srv):
|
|
t = srv.app.token
|
|
st, loc, _ = req(srv, "POST", "/run/new/0", {"_t": t, "operator": "bernd", "p_host": "localhost", "p_count": "1"})
|
|
assert st == 303 and loc.startswith("/run/0/") and loc.endswith("#current")
|
|
assert req(srv, "POST", loc.split("#")[0] + "/ping/start", {"_t": t})[0] == 303
|
|
st, _, body = req(srv, "GET", loc.split("#")[0])
|
|
assert "ping -c 1 localhost" in body and "Run check" in body
|
|
req(srv, "POST", loc.split("#")[0] + "/ping/verify", {"_t": t, "yes": "1"})
|
|
st, _, body = req(srv, "GET", loc.split("#")[0])
|
|
if "passed" in body and "failed" not in body.split("Threat")[0].replace("Mark failed", ""):
|
|
assert req(srv, "POST", loc.split("#")[0] + "/ping/done", {"_t": t})[0] == 303
|
|
assert "All steps done" in req(srv, "GET", loc.split("#")[0])[2]
|
|
|
|
|
|
def test_unsafe_parameter_rejected_and_escaped(srv):
|
|
t = srv.app.token
|
|
st, _, body = req(srv, "POST", "/run/new/0", {"_t": t, "operator": "bernd", "p_host": "<script>x;"})
|
|
assert st == 400 and "<script>x" not in body
|
|
|
|
|
|
def test_secret_in_note_refused(srv):
|
|
t = srv.app.token
|
|
_, loc, _ = req(srv, "POST", "/run/new/0", {"_t": t, "operator": "bernd"})
|
|
req(srv, "POST", loc.split("#")[0] + "/ping/start", {"_t": t})
|
|
st, _, body = req(srv, "POST", loc.split("#")[0] + "/ping/fail", {"_t": t, "note": "hvs.ABCDEFGHIJKLMN"})
|
|
assert st == 400 and "secret-looking" in body
|
|
|
|
|
|
def test_manual_check_cycle_and_blocked_reason(srv):
|
|
"""Yes/Not-yet buttons, 'Mark done' to lock in, numbering, and why later steps are blocked."""
|
|
import shutil
|
|
t = srv.app.token
|
|
# use the showcase pack, which starts with a manual step
|
|
d = Path(srv.app.packs[0].path).parents[2] / "runbooks" / "show"; d.mkdir(parents=True)
|
|
shutil.copy(ROOT / "runbooks/showcase-lighthouse/pack.yaml", d / "pack.yaml")
|
|
from rtut import pack as P
|
|
srv.app.packs.append(P.load(d / "pack.yaml"))
|
|
i = len(srv.app.packs) - 1
|
|
_, loc, _ = req(srv, "POST", f"/run/new/{i}", {"_t": t, "operator": "bernd"})
|
|
_, _, body = req(srv, "GET", loc.split("#")[0])
|
|
assert "0 of 8 steps done" in body and "1. Read the poem" in body
|
|
assert "Waiting for: step 1" in body # blocked steps explain themselves
|
|
req(srv, "POST", loc.split("#")[0] + "/read-the-sign/start", {"_t": t})
|
|
_, _, body = req(srv, "GET", loc.split("#")[0])
|
|
assert "Yes, it is" in body and "Not yet" in body and "/read-the-sign/done" not in body
|
|
req(srv, "POST", loc.split("#")[0] + "/read-the-sign/verify", {"_t": t, "ok": "0"})
|
|
_, _, body = req(srv, "GET", loc.split("#")[0])
|
|
assert "Check not passed" in body and "/read-the-sign/done" not in body
|
|
req(srv, "POST", loc.split("#")[0] + "/read-the-sign/verify", {"_t": t, "ok": "1"})
|
|
_, _, body = req(srv, "GET", loc.split("#")[0])
|
|
assert "Check passed" in body and "/read-the-sign/done" in body
|
|
req(srv, "POST", loc.split("#")[0] + "/read-the-sign/done", {"_t": t})
|
|
_, _, body = req(srv, "GET", loc.split("#")[0])
|
|
assert "1 of 8 steps done" in body and "Start this step" in body
|
|
|
|
|
|
def _showcase_run(srv):
|
|
import shutil
|
|
from rtut import pack as P
|
|
t = srv.app.token
|
|
d = Path(srv.app.packs[0].path).parents[2] / "runbooks" / "show2"; d.mkdir(parents=True)
|
|
shutil.copy(ROOT / "runbooks/showcase-lighthouse/pack.yaml", d / "pack.yaml")
|
|
srv.app.packs.append(P.load(d / "pack.yaml"))
|
|
i = len(srv.app.packs) - 1
|
|
st, loc, _ = req(srv, "POST", f"/run/new/{i}", {"_t": t, "operator": "bernd"})
|
|
return t, loc, st
|
|
|
|
|
|
def test_redirects_keep_focus_on_current_step(srv):
|
|
t, loc, _ = _showcase_run(srv)
|
|
assert loc.endswith("#current")
|
|
base = loc.split("#")[0]
|
|
st, redirect, _ = req(srv, "POST", base + "/read-the-sign/start", {"_t": t})
|
|
assert st == 303 and redirect.endswith("#current")
|
|
_, _, body = req(srv, "GET", base)
|
|
assert body.count("id=current") == 1 and "scrollIntoView" in body
|
|
|
|
|
|
def test_command_and_copy_appear_only_when_step_is_started(srv):
|
|
t, loc, _ = _showcase_run(srv)
|
|
base = loc.split("#")[0]
|
|
_, _, body = req(srv, "GET", base)
|
|
assert "<button type=button data-copy>" not in body and "<pre>" not in body # nothing started yet
|
|
req(srv, "POST", base + "/read-the-sign/start", {"_t": t})
|
|
_, _, body = req(srv, "GET", base)
|
|
assert "<button type=button data-copy>" in body # started: command and Copy appear
|
|
assert "<pre>printf" in body
|
|
req(srv, "POST", base + "/read-the-sign/verify", {"_t": t, "ok": "1"})
|
|
req(srv, "POST", base + "/read-the-sign/done", {"_t": t, "note": "read it twice"})
|
|
_, _, body = req(srv, "GET", base)
|
|
assert "<button type=button data-copy>" not in body # done: no copy button
|
|
assert "<pre>printf" in body and "Comment: read it twice" in body # command kept as a record, comment shown
|
|
|
|
|
|
def test_action_buttons_left_of_the_comment(srv):
|
|
t, loc, _ = _showcase_run(srv)
|
|
base = loc.split("#")[0]
|
|
req(srv, "POST", base + "/read-the-sign/start", {"_t": t})
|
|
req(srv, "POST", base + "/read-the-sign/verify", {"_t": t, "ok": "1"})
|
|
_, _, body = req(srv, "GET", base)
|
|
row = body[body.index("<p class=row>"):]
|
|
assert row.index("Mark done") < row.index("Mark failed") < row.index("name=note")
|
|
|
|
|
|
def test_fail_without_comment_warns_first_then_is_allowed(srv):
|
|
t, loc, _ = _showcase_run(srv)
|
|
base = loc.split("#")[0]
|
|
req(srv, "POST", base + "/read-the-sign/start", {"_t": t})
|
|
_, _, body = req(srv, "GET", base)
|
|
assert "data-fail" in body and "data-warn hidden" in body and "dataset.warned" in body # client-side first-click warning
|
|
st, _, _ = req(srv, "POST", base + "/read-the-sign/fail", {"_t": t, "note": ""}) # the second click submits
|
|
assert st == 303
|
|
_, _, body = req(srv, "GET", base)
|
|
assert "(no comment)" in body and "Retry" in body
|
|
|
|
|
|
def test_state_pill_is_top_right_after_the_numbered_title(srv):
|
|
t, loc, _ = _showcase_run(srv)
|
|
_, _, body = req(srv, "GET", loc.split("#")[0])
|
|
head = body[body.index("<span class=head>"):]
|
|
assert head.index("1. Read the poem") < head.index("class=pill") # number and title first, state last
|
|
assert "justify-content:space-between" in body # pill pushed to the right edge
|
|
|
|
|
|
def test_done_steps_fold_to_the_title_line_and_others_stay_open(srv):
|
|
t, loc, _ = _showcase_run(srv)
|
|
base = loc.split("#")[0]
|
|
_, _, body = req(srv, "GET", base)
|
|
assert "<details id=current class='step todo' open>" in body # active step is open
|
|
req(srv, "POST", base + "/read-the-sign/start", {"_t": t})
|
|
req(srv, "POST", base + "/read-the-sign/verify", {"_t": t, "ok": "1"})
|
|
_, _, body = req(srv, "GET", base)
|
|
assert "<details id=current class='step running' open>" in body
|
|
req(srv, "POST", base + "/read-the-sign/done", {"_t": t, "note": "fine"})
|
|
_, _, body = req(srv, "GET", base)
|
|
assert "<details class='step done'>" in body # done: no 'open' attribute, so folded
|
|
assert "1. Read the poem" in body.split("<details class='step done'>")[1].split("</summary>")[0] # title line stays in the summary
|
|
rest = body.split("<details class='step done'>")[1]
|
|
step_end = min(x for x in (rest.find("<details id=current"), rest.find("<details class='step"), rest.find("<div id=close")) if x >= 0)
|
|
assert "Comment: fine" in rest[:step_end] # the rest is inside the fold
|
|
|
|
|
|
def _finish_showcase(srv, t, base):
|
|
import subprocess
|
|
from rtut import run as R
|
|
rid = base.rsplit("/", 1)[1]
|
|
r = R.load(srv.app.packs[int(base.split("/")[2])], rid)
|
|
for s in r.pack.data["steps"]:
|
|
sid = s["id"]
|
|
req(srv, "POST", f"{base}/{sid}/start", {"_t": t})
|
|
cmd = R.command_for(r, sid)
|
|
if cmd:
|
|
subprocess.run(cmd, shell=True, check=True, capture_output=True)
|
|
if s["verify"]["expect"] == "manual":
|
|
req(srv, "POST", f"{base}/{sid}/verify", {"_t": t, "ok": "1"})
|
|
else:
|
|
req(srv, "POST", f"{base}/{sid}/verify", {"_t": t, "yes": "1"})
|
|
ev = {f"e_{n}": "1" for n in s.get("evidence", [])}
|
|
req(srv, "POST", f"{base}/{sid}/done", {"_t": t, **ev})
|
|
|
|
|
|
def test_feedback_form_after_all_steps_done_and_recording(srv):
|
|
t, loc, _ = _showcase_run(srv)
|
|
base = loc.split("#")[0]
|
|
_, _, body = req(srv, "GET", base)
|
|
assert "<h2>How was this runbook?</h2>" not in body # not before the run is complete
|
|
_finish_showcase(srv, t, base)
|
|
_, _, body = req(srv, "GET", base)
|
|
assert "All steps done." in body and "<h2>How was this runbook?</h2>" in body
|
|
assert body.index("All steps done.") < body.index("<h2>How was this runbook?</h2>")
|
|
assert "🙂" in body and "😐" in body and "🙁" in body and "name=comment" in body
|
|
st, loc2, _ = req(srv, "POST", base + "/feedback", {"_t": t, "rating": "neutral", "comment": "step 3 was unclear"})
|
|
assert st == 303 and loc2.endswith("#feedback")
|
|
_, _, body = req(srv, "GET", base)
|
|
assert "Thank you." in body and "step 3 was unclear" in body
|
|
assert "<h2>How was this runbook?</h2>" not in body # the section closed once feedback was collected
|
|
assert "Back to the runbook list" not in body # the sidebar's Home replaces that link
|
|
assert "Change my feedback" in body # the form is still reachable, folded away
|
|
assert req(srv, "GET", "/")[0] == 200
|
|
|
|
|
|
def test_feedback_validation(srv):
|
|
t, loc, _ = _showcase_run(srv)
|
|
base = loc.split("#")[0]
|
|
_finish_showcase(srv, t, base)
|
|
assert req(srv, "POST", base + "/feedback", {"_t": t, "rating": "amazing"})[0] == 400
|
|
st, _, body = req(srv, "POST", base + "/feedback", {"_t": t, "rating": "good", "comment": "hvs.ABCDEFGHIJKLMN"})
|
|
assert st == 400 and "secret-looking" in body
|
|
assert req(srv, "POST", base + "/feedback", {"rating": "good"})[0] == 403 # CSRF token required
|
|
|
|
|
|
def test_score_runs_list_and_finalize_flow(srv):
|
|
t, loc, _ = _showcase_run(srv)
|
|
base = loc.split("#")[0]
|
|
i = int(base.split("/")[2])
|
|
_, _, body = req(srv, "GET", f"/pack/{i}")
|
|
assert "tried 1 · succeeded 0" in body and "stalled 1" in body and "Runs" in body
|
|
_, _, body = req(srv, "GET", base)
|
|
assert "Abandon run" in body and "Finish: succeeded" not in body # cannot succeed before every step is done
|
|
_finish_showcase(srv, t, base)
|
|
_, _, body = req(srv, "GET", base)
|
|
assert "Finish: succeeded" in body
|
|
assert req(srv, "POST", base + "/finalize", {"_t": t, "outcome": "succeeded", "note": "smooth"})[0] == 303
|
|
_, _, body = req(srv, "GET", base)
|
|
assert "Run finished: succeeded" in body and "smooth" in body
|
|
assert "Record pack as exercised" not in body and "would set it to" not in body # retired: maturity is derived from runs
|
|
assert "Start this step" not in body # read-only now
|
|
_, _, body = req(srv, "GET", "/")
|
|
assert "succeeded 1" in body
|
|
assert req(srv, "POST", base + "/exercise", {"_t": t})[0] == 400 # the route is gone
|
|
|
|
|
|
def test_abandon_and_actions_refused_after_finish(srv):
|
|
t, loc, _ = _showcase_run(srv)
|
|
base = loc.split("#")[0]
|
|
assert req(srv, "POST", base + "/finalize", {"_t": t, "outcome": "abandoned"})[0] == 303
|
|
st, _, body = req(srv, "POST", base + "/read-the-sign/start", {"_t": t})
|
|
assert st == 400 and "finished" in body
|
|
|
|
|
|
def test_capture_and_pasted_output_check(srv):
|
|
t, loc, _ = _showcase_run(srv)
|
|
base = loc.split("#")[0]
|
|
st, _, body = req(srv, "POST", base + "/meet-the-keeper/capture", {"_t": t, "output": "x"})
|
|
assert st == 400 and "start the step" in body
|
|
for sid in ("read-the-sign",):
|
|
req(srv, "POST", f"{base}/{sid}/start", {"_t": t})
|
|
req(srv, "POST", f"{base}/{sid}/verify", {"_t": t, "ok": "1"})
|
|
req(srv, "POST", f"{base}/{sid}/done", {"_t": t})
|
|
req(srv, "POST", base + "/meet-the-keeper/start", {"_t": t})
|
|
_, _, body = req(srv, "GET", base)
|
|
assert "Capture command and output" in body and "Check against this output" in body
|
|
assert req(srv, "POST", base + "/meet-the-keeper/capture", {"_t": t, "command": "echo hi", "output": "hello there"})[0] == 303
|
|
_, _, body = req(srv, "GET", base)
|
|
assert "hello there" in body and "$ echo hi" in body and "1 saved" in body
|
|
st, _, body = req(srv, "POST", base + "/meet-the-keeper/verify-pasted", {"_t": t, "output": "wrong text"})
|
|
assert st == 303
|
|
_, _, body = req(srv, "GET", base)
|
|
assert "Check not passed" in body
|
|
req(srv, "POST", base + "/meet-the-keeper/verify-pasted", {"_t": t, "output": "Ada climbs the stairs. Motto on the wall - x"})
|
|
_, _, body = req(srv, "GET", base)
|
|
assert "Check passed" in body
|
|
st, _, body = req(srv, "POST", base + "/meet-the-keeper/capture", {"_t": t, "output": "hvs.ABCDEFGHIJKLMN"})
|
|
assert st == 400 and "secret-looking" in body
|