runbook-tutorials/tests/test_ui_v2.py

import http.client
import shutil
import subprocess
import threading
import uuid
from pathlib import Path
from urllib.parse import urlencode

import pytest

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

ROOT = Path(__file__).resolve().parent.parent
PHRASE = "I understand this cannot be undone"
_scratch = []


@pytest.fixture
def s2(tmp_path):
    d = tmp_path / "runbooks" / "lh"; d.mkdir(parents=True)
    shutil.copy(ROOT / "runbooks/showcase-lighthouse-v2/pack.yaml", d / "pack.yaml")
    s = ui.serve([tmp_path], 0)
    threading.Thread(target=s.serve_forever, daemon=True).start()
    yield s
    s.shutdown()
    while _scratch:
        shutil.rmtree(_scratch.pop(), ignore_errors=True)


def req(s, method, path, form=None, host="127.0.0.1"):
    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": host, "Content-Type": "application/x-www-form-urlencoded"})
    r = c.getresponse()
    return r.status, r.getheader("Location"), r.read().decode()


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


def new_run(s, mode="manual", **extra):
    logdir = f"/tmp/rtut-uv2-{uuid.uuid4().hex[:8]}"
    _scratch.append(logdir)
    st, loc, _ = post(s, "/run/new/0", operator="bernd", mode=mode, p_logdir=logdir, **extra)
    assert st == 303, loc
    return loc.split("#")[0], logdir


def page(s, base, query=""):
    st, _, body = req(s, "GET", base + query)
    assert st == 200
    return body


def do_step(s, base, sid, alt=None, phrase=None, run_cmd=True, cmd_of=None):
    """What an operator does through the UI: start, do the work, check, mark done."""
    form = {}
    if alt: form["alt"] = alt
    if phrase: form["phrase"] = phrase
    assert post(s, f"{base}/{sid}/start", **form)[0] == 303
    run = R1.load(s.app.packs[0], base.rsplit("/", 1)[1])
    cmd = R.command_for(run, sid)
    if cmd and run_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


def pass_gate(s, base, gid):
    run = R1.load(s.app.packs[0], base.rsplit("/", 1)[1])
    for i, ck in enumerate(R.flow(run).gates[gid]["checks"]):
        exp = ck.get("expect") or (None if ck.get("datapoints_ok") else "manual")
        assert (post(s, f"{base}/gate/{gid}/check", idx=str(i), ok="1") if exp == "manual" else post(s, f"{base}/gate/{gid}/check", idx=str(i), yes="1"))[0] == 303


# ---- starting -----------------------------------------------------------------------------------------------

def test_v02_pack_is_listed_and_has_a_start_form_with_modes_and_preauthorization(s2):
    _, _, body = req(s2, "GET", "/")
    assert "/pack/0" in body and "cannot be run yet" not in body
    body = page(s2, "/pack/0")
    assert "Keep in mind" in body and "Start a run" in body
    assert "name='p_keeper_name'" in body and "<span class=chip>actor: keeper</span>" in body           # datapoints with their dimension tags
    assert "name=mode value=manual checked" in body and "name=mode value=auto-confirm" in body
    assert "Pre-authorize irreversible steps" in body and "name='pre_close-log'" in body and "manual-only" in body
    assert "name='p_lamp_lit'" not in body                                                              # produced datapoints are not questions


def test_flow_view_structure_framing_first_next_action_and_parallel_items(s2):
    base, _ = new_run(s2)
    body = page(s2, base)
    assert "Guided flow" in body and "Data" in body and "Overview" in body
    assert body.index("Keep in mind") < body.index("Dusk - prepare")                                    # framing comes first, not last
    assert "Manual mode." in body and "Hand over to auto-confirm" in body
    assert "<b>Next:</b> <a href='#item-read-the-sign'>Read the poem over the door</a>" in body and "(recommended)" in body
    assert "Also possible in parallel" in body and "#item-light-lamp" in body and "#item-read-the-sky" in body
    assert "<b>Goal:</b> The lamp is lit" in body
    assert "Questions for this stage" in body and "name='d_keeper_name'" in body
    assert "Gate: The lamp burns steadily" in body and "quality goal: safe-navigation" in body and "blocking" in body
    assert "1. Read the poem over the door" in body and "[Lighthouse keeper · the-lighthouse]" in body
    assert "<button type=button data-copy>" not in body                                                 # nothing started: no command, no Copy


def test_blocked_step_says_why_and_offers_an_override_dialog_with_the_downstream_risk(s2):
    base, _ = new_run(s2)
    body = page(s2, base)
    assert "Waiting for: gate lamp-check is not complete" in body
    assert "Start anyway (override)" in body and "Downstream risk:" in body and "name=override_reason" in body and "name=acknowledge_risk" in body


def test_starting_and_running_a_parallel_step_shows_command_and_copy_only_then(s2):
    base, _ = new_run(s2, p_weather="fog")
    assert post(s2, f"{base}/light-lamp/start")[0] == 303
    body = page(s2, base)
    assert "<b>Continue:</b> <a href='#item-light-lamp'>Light the lamp</a>" in body
    assert "<pre>echo &quot;Ada lights the paraffin lamp.&quot;</pre><button type=button data-copy>" in body
    assert "Yes, it is" in body and "Mark failed" in body and "id=current" in body


# ---- alternatives, irreversible steps, overrides, gates ------------------------------------------------------------

def open_night(s, base):
    for sid in ("read-the-sign", "light-lamp", "read-the-sky"):
        do_step(s, base, sid)
    pass_gate(s, base, "lamp-check")


def test_alternatives_are_recommended_by_datapoint_and_can_be_overridden_by_the_operator(s2):
    base, _ = new_run(s2, p_weather="fog")
    open_night(s2, base)
    body = page(s2, base)
    assert "<b>Choose one:</b>" in body and "value='sound-horn' checked" in body and "Sound the fog horn <span class=note>(recommended)</span>" in body
    assert post(s2, f"{base}/warn/start", alt="wave-flags")[0] == 303
    body = page(s2, base)
    assert "Signal with flags" in body and "signals with flags" in body


def test_irreversible_step_needs_the_typed_phrase_and_shows_it(s2):
    base, _ = new_run(s2, p_weather="fog")
    open_night(s2, base)
    do_step(s2, base, "warn"); do_step(s2, base, "write-logbook")
    body = page(s2, base)
    assert "irreversible" in body and f"<code>{PHRASE}</code>" in body and "name=phrase" in body
    st, _, err = post(s2, f"{base}/close-log/start")
    assert st == 400 and "confirmation phrase" in err
    assert post(s2, f"{base}/close-log/start", phrase="yes please")[0] == 400
    assert post(s2, f"{base}/close-log/start", phrase=PHRASE)[0] == 303
    assert "Continue:" in page(s2, base)


def test_override_needs_reason_and_acknowledgement_and_is_recorded_and_shown(s2):
    base, _ = new_run(s2)
    st, _, err = post(s2, f"{base}/warn/start", alt="sound-horn")
    assert st == 400 and "Risk:" in err
    st, _, err = post(s2, f"{base}/warn/start", alt="sound-horn", override_reason="just to see")
    assert st == 400 and "acknowledge" in err
    assert post(s2, f"{base}/warn/start", alt="sound-horn", override_reason="just to see", acknowledge_risk="1")[0] == 303
    body = page(s2, base)
    assert "1 deviation(s) recorded" in body and "will never count towards maturity" in body
    ov = page(s2, base, "?view=overview")
    assert "<h2>Deviations</h2>" in ov and ">override<" in ov and "just to see" in ov and "warn" in ov


def test_gate_checks_waiver_and_state_through_the_ui(s2):
    base, _ = new_run(s2)
    for sid in ("read-the-sign", "light-lamp", "read-the-sky"):
        do_step(s2, base, sid)
    body = page(s2, base)
    assert "Yes, it is" in body and "Not yet" in body and "datapoints ok: lamp_lit" in body
    assert post(s2, f"{base}/gate/lamp-check/check", idx="0", ok="0")[0] == 303
    assert post(s2, f"{base}/gate/lamp-check/check", idx="1", yes="1")[0] == 303
    body = page(s2, base)
    assert "Waive this gate (declared waivable by Harbour master)" in body and "Ships may steer by a light that is not burning." in body
    st, _, err = post(s2, f"{base}/gate/lamp-check/waive", reason="  ")
    assert st == 400 and "reason" in err
    assert post(s2, f"{base}/gate/lamp-check/waive", reason="harbour master accepts")[0] == 303
    body = page(s2, base)
    assert "Waived: harbour master accepts" in body and "1 deviation(s)" in body
    assert "Start this step" in body                                                                # the night can begin after a waiver


def test_a_gate_that_is_not_waivable_offers_no_waiver(s2):
    base, _ = new_run(s2)
    assert "Waive this gate" in page(s2, base).split("Gate: Nothing left behind")[0]               # lamp-check is waivable
    assert "Waive this gate" not in page(s2, base).split("Gate: Nothing left behind")[1]           # night-on-record is not
    assert post(s2, f"{base}/gate/night-on-record/waive", reason="x")[0] == 400


# ---- retry policy -----------------------------------------------------------------------------------------------------

def test_retry_of_non_idempotent_step_asks_for_the_partial_application_confirmation(s2):
    base, _ = new_run(s2, p_weather="fog")
    open_night(s2, base)
    do_step(s2, base, "warn")
    assert post(s2, f"{base}/write-logbook/start")[0] == 303
    assert post(s2, f"{base}/write-logbook/fail", note="disk full")[0] == 303
    body = page(s2, base)
    assert "name=acknowledge_partial" in body and "did not partly apply" in body
    st, _, err = post(s2, f"{base}/write-logbook/retry")
    assert st == 400 and "did not partly apply" in err
    assert post(s2, f"{base}/write-logbook/retry", acknowledge_partial="1")[0] == 303
    # an idempotent step needs no confirmation
    base2, _ = new_run(s2)
    post(s2, f"{base2}/read-the-sign/start"); post(s2, f"{base2}/read-the-sign/fail", note="")
    body2 = page(s2, base2)
    assert "Failed: (no comment)" in body2 and "name=acknowledge_partial" not in body2                 # idempotent: free retry
    assert post(s2, f"{base2}/read-the-sign/retry")[0] == 303


# ---- data view --------------------------------------------------------------------------------------------------------

def test_data_view_groups_filters_and_edits_datapoints_independently_of_the_flow(s2):
    base, logdir = new_run(s2)
    body = page(s2, base, "?view=data")
    assert "independent of the flow" in body and "<b>Group by</b>" in body
    assert "href='?view=data&amp;group=actor'" in body
    assert "Scratch folder" in body and "actor: keeper" in body
    grouped = page(s2, base, "?view=data&group=actor")
    assert "<h3>actor: keeper</h3>" in grouped and "<h3>actor: (none)</h3>" in grouped
    only = page(s2, base, "?view=data&filter=kind:identity")
    assert "Keeper&#x27;s name" in only and "Motto" in only and "Scratch folder" not in only and "clear filter" in only
    assert "Where it lives" in body and f"{logdir}/logbook.txt" in body                             # artefact location resolved
    assert "<h2>Actors</h2>" in body and "Harbour master" in body and "approve-custody" not in body
    # editing here needs no step to be running
    st, loc, _ = post(s2, f"{base}/data", d_weather="storm", next=f"{base}?view=data")
    assert st == 303 and loc == f"{base}?view=data"
    assert "selected>storm" in page(s2, base, "?view=data")
    st, _, err = post(s2, f"{base}/data", d_weather="hail")
    assert st == 400 and "weather" in err
    st, loc, _ = post(s2, f"{base}/data", d_motto="Hold fast", next="https://evil.example/x")
    assert st == 303 and loc == base                                                                # an unsafe next is ignored
    st, _, err = post(s2, f"{base}/data", d_motto="<script>alert(1)</script>")
    assert st == 400 and "<script>alert(1)" not in err


def test_produced_datapoints_are_shown_not_editable_and_state_changes_are_visible(s2):
    base, _ = new_run(s2)
    body = page(s2, base, "?view=data")
    assert "name='d_lamp_lit'" not in body and ">nil<" in body and "produced by light-lamp" in body
    do_step(s2, base, "light-lamp")
    body = page(s2, base, "?view=data")
    assert ">ok<" in body and "confirmed by light-lamp" in body


# ---- overview -----------------------------------------------------------------------------------------------------------

def test_overview_shows_progress_stages_gates_datapoints_and_no_deviations(s2):
    base, _ = new_run(s2)
    body = page(s2, base, "?view=overview")
    assert "0 of 8 done" in page(s2, base) and "<h2>Stages</h2>" in body
    assert "Dusk - prepare" in body and "Night watch - give warning" in body and "Dawn - close the night" in body
    assert "0/3" in body and "<h2>Datapoints</h2>" in body and ">set<" in body and ">nil<" in body
    assert "None. A run without deviations can count towards maturity." in body
    do_step(s2, base, "light-lamp")
    assert "1 of 8 done" in page(s2, base)


# ---- auto-confirm -----------------------------------------------------------------------------------------------------

def test_auto_confirm_mode_confirms_waits_and_can_be_taken_over(s2):
    base, _ = new_run(s2, mode="auto-confirm", p_weather="fog")
    body = page(s2, base)
    assert "Auto-confirm mode." in body and "never runs commands" in body and "Run until it waits" in body and "Take over (manual)" in body
    assert "Overrides are made in manual mode" in body                                                # blocked items cannot be overridden here
    assert "Start anyway" not in body
    assert post(s2, f"{base}/tick", n="50")[0] == 303
    body = page(s2, base)
    assert "Last: waiting write-logbook" in body or "Last: awaiting-operator" in body
    run = R1.load(s2.app.packs[0], base.rsplit("/", 1)[1])
    assert run.data["steps"]["read-the-sign"]["state"] == "done" and run.data["steps"]["warn"]["state"] == "done"
    assert run.data["steps"]["read-the-sign"]["verification"].get("auto_confirmed")
    assert post(s2, f"{base}/mode", to="manual")[0] == 303
    body = page(s2, base)
    assert "Manual mode." in body and "Hand over to auto-confirm" in body
    assert "auto-confirm → manual" in page(s2, base, "?view=overview")
    assert post(s2, f"{base}/mode", to="manual")[0] == 400                                            # already manual


def test_auto_confirm_stop_is_shown_and_never_confirms_manual_only_steps(s2):
    base, logdir = new_run(s2, mode="auto-confirm", p_weather="fog")
    post(s2, f"{base}/tick", n="50")                     # the engine confirms what it can, then waits for the work
    # the operator does the work between ticks
    run = R1.load(s2.app.packs[0], base.rsplit("/", 1)[1])
    subprocess.run(R.command_for(run, "write-logbook"), shell=True, check=True, capture_output=True)
    assert post(s2, f"{base}/tick", n="50")[0] == 303
    body = page(s2, base)
    assert "Last: awaiting-operator close-log" in body and "step is manual-only" in body
    assert "Manual-only: auto-confirm will not start this step" in body


# ---- captures, feedback, closing ------------------------------------------------------------------------------------

def test_captures_and_pasted_output_verification_work_in_v02_runs(s2):
    base, _ = new_run(s2, p_weather="fog")
    post(s2, f"{base}/read-the-sky/start")
    body = page(s2, base)
    assert "Capture command and output" in body and "Check against this output" in body
    assert post(s2, f"{base}/read-the-sky/capture", command="echo x", output="Weather tonight - fog.")[0] == 303
    assert post(s2, f"{base}/read-the-sky/verify-pasted", output="something else")[0] == 303
    assert "Check not passed" in page(s2, base)
    assert post(s2, f"{base}/read-the-sky/verify-pasted", output="Weather tonight - fog.")[0] == 303
    assert "Check passed" in page(s2, base)
    st, _, err = post(s2, f"{base}/read-the-sky/capture", output="hvs.ABCDEFGHIJKLMN")
    assert st == 400 and "secret-looking" in err


def test_finishing_a_completed_run_feedback_and_the_read_only_result(s2):
    base, logdir = new_run(s2, p_weather="fog")
    open_night(s2, base)
    do_step(s2, base, "warn"); do_step(s2, base, "write-logbook")
    do_step(s2, base, "close-log", phrase=PHRASE)
    pass_gate(s2, base, "night-on-record")
    body = page(s2, base)
    assert "Everything is done." in body and "Finish: succeeded" in body and "<h2>How was this runbook?</h2>" in body
    assert post(s2, f"{base}/feedback", rating="good", comment="clear")[0] == 303
    assert "Back to the runbook list" not in page(s2, base) and "Change my feedback" in page(s2, base)   # still open: changeable
    assert post(s2, f"{base}/finalize", outcome="succeeded", note="smooth")[0] == 303
    body = page(s2, base)
    assert "Run finished: succeeded" in body and "Start this step" not in body
    assert post(s2, f"{base}/read-the-sign/start")[0] == 400


def test_abandoning_and_failed_outcome_rules(s2):
    base, _ = new_run(s2)
    body = page(s2, base)
    assert "Abandon run" in body and "Finish: succeeded" not in body and "Finish: failed" not in body
    assert post(s2, f"{base}/finalize", outcome="succeeded")[0] == 400
    assert post(s2, f"{base}/finalize", outcome="abandoned")[0] == 303
    assert "Run finished: abandoned" in page(s2, base)


# ---- safety --------------------------------------------------------------------------------------------------------------

def test_csrf_host_and_unknown_actions(s2):
    base, _ = new_run(s2)
    assert req(s2, "POST", f"{base}/mode", {"to": "manual"})[0] == 403
    assert req(s2, "GET", base, host="evil.example")[0] == 403
    assert post(s2, f"{base}/read-the-sign/explode")[0] == 400
    assert post(s2, f"{base}/nope/start")[0] == 400
    assert req(s2, "GET", base + "?view=<script>")[0] == 200                                        # an unknown view falls back to the flow