runbook-tutorials/tests/test_score.py

import shutil
import subprocess
from datetime import datetime, timedelta, timezone
from pathlib import Path

import pytest

from rtut import pack as P, run as R

ROOT = Path(__file__).resolve().parent.parent


@pytest.fixture
def pk(tmp_path):
    d = tmp_path / "runbooks" / "show"; d.mkdir(parents=True)
    shutil.copy(ROOT / "runbooks/showcase-lighthouse/pack.yaml", d / "pack.yaml")
    return P.load(d / "pack.yaml")


def play_all(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", [])})


def test_finalize_rules(pk):
    r = R.create(pk, "bernd")
    with pytest.raises(R.RunError):
        R.finalize(r, "succeeded")                 # nothing done yet
    with pytest.raises(R.RunError):
        R.finalize(r, "failed")                    # nothing failed: abandon instead
    with pytest.raises(R.RunError):
        R.finalize(r, "great")
    R.start(r, "read-the-sign"); R.fail(r, "read-the-sign", "could not read it")
    R.finalize(r, "failed", "gave up on step 1")
    assert r.data["outcome"] == "failed" and r.data["outcome_note"]
    with pytest.raises(R.RunError):
        R.finalize(r, "abandoned")                 # already finished


def test_finished_run_is_read_only_and_feedback_is_locked_too(pk):
    r = R.create(pk, "bernd"); R.finalize(r, "abandoned")
    for call in (lambda: R.start(r, "read-the-sign"), lambda: R.add_capture(r, "read-the-sign", "x")):
        with pytest.raises(R.RunError, match="finished"):
            call()
    with pytest.raises(R.RunError, match="closed and locked"):
        R.add_feedback(r, "poor", "gave up: unclear")            # feedback goes in before closing, or with the closing form


def test_scorecard_counts_every_outcome_and_stale_open_runs(pk):
    ok = R.create(pk, "a"); play_all(ok); R.finalize(ok, "succeeded", rating="good")
    bad = R.create(pk, "b"); R.start(bad, "read-the-sign"); R.fail(bad, "read-the-sign", "x"); R.finalize(bad, "failed")
    gone = R.create(pk, "c"); R.finalize(gone, "abandoned")
    stalled = R.create(pk, "d")
    stale = R.create(pk, "e")
    stale.data["updated"] = (datetime.now(timezone.utc) - timedelta(days=9)).isoformat()
    stale.path.write_text(__import__("json").dumps(stale.data))
    c = R.scorecard(pk)
    assert (c["tried"], c["succeeded"], c["failed"], c["abandoned"], c["stalled"]) == (5, 1, 1, 2, 1)
    assert c["feedback"]["good"] == 1


def test_capture_documentation_and_limits(pk):
    r = R.create(pk, "bernd")
    with pytest.raises(R.RunError, match="start the step"):
        R.add_capture(r, "read-the-sign", "out")
    R.start(r, "read-the-sign")
    R.add_capture(r, "read-the-sign", "Tower of stone", command="printf poem")
    assert R.load(pk, r.data["run_id"]).state("read-the-sign")["captures"][0]["output"] == "Tower of stone"
    with pytest.raises(R.RunError, match="too large"):
        R.add_capture(r, "read-the-sign", "x" * 20001)
    with pytest.raises(R.RunError, match="secret-looking"):
        R.add_capture(r, "read-the-sign", "hvs.ABCDEFGHIJKLMN")
    with pytest.raises(R.RunError, match="nothing"):
        R.add_capture(r, "read-the-sign", "  ")


def test_verify_against_pasted_output(pk):
    r = R.create(pk, "bernd", {"keeper": "Ada"})
    R.start(r, "read-the-sign")
    with pytest.raises(R.RunError, match="output-contains"):
        R.verify_pasted(r, "read-the-sign", "x")                  # manual step: not applicable
    R.complete  # (step 2 is output-contains)
    R.verify(r, "read-the-sign", manual_ok=True); R.complete(r, "read-the-sign")
    R.start(r, "meet-the-keeper")
    assert not R.verify_pasted(r, "meet-the-keeper", "nothing useful", "echo ...")["passed"]
    ok = R.verify_pasted(r, "meet-the-keeper", "Ada climbs the stairs. Motto on the wall - x")
    assert ok["passed"] and ok["source"] == "pasted"
    caps = R.load(pk, r.data["run_id"]).state("meet-the-keeper")["captures"]
    assert len(caps) == 2 and caps[0]["source"] == "pasted-for-verification"


def test_runs_of_different_packs_in_one_repository_are_not_mixed(tmp_path):
    a = tmp_path / "runbooks" / "show"; a.mkdir(parents=True)
    b = tmp_path / "runbooks" / "hello"; b.mkdir(parents=True)
    shutil.copy(ROOT / "runbooks/showcase-lighthouse/pack.yaml", a / "pack.yaml")
    shutil.copy(ROOT / "runbooks/example-hello/pack.yaml", b / "pack.yaml")
    pa, pb = P.load(a / "pack.yaml"), P.load(b / "pack.yaml")
    ra = R.create(pa, "x"); R.create(pa, "y"); R.create(pb, "z")
    R.add_feedback(ra, "good", "only for the showcase")
    assert R.scorecard(pa)["tried"] == 2 and R.scorecard(pb)["tried"] == 1
    assert R.list_runs(pb) != R.list_runs(pa) and len(R.list_runs(pb)) == 1
    assert R.scorecard(pb)["feedback"]["good"] == 0 and R.all_feedback(pb) == []


# ---- durations -----------------------------------------------------------------------------------------------

def test_fmt_duration():
    assert [R.fmt_duration(x) for x in (None, 0, 45, 60, 200, 3600, 3900, 90000)] == ["-", "0 s", "45 s", "1 min", "3 min 20 s", "1 h", "1 h 5 min", "1 d 1 h"]


def test_run_times_and_average_of_succeeded_runs_only(pk):
    from datetime import datetime, timedelta, timezone
    import json
    def timed(operator, seconds, outcome):
        r = R.create(pk, operator)
        start = datetime(2026, 9, 29, 10, 0, tzinfo=timezone.utc)
        r.data["created"] = start.isoformat()
        r.data["outcome"] = outcome
        r.data["finished"] = (start + timedelta(seconds=seconds)).isoformat()
        first = next(iter(r.data["steps"]))
        r.data["steps"][first].update(state="done", started=(start + timedelta(seconds=10)).isoformat(),
                                      finished=(start + timedelta(seconds=40)).isoformat())
        r.path.write_text(json.dumps(r.data))
        return r
    a = timed("a", 600, "succeeded"); timed("b", 1200, "succeeded"); timed("c", 9000, "failed"); timed("d", 300, "abandoned")
    t = R.run_times(R.load(pk, a.data["run_id"]))
    assert t["elapsed"] == 600 and t["finished_run"] and t["step_seconds"] == {next(iter(a.data["steps"])): 30.0} and t["working"] == 30.0
    c = R.scorecard(pk)
    assert c["avg_seconds"] == 900 and c["median_seconds"] == 900 and c["fastest_seconds"] == 600 and c["timed_runs"] == 2   # the failed 9000 s run is not counted
    assert R.scorecard(pk)["succeeded"] == 2


def test_unfinished_run_elapsed_runs_to_now_and_no_average_without_succeeded_runs(pk):
    from datetime import datetime, timedelta, timezone
    r = R.create(pk, "x")
    t = R.run_times(r, now=datetime.now(timezone.utc) + timedelta(seconds=90))
    assert not t["finished_run"] and 89 <= t["elapsed"] <= 91
    assert R.scorecard(pk)["avg_seconds"] is None and R.scorecard(pk)["timed_runs"] == 0


# ---- a closed run is locked and sealed --------------------------------------------------------------------------

def test_closing_records_feedback_with_the_outcome_and_validates_before_changing_anything(pk):
    r = R.create(pk, "bernd")
    with pytest.raises(R.RunError, match="rating must be"):
        R.finalize(r, "abandoned", rating="great")
    assert not r.data.get("outcome") and "seal" not in r.data                       # a refused close changed nothing
    R.finalize(r, "abandoned", "gave up", rating="poor", comment="step 2 was unclear")
    assert R.latest_feedback(r)["comment"] == "step 2 was unclear" and r.data["outcome"] == "abandoned"
    assert R.load(pk, r.data["run_id"]).data["feedback"][0]["rating"] == "poor"


def test_a_closed_record_cannot_be_saved_with_changes(pk):
    r = R.create(pk, "bernd"); R.finalize(r, "abandoned")
    r2 = R.load(pk, r.data["run_id"])
    r2.data["steps"]["read-the-sign"]["state"] = "done"
    with pytest.raises(R.RunError, match="closed and locked"):
        r2.save()
    r3 = R.load(pk, r.data["run_id"])
    r3.data["operator"] = "someone-else"
    with pytest.raises(R.RunError, match="operator"):
        r3.save()
    assert R.load(pk, r.data["run_id"]).data["operator"] == "bernd"                # nothing reached the disk
    assert R.seal_state(R.load(pk, r.data["run_id"])) == "sealed"                  # and the seal still holds


def test_closing_seals_the_file_read_only_and_tampering_is_detected(pk):
    import json, stat
    r = R.create(pk, "bernd")
    assert R.seal_state(r) == "open" and r.path.stat().st_mode & stat.S_IWUSR
    R.finalize(r, "abandoned")
    assert R.seal_state(r) == "sealed" and not (r.path.stat().st_mode & (stat.S_IWUSR | stat.S_IWGRP | stat.S_IWOTH))
    assert len(r.data["seal"]) == 64
    r.path.chmod(0o644)                                                              # someone edits the file by hand
    data = json.loads(r.path.read_text()); data["operator"] = "forged"
    r.path.write_text(json.dumps(data))
    assert R.seal_state(R.load(pk, r.data["run_id"])) == "broken"


def test_a_run_closed_before_sealing_existed_is_reported_as_unsealed_not_broken(pk):
    import json
    r = R.create(pk, "bernd"); R.finalize(r, "abandoned")
    r.path.chmod(0o644)
    data = json.loads(r.path.read_text()); data.pop("seal"); r.path.write_text(json.dumps(data))
    assert R.seal_state(R.load(pk, r.data["run_id"])) == "unsealed"


def test_a_tampered_run_no_longer_counts_as_maturity_evidence(pk):
    import json
    from rtut import maturity as M
    M.add_review(pk, "alice")
    r = R.create(pk, "bob", {"logdir": "/tmp/rtut-lock-mat"}, {"project": "lab", "environment": "rehearsal"})
    for s in pk.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")
    assert M.maturity(pk)["evidence"]["rehearsal_runs"] == 1
    r.path.chmod(0o644)
    data = json.loads(r.path.read_text()); data["operator"] = "forged"; r.path.write_text(json.dumps(data))
    assert M.maturity(pk)["evidence"]["clean_runs"] == 0                             # changed after closing: not evidence