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
247 lines
13 KiB
Python
247 lines
13 KiB
Python
import http.client
|
|
import shutil
|
|
import threading
|
|
from pathlib import Path
|
|
from urllib.parse import urlencode
|
|
|
|
import pytest
|
|
import yaml
|
|
|
|
from rtut import project as J, projectrun as PR, run as R1, run2, ui
|
|
|
|
ROOT = Path(__file__).resolve().parent.parent
|
|
|
|
|
|
def v2_pack(pid, manual_only_last=False):
|
|
steps = [{"id": "first", "title": "First step", "actor": "op", "owner": "t", "uses": ["site"], "command": "echo {{site}}",
|
|
"verify": {"done_when": "you saw the site", "expect": "manual"}},
|
|
{"id": "second", "title": "Second step", "actor": "op", "owner": "t", "verify": {"done_when": "second done", "expect": "manual"}}]
|
|
if manual_only_last:
|
|
steps.append({"id": "careful", "title": "Careful step", "actor": "op", "owner": "t", "auto_confirm": "manual-only",
|
|
"verify": {"done_when": "you did it", "expect": "manual"}})
|
|
return {"spec": "runbook-pack/v0.2", "id": pid, "title": f"Pack {pid}", "owner": "t", "engine": "native",
|
|
"framing": ["keep in mind"], "actors": [{"id": "op", "role": "Operator", "description": "does it"}],
|
|
"datapoints": [{"id": "site", "label": "Site", "type": "string", "pattern": "[a-z]+", "default": "north", "ask_in": "main"}],
|
|
"stages": [{"id": "main", "title": "Main", "goal": "do the things", "steps": steps}]}
|
|
|
|
|
|
def write(path, data):
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(yaml.safe_dump(data, sort_keys=False))
|
|
|
|
|
|
@pytest.fixture
|
|
def srv(tmp_path):
|
|
write(tmp_path / "runbooks/alpha/pack.yaml", v2_pack("t.alpha"))
|
|
write(tmp_path / "runbooks/beta/pack.yaml", v2_pack("t.beta", manual_only_last=True))
|
|
write(tmp_path / "projects/demo/project.yaml", {
|
|
"spec": "project/v0.1", "id": "demo", "title": "Demo project", "environment": "rehearsal", "attributes": {"site": "south"},
|
|
"runbooks": [{"pack": "t.alpha"}, {"pack": "t.beta", "after": ["t.alpha"]}]})
|
|
s = ui.serve([tmp_path], 0)
|
|
threading.Thread(target=s.serve_forever, daemon=True).start()
|
|
s.repo = tmp_path
|
|
yield s
|
|
s.shutdown()
|
|
|
|
|
|
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_pr(s, mode="manual"):
|
|
st, loc, _ = post(s, "/project/demo/runall", mode=mode)
|
|
assert st == 303 and loc.startswith("/project-run/")
|
|
return loc
|
|
|
|
|
|
def finish_run(s, run_url):
|
|
"""Complete and close the member run behind a /run/<i>/<id> URL, as an operator would in that page."""
|
|
parts = run_url.split("#")[0].split("/")
|
|
i, rid = int(parts[2]), parts[3]
|
|
run = R1.load(s.app.packs[i], rid)
|
|
for sid in run2.flow(run).steps:
|
|
assert post(s, f"/run/{i}/{rid}/{sid}/start")[0] == 303
|
|
assert post(s, f"/run/{i}/{rid}/{sid}/verify", ok="1")[0] == 303
|
|
assert post(s, f"/run/{i}/{rid}/{sid}/done")[0] == 303
|
|
assert post(s, f"/run/{i}/{rid}/finalize", outcome="succeeded")[0] == 303
|
|
|
|
|
|
# ---- starting -------------------------------------------------------------------------------------------------------------
|
|
|
|
def test_project_page_offers_the_whole_project_run_with_both_modes(srv):
|
|
_, _, body = req(srv, "GET", "/project/demo")
|
|
assert "<h2>Run the whole project</h2>" in body and "name=mode value='manual' checked" in body and "name=mode value='auto-confirm'" in body
|
|
assert "<b>manual</b>: You start each member yourself" in body and "<b>auto-confirm</b>: The engine starts the next ready member" in body
|
|
assert "Running as <b>tester-login</b>" in body and "Start project run" in body and "snapshotted" in body
|
|
|
|
|
|
def test_missing_attributes_block_the_start_and_say_which(srv):
|
|
p = v2_pack("t.alpha"); p["datapoints"][0].pop("default")
|
|
write(srv.repo / "runbooks/alpha/pack.yaml", p)
|
|
d = yaml.safe_load((srv.repo / "projects/demo/project.yaml").read_text()); d["attributes"] = {}
|
|
write(srv.repo / "projects/demo/project.yaml", d)
|
|
_, _, body = req(srv, "GET", "/project/demo")
|
|
assert "Give the project the attributes it still needs first: site" in body and "Start project run" not in body
|
|
st, _, err = post(srv, "/project/demo/runall", mode="manual")
|
|
assert st == 400 and "attributes still needed: site" in err
|
|
|
|
|
|
# ---- manual --------------------------------------------------------------------------------------------------------------------
|
|
|
|
def test_manual_project_run_page_shows_next_member_and_walks_them_in_order(srv):
|
|
loc = start_pr(srv)
|
|
_, _, body = req(srv, "GET", loc)
|
|
assert "<h1>Project run: Demo project</h1>" in body and "0 of 2 members done" in body
|
|
assert "<b>Next:</b> t.alpha" in body and "Manual mode." in body and "Hand over to auto-confirm" in body
|
|
assert "waiting for t.alpha" in body and "Start anyway" in body and ">ready<" in body and ">waiting<" in body
|
|
st, run_loc, _ = post(srv, f"{loc}/start", key="t.alpha")
|
|
assert st == 303 and run_loc.startswith("/run/") and run_loc.endswith("#current")
|
|
_, _, runpage = req(srv, "GET", run_loc.split("#")[0])
|
|
assert f"part of <a href='{loc}'>project run" in runpage and "project <a href='/project/demo'>demo</a>" in runpage
|
|
_, _, body = req(srv, "GET", loc)
|
|
assert "<b>Continue:</b> t.alpha" in body and "open run" in body and ">running<" in body
|
|
assert "<button>Start</button>" not in body # one member at a time
|
|
finish_run(srv, run_loc)
|
|
_, _, body = req(srv, "GET", loc)
|
|
assert "1 of 2 members done" in body and "<b>Next:</b> t.beta" in body and ">succeeded<" in body
|
|
|
|
|
|
def test_out_of_order_start_needs_reason_and_acknowledgement_and_shows_as_a_deviation(srv):
|
|
loc = start_pr(srv)
|
|
st, _, err = post(srv, f"{loc}/start", key="t.beta")
|
|
assert st == 400 and "waits for t.alpha" in err
|
|
st, _, err = post(srv, f"{loc}/start", key="t.beta", override_reason="beta first")
|
|
assert st == 400 and "acknowledge" in err
|
|
assert post(srv, f"{loc}/start", key="t.beta", override_reason="beta first", acknowledge_risk="1")[0] == 303
|
|
_, _, body = req(srv, "GET", loc)
|
|
assert "⚑ Deviations: override t.beta (beta first)" in body and "started t.beta (override)" in body
|
|
|
|
|
|
def test_only_one_project_run_is_active_and_the_page_links_to_it(srv):
|
|
loc = start_pr(srv)
|
|
_, _, body = req(srv, "GET", "/project/demo")
|
|
assert f"<a href='{loc}'>continue it</a>" in body and "Start project run" not in body
|
|
st, _, err = post(srv, "/project/demo/runall", mode="manual")
|
|
assert st == 400 and "still open" in err
|
|
|
|
|
|
# ---- auto-confirm -----------------------------------------------------------------------------------------------------------------
|
|
|
|
def test_auto_confirm_runs_members_until_a_person_is_needed_then_continues(srv):
|
|
loc = start_pr(srv, "auto-confirm")
|
|
_, _, body = req(srv, "GET", loc)
|
|
assert "Auto-confirm mode." in body and "Run until it waits" in body and "Take over (manual)" in body and "never runs commands" in body
|
|
assert "<button>Start</button>" not in body # the engine starts members, not you
|
|
assert post(srv, f"{loc}/tick", n="300")[0] == 303
|
|
_, _, body = req(srv, "GET", loc)
|
|
assert "1 of 2 members done" in body and "Last: awaiting-operator t.beta" in body and "manual-only" in body
|
|
pr = PR.load(srv.app.registry(), loc.rsplit("/", 1)[1])
|
|
beta = R1.load(srv.app.registry().packs["t.beta"], next(m["run_id"] for m in pr.data["members"] if m["pack"] == "t.beta"))
|
|
assert beta.data["mode"] == "auto-confirm" and beta.data["context"]["project_run"] == pr.id
|
|
i = srv.app.pack_index("t.beta")
|
|
assert post(srv, f"/run/{i}/{beta.data['run_id']}/careful/start")[0] == 303 # the person takes the careful step
|
|
assert post(srv, f"/run/{i}/{beta.data['run_id']}/careful/verify", ok="1")[0] == 303
|
|
assert post(srv, f"/run/{i}/{beta.data['run_id']}/careful/done")[0] == 303
|
|
assert post(srv, f"{loc}/tick", n="300")[0] == 303
|
|
_, _, body = req(srv, "GET", loc)
|
|
assert "2 of 2 members done" in body and "Finish: succeeded" in body
|
|
|
|
|
|
def test_take_over_and_hand_over(srv):
|
|
loc = start_pr(srv, "auto-confirm")
|
|
assert post(srv, f"{loc}/mode", to="manual")[0] == 303
|
|
_, _, body = req(srv, "GET", loc)
|
|
assert "Manual mode." in body and "Hand over to auto-confirm" in body and "switched to manual mode" in body
|
|
st, _, err = post(srv, f"{loc}/mode", to="manual")
|
|
assert st == 400 and "already in manual" in err
|
|
|
|
|
|
# ---- closing and locking ------------------------------------------------------------------------------------------------------------
|
|
|
|
def test_closing_the_project_run_locks_it_and_abandons_the_open_member_run(srv):
|
|
loc = start_pr(srv)
|
|
post(srv, f"{loc}/start", key="t.alpha")
|
|
_, _, body = req(srv, "GET", loc)
|
|
assert "Finish: succeeded" not in body and "Abandon project run" in body and "Finish: failed" in body
|
|
st, _, err = post(srv, f"{loc}/finalize", outcome="succeeded")
|
|
assert st == 400 and "succeeds only when" in err
|
|
assert post(srv, f"{loc}/finalize", outcome="abandoned", note="not now")[0] == 303
|
|
_, _, body = req(srv, "GET", loc)
|
|
assert "Project run finished: abandoned" in body and "Comment: not now" in body and "closed and locked" in body
|
|
assert "Close this project run" not in body and "<button>Start</button>" not in body and "Hand over" not in body
|
|
reg = srv.app.registry()
|
|
pr = PR.load(reg, loc.rsplit("/", 1)[1])
|
|
member = R1.load(reg.packs["t.alpha"], pr.data["members"][0]["run_id"])
|
|
assert member.data["outcome"] == "abandoned"
|
|
for path, form in ((f"{loc}/start", {"key": "t.beta"}), (f"{loc}/tick", {"n": "1"}), (f"{loc}/mode", {"to": "auto-confirm"}), (f"{loc}/finalize", {"outcome": "abandoned"})):
|
|
st, _, err = post(srv, path, **form)
|
|
assert st == 400, path
|
|
_, _, body = req(srv, "GET", "/project/demo")
|
|
assert "Start project run" in body and f"<a href='{loc}'>" in body # a new one may start; the old one is in the history
|
|
|
|
|
|
def test_a_tampered_closed_project_run_is_called_out(srv):
|
|
import json
|
|
loc = start_pr(srv)
|
|
post(srv, f"{loc}/finalize", outcome="abandoned")
|
|
pr = PR.load(srv.app.registry(), loc.rsplit("/", 1)[1])
|
|
pr.path.chmod(0o644)
|
|
d = json.loads(pr.path.read_text()); d["operator"] = "forged"; pr.path.write_text(json.dumps(d))
|
|
assert "changed outside the app after it was sealed" in req(srv, "GET", loc)[2]
|
|
|
|
|
|
# ---- history, activity, snapshot, safety --------------------------------------------------------------------------------------------
|
|
|
|
def test_history_table_and_activity_feed_include_the_project_run(srv):
|
|
loc = start_pr(srv)
|
|
post(srv, f"{loc}/start", key="t.alpha")
|
|
_, _, body = req(srv, "GET", "/project/demo")
|
|
assert "<h3>Project runs</h3>" in body and ">in progress<" in body and "0 of 2 members" in body
|
|
_, _, act = req(srv, "GET", "/activity")
|
|
assert "project run: started the project run (manual mode)" in act and "project run: started t.alpha" in act and f"href='{loc}'" in act
|
|
|
|
|
|
def test_attribute_changes_after_the_start_are_shown_and_do_not_reach_later_members(srv):
|
|
loc = start_pr(srv)
|
|
post(srv, f"{loc}/start", key="t.alpha")
|
|
reg = srv.app.registry()
|
|
J.set_attributes(reg, reg.projects["demo"], {"site": "east"})
|
|
_, _, body = req(srv, "GET", loc)
|
|
assert "attributes have changed since this project run started" in body and "keeps the values it started with" in body
|
|
pr = PR.load(srv.app.registry(), loc.rsplit("/", 1)[1])
|
|
assert pr.data["snapshot"]["t.beta"] == {"site": "south"}
|
|
|
|
|
|
def test_sidebar_marks_projects_on_a_project_run_page_and_csrf_and_unknown_ids(srv):
|
|
loc = start_pr(srv)
|
|
_, _, body = req(srv, "GET", loc)
|
|
assert "'/project-run/']" in body # the alias list includes project runs
|
|
assert req(srv, "POST", f"{loc}/tick", {"n": "1"})[0] == 403
|
|
assert req(srv, "POST", "/project/demo/runall", {"mode": "manual"})[0] == 403
|
|
assert req(srv, "GET", "/project-run/nope")[0] == 404
|
|
assert post(srv, "/project-run/nope/tick", n="1")[0] == 400
|
|
|
|
|
|
def test_a_closed_project_run_record_can_be_downloaded(srv):
|
|
import json
|
|
from rtut import run as RR
|
|
loc = start_pr(srv)
|
|
c = http.client.HTTPConnection("127.0.0.1", srv.server_address[1])
|
|
c.request("GET", f"{loc}/record.json", None, {"Host": "127.0.0.1"})
|
|
assert c.getresponse().status == 409 # open: no final record yet
|
|
post(srv, f"{loc}/finalize", outcome="abandoned")
|
|
assert f"<a href='{loc}/record.json'>Download the project run record (JSON)</a>" in req(srv, "GET", loc)[2]
|
|
c = http.client.HTTPConnection("127.0.0.1", srv.server_address[1])
|
|
c.request("GET", f"{loc}/record.json", None, {"Host": "127.0.0.1"})
|
|
r = c.getresponse(); body = r.read()
|
|
assert r.status == 200 and r.getheader("X-Rtut-Seal") == "sealed" and "attachment" in r.getheader("Content-Disposition")
|
|
rec = json.loads(body)
|
|
assert rec["project"] == "demo" and rec["outcome"] == "abandoned" and RR.record_seal_state(rec) == "sealed"
|