runbook-tutorials/tests/test_ui_round3.py

import http.client
import json
import shutil
import subprocess
import threading
import uuid
from datetime import datetime, timedelta, timezone
from pathlib import Path
from urllib.parse import urlencode

import pytest
import yaml

from rtut import run as R1, run2 as R, ui

ROOT = Path(__file__).resolve().parent.parent
_scratch = []


@pytest.fixture
def srv(tmp_path):
    for name, src in (("v1", "showcase-lighthouse"), ("v2", "showcase-lighthouse-v2")):
        d = tmp_path / "runbooks" / name; d.mkdir(parents=True)
        shutil.copy(ROOT / f"runbooks/{src}/pack.yaml", d / "pack.yaml")
    (tmp_path / "projects/watch").mkdir(parents=True)
    (tmp_path / "projects/watch/project.yaml").write_text(yaml.safe_dump({
        "spec": "project/v0.1", "id": "watch", "title": "Watch project", "environment": "rehearsal",
        "attribute_sets": ["user-attributes"], "attributes": {"logdir": "/tmp/rtut-r3-proj"},
        "runbooks": [{"pack": "rtut.showcase-lighthouse"}]}))
    s = ui.serve([tmp_path], 0)
    threading.Thread(target=s.serve_forever, daemon=True).start()
    s.repo = tmp_path
    yield s
    s.shutdown()
    while _scratch:
        shutil.rmtree(_scratch.pop(), ignore_errors=True)


def raw(s, method, path, form=None):
    c = http.client.HTTPConnection("127.0.0.1", s.server_address[1])
    body = urlencode(form) if form is not None else None
    c.request(method, path, body, {"Host": "127.0.0.1", "Content-Type": "application/x-www-form-urlencoded"})
    r = c.getresponse()
    return r.status, r.getheader("Location"), r.read().decode(errors="replace"), {k.lower(): v for k, v in r.getheaders()}


def req(s, method, path, form=None):
    return raw(s, method, path, form)[:3]


def post(s, path, **form):
    return req(s, "POST", path, {"_t": s.app.token, **form})


def pidx(s, pack_id):
    return s.app.pack_index(pack_id)


def new_v2(s, **extra):
    logdir = f"/tmp/rtut-r3-{uuid.uuid4().hex[:8]}"
    _scratch.append(logdir)
    st, loc, _ = post(s, f"/run/new/{pidx(s, 'rtut.showcase-lighthouse-v2')}", p_logdir=logdir, **extra)   # note: no operator posted
    assert st == 303
    return loc.split("#")[0], logdir


def step(s, base, sid, **form):
    assert post(s, f"{base}/{sid}/start", **form)[0] == 303
    run = R1.load(s.app.packs[int(base.split("/")[2])], base.rsplit("/", 1)[1])          # reload: the chosen alternative is known now
    cmd = R.command_for(run, sid)
    if cmd:
        subprocess.run(cmd, shell=True, check=True, capture_output=True)
    eff = R.step_def(run, sid)
    manual = (eff["verify"].get("expect") or "manual") == "manual" and not eff["verify"].get("datapoints_ok")
    assert (post(s, f"{base}/{sid}/verify", ok="1") if manual else post(s, f"{base}/{sid}/verify", yes="1"))[0] == 303
    assert post(s, f"{base}/{sid}/done")[0] == 303


# ---- (a) the active user instead of asking for a name ----------------------------------------------------------

def test_user_page_shows_source_and_saves(srv):
    _, _, body = req(srv, "GET", "/user")
    assert "tester-login" in body and "your operating-system login" in body and "user-attributes" in body
    assert post(srv, "/user", operator="bernd", display_name="Bernd W.")[0] == 303
    _, _, body = req(srv, "GET", "/user")
    assert "<b>bernd</b>" in body and "your saved settings" in body and "Bernd W." in body
    st, _, body = post(srv, "/user", operator="two words")
    assert st == 400 and "one word" in body


def test_forms_use_the_active_user_and_do_not_ask_for_a_name(srv):
    post(srv, "/user", operator="bernd", display_name="Bernd W.")
    i1, i2 = pidx(srv, "rtut.showcase-lighthouse"), pidx(srv, "rtut.showcase-lighthouse-v2")
    for path in (f"/pack/{i1}", f"/pack/{i2}"):
        _, _, body = req(srv, "GET", path)
        assert "Running as <b>Bernd W. (bernd)</b>" in body and "<a href='/user'>change</a>" in body
        assert "Your name (one word)" not in body and "<input name=operator" not in body
    _, _, body = req(srv, "GET", "/project/watch")
    assert "Start run as bernd" in body and "name=operator value='bernd'" in body


def test_a_run_started_without_a_posted_name_belongs_to_the_active_user_and_reviews_too(srv):
    post(srv, "/user", operator="carol")
    base, _ = new_v2(srv)
    run = R1.load(srv.app.packs[pidx(srv, "rtut.showcase-lighthouse-v2")], base.rsplit("/", 1)[1])
    assert run.data["operator"] == "carol"
    i1 = pidx(srv, "rtut.showcase-lighthouse")
    assert post(srv, f"/pack/{i1}/review", note="ok")[0] == 303
    _, _, body = req(srv, "GET", f"/pack/{i1}")
    assert "carol" in body and "Reviewed: <b>yes</b>" in body


