tegwick
6d70f4d97d
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
204 lines
10 KiB
Python
204 lines
10 KiB
Python
import http.client
|
|
import json
|
|
import shutil
|
|
import subprocess
|
|
import threading
|
|
import uuid
|
|
from pathlib import Path
|
|
from urllib.parse import urlencode
|
|
|
|
import pytest
|
|
|
|
from rtut import run as R1, run2 as R, ui
|
|
|
|
ROOT = Path(__file__).resolve().parent.parent
|
|
_scratch = []
|
|
|
|
|
|
@pytest.fixture
|
|
def srv(tmp_path):
|
|
for name, src in (("v1", "showcase-lighthouse"), ("v2", "showcase-lighthouse-v2")):
|
|
d = tmp_path / "runbooks" / name; d.mkdir(parents=True)
|
|
shutil.copy(ROOT / f"runbooks/{src}/pack.yaml", d / "pack.yaml")
|
|
s = ui.serve([tmp_path], 0)
|
|
threading.Thread(target=s.serve_forever, daemon=True).start()
|
|
s.repo = tmp_path
|
|
yield s
|
|
s.shutdown()
|
|
while _scratch:
|
|
shutil.rmtree(_scratch.pop(), ignore_errors=True)
|
|
|
|
|
|
def req(s, method, path, form=None):
|
|
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": "127.0.0.1", "Content-Type": "application/x-www-form-urlencoded"})
|
|
r = c.getresponse()
|
|
return r.status, r.getheader("Location"), r.read().decode()
|
|
|
|
|
|
def post(s, path, **form):
|
|
return req(s, "POST", path, {"_t": s.app.token, **form})
|
|
|
|
|
|
def start(s, pack_id):
|
|
i = s.app.pack_index(pack_id)
|
|
logdir = f"/tmp/rtut-lock-{uuid.uuid4().hex[:8]}"
|
|
_scratch.append(logdir)
|
|
st, loc, _ = post(s, f"/run/new/{i}", operator="bernd", p_logdir=logdir)
|
|
assert st == 303
|
|
return loc.split("#")[0], i
|
|
|
|
|
|
def run_of(s, base):
|
|
return R1.load(s.app.packs[int(base.split("/")[2])], base.rsplit("/", 1)[1])
|
|
|
|
|
|
# ---- Home headlines are shortcuts ---------------------------------------------------------------------------
|
|
|
|
def test_home_headlines_link_to_the_sidebar_sections_and_the_section_pages_do_not(srv):
|
|
_, _, home = req(srv, "GET", "/")
|
|
assert "<h1 class=hl><a href='/projects'>Projects</a></h1>" in home
|
|
assert "<h2 class=hl><a href='/projects'>Project templates</a></h2>" in home
|
|
assert "<a href='/activity'>Latest activities</a>" in home and "<h2 class=hl><a href='/runbooks'>Runbooks</a></h2>" in home
|
|
_, _, projects = req(srv, "GET", "/projects")
|
|
assert "<h1>Projects</h1>" in projects and "class=hl" not in projects # on its own page the headline is just a headline
|
|
assert "<h1>Runbooks</h1>" in req(srv, "GET", "/runbooks")[2]
|
|
assert "<h1>Latest activities" in req(srv, "GET", "/activity")[2]
|
|
|
|
|
|
# ---- no "back to the runbook list" -------------------------------------------------------------------------------
|
|
|
|
def test_there_is_no_back_to_the_runbook_list_link_in_either_run_view(srv):
|
|
for pack in ("rtut.showcase-lighthouse", "rtut.showcase-lighthouse-v2"):
|
|
base, i = start(srv, pack)
|
|
assert post(srv, f"{base}/feedback", rating="good", comment="fine")[0] == 303
|
|
assert "Back to the runbook list" not in req(srv, "GET", base)[2]
|
|
assert post(srv, f"{base}/finalize", outcome="abandoned")[0] == 303
|
|
assert "Back to the runbook list" not in req(srv, "GET", base)[2]
|
|
|
|
|
|
# ---- closed runs are locked ------------------------------------------------------------------------------------------
|
|
|
|
@pytest.mark.parametrize("pack", ["rtut.showcase-lighthouse", "rtut.showcase-lighthouse-v2"])
|
|
def test_a_closed_run_shows_its_feedback_read_only_without_a_change_option(srv, pack):
|
|
base, _ = start(srv, pack)
|
|
assert post(srv, f"{base}/feedback", rating="neutral", comment="a bit long")[0] == 303
|
|
assert "Change my feedback" in req(srv, "GET", base)[2] # while open it can still be changed
|
|
assert post(srv, f"{base}/finalize", outcome="abandoned")[0] == 303
|
|
body = req(srv, "GET", base)[2]
|
|
assert "Change my feedback" not in body and f"action='{base}/feedback'" not in body
|
|
assert "<h2>Feedback</h2>" in body and "a bit long" in body and "closed and locked, so its feedback can no longer be changed" in body
|
|
st, _, err = post(srv, f"{base}/feedback", rating="good", comment="changed my mind")
|
|
assert st == 400 and "closed and locked" in err
|
|
assert run_of(srv, base).data["feedback"][0]["comment"] == "a bit long" # nothing reached the record
|
|
|
|
|
|
@pytest.mark.parametrize("pack", ["rtut.showcase-lighthouse", "rtut.showcase-lighthouse-v2"])
|
|
def test_a_closed_run_without_feedback_shows_no_feedback_section(srv, pack):
|
|
base, _ = start(srv, pack)
|
|
assert post(srv, f"{base}/finalize", outcome="abandoned")[0] == 303
|
|
body = req(srv, "GET", base)[2]
|
|
assert "<h2>Feedback</h2>" not in body and "<h2>How was this runbook?</h2>" not in body and "Change my feedback" not in body
|
|
|
|
|
|
@pytest.mark.parametrize("pack", ["rtut.showcase-lighthouse", "rtut.showcase-lighthouse-v2"])
|
|
def test_the_closing_form_takes_optional_feedback_so_abandoned_runs_are_not_silent(srv, pack):
|
|
base, _ = start(srv, pack)
|
|
body = req(srv, "GET", base)[2]
|
|
assert "How was this runbook? <span class=note>(optional; it cannot be added after closing)</span>" in body and "name=fb_comment" in body
|
|
assert post(srv, f"{base}/finalize", outcome="abandoned", rating="poor", fb_comment="step 1 was unclear")[0] == 303
|
|
body = req(srv, "GET", base)[2]
|
|
assert "<h2>Feedback</h2>" in body and "step 1 was unclear" in body
|
|
assert run_of(srv, base).data["feedback"][0]["rating"] == "poor"
|
|
|
|
|
|
def test_the_closing_form_does_not_ask_again_when_feedback_was_already_given(srv):
|
|
base, _ = start(srv, "rtut.showcase-lighthouse-v2")
|
|
post(srv, f"{base}/feedback", rating="good", comment="ok")
|
|
assert "name=fb_comment" not in req(srv, "GET", base)[2]
|
|
|
|
|
|
def test_a_bad_rating_on_closing_refuses_the_whole_close(srv):
|
|
base, _ = start(srv, "rtut.showcase-lighthouse-v2")
|
|
st, _, err = post(srv, f"{base}/finalize", outcome="abandoned", rating="great")
|
|
assert st == 400 and "rating must be" in err
|
|
assert not run_of(srv, base).data.get("outcome") # still open
|
|
|
|
|
|
@pytest.mark.parametrize("pack", ["rtut.showcase-lighthouse", "rtut.showcase-lighthouse-v2"])
|
|
def test_a_closed_run_says_it_is_locked_and_refuses_every_change(srv, pack):
|
|
base, _ = start(srv, pack)
|
|
post(srv, f"{base}/finalize", outcome="abandoned")
|
|
body = req(srv, "GET", base)[2]
|
|
assert "closed and locked" in body and "its record is sealed and cannot be changed" in body
|
|
first = "read-the-sign"
|
|
for path, form in ((f"{base}/{first}/start", {}), (f"{base}/mode", {"to": "auto-confirm"}), (f"{base}/data", {"d_weather": "clear"}),
|
|
(f"{base}/{first}/capture", {"output": "x"}), (f"{base}/finalize", {"outcome": "succeeded"})):
|
|
st, _, err = post(srv, path, **form)
|
|
assert st in (400,) and ("closed" in err or "finished" in err or "already" in err), path
|
|
assert run_of(srv, base).data["outcome"] == "abandoned"
|
|
|
|
|
|
def test_the_finished_comment_is_shown_as_a_comment_not_as_a_stray_dash(srv):
|
|
base, _ = start(srv, "rtut.showcase-lighthouse-v2")
|
|
post(srv, f"{base}/finalize", outcome="abandoned", note="gave up on step two")
|
|
body = req(srv, "GET", base)[2]
|
|
assert "<p>Comment: gave up on step two</p>" in body and "<p> - " not in body
|
|
|
|
|
|
def test_the_record_file_is_read_only_after_closing_and_tampering_is_called_out(srv):
|
|
base, _ = start(srv, "rtut.showcase-lighthouse-v2")
|
|
post(srv, f"{base}/finalize", outcome="abandoned")
|
|
r = run_of(srv, base)
|
|
assert r.path.stat().st_mode & 0o222 == 0 and R1.seal_state(r) == "sealed"
|
|
assert "changed outside the app" not in req(srv, "GET", base)[2]
|
|
r.path.chmod(0o644)
|
|
data = json.loads(r.path.read_text()); data["operator"] = "forged"; r.path.write_text(json.dumps(data))
|
|
body = req(srv, "GET", base)[2]
|
|
assert "changed outside the app after it was sealed" in body and "no longer counts towards maturity" in body
|
|
|
|
|
|
# ---- evidence export ------------------------------------------------------------------------------------------------------
|
|
|
|
def raw(s, path):
|
|
c = http.client.HTTPConnection("127.0.0.1", s.server_address[1])
|
|
c.request("GET", path, None, {"Host": "127.0.0.1"})
|
|
r = c.getresponse()
|
|
return r.status, {k.lower(): v for k, v in r.getheaders()}, r.read()
|
|
|
|
|
|
@pytest.mark.parametrize("pack", ["rtut.showcase-lighthouse", "rtut.showcase-lighthouse-v2"])
|
|
def test_a_closed_run_record_can_be_downloaded_and_its_seal_checked_elsewhere(srv, pack, tmp_path, capsys):
|
|
from rtut.__main__ import main
|
|
base, _ = start(srv, pack)
|
|
st, _, body = raw(srv, f"{base}/record.json")
|
|
assert st == 409 and b"once the run is closed" in body # an open run has no final record
|
|
assert "Download the run record" not in req(srv, "GET", base)[2]
|
|
post(srv, f"{base}/finalize", outcome="abandoned", note="for the record")
|
|
st, hdr, body = raw(srv, f"{base}/record.json")
|
|
rid = base.rsplit("/", 1)[1]
|
|
assert st == 200 and hdr["content-type"] == "application/json" and hdr["x-rtut-seal"] == "sealed"
|
|
assert hdr["content-disposition"] == f'attachment; filename="run-{rid}.json"' and hdr["x-content-type-options"] == "nosniff"
|
|
assert "sandbox" in hdr["content-security-policy"]
|
|
assert body == run_of(srv, base).path.read_bytes() # exactly as stored
|
|
record = json.loads(body)
|
|
assert record["run_id"] == rid and record["outcome"] == "abandoned" and R1.record_seal_state(record) == "sealed"
|
|
assert f"<a href='{base}/record.json'>Download the run record (JSON)</a>" in req(srv, "GET", base)[2]
|
|
f = tmp_path / "exported.json"; f.write_bytes(body)
|
|
assert main(["seal", str(f)]) == 0 and capsys.readouterr().out.strip() == "sealed" # anyone can verify it
|
|
record["operator"] = "forged"; f.write_text(json.dumps(record))
|
|
assert main(["seal", str(f)]) == 1 and capsys.readouterr().out.strip() == "broken"
|
|
|
|
|
|
def test_a_tampered_record_is_still_downloadable_but_flagged_in_the_header(srv):
|
|
base, _ = start(srv, "rtut.showcase-lighthouse-v2")
|
|
post(srv, f"{base}/finalize", outcome="abandoned")
|
|
r = run_of(srv, base); r.path.chmod(0o644)
|
|
data = json.loads(r.path.read_text()); data["operator"] = "forged"; r.path.write_text(json.dumps(data))
|
|
st, hdr, _ = raw(srv, f"{base}/record.json")
|
|
assert st == 200 and hdr["x-rtut-seal"] == "broken" # evidence, with the warning attached
|
|
|
|
|
|
def test_unknown_run_record_is_not_found(srv):
|
|
assert raw(srv, "/run/0/nope/record.json")[0] == 404
|