coulomb/runbook-tutorials
5
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Run a whole project: project runs in manual and auto-confirm mode (RBT-WP-0008-T10)
Some checks failed
CI Smoke / host-smoke (push) Successful in 0s
Details
CI Smoke / container-smoke (push) Successful in 3s
Details
Work Records / validate (push) Has been cancelled
Details

Browse source 
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
This commit is contained in:
tegwick 2026-09-29 20:08:17 +02:00
parent 1d1087fa7f
commit eff75a303c
14 changed files with 1112 additions and 15 deletions
Download patch file Download diff file Expand all files Collapse all files

264
tests/test_projectrun.py Normal file
View file

@ -0,0 +1,264 @@
import copy
import shutil
import stat
from datetime import datetime, timedelta, timezone
from pathlib import Path
import pytest
import yaml
from rtut import project as J, projectrun as PR, run as R1, run2
from rtut.attrs import Registry
ROOT = Path(__file__).resolve().parent.parent
def v2_pack(pid, extra_step=None):
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 extra_step:
steps.append(extra_step)
return {"spec": "runbook-pack/v0.2", "id": pid, "title": f"Pack {pid}", "owner": "t", "exercise_status": "unexercised", "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))
def make_repo(tmp_path, members=None, v1=False, alpha_extra=None, attrs=None):
write(tmp_path / "runbooks/alpha/pack.yaml", v2_pack("t.alpha", alpha_extra))
write(tmp_path / "runbooks/beta/pack.yaml", v2_pack("t.beta"))
if v1:
shutil.copy(ROOT / "runbooks/example-hello/pack.yaml", tmp_path / "runbooks/hello.yaml")
(tmp_path / "runbooks/hello").mkdir()
(tmp_path / "runbooks/hello.yaml").rename(tmp_path / "runbooks/hello/pack.yaml")
members = members or [{"pack": "t.alpha"}, {"pack": "t.beta", "after": ["t.alpha"]}]
write(tmp_path / "projects/demo/project.yaml", {"spec": "project/v0.1", "id": "demo", "title": "Demo project", "environment": "rehearsal",
"attributes": attrs or {"site": "south"}, "runbooks": members})
return Registry([tmp_path])
def do_all_manual(run):
for sid in run2.flow(run).steps:
run2.start(run, sid); run2.verify(run, sid, manual_ok=True); run2.complete(run, sid)
def finish_member(reg, doc, pr, key, outcome="succeeded"):
row = next(m for m in pr.data["members"] if m["key"] == key)
run = R1.load(reg.packs[row["pack"]], row["run_id"])
do_all_manual(run)
R1.finalize(run, outcome)
# ---- create ------------------------------------------------------------------------------------------------------------
def test_create_snapshots_attributes_records_events_and_refuses_bad_starts(tmp_path):
reg = make_repo(tmp_path)
doc = reg.projects["demo"]
pr = PR.create(reg, doc, "bernd")
assert pr.data["mode"] == "manual" and pr.data["snapshot"]["t.alpha"] == {"site": "south"} and pr.data["events"][0]["text"].startswith("started")
assert pr.path.parent.name == "project-runs" and PR.active_run(reg, "demo").id == pr.id
with pytest.raises(R1.RunError, match="still open"):
PR.create(reg, doc, "bernd") # one active project run at a time
with pytest.raises(R1.RunError, match="mode must be"):
PR.create(make_repo(tmp_path / "x"), Registry([tmp_path / "x"]).projects["demo"], "bernd", "autopilot")
with pytest.raises(R1.RunError, match="template is not run"):
PR.create(reg, Registry([ROOT]).templates["lighthouse-watch"], "bernd")
def test_missing_required_attributes_refuse_the_start(tmp_path):
p = v2_pack("t.alpha"); p["datapoints"][0].pop("default")
write(tmp_path / "runbooks/alpha/pack.yaml", p)
write(tmp_path / "projects/demo/project.yaml", {"spec": "project/v0.1", "id": "demo", "title": "Demo project", "environment": "rehearsal",
"runbooks": [{"pack": "t.alpha"}]})
reg = Registry([tmp_path])
with pytest.raises(R1.RunError, match="attributes still needed: site"):
PR.create(reg, reg.projects["demo"], "bernd")
# ---- manual mode ------------------------------------------------------------------------------------------------------------
def test_manual_project_run_goes_member_by_member_in_after_order(tmp_path):
reg = make_repo(tmp_path); doc = reg.projects["demo"]
pr = PR.create(reg, doc, "bernd")
v = PR.view(reg, pr)
assert [r["status"] for r in v["members"]] == ["ready", "waiting"] and v["members"][1]["blocked_by"] == ["t.alpha"] and v["ready"] == ["t.alpha"]
with pytest.raises(R1.RunError, match="waits for t.alpha"):
PR.start_member(reg, doc, pr, "t.beta") # out of order: refused with the reason
run = PR.start_member(reg, doc, pr, "t.alpha")
assert run.data["context"] == {"project": "demo", "environment": "rehearsal", "project_run": pr.id}
assert run.data["params"]["site"] == "south" and run.data["mode"] == "manual"
with pytest.raises(R1.RunError, match="already has a run"):
PR.start_member(reg, doc, pr, "t.alpha")
with pytest.raises(R1.RunError, match="finish or abandon the open run"):
PR.start_member(reg, doc, pr, "t.beta", override_reason="x", acknowledge_risk=True)
finish_member(reg, doc, pr, "t.alpha")
v = PR.view(reg, pr)
assert [r["status"] for r in v["members"]] == ["succeeded", "ready"] and v["done"] == 1 and v["open"] is None
PR.start_member(reg, doc, pr, "t.beta")
finish_member(reg, doc, pr, "t.beta")
assert PR.view(reg, pr)["done"] == 2
PR.finalize(reg, pr, "succeeded", "all good")
assert pr.data["outcome"] == "succeeded" and pr.data["events"][-1]["text"] == "closed the project run as succeeded"
def test_starting_out_of_order_needs_a_reason_and_acknowledgement_and_is_recorded(tmp_path):
reg = make_repo(tmp_path); doc = reg.projects["demo"]
pr = PR.create(reg, doc, "bernd")
with pytest.raises(R1.RunError, match="acknowledge"):
PR.start_member(reg, doc, pr, "t.beta", override_reason="testing beta first")
PR.start_member(reg, doc, pr, "t.beta", override_reason="testing beta first", acknowledge_risk=True)
dev = pr.data["deviations"][0]
assert dev["kind"] == "override" and dev["item"] == "t.beta" and dev["unmet"] == ["t.alpha is not done"] and dev["reason"] == "testing beta first"
assert pr.data["events"][-1]["text"] == "started t.beta (override)"
def test_snapshot_keeps_member_runs_consistent_when_the_project_is_edited_meanwhile(tmp_path):
reg = make_repo(tmp_path); doc = reg.projects["demo"]
pr = PR.create(reg, doc, "bernd")
PR.start_member(reg, doc, pr, "t.alpha"); finish_member(reg, doc, pr, "t.alpha")
J.set_attributes(reg, doc, {"site": "east"}) # the project changes while the project run is open
reg = Registry([tmp_path]); doc = reg.projects["demo"]
assert PR.attributes_changed(reg, doc, pr) == ["t.alpha", "t.beta"]
run = PR.start_member(reg, doc, pr, "t.beta")
assert run.data["params"]["site"] == "south" # the value the project run started with
assert J.member_values(reg, doc.data, doc.data["runbooks"][1])["site"] == "east"
# ---- auto-confirm --------------------------------------------------------------------------------------------------------------
def test_auto_confirm_runs_the_whole_project_member_by_member_and_closes_each_run(tmp_path):
reg = make_repo(tmp_path); doc = reg.projects["demo"]
pr = PR.create(reg, doc, "bernd", "auto-confirm")
last = PR.run_until_waits(reg, doc, pr)
assert last["action"] == "finished"
v = PR.view(reg, pr)
assert [r["status"] for r in v["members"]] == ["succeeded", "succeeded"] and v["done"] == 2
a = R1.load(reg.packs["t.alpha"], v["members"][0]["run_id"])
assert a.data["mode"] == "auto-confirm" and a.data["outcome_note"] == "closed by auto-confirm" and R1.seal_state(a) == "sealed"
assert [e["text"] for e in pr.data["events"] if "started" in e["text"] or "finished" in e["text"]][:3] == [
"started the project run (auto-confirm mode)", "started t.alpha", "t.alpha finished (succeeded)"]
PR.finalize(reg, pr, "succeeded")
def test_auto_confirm_stops_at_a_manual_member_and_continues_after_the_operator_finishes_it(tmp_path):
reg = make_repo(tmp_path, [{"pack": "t.alpha"}, {"pack": "rtut.example-hello", "after": ["t.alpha"]}, {"pack": "t.beta", "after": ["rtut.example-hello"]}], v1=True,
attrs={"site": "south", "host": "localhost"})
doc = reg.projects["demo"]
pr = PR.create(reg, doc, "bernd", "auto-confirm")
last = PR.run_until_waits(reg, doc, pr)
assert last["action"] == "awaiting-operator" and last["member"] == "rtut.example-hello" and "manual mode" in last["detail"]
hello = R1.load(reg.packs["rtut.example-hello"], next(m["run_id"] for m in pr.data["members"] if m["pack"] == "rtut.example-hello"))
assert hello.pack.spec == "runbook-pack/v0.1" and "mode" not in hello.data # a v0.1 run is always manual: it cannot be auto-confirmed
R1.start(hello, "ping"); R1.verify(hello, "ping", confirm=True); R1.complete(hello, "ping"); R1.finalize(hello, "succeeded")
assert PR.run_until_waits(reg, doc, pr)["action"] == "finished"
assert [r["status"] for r in PR.view(reg, pr)["members"]] == ["succeeded"] * 3
def test_auto_confirm_stops_when_a_member_needs_a_person(tmp_path):
step = {"id": "careful", "title": "Careful step", "actor": "op", "owner": "t", "auto_confirm": "manual-only",
"verify": {"done_when": "you did it", "expect": "manual"}}
reg = make_repo(tmp_path, alpha_extra=step); doc = reg.projects["demo"]
pr = PR.create(reg, doc, "bernd", "auto-confirm")
last = PR.run_until_waits(reg, doc, pr)
assert last["action"] == "awaiting-operator" and last["member"] == "t.alpha" and "manual-only" in last["detail"]
assert PR.view(reg, pr)["members"][0]["status"] == "running" and PR.view(reg, pr)["members"][1]["status"] == "waiting"
def test_a_pushed_through_member_lets_dependents_proceed_but_the_project_cannot_succeed(tmp_path):
step = {"id": "shaky", "title": "Shaky step", "actor": "op", "owner": "t", "timeout": {"seconds": 1, "on_timeout": "push-through"},
"verify": {"done_when": "never true", "command": "false", "expect": "exit-0"}}
reg = make_repo(tmp_path, alpha_extra=step); doc = reg.projects["demo"]
pr = PR.create(reg, doc, "bernd", "auto-confirm")
for _ in range(40):
last = PR.tick(reg, doc, pr)
if last["action"] == "waiting":
run = R1.load(reg.packs["t.alpha"], pr.data["members"][0]["run_id"])
run.data["steps"]["shaky"]["started"] = (datetime.now(timezone.utc) - timedelta(seconds=5)).isoformat(); run.save()
if last["action"] in ("finished", "blocked", "awaiting-operator", "stopped"):
break
v = PR.view(reg, pr)
assert v["members"][0]["status"] == "pushed" and v["members"][0]["deviations"] >= 1 # closed as failed, but pushed through
assert v["members"][1]["status"] in ("succeeded", "running", "ready") # beta was allowed to proceed
assert v["problems"] == ["t.alpha"]
with pytest.raises(R1.RunError, match="every member run succeeded"):
PR.finalize(reg, pr, "succeeded")
def test_a_failed_member_blocks_its_dependents(tmp_path):
reg = make_repo(tmp_path); doc = reg.projects["demo"]
pr = PR.create(reg, doc, "bernd")
run = PR.start_member(reg, doc, pr, "t.alpha")
run2.start(run, "first"); run2.fail(run, "first", "broke")
R1.finalize(run, "failed")
v = PR.view(reg, pr)
assert v["members"][0]["status"] == "failed" and v["members"][1]["status"] == "waiting" and v["problems"] == ["t.alpha"] and v["ready"] == []
pr2 = PR.create(make_repo(tmp_path / "y"), Registry([tmp_path / "y"]).projects["demo"], "bernd", "auto-confirm")
assert PR.tick(Registry([tmp_path / "y"]), Registry([tmp_path / "y"]).projects["demo"], pr2)["action"] == "member-started"
# ---- modes, closing, locking ------------------------------------------------------------------------------------------------------------
def test_switching_mode_carries_the_open_member_run_along(tmp_path):
reg = make_repo(tmp_path); doc = reg.projects["demo"]
pr = PR.create(reg, doc, "bernd")
run = PR.start_member(reg, doc, pr, "t.alpha")
PR.switch_mode(reg, pr, "auto-confirm")
assert pr.data["mode"] == "auto-confirm" and R1.load(reg.packs["t.alpha"], run.data["run_id"]).data["mode"] == "auto-confirm"
PR.switch_mode(reg, pr, "manual")
assert R1.load(reg.packs["t.alpha"], run.data["run_id"]).data["mode"] == "manual"
with pytest.raises(R1.RunError, match="already in manual"):
PR.switch_mode(reg, pr, "manual")
def test_closing_abandons_the_open_member_run_locks_seals_and_refuses_further_work(tmp_path):
reg = make_repo(tmp_path); doc = reg.projects["demo"]
pr = PR.create(reg, doc, "bernd")
run = PR.start_member(reg, doc, pr, "t.alpha")
with pytest.raises(R1.RunError, match="succeeds only when"):
PR.finalize(reg, pr, "succeeded")
PR.finalize(reg, pr, "abandoned", "not now")
closed = R1.load(reg.packs["t.alpha"], run.data["run_id"])
assert closed.data["outcome"] == "abandoned" and "project run closed as abandoned" in closed.data["outcome_note"]
assert pr.data["outcome"] == "abandoned" and PR.seal_state(pr) == "sealed" and not (pr.path.stat().st_mode & 0o222)
assert PR.active_run(reg, "demo") is None
for call in (lambda: PR.start_member(reg, doc, pr, "t.beta"), lambda: PR.finalize(reg, pr, "abandoned"), lambda: PR.switch_mode(reg, pr, "auto-confirm")):
with pytest.raises(R1.RunError, match="closed and locked"):
call()
pr.data["operator"] = "forged"
with pytest.raises(R1.RunError, match="closed and locked"):
pr.save()
fresh = PR.load(reg, pr.id)
assert fresh.data["operator"] == "bernd"
def test_failed_needs_a_problem_or_an_open_member_otherwise_abandon(tmp_path):
reg = make_repo(tmp_path); doc = reg.projects["demo"]
pr = PR.create(reg, doc, "bernd") # nothing started, nothing wrong
with pytest.raises(R1.RunError, match="only when a member failed"):
PR.finalize(reg, pr, "failed")
PR.start_member(reg, doc, pr, "t.alpha") # an open member run may be given up as failed
PR.finalize(reg, pr, "failed", "gave up")
assert pr.data["outcome"] == "failed"
def test_a_project_run_edited_on_disk_after_closing_is_reported_broken(tmp_path):
import json
reg = make_repo(tmp_path); doc = reg.projects["demo"]
pr = PR.create(reg, doc, "bernd"); PR.finalize(reg, pr, "abandoned")
pr.path.chmod(0o644)
d = json.loads(pr.path.read_text()); d["operator"] = "forged"; pr.path.write_text(json.dumps(d))
assert PR.seal_state(PR.load(reg, pr.id)) == "broken"
def test_listing_and_times(tmp_path):
reg = make_repo(tmp_path); doc = reg.projects["demo"]
a = PR.create(reg, doc, "bernd"); PR.finalize(reg, a, "abandoned")
b = PR.create(reg, doc, "bernd")
assert [p.id for p in PR.list_runs(reg, "demo")] == [b.id, a.id] and PR.active_run(reg, "demo").id == b.id
assert PR.list_runs(reg, "nope") == [] and PR.times(a)["finished"] and PR.times(b)["finished"] is None

230
tests/test_ui_projectrun.py Normal file
View file

@ -0,0 +1,230 @@
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", "exercise_status": "unexercised", "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

2
tests/test_ui_shell.py
View file

@ -71,7 +71,7 @@ def test_the_theme_can_be_chosen_and_is_remembered_and_dark_styles_exist(srv):
assert "localStorage.getItem('rtut-theme')" in head and "URLSearchParams" in head # applied before painting; ?theme=dark also works
assert "system:'light',light:'dark',dark:'system'" in body # the button cycles system, light, dark
assert "aria-current" in body # the current section is marked
assert "'/attribute-sets':['/attribute-set/']" in body and "'/projects':['/project/','/template/']" in body # detail pages mark their section too
assert "'/attribute-sets':['/attribute-set/']" in body and "'/projects':['/project/','/template/','/project-run/']" in body # detail pages mark their section too
def test_the_sidebar_shows_the_saved_name(srv):