def test_user_attributes_are_reusable_by_projects_through_the_app_level_set(srv):
    post(srv, "/user", operator="dave")
    _, _, body = req(srv, "GET", "/project/watch")                      # the project lists attribute_sets: [user-attributes]
    assert "This project has problems" not in body and "Watch project" in body
    from rtut import project as J
    reg = srv.app.registry()
    assert J.validate(reg, reg.projects["watch"]) == []                                  # the set is found although no repository defines it
    member = reg.projects["watch"].data["runbooks"][0]
    res = J.member_attributes(reg, reg.projects["watch"].data, member)
    assert "operator" in res["unused"]                                                    # v0.1 showcase has no operator input: reported, harmless
    assert "Unused attributes: " in body and "operator" in body.split("Unused attributes: ")[1].split("</p>")[0]


# ---- (b) the irreversible example is irreversible -------------------------------------------------------------------

def test_the_showcase_logbook_entry_carries_the_time_and_close_log_says_why_it_cannot_be_undone(srv):
    base, logdir = new_v2(srv)
    run = R1.load(srv.app.packs[pidx(srv, "rtut.showcase-lighthouse-v2")], base.rsplit("/", 1)[1])
    assert "$(date -u" in run.pack.data["stages"][1]["steps"][1]["command"]
    _, _, body = req(srv, "GET", base)
    assert "cannot be undone" in body and "to the second" in body


# ---- (c) latest activities ---------------------------------------------------------------------------------------------

def test_main_page_lists_latest_activities_newest_first_with_links(srv):
    _, _, body = req(srv, "GET", "/")
    assert "Latest activities" in body and "Nothing has happened yet" in body
    post(srv, "/user", operator="erin")
    base, _ = new_v2(srv, p_weather="fog")
    step(srv, base, "read-the-sign")
    step(srv, base, "light-lamp")
    _, _, body = req(srv, "GET", "/")
    feed = body.split("Latest activities")[1].split("<h2>Runbooks</h2>")[0]
    assert "<b>erin</b> started a run (manual mode)" in feed and "finished step “Light the lamp”" in feed
    assert feed.index("Light the lamp") < feed.index("Read the poem over the door") < feed.index("started a run")   # newest first
    assert f"<a href='{base}'>open</a>" in feed and "(times in UTC)" in body


# ---- (d) changing values that steps already used --------------------------------------------------------------------------

def test_changing_a_value_in_use_needs_an_override_shows_the_lock_and_is_recorded(srv):
    base, _ = new_v2(srv, p_weather="fog")
    _, _, body = req(srv, "GET", base + "?view=data")
    assert "in use by" not in body and "Changing a value that steps already used" not in body        # nothing used yet: no lock, no override
    assert post(srv, f"{base}/data", d_weather="clear")[0] == 303                                        # free change before use
    step(srv, base, "read-the-sky")
    _, _, body = req(srv, "GET", base + "?view=data")
    assert "in use by read-the-sky" in body and "Changing a value that steps already used (override)" in body
    st, _, err = post(srv, f"{base}/data", d_weather="storm")
    assert st == 400 and "already in use" in err and "read-the-sky" in err
    st, _, err = post(srv, f"{base}/data", d_weather="storm", override_reason="forecast changed")
    assert st == 400 and "acknowledge" in err
    run = R1.load(srv.app.packs[pidx(srv, "rtut.showcase-lighthouse-v2")], base.rsplit("/", 1)[1])
    assert run.data["params"]["weather"] == "clear" and run.data["deviations"] == []                  # refused: nothing changed
    assert post(srv, f"{base}/data", d_weather="storm", override_reason="forecast changed", acknowledge_risk="1")[0] == 303
    _, _, ov = req(srv, "GET", base + "?view=overview")
    assert ">changed-after-use<" in ov and "forecast changed" in ov and "read-the-sky already used weather" in ov


def test_a_refused_save_applies_none_of_its_changes(srv):
    base, _ = new_v2(srv, p_weather="fog")
    step(srv, base, "read-the-sky")
    st, _, _ = post(srv, f"{base}/data", d_weather="storm", d_motto="Hold fast")          # weather is locked, motto is free
    assert st == 400
    run = R1.load(srv.app.packs[pidx(srv, "rtut.showcase-lighthouse-v2")], base.rsplit("/", 1)[1])
    assert run.data["params"]["motto"] != "Hold fast" and run.data["params"]["weather"] == "fog"


