runbook-tutorials/tests/test_v2.py

import copy
from pathlib import Path

import pytest
import yaml

from rtut import pack as P, run as R, v2

ROOT = Path(__file__).resolve().parent.parent
FILE = ROOT / "runbooks/showcase-lighthouse-v2/pack.yaml"
GOOD = yaml.safe_load(FILE.read_text())


def errs(mutate):
    d = copy.deepcopy(GOOD)
    mutate(d)
    return v2.validate(d)


def step(d, sid):
    for st in d["stages"]:
        for s in st.get("steps", []):
            if s["id"] == sid:
                return s
    raise KeyError(sid)


def gate(d, gid):
    return next(g for st in d["stages"] for g in st.get("gates", []) if g["id"] == gid)


def test_showcase_v02_is_valid_and_loads_through_pack_load():
    pk = P.load(FILE)
    assert pk.ok, pk.errors
    assert pk.spec == "runbook-pack/v0.2"


def test_v02_packs_run_through_run_create(tmp_path):
    import shutil
    import uuid
    d = tmp_path / "runbooks" / "lh"; d.mkdir(parents=True)          # a copy: never write run records into the real repo
    shutil.copy(FILE, d / "pack.yaml")
    r = R.create(P.load(d / "pack.yaml"), "tester", {"logdir": f"/tmp/rtut-{uuid.uuid4().hex[:8]}"})
    assert r.data["spec"] == "runbook-run/v0.2" and r.data["mode"] == "manual"


CASES = [
    ("unknown actor", lambda d: step(d, "light-lamp").update(actor="nobody"), "unknown actor"),
    ("unknown datapoint in uses", lambda d: step(d, "light-lamp").update(uses=["nope"]), "uses unknown datapoint"),
    ("placeholder not in uses", lambda d: step(d, "light-lamp").update(uses=["keeper_name"]), "not listed in uses"),
    ("undefined placeholder", lambda d: step(d, "light-lamp").update(command="echo {{ghost}}"), "undefined placeholder"),
    ("requires unknown", lambda d: step(d, "write-logbook").update(requires=["gate:ghost"]), "requires unknown item"),
    ("duplicate id", lambda d: d["stages"][0]["gates"][0].update(id="dusk"), "duplicate id"),
    ("irreversible without phrase", lambda d: (d.pop("confirm_phrase"), None), "confirm_phrase"),
    ("waivable by unknown actor", lambda d: gate(d, "lamp-check").update(waivable={"by": "ghost", "needs_reason": True}), "unknown actor"),
    ("datapoints_ok unknown", lambda d: gate(d, "lamp-check")["checks"].append({"datapoints_ok": ["ghost"]}), "unknown datapoint"),
    ("ask_in unknown stage", lambda d: d["datapoints"][0].update(ask_in="ghost"), "unknown stage"),
    ("artefact produced_by unknown", lambda d: d["artefacts"][0].update(produced_by="ghost"), "unknown step"),
    ("recommended_when bad value", lambda d: step(d, "warn")["one_of"][0].update(recommended_when={"weather": ["hail"]}), "not in values"),
    ("step with one_of and command", lambda d: step(d, "warn").update(command="echo x"), "must not also have its own command"),
    ("secret marker", lambda d: d["framing"].append("token hvs.ABCDEFGHIJKLMN"), "secret-looking"),
    ("exit-0 without command", lambda d: step(d, "write-logbook")["verify"].pop("command"), "needs a command"),
    ("secret datapoint", lambda d: d["datapoints"][0].update(secret=True), "secret"),
    ("gate without goal", lambda d: gate(d, "lamp-check").pop("goal"), "goal"),
    ("waiver must need a reason", lambda d: gate(d, "lamp-check").update(waivable={"by": "harbourmaster", "needs_reason": False}), "needs_reason"),
    ("bad timeout action", lambda d: gate(d, "lamp-check").update(timeout={"seconds": 5, "on_timeout": "explode"}), "on_timeout"),
    ("bad auto_confirm value", lambda d: step(d, "close-log").update(auto_confirm="always"), "auto_confirm"),
]


@pytest.mark.parametrize("name,fn,frag", CASES, ids=[c[0] for c in CASES])
def test_invalid(name, fn, frag):
    out = errs(fn)
    assert any(frag in e for e in out), out


def test_cycle_detected():
    out = errs(lambda d: d["stages"][0].update(requires=["night"]))
    assert any("cycle" in e for e in out), out


def test_child_requiring_its_parent_is_a_cycle():
    def m(d):
        step(d, "read-the-sky").update(requires=["dusk"])
    assert any("cycle" in e for e in errs(m))


def test_depth_limit():
    def m(d):
        inner = {"id": "l4", "title": "Level four", "goal": "too deep", "steps": [copy.deepcopy(step(d, "read-the-sign"))]}
        inner["steps"][0]["id"] = "deep-step"
        l3 = {"id": "l3", "title": "Level three", "goal": "three", "stages": [inner]}
        l2 = {"id": "l2", "title": "Level two", "goal": "two", "stages": [l3]}
        d["stages"].append({"id": "l1", "title": "Level one", "goal": "one", "stages": [l2]})
    assert any("nesting deeper" in e for e in errs(m))


def test_three_levels_allowed():
    def m(d):
        l3 = {"id": "l3", "title": "Level three", "goal": "three", "steps": [copy.deepcopy(step(d, "read-the-sign"))]}
        l3["steps"][0]["id"] = "deep-step"
        l2 = {"id": "l2", "title": "Level two", "goal": "two", "stages": [l3]}
        d["stages"].append({"id": "l1", "title": "Level one", "goal": "one", "stages": [l2]})
    assert errs(m) == []


def test_legacy_v02_needs_only_schema():
    d = {"spec": "runbook-pack/v0.2", "id": "x.y", "title": "Legacy x", "owner": "o",
         "engine": "legacy-command", "legacy": {"command": ["true"]}}
    assert v2.validate(d) == []


def test_gate_cannot_confirm_a_datapoint_it_requires():
    out = errs(lambda d: next(x for x in d["datapoints"] if x["id"] == "lamp_lit").update(confirmed_by="lamp-check"))
    assert any("circular" in e for e in out), out


def test_idempotency_is_declared_validated_and_summarized():
    from rtut import flow
    assert P.load(FILE).ok
    assert flow.pack_idempotency(GOOD) == "non-idempotent"          # write-logbook appends
    d = copy.deepcopy(GOOD)
    for st in d["stages"]:
        for s in st.get("steps", []):
            s["idempotent"] = "idempotent"
            s.pop("idempotent_note", None)
    assert flow.pack_idempotency(d) == "idempotent" and v2.validate(d) == []
    step(d, "close-log")["idempotent"] = "conditional"
    assert any("idempotent_note" in e for e in v2.validate(d))
    step(d, "close-log")["idempotent_note"] = "only while the log file exists"
    assert v2.validate(d) == [] and flow.pack_idempotency(d) == "conditional"
    step(d, "close-log").pop("idempotent")
    assert flow.pack_idempotency(d) == "unknown"                     # one undeclared step makes the whole unknown
    assert errs(lambda x: step(x, "warn").update(idempotent="sometimes"))
    assert errs(lambda x: step(x, "warn").update(idempotent=True))       # a YAML "yes" is refused, not silently accepted


def test_v02_accepts_the_deprecated_exercise_status_and_does_not_require_it():
    d = copy.deepcopy(GOOD)
    assert "exercise_status" not in d and v2.validate(d) == []
    d["exercise_status"] = "exercised 2026-01-01 by someone"
    assert v2.validate(d) == []