tegwick
6d3414d8b7
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
264 lines
15 KiB
Python
264 lines
15 KiB
Python
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", "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
|