def test_in_auto_confirm_mode_used_values_cannot_be_changed_until_taking_over(srv):
    base, _ = new_v2(srv, mode="auto-confirm", p_weather="fog")
    post(srv, f"{base}/tick", n="3")
    run = R1.load(srv.app.packs[pidx(srv, "rtut.showcase-lighthouse-v2")], base.rsplit("/", 1)[1])
    if not R.datapoint_users(run, "weather"):
        run.data["steps"]["read-the-sky"]["state"] = "done"; run.save()
    _, _, body = req(srv, "GET", base + "?view=data")
    assert "can only be changed in manual mode" in body and "name=override_reason" not in body
    st, _, err = post(srv, f"{base}/data", d_weather="clear", override_reason="x", acknowledge_risk="1")
    assert st == 400 and "take over" in err


# ---- (e) artefacts can be opened and downloaded ---------------------------------------------------------------------------

def finish_logbook(srv, base):
    for sid in ("read-the-sign", "light-lamp", "read-the-sky"):
        step(srv, base, sid)
    for i in (0, 1):
        ck = post(srv, f"{base}/gate/lamp-check/check", idx=str(i), ok="1") if i == 0 else post(srv, f"{base}/gate/lamp-check/check", idx="1", yes="1")
        assert ck[0] == 303
    step(srv, base, "warn")
    step(srv, base, "write-logbook")


def test_artefact_is_linked_and_can_be_opened_and_downloaded(srv):
    base, logdir = new_v2(srv, p_weather="fog")
    _, _, body = req(srv, "GET", base + "?view=data")
    assert "the file does not exist (yet)" in body and "artefact/logbook-file" not in body           # nothing to open before it exists
    assert raw(srv, "GET", f"{base}/artefact/logbook-file")[0] == 403
    finish_logbook(srv, base)
    _, _, body = req(srv, "GET", base + "?view=data")
    assert f"<a href='{base}/artefact/logbook-file' target=_blank rel=noopener>Open</a>" in body and "?download=1'>Download</a>" in body
    st, _, text, hdr = raw(srv, "GET", f"{base}/artefact/logbook-file")
    assert st == 200 and "Night watch by Ada at " in text and hdr["content-type"].startswith("text/plain")
    assert hdr["x-content-type-options"] == "nosniff" and "sandbox" in hdr["content-security-policy"] and "content-disposition" not in hdr
    st, _, text2, hdr = raw(srv, "GET", f"{base}/artefact/logbook-file?download=1")
    assert st == 200 and text2 == text and 'attachment; filename="artefact-logbook-file"' in hdr["content-disposition"]


def test_artefact_route_refuses_unknown_ids_secrets_and_v01_runs(srv):
    base, logdir = new_v2(srv, p_weather="fog")
    finish_logbook(srv, base)
    assert raw(srv, "GET", f"{base}/artefact/nope")[0] == 404
    (Path(logdir) / "logbook.txt").write_text("token " + "hvs." + "ABCDEFGHIJKLMN")
    st, _, body, _ = raw(srv, "GET", f"{base}/artefact/logbook-file")
    assert st == 403 and "secret" in body
    _, _, dbody = req(srv, "GET", base + "?view=data")
    assert "looks like it contains a secret" in dbody and "artefact/logbook-file" not in dbody
    st, loc, _ = post(srv, f"/run/new/{pidx(srv, 'rtut.showcase-lighthouse')}", p_logdir="/tmp/rtut-r3-v1")
    assert raw(srv, "GET", loc.split("#")[0] + "/artefact/logbook-file")[0] == 404


# ---- (f) timestamps, durations and the average ---------------------------------------------------------------------------------

def test_run_shows_start_finish_and_duration_and_steps_show_how_long_they_took(srv):
    base, _ = new_v2(srv, p_weather="fog")
    _, _, body = req(srv, "GET", base)
    assert "Started 20" in body and "UTC · running for <b>" in body
    step(srv, base, "read-the-sign")
    _, _, body = req(srv, "GET", base)
    assert "took 0 s" in body or "took 1 s" in body
    assert post(srv, f"{base}/finalize", outcome="abandoned")[0] == 303
    _, _, body = req(srv, "GET", base)
    assert "finished 20" in body and "· took <b>" in body and "working" in body


def test_pack_page_and_list_show_the_average_duration_of_successful_runs(srv):
    i1 = pidx(srv, "rtut.showcase-lighthouse")
    pack = srv.app.packs[i1]
    for secs, who in ((300, "a"), (900, "b")):
        r = R1.create(pack, who)
        start = datetime(2026, 9, 29, 10, 0, tzinfo=timezone.utc)
        r.data.update(created=start.isoformat(), outcome="succeeded", finished=(start + timedelta(seconds=secs)).isoformat())
        r.path.write_text(json.dumps(r.data))
    slow = R1.create(pack, "c")
    slow.data.update(created="2026-09-29T10:00:00+00:00", outcome="failed", finished="2026-09-29T20:00:00+00:00")
    slow.path.write_text(json.dumps(slow.data))
    _, _, body = req(srv, "GET", f"/pack/{i1}")
    assert "average 10 min per successful run (fastest 5 min, 2 timed)" in body            # the failed 10-hour run is not in the average
    _, _, body = req(srv, "GET", "/")
    assert "average 10 min per successful run" in body
    assert "took 5 min" in req(srv, "GET", f"/pack/{i1}")[2]                                # the runs list shows each duration