tegwick
8d62c1fbe8
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
263 lines
14 KiB
Python
263 lines
14 KiB
Python
import http.client
|
|
import json
|
|
import shutil
|
|
import subprocess
|
|
import threading
|
|
import uuid
|
|
from datetime import datetime, timedelta, timezone
|
|
from pathlib import Path
|
|
from urllib.parse import urlencode
|
|
|
|
import pytest
|
|
import yaml
|
|
|
|
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")
|
|
(tmp_path / "projects/watch").mkdir(parents=True)
|
|
(tmp_path / "projects/watch/project.yaml").write_text(yaml.safe_dump({
|
|
"spec": "project/v0.1", "id": "watch", "title": "Watch project", "environment": "rehearsal",
|
|
"attribute_sets": ["user-attributes"], "attributes": {"logdir": "/tmp/rtut-r3-proj"},
|
|
"runbooks": [{"pack": "rtut.showcase-lighthouse"}]}))
|
|
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 raw(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(errors="replace"), {k.lower(): v for k, v in r.getheaders()}
|
|
|
|
|
|
def req(s, method, path, form=None):
|
|
return raw(s, method, path, form)[:3]
|
|
|
|
|
|
def post(s, path, **form):
|
|
return req(s, "POST", path, {"_t": s.app.token, **form})
|
|
|
|
|
|
def pidx(s, pack_id):
|
|
return s.app.pack_index(pack_id)
|
|
|
|
|
|
def new_v2(s, **extra):
|
|
logdir = f"/tmp/rtut-r3-{uuid.uuid4().hex[:8]}"
|
|
_scratch.append(logdir)
|
|
st, loc, _ = post(s, f"/run/new/{pidx(s, 'rtut.showcase-lighthouse-v2')}", p_logdir=logdir, **extra) # note: no operator posted
|
|
assert st == 303
|
|
return loc.split("#")[0], logdir
|
|
|
|
|
|
def step(s, base, sid, **form):
|
|
assert post(s, f"{base}/{sid}/start", **form)[0] == 303
|
|
run = R1.load(s.app.packs[int(base.split("/")[2])], base.rsplit("/", 1)[1]) # reload: the chosen alternative is known now
|
|
cmd = R.command_for(run, sid)
|
|
if cmd:
|
|
subprocess.run(cmd, shell=True, check=True, capture_output=True)
|
|
eff = R.step_def(run, sid)
|
|
manual = (eff["verify"].get("expect") or "manual") == "manual" and not eff["verify"].get("datapoints_ok")
|
|
assert (post(s, f"{base}/{sid}/verify", ok="1") if manual else post(s, f"{base}/{sid}/verify", yes="1"))[0] == 303
|
|
assert post(s, f"{base}/{sid}/done")[0] == 303
|
|
|
|
|
|
# ---- (a) the active user instead of asking for a name ----------------------------------------------------------
|
|
|
|
def test_user_page_shows_source_and_saves(srv):
|
|
_, _, body = req(srv, "GET", "/user")
|
|
assert "tester-login" in body and "your operating-system login" in body and "user-attributes" in body
|
|
assert post(srv, "/user", operator="bernd", display_name="Bernd W.")[0] == 303
|
|
_, _, body = req(srv, "GET", "/user")
|
|
assert "<b>bernd</b>" in body and "your saved settings" in body and "Bernd W." in body
|
|
st, _, body = post(srv, "/user", operator="two words")
|
|
assert st == 400 and "one word" in body
|
|
|
|
|
|
def test_forms_use_the_active_user_and_do_not_ask_for_a_name(srv):
|
|
post(srv, "/user", operator="bernd", display_name="Bernd W.")
|
|
i1, i2 = pidx(srv, "rtut.showcase-lighthouse"), pidx(srv, "rtut.showcase-lighthouse-v2")
|
|
for path in (f"/pack/{i1}", f"/pack/{i2}"):
|
|
_, _, body = req(srv, "GET", path)
|
|
assert "Running as <b>Bernd W. (bernd)</b>" in body and "<a href='/user'>change</a>" in body
|
|
assert "Your name (one word)" not in body and "<input name=operator" not in body
|
|
_, _, body = req(srv, "GET", "/project/watch")
|
|
assert "Start run as bernd" in body and "name=operator value='bernd'" in body
|
|
|
|
|
|
def test_a_run_started_without_a_posted_name_belongs_to_the_active_user_and_reviews_too(srv):
|
|
post(srv, "/user", operator="carol")
|
|
base, _ = new_v2(srv)
|
|
run = R1.load(srv.app.packs[pidx(srv, "rtut.showcase-lighthouse-v2")], base.rsplit("/", 1)[1])
|
|
assert run.data["operator"] == "carol"
|
|
i1 = pidx(srv, "rtut.showcase-lighthouse")
|
|
assert post(srv, f"/pack/{i1}/review", note="ok")[0] == 303
|
|
_, _, body = req(srv, "GET", f"/pack/{i1}")
|
|
assert "carol" in body and "Reviewed: <b>yes</b>" in body
|
|
|
|
|
|
def test_user_attributes_are_reusable_by_projects_through_the_app_level_set(srv):
|
|
post(srv, "/user", operator="dave")
|
|
_, _, body = req(srv, "GET", "/project/watch") # the project lists attribute_sets: [user-attributes]
|
|
assert "This project has problems" not in body and "Watch project" in body
|
|
from rtut import project as J
|
|
reg = srv.app.registry()
|
|
assert J.validate(reg, reg.projects["watch"]) == [] # the set is found although no repository defines it
|
|
member = reg.projects["watch"].data["runbooks"][0]
|
|
res = J.member_attributes(reg, reg.projects["watch"].data, member)
|
|
assert "operator" in res["unused"] # v0.1 showcase has no operator input: reported, harmless
|
|
assert "Unused attributes: " in body and "operator" in body.split("Unused attributes: ")[1].split("</p>")[0]
|
|
|
|
|
|
# ---- (b) the irreversible example is irreversible -------------------------------------------------------------------
|
|
|
|
def test_the_showcase_logbook_entry_carries_the_time_and_close_log_says_why_it_cannot_be_undone(srv):
|
|
base, logdir = new_v2(srv)
|
|
run = R1.load(srv.app.packs[pidx(srv, "rtut.showcase-lighthouse-v2")], base.rsplit("/", 1)[1])
|
|
assert "$(date -u" in run.pack.data["stages"][1]["steps"][1]["command"]
|
|
_, _, body = req(srv, "GET", base)
|
|
assert "cannot be undone" in body and "to the second" in body
|
|
|
|
|
|
# ---- (c) latest activities ---------------------------------------------------------------------------------------------
|
|
|
|
def test_main_page_lists_latest_activities_newest_first_with_links(srv):
|
|
_, _, body = req(srv, "GET", "/")
|
|
assert "Latest activities" in body and "Nothing has happened yet" in body
|
|
post(srv, "/user", operator="erin")
|
|
base, _ = new_v2(srv, p_weather="fog")
|
|
step(srv, base, "read-the-sign")
|
|
step(srv, base, "light-lamp")
|
|
_, _, body = req(srv, "GET", "/")
|
|
feed = body.split("Latest activities")[1].split("<h2>Runbooks</h2>")[0]
|
|
assert "<b>erin</b> started a run (manual mode)" in feed and "finished step “Light the lamp”" in feed
|
|
assert feed.index("Light the lamp") < feed.index("Read the poem over the door") < feed.index("started a run") # newest first
|
|
assert f"<a href='{base}'>open</a>" in feed
|
|
|
|
|
|
# ---- (d) changing values that steps already used --------------------------------------------------------------------------
|
|
|
|
def test_changing_a_value_in_use_needs_an_override_shows_the_lock_and_is_recorded(srv):
|
|
base, _ = new_v2(srv, p_weather="fog")
|
|
_, _, body = req(srv, "GET", base + "?view=data")
|
|
assert "in use by" not in body and "Changing a value that steps already used" not in body # nothing used yet: no lock, no override
|
|
assert post(srv, f"{base}/data", d_weather="clear")[0] == 303 # free change before use
|
|
step(srv, base, "read-the-sky")
|
|
_, _, body = req(srv, "GET", base + "?view=data")
|
|
assert "in use by read-the-sky" in body and "Changing a value that steps already used (override)" in body
|
|
st, _, err = post(srv, f"{base}/data", d_weather="storm")
|
|
assert st == 400 and "already in use" in err and "read-the-sky" in err
|
|
st, _, err = post(srv, f"{base}/data", d_weather="storm", override_reason="forecast changed")
|
|
assert st == 400 and "acknowledge" in err
|
|
run = R1.load(srv.app.packs[pidx(srv, "rtut.showcase-lighthouse-v2")], base.rsplit("/", 1)[1])
|
|
assert run.data["params"]["weather"] == "clear" and run.data["deviations"] == [] # refused: nothing changed
|
|
assert post(srv, f"{base}/data", d_weather="storm", override_reason="forecast changed", acknowledge_risk="1")[0] == 303
|
|
_, _, ov = req(srv, "GET", base + "?view=overview")
|
|
assert ">changed-after-use<" in ov and "forecast changed" in ov and "read-the-sky already used weather" in ov
|
|
|
|
|
|
def test_a_refused_save_applies_none_of_its_changes(srv):
|
|
base, _ = new_v2(srv, p_weather="fog")
|
|
step(srv, base, "read-the-sky")
|
|
st, _, _ = post(srv, f"{base}/data", d_weather="storm", d_motto="Hold fast") # weather is locked, motto is free
|
|
assert st == 400
|
|
run = R1.load(srv.app.packs[pidx(srv, "rtut.showcase-lighthouse-v2")], base.rsplit("/", 1)[1])
|
|
assert run.data["params"]["motto"] != "Hold fast" and run.data["params"]["weather"] == "fog"
|
|
|
|
|
|
def test_in_auto_confirm_mode_used_values_cannot_be_changed_until_taking_over(srv):
|
|
base, _ = new_v2(srv, mode="auto-confirm", p_weather="fog")
|
|
post(srv, f"{base}/tick", n="3")
|
|
run = R1.load(srv.app.packs[pidx(srv, "rtut.showcase-lighthouse-v2")], base.rsplit("/", 1)[1])
|
|
if not R.datapoint_users(run, "weather"):
|
|
run.data["steps"]["read-the-sky"]["state"] = "done"; run.save()
|
|
_, _, body = req(srv, "GET", base + "?view=data")
|
|
assert "can only be changed in manual mode" in body and "name=override_reason" not in body
|
|
st, _, err = post(srv, f"{base}/data", d_weather="clear", override_reason="x", acknowledge_risk="1")
|
|
assert st == 400 and "take over" in err
|
|
|
|
|
|
# ---- (e) artefacts can be opened and downloaded ---------------------------------------------------------------------------
|
|
|
|
def finish_logbook(srv, base):
|
|
for sid in ("read-the-sign", "light-lamp", "read-the-sky"):
|
|
step(srv, base, sid)
|
|
for i in (0, 1):
|
|
ck = post(srv, f"{base}/gate/lamp-check/check", idx=str(i), ok="1") if i == 0 else post(srv, f"{base}/gate/lamp-check/check", idx="1", yes="1")
|
|
assert ck[0] == 303
|
|
step(srv, base, "warn")
|
|
step(srv, base, "write-logbook")
|
|
|
|
|
|
def test_artefact_is_linked_and_can_be_opened_and_downloaded(srv):
|
|
base, logdir = new_v2(srv, p_weather="fog")
|
|
_, _, body = req(srv, "GET", base + "?view=data")
|
|
assert "the file does not exist (yet)" in body and "artefact/logbook-file" not in body # nothing to open before it exists
|
|
assert raw(srv, "GET", f"{base}/artefact/logbook-file")[0] == 403
|
|
finish_logbook(srv, base)
|
|
_, _, body = req(srv, "GET", base + "?view=data")
|
|
assert f"<a href='{base}/artefact/logbook-file' target=_blank rel=noopener>Open</a>" in body and "?download=1'>Download</a>" in body
|
|
st, _, text, hdr = raw(srv, "GET", f"{base}/artefact/logbook-file")
|
|
assert st == 200 and "Night watch by Ada at " in text and hdr["content-type"].startswith("text/plain")
|
|
assert hdr["x-content-type-options"] == "nosniff" and "sandbox" in hdr["content-security-policy"] and "content-disposition" not in hdr
|
|
st, _, text2, hdr = raw(srv, "GET", f"{base}/artefact/logbook-file?download=1")
|
|
assert st == 200 and text2 == text and 'attachment; filename="artefact-logbook-file"' in hdr["content-disposition"]
|
|
|
|
|
|
def test_artefact_route_refuses_unknown_ids_secrets_and_v01_runs(srv):
|
|
base, logdir = new_v2(srv, p_weather="fog")
|
|
finish_logbook(srv, base)
|
|
assert raw(srv, "GET", f"{base}/artefact/nope")[0] == 404
|
|
(Path(logdir) / "logbook.txt").write_text("token " + "hvs." + "ABCDEFGHIJKLMN")
|
|
st, _, body, _ = raw(srv, "GET", f"{base}/artefact/logbook-file")
|
|
assert st == 403 and "secret" in body
|
|
_, _, dbody = req(srv, "GET", base + "?view=data")
|
|
assert "looks like it contains a secret" in dbody and "artefact/logbook-file" not in dbody
|
|
st, loc, _ = post(srv, f"/run/new/{pidx(srv, 'rtut.showcase-lighthouse')}", p_logdir="/tmp/rtut-r3-v1")
|
|
assert raw(srv, "GET", loc.split("#")[0] + "/artefact/logbook-file")[0] == 404
|
|
|
|
|
|
# ---- (f) timestamps, durations and the average ---------------------------------------------------------------------------------
|
|
|
|
def test_run_shows_start_finish_and_duration_and_steps_show_how_long_they_took(srv):
|
|
base, _ = new_v2(srv, p_weather="fog")
|
|
_, _, body = req(srv, "GET", base)
|
|
assert "Started 20" in body and "UTC · running for <b>" in body
|
|
step(srv, base, "read-the-sign")
|
|
_, _, body = req(srv, "GET", base)
|
|
assert "took 0 s" in body or "took 1 s" in body
|
|
assert post(srv, f"{base}/finalize", outcome="abandoned")[0] == 303
|
|
_, _, body = req(srv, "GET", base)
|
|
assert "finished 20" in body and "· took <b>" in body and "working" in body
|
|
|
|
|
|
def test_pack_page_and_list_show_the_average_duration_of_successful_runs(srv):
|
|
i1 = pidx(srv, "rtut.showcase-lighthouse")
|
|
pack = srv.app.packs[i1]
|
|
for secs, who in ((300, "a"), (900, "b")):
|
|
r = R1.create(pack, who)
|
|
start = datetime(2026, 9, 29, 10, 0, tzinfo=timezone.utc)
|
|
r.data.update(created=start.isoformat(), outcome="succeeded", finished=(start + timedelta(seconds=secs)).isoformat())
|
|
r.path.write_text(json.dumps(r.data))
|
|
slow = R1.create(pack, "c")
|
|
slow.data.update(created="2026-09-29T10:00:00+00:00", outcome="failed", finished="2026-09-29T20:00:00+00:00")
|
|
slow.path.write_text(json.dumps(slow.data))
|
|
_, _, body = req(srv, "GET", f"/pack/{i1}")
|
|
assert "average 10 min per successful run (fastest 5 min, 2 timed)" in body # the failed 10-hour run is not in the average
|
|
_, _, body = req(srv, "GET", "/")
|
|
assert "average 10 min per successful run" in body
|
|
assert "took 5 min" in req(srv, "GET", f"/pack/{i1}")[2] # the runs list shows each duration
|