tegwick
6579618737
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
315 lines
18 KiB
Python
315 lines
18 KiB
Python
import shutil
|
|
import subprocess
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
import yaml
|
|
|
|
from rtut import project as J, run as R
|
|
from rtut.attrs import Registry, fold, load_set, normalize, set_layers
|
|
|
|
ROOT = Path(__file__).resolve().parent.parent
|
|
|
|
|
|
def write(path: Path, data: dict):
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(yaml.safe_dump(data, sort_keys=False))
|
|
|
|
|
|
@pytest.fixture
|
|
def repo(tmp_path):
|
|
for name in ("showcase-lighthouse", "example-hello"):
|
|
(tmp_path / "runbooks" / name).mkdir(parents=True)
|
|
shutil.copy(ROOT / f"runbooks/{name}/pack.yaml", tmp_path / "runbooks" / name / "pack.yaml")
|
|
for f in (ROOT / "attribute-sets").glob("*.yaml"):
|
|
write(tmp_path / "attribute-sets" / f.name, yaml.safe_load(f.read_text()))
|
|
shutil.copy(ROOT / "project-templates/lighthouse-watch.yaml", tmp_path / "project-templates/lighthouse-watch.yaml") \
|
|
if (tmp_path / "project-templates").mkdir(exist_ok=True) is None else None
|
|
return tmp_path
|
|
|
|
|
|
def project(repo, **over):
|
|
d = {"spec": "project/v0.1", "id": "north-watch", "title": "North watch", "environment": "rehearsal",
|
|
"attribute_sets": ["stormy-coast"],
|
|
"attributes": {"keeper": "Ada", "logdir": "/tmp/rtut-proj-north"},
|
|
"runbooks": [{"pack": "rtut.showcase-lighthouse"}, {"pack": "rtut.example-hello", "after": ["rtut.showcase-lighthouse"]}]}
|
|
d.update(over)
|
|
write(repo / "projects" / d["id"] / "project.yaml", d)
|
|
return Registry([repo])
|
|
|
|
|
|
# ---- attribute sets ---------------------------------------------------------------------------------------
|
|
|
|
def test_shipped_examples_are_valid():
|
|
reg = Registry([ROOT])
|
|
assert all(d.ok for d in reg.sets.values()), [d.errors for d in reg.sets.values()]
|
|
assert reg.templates["lighthouse-watch"].ok and set(reg.sets) >= {"harbour-defaults", "stormy-coast"}
|
|
|
|
|
|
def test_extends_order_provenance_and_modified_flag(repo):
|
|
reg = Registry([repo])
|
|
f = fold(set_layers(reg, "stormy-coast"))
|
|
assert f["weather"]["value"] == "storm" and f["weather"]["source"] == "set:stormy-coast"
|
|
assert f["signal"]["value"] == "flags" and f["signal"]["modified"] is True # replaced a base value
|
|
assert [h["source"] for h in f["signal"]["history"]] == ["set:harbour-defaults", "set:stormy-coast"]
|
|
assert f["motto"]["value"] and f["motto"]["modified"] is False
|
|
assert f["keeper"]["state"] == "nil" and f["keeper"]["requirement"] == "required"
|
|
|
|
|
|
def test_attribute_set_errors(tmp_path):
|
|
write(tmp_path / "attribute-sets/a.yaml", {"spec": "attribute-set/v0.1", "id": "a", "title": "Set A", "extends": ["b"], "attributes": {}})
|
|
write(tmp_path / "attribute-sets/b.yaml", {"spec": "attribute-set/v0.1", "id": "b", "title": "Set B", "extends": ["a"], "attributes": {}})
|
|
write(tmp_path / "attribute-sets/c.yaml", {"spec": "attribute-set/v0.1", "id": "c", "title": "Set C", "attributes": {"token": "hvs.ABCDEFGHIJKLMN"}})
|
|
write(tmp_path / "attribute-sets/d.yaml", {"spec": "attribute-set/v0.1", "id": "d", "title": "Set D", "attributes": {"Bad-Name": "x"}})
|
|
reg = Registry([tmp_path])
|
|
with pytest.raises(ValueError, match="cycle"):
|
|
set_layers(reg, "a")
|
|
with pytest.raises(ValueError, match="unknown attribute set"):
|
|
set_layers(reg, "zzz")
|
|
assert not reg.sets["c"].ok and any("secret" in e for e in reg.sets["c"].errors)
|
|
assert not reg.sets["d"].ok
|
|
assert not load_set(tmp_path / "attribute-sets/d.yaml").ok
|
|
|
|
|
|
def test_normalize_scalar_and_object():
|
|
n = normalize({"a": "x", "b": {"requirement": "required"}, "c": 3})
|
|
assert n["a"] == {"value": "x", "requirement": None} and n["b"]["value"] is None and n["c"]["value"] == 3
|
|
|
|
|
|
# ---- resolution against runbooks -----------------------------------------------------------------------------
|
|
|
|
def test_member_attribute_states_requirements_and_layers(repo):
|
|
reg = project(repo, attributes={"logdir": "/etc", "surplus": "x"}) # invalid for the runbook, and unused
|
|
d = reg.projects["north-watch"].data
|
|
rows = {r["id"]: r for r in J.member_attributes(reg, d, d["runbooks"][0])["rows"]}
|
|
assert rows["weather"]["state"] == "set" and rows["weather"]["source"] == "set:stormy-coast"
|
|
assert rows["logdir"]["state"] == "err" and "rtut-" in rows["logdir"]["error"] or rows["logdir"]["state"] == "err"
|
|
assert rows["keeper"]["requirement"] == "required" # the set declares it required; the pack has a default
|
|
assert rows["keeper"]["state"] == "nil" and rows["keeper"]["default"] == "Ada"
|
|
assert J.member_attributes(reg, d, d["runbooks"][0])["unused"] == ["surplus"]
|
|
|
|
|
|
def test_target_and_member_layers_override_the_project(repo):
|
|
reg = project(repo, targets={"cape": {"attributes": {"weather": "fog"}}},
|
|
runbooks=[{"pack": "rtut.showcase-lighthouse", "target": "cape", "attributes": {"motto": "Hold fast"}}])
|
|
d = reg.projects["north-watch"].data
|
|
rows = {r["id"]: r for r in J.member_attributes(reg, d, d["runbooks"][0])["rows"]}
|
|
assert rows["weather"]["value"] == "fog" and rows["weather"]["source"] == "target:cape" and rows["weather"]["modified"]
|
|
assert rows["motto"]["value"] == "Hold fast" and rows["motto"]["source"].startswith("member:")
|
|
|
|
|
|
def test_missing_required_lists_what_a_member_still_needs(repo):
|
|
reg = project(repo, attributes={"logdir": "/tmp/rtut-proj-north"}) # keeper has a pack default, so not missing
|
|
d = reg.projects["north-watch"].data
|
|
assert J.missing_required(reg, d) == []
|
|
reg = project(repo, runbooks=[{"pack": "rtut.example-hello"}], attribute_sets=[], attributes={"host": "h"})
|
|
write(repo / "runbooks/example-hello/pack.yaml", {**yaml.safe_load((ROOT / "runbooks/example-hello/pack.yaml").read_text()),
|
|
"parameters": [{"id": "host", "label": "Host", "type": "string"}, {"id": "count", "label": "Count", "type": "int", "default": 1}]})
|
|
reg = Registry([repo]); d = reg.projects["north-watch"].data; d["attributes"] = {}
|
|
assert [m["attribute"] for m in J.missing_required(reg, d)] == ["host"]
|
|
|
|
|
|
# ---- validation ------------------------------------------------------------------------------------------------
|
|
|
|
@pytest.mark.parametrize("over,frag", [
|
|
({"runbooks": [{"pack": "nope.nothing"}]}, "not found"),
|
|
({"attribute_sets": ["ghost"]}, "unknown attribute set"),
|
|
({"runbooks": [{"pack": "rtut.showcase-lighthouse", "target": "ghost"}]}, "unknown target"),
|
|
({"runbooks": [{"pack": "rtut.showcase-lighthouse", "after": ["rtut.example-hello"]}]}, "not a member"),
|
|
({"runbooks": [{"pack": "rtut.showcase-lighthouse", "after": ["rtut.example-hello"]},
|
|
{"pack": "rtut.example-hello", "after": ["rtut.showcase-lighthouse"]}]}, "cycle"),
|
|
({"runbooks": [{"pack": "rtut.showcase-lighthouse"}, {"pack": "rtut.showcase-lighthouse"}]}, "twice"),
|
|
])
|
|
def test_project_validation(repo, over, frag):
|
|
reg = project(repo, **over)
|
|
errs = J.validate(reg, reg.projects["north-watch"])
|
|
assert any(frag in e for e in errs), errs
|
|
|
|
|
|
def test_project_needs_an_environment_and_valid_class(repo):
|
|
write(repo / "projects/x/project.yaml", {"spec": "project/v0.1", "id": "x", "title": "Project x", "runbooks": [{"pack": "rtut.example-hello"}]})
|
|
write(repo / "projects/y/project.yaml", {"spec": "project/v0.1", "id": "y", "title": "Project y", "environment": "prod", "runbooks": [{"pack": "rtut.example-hello"}]})
|
|
reg = Registry([repo])
|
|
assert not reg.projects["x"].ok and not reg.projects["y"].ok
|
|
|
|
|
|
# ---- runs and state ----------------------------------------------------------------------------------------------
|
|
|
|
def finish(r):
|
|
for s in r.pack.data["steps"]:
|
|
sid = s["id"]; R.start(r, sid)
|
|
cmd = R.command_for(r, sid)
|
|
if cmd:
|
|
subprocess.run(cmd, shell=True, check=True, capture_output=True)
|
|
R.verify(r, sid, confirm=True, manual_ok=True if s["verify"]["expect"] == "manual" else None)
|
|
R.complete(r, sid, {n: "1" for n in s.get("evidence", [])})
|
|
R.finalize(r, "succeeded")
|
|
|
|
|
|
def test_start_run_uses_attributes_and_records_context(repo):
|
|
reg = project(repo)
|
|
r = J.start_run(reg, reg.projects["north-watch"], "rtut.showcase-lighthouse", "bob")
|
|
assert r.data["params"]["weather"] == "storm" and r.data["params"]["signal"] == "flags" and r.data["params"]["logdir"] == "/tmp/rtut-proj-north"
|
|
assert r.data["context"] == {"project": "north-watch", "environment": "rehearsal"}
|
|
with pytest.raises(R.RunError, match="manual mode only"):
|
|
J.start_run(reg, reg.projects["north-watch"], "rtut.showcase-lighthouse", "bob", mode="auto-confirm")
|
|
|
|
|
|
def test_project_state_planned_applying_applied_and_next(repo):
|
|
reg = project(repo)
|
|
doc = reg.projects["north-watch"]
|
|
st = J.state(reg, doc)
|
|
assert st["state"] == "planned" and st["next"] == ["rtut.showcase-lighthouse"] # the second member waits for the first
|
|
r = J.start_run(reg, doc, "rtut.showcase-lighthouse", "bob")
|
|
st = J.state(reg, doc)
|
|
assert st["state"] == "applying" and st["members"][0]["status"] == "applying" and st["next"] == []
|
|
finish(r)
|
|
st = J.state(reg, doc)
|
|
assert st["members"][0]["status"] == "applied" and st["state"] == "applying" and st["next"] == ["rtut.example-hello"]
|
|
finish(J.start_run(reg, doc, "rtut.example-hello", "bob"))
|
|
assert J.state(reg, doc)["state"] == "applied"
|
|
|
|
|
|
def test_project_goes_stale_when_a_member_changes_and_failed_members_are_shown(repo):
|
|
reg = project(repo, runbooks=[{"pack": "rtut.showcase-lighthouse"}])
|
|
doc = reg.projects["north-watch"]
|
|
finish(J.start_run(reg, doc, "rtut.showcase-lighthouse", "bob"))
|
|
assert J.state(reg, doc)["state"] == "applied"
|
|
p = repo / "runbooks/showcase-lighthouse/pack.yaml"
|
|
p.write_text(p.read_text() + "\n# revised\n")
|
|
reg = Registry([repo])
|
|
st = J.state(reg, reg.projects["north-watch"])
|
|
assert st["state"] == "stale" and st["members"][0]["status"] == "stale"
|
|
r = J.start_run(reg, reg.projects["north-watch"], "rtut.showcase-lighthouse", "bob")
|
|
R.finalize(r, "abandoned")
|
|
assert J.state(reg, reg.projects["north-watch"])["members"][0]["status"] == "failed"
|
|
|
|
|
|
def test_runs_of_other_projects_do_not_count(repo):
|
|
reg = project(repo, runbooks=[{"pack": "rtut.showcase-lighthouse"}])
|
|
write(repo / "projects/other/project.yaml", {**reg.projects["north-watch"].data, "id": "other", "title": "Other"})
|
|
reg = Registry([repo])
|
|
finish(J.start_run(reg, reg.projects["other"], "rtut.showcase-lighthouse", "bob"))
|
|
assert J.state(reg, reg.projects["north-watch"])["state"] == "planned"
|
|
|
|
|
|
# ---- templates ------------------------------------------------------------------------------------------------------
|
|
|
|
def test_instantiate_template_records_origin_merges_values_and_lists_needs(repo):
|
|
reg = Registry([repo])
|
|
out = J.instantiate(reg, "lighthouse-watch", "harbour-north", "rehearsal", {"keeper": "Ada", "logdir": "/tmp/rtut-harbour"},
|
|
title="North harbour watch", description="Our own harbour", goal="Both lights keep watch")
|
|
made = yaml.safe_load(out.read_text())
|
|
assert made["spec"] == "project/v0.1" and made["template"] == "lighthouse-watch" and made["environment"] == "rehearsal"
|
|
assert made["attributes"]["keeper"] == "Ada"
|
|
assert made["title"] == "North harbour watch" and made["description"] == "Our own harbour" and made["goal"] == "Both lights keep watch"
|
|
assert "The night watch is played through" not in out.read_text() and "A watch at a lighthouse" not in out.read_text() # nothing of the template's text is copied
|
|
assert set(made["targets"]) == {"north-light", "south-light"}
|
|
reg = Registry([repo])
|
|
assert reg.projects["harbour-north"].ok and J.validate(reg, reg.projects["harbour-north"]) == []
|
|
r = J.start_run(reg, reg.projects["harbour-north"], "rtut.showcase-lighthouse", "bob", target="north-light")
|
|
assert r.data["params"]["weather"] == "fog" and r.data["context"]["target"] == "north-light" # target attributes applied
|
|
with pytest.raises(ValueError, match="already exists"):
|
|
J.instantiate(reg, "lighthouse-watch", "harbour-north", "rehearsal")
|
|
with pytest.raises(ValueError, match="environment"):
|
|
J.instantiate(reg, "lighthouse-watch", "harbour-x", "prod")
|
|
with pytest.raises(R.RunError, match="template is not run"):
|
|
J.start_run(reg, reg.templates["lighthouse-watch"], "rtut.showcase-lighthouse", "bob")
|
|
|
|
|
|
def test_set_attributes_validates_against_the_consuming_runbooks(repo):
|
|
reg = project(repo, attributes={})
|
|
doc = reg.projects["north-watch"]
|
|
J.set_attributes(reg, doc, {"keeper": "Bea", "logdir": "/tmp/rtut-set-here", "unrelated": "x"})
|
|
made = yaml.safe_load(doc.path.read_text())["attributes"]
|
|
assert made["keeper"] == "Bea" and made["logdir"] == "/tmp/rtut-set-here" and made["unrelated"] == "x"
|
|
with pytest.raises(ValueError, match="rtut-|violates"):
|
|
J.set_attributes(reg, doc, {"logdir": "/etc"})
|
|
with pytest.raises(ValueError, match="secret"):
|
|
J.set_attributes(reg, doc, {"keeper": "hvs.ABCDEFGHIJKLMN"})
|
|
with pytest.raises(ValueError, match="bad attribute name"):
|
|
J.set_attributes(reg, doc, {"Bad-Name": "x"})
|
|
assert yaml.safe_load(doc.path.read_text())["attributes"]["logdir"] == "/tmp/rtut-set-here" # refused values change nothing
|
|
reg = Registry([repo])
|
|
with pytest.raises(ValueError, match="template"):
|
|
J.set_attributes(reg, reg.templates["lighthouse-watch"], {"keeper": "x"})
|
|
|
|
|
|
def test_attribute_notes_survive_resolution(repo):
|
|
reg = Registry([repo])
|
|
assert "Who keeps this light" in fold(set_layers(reg, "harbour-defaults"))["keeper"]["note"]
|
|
|
|
|
|
def test_project_without_a_given_name_uses_its_id_never_the_templates_title(repo):
|
|
reg = Registry([repo])
|
|
out = J.instantiate(reg, "lighthouse-watch", "plain-one", "staging")
|
|
made = yaml.safe_load(out.read_text())
|
|
assert made["title"] == "plain-one" and "goal" not in made and "description" not in made
|
|
with pytest.raises(ValueError, match="3 characters"):
|
|
J.instantiate(reg, "lighthouse-watch", "ab", "staging")
|
|
with pytest.raises(ValueError, match="secret"):
|
|
J.instantiate(reg, "lighthouse-watch", "leaky-one", "staging", description="token hvs.ABCDEFGHIJKLMN")
|
|
|
|
|
|
# ---- managing attribute sets and editing projects --------------------------------------------------------------------
|
|
|
|
def test_used_by_finds_projects_templates_targets_and_extending_sets(repo):
|
|
from rtut.attrs import used_by
|
|
reg = project(repo) # north-watch uses stormy-coast, which extends harbour-defaults
|
|
u = used_by(reg, "stormy-coast")
|
|
assert u == {"projects": ["north-watch"], "templates": [], "sets": []}
|
|
u = used_by(reg, "harbour-defaults")
|
|
assert u["templates"] == ["lighthouse-watch"] and u["sets"] == ["stormy-coast"]
|
|
|
|
|
|
def test_save_set_writes_validates_and_normalizes(repo):
|
|
from rtut.attrs import save_set
|
|
reg = Registry([repo])
|
|
path = save_set(reg, repo, "team-defaults", "Team defaults", ["harbour-defaults"],
|
|
{"lamp": {"value": "electric"}, "keeper": {"requirement": "required", "note": "who"}, "empty_note": {"note": "later"}})
|
|
made = yaml.safe_load(path.read_text())
|
|
assert path == repo / "attribute-sets/team-defaults.yaml" and made["extends"] == ["harbour-defaults"]
|
|
assert made["attributes"]["lamp"] == "electric" # a lone value is written as a scalar
|
|
assert made["attributes"]["keeper"] == {"requirement": "required", "note": "who"}
|
|
reg = Registry([repo])
|
|
assert reg.sets["team-defaults"].ok and fold(set_layers(reg, "team-defaults"))["lamp"]["value"] == "electric"
|
|
|
|
|
|
@pytest.mark.parametrize("kwargs,frag", [
|
|
({"set_id": "Bad Id"}, "lower-case"), ({"title": "ab"}, "3 characters"), ({"extends": ["ghost"]}, "unknown attribute set"),
|
|
({"extends": ["team-x"]}, "itself"), ({"attributes": {"Bad-Name": {"value": "x"}}}, "bad attribute name"),
|
|
({"attributes": {"a": {"value": "hvs.ABCDEFGHIJKLMN"}}}, "secret"), ({"attributes": {"a": {"requirement": "maybe", "value": "x"}}}, "requirement"),
|
|
({"attributes": {"a": {}}}, "give a value"),
|
|
])
|
|
def test_save_set_refusals(repo, kwargs, frag):
|
|
from rtut.attrs import save_set
|
|
args = {"set_id": "team-x", "title": "Team x", "extends": [], "attributes": {"a": {"value": "1"}}, **kwargs}
|
|
with pytest.raises(ValueError, match=frag):
|
|
save_set(Registry([repo]), repo, **args)
|
|
assert not (repo / "attribute-sets/team-x.yaml").exists()
|
|
|
|
|
|
def test_save_set_refuses_a_cycle(repo):
|
|
from rtut.attrs import save_set
|
|
reg = Registry([repo])
|
|
with pytest.raises(ValueError, match="circular"):
|
|
save_set(reg, repo, "harbour-defaults", "Harbour defaults", ["stormy-coast"], {"lamp": {"value": "x"}}, reg.sets["harbour-defaults"].path)
|
|
|
|
|
|
def test_update_project_renames_and_keeps_everything_else(repo):
|
|
reg = project(repo)
|
|
doc = reg.projects["north-watch"]
|
|
before = yaml.safe_load(doc.path.read_text())
|
|
J.update_project(doc, "North harbour watch", "Our harbour", "Both lights burn")
|
|
after = yaml.safe_load(doc.path.read_text())
|
|
assert after["title"] == "North harbour watch" and after["description"] == "Our harbour" and after["goal"] == "Both lights burn"
|
|
assert after["runbooks"] == before["runbooks"] and after["attributes"] == before["attributes"] and after["id"] == "north-watch"
|
|
assert list(after)[:5] == ["spec", "id", "title", "description", "goal"]
|
|
J.update_project(Registry([repo]).projects["north-watch"], "North harbour watch", "", "")
|
|
assert "description" not in yaml.safe_load(doc.path.read_text())
|
|
for bad, frag in (("ab", "3 characters"), ("x" * 5 + "hvs.ABCDEFGHIJKLMN", "secret")):
|
|
with pytest.raises(ValueError, match=frag):
|
|
J.update_project(Registry([repo]).projects["north-watch"], bad)
|
|
reg = Registry([repo])
|
|
with pytest.raises(ValueError, match="only a project"):
|
|
J.update_project(reg.templates["lighthouse-watch"], "Renamed template")
|