tegwick
a6ee6a0a6e
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
103 lines
4.8 KiB
Python
103 lines
4.8 KiB
Python
import os
|
|
import shutil
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from rtut import activity, artefacts as A, maturity as M, pack as P, run as R, run2
|
|
|
|
ROOT = Path(__file__).resolve().parent.parent
|
|
|
|
|
|
def repo_with(tmp_path, *names):
|
|
packs = []
|
|
for n in names:
|
|
d = tmp_path / "runbooks" / n; d.mkdir(parents=True)
|
|
shutil.copy(ROOT / ("runbooks/showcase-lighthouse-v2/pack.yaml" if n.endswith("2") else "runbooks/showcase-lighthouse/pack.yaml"), d / "pack.yaml")
|
|
packs.append(P.load(d / "pack.yaml"))
|
|
return packs
|
|
|
|
|
|
# ---- activity -----------------------------------------------------------------------------------------------------
|
|
|
|
def test_latest_activities_are_newest_first_across_packs_and_cover_the_main_events(tmp_path):
|
|
v1, v2 = repo_with(tmp_path, "one", "two2")
|
|
r = R.create(v1, "ann")
|
|
R.start(r, "read-the-sign"); R.verify(r, "read-the-sign", manual_ok=True); R.complete(r, "read-the-sign")
|
|
R.add_feedback(r, "good", "smooth")
|
|
M.add_review(v1, "bob", "read it")
|
|
q = run2.create(v2, "cy", {"logdir": "/tmp/rtut-act-x"})
|
|
run2.set_datapoint(q, "weather", "clear")
|
|
run2.start(q, "read-the-sky"); run2.fail(q, "read-the-sky", "x")
|
|
R.finalize(r, "abandoned")
|
|
ev = activity.latest([v1, v2], 50)
|
|
assert [e["at"] for e in ev] == sorted((e["at"] for e in ev), reverse=True)
|
|
text = " | ".join(f"{e['who']}: {e['text']}" for e in ev)
|
|
for frag in ("ann: started a run (manual mode)", "ann: finished step “Read the poem over the door”", "ann: rated the run good: smooth",
|
|
"bob: reviewed the runbook: read it", "cy: started a run (manual mode)", "cy: failed step “Read the sky”",
|
|
"ann: finished the run as abandoned"):
|
|
assert frag in text, frag
|
|
assert all(e["link"] for e in ev) and {e["pack"] for e in ev} == {v1.data["title"], v2.data["title"]}
|
|
assert len(activity.latest([v1, v2], 3)) == 3
|
|
|
|
|
|
def test_gate_events_use_the_gate_title(tmp_path):
|
|
(v2,) = repo_with(tmp_path, "two2")
|
|
q = run2.create(v2, "cy", {"logdir": "/tmp/rtut-act-z"})
|
|
for sid in ("read-the-sign", "light-lamp", "read-the-sky"):
|
|
run2.start(q, sid); run2.verify(q, sid, confirm=True, manual_ok=True); run2.complete(q, sid)
|
|
run2.gate_check(q, "lamp-check", 0, manual_ok=True); run2.gate_check(q, "lamp-check", 1)
|
|
text = " | ".join(e["text"] for e in activity.latest([v2], 30))
|
|
assert "gate “The lamp burns steadily” passed" in text and "lamp-check" not in text
|
|
|
|
|
|
def test_overrides_and_changed_values_are_visible_in_the_feed(tmp_path):
|
|
(v2,) = repo_with(tmp_path, "two2")
|
|
q = run2.create(v2, "cy", {"logdir": "/tmp/rtut-act-y"})
|
|
run2.start(q, "warn", alt="sound-horn", override_reason="want to see", acknowledge_risk=True)
|
|
text = " | ".join(e["text"] for e in activity.latest([v2], 20))
|
|
assert "override: warn (want to see)" in text
|
|
|
|
|
|
# ---- artefact serving rules ----------------------------------------------------------------------------------------
|
|
|
|
def test_a_plain_text_file_may_be_served(tmp_path):
|
|
f = tmp_path / "logbook.txt"; f.write_text("Night watch by Ada\n")
|
|
path, why = A.check(str(f))
|
|
assert path == f.resolve() and why == ""
|
|
data, ctype, is_text = A.content(path)
|
|
assert data == b"Night watch by Ada\n" and ctype.startswith("text/plain") and is_text
|
|
|
|
|
|
@pytest.mark.parametrize("make,frag", [
|
|
(lambda t: None, "not known yet"),
|
|
(lambda t: str(t / "missing.txt"), "does not exist"),
|
|
(lambda t: str(t), "not a regular file"),
|
|
(lambda t: "relative/file.txt", "absolute"),
|
|
(lambda t: "https://example.org/x", "link"),
|
|
])
|
|
def test_refused_locations(tmp_path, make, frag):
|
|
path, why = A.check(make(tmp_path))
|
|
assert path is None and frag in why
|
|
|
|
|
|
def test_sensitive_files_secrets_and_big_files_are_never_served(tmp_path):
|
|
ssh = tmp_path / ".ssh"; ssh.mkdir(); (ssh / "notes.txt").write_text("hello")
|
|
assert "sensitive" in A.check(str(ssh / "notes.txt"))[1]
|
|
key = tmp_path / "server.key"; key.write_text("x")
|
|
assert "sensitive" in A.check(str(key))[1]
|
|
runs = tmp_path / "runbook-runs"; runs.mkdir(); (runs / "r.json").write_text("{}")
|
|
assert "sensitive" in A.check(str(runs / "r.json"))[1]
|
|
leaky = tmp_path / "out.txt"; leaky.write_text("token " + "hvs." + "ABCDEFGHIJKLMN")
|
|
assert "secret" in A.check(str(leaky))[1]
|
|
big = tmp_path / "big.bin"; big.write_bytes(b"0" * (A.MAX_BYTES + 1))
|
|
assert "larger" in A.check(str(big))[1]
|
|
link = tmp_path / "link.txt"; os.symlink(str(ssh / "notes.txt"), link)
|
|
assert "sensitive" in A.check(str(link))[1] # a symlink is judged by where it really points
|
|
|
|
|
|
def test_binary_files_are_offered_only_as_downloads(tmp_path):
|
|
f = tmp_path / "blob.bin"; f.write_bytes(b"\xff\xfe\x00\x01")
|
|
path, _ = A.check(str(f))
|
|
_, ctype, is_text = A.content(path)
|
|
assert not is_text and "text/plain" not in ctype
|