clanker-saddle/test/bundles/t7log.py

"""The t7log bundle (goal 41 T7, ux-notes structural bet 3): the task
page <-> event log unification. The two surfaces are ONE investigation
flow: the task page mounts the log as its "Event log" tab (the same
task-log-feed partial the standalone /tasks/{id}/log route renders),
the entrypoint publishes both slots, and no link to the log opens a
new tab any more.

Fixture-driven like the pages bundle: a task + session are inserted as
rows, events land through the push wire (/api/saddle/chunks), and the
assertions read page HTML, entrypoint slot state and the SSE stream.
"""
from harness import *  # noqa: F401,F403 — the shared wire helpers


# =============================================================================
# Bundle: t7log — task page <-> event log unification.
# =============================================================================

def run(ctx):
    auth, fake, fake_id, fake_token = ctx.auth, ctx.fake, ctx.fake_id, ctx.fake_token

    print("== B1: the task page embeds the log as its second tab ==")
    # A working task with one session, then a real event through the
    # push wire (the same path the daemon writes).
    conn = db()
    conn.execute(
        "insert into tasks (project_id, title, status, priority, created_at)"
        " values (1, 'T7 unified log task', 'working', 5,"
        " strftime('%s','now'))")
    task_id = conn.execute("select max(id) from tasks").fetchone()[0]
    conn.execute(
        "insert into sessions (task_id, saddle_id, started_at)"
        " values (?, ?, strftime('%s','now'))", (task_id, fake_id))
    conn.commit()
    session_id = conn.execute("select max(id) from sessions").fetchone()[0]
    conn.close()

    status, body = saddle_api("POST", "/api/saddle/chunks", fake_token, {
        "session_id": session_id,
        "events": [{"seq": 101, "kind": "comment", "actor": "user:admin",
                    "payload": {"body": "T7 probe comment",
                                "author": "user:admin"}}]})
    check(status == 200 and int(body.get("stored") or 0) == 1,
          "the fixture event stores through the push wire")

    # The tab markup: Spry's CSS tabs (radios first, then .tab-list
    # labels, then .tab-panels), Overview checked by default.
    task_page = http_get_page(auth, "/tasks/%d" % task_id)
    check('class="tabs task-tabs"' in task_page,
          "the task page carries the tab strip")
    check(task_page.index('id="task-view-overview"')
          < task_page.index('class="tab-list"')
          < task_page.index('class="tab-panels"'),
          "the radios lead the tab markup (Spry's nth-of-type contract)")
    check('id="task-view-overview" checked' in task_page,
          "Overview is the default (checked) tab")
    check('<label for="task-view-overview">Overview</label>' in task_page
          and '<label for="task-view-log">Event log</label>' in task_page,
          "the tab list names Overview and Event log")
    check('class="task-overview-panel"' in task_page
          and 'class="task-log-panel"' in task_page,
          "the panels carry the overview and the log")

    print("== B2: one shared shell, both surfaces ==")
    # The SAME partial ships both pages: the interactive hooks the
    # client engine binds appear exactly once on each, and the
    # compiled shell markup (the partial's) rides both.
    task_page = http_get_page(auth, "/tasks/%d" % task_id)
    log_page = http_get_page(auth, "/tasks/%d/log" % task_id)
    for surface, name in ((task_page, "the task page"),
                          (log_page, "the log page")):
        check('class="log-shell"' in surface and "data-log-filter" in surface,
              "%s renders the shared log shell" % name)
    # Hook multiplicity: each interactive hook appears exactly as
    # many times as the ONE shared shell carries it (5 filter
    # buttons, 1 feed/more/search/jump) — proving neither surface
    # ships a second, hand-copied shell.
    for hook, times in (("data-log-feed", 1), ("data-log-more", 1),
                        ("data-log-filter", 5), ("data-log-search", 1),
                        ("data-log-jump", 1)):
        check(task_page.count(hook) == times
              and log_page.count(hook) == times,
              "the %s hook appears exactly %d time(s) on both surfaces"
              % (hook, times))

    print("== B3: no log link opens a new tab any more ==")
    # The Progress cluster's control is a <label> toggling the radio
    # in place; no link element on the page navigates to the log
    # route any more (comments may mention it, markup may not).
    check('<label class="tab-link" for="task-view-log">Event log</label>'
          in task_page,
          "the Progress cluster's Event log control toggles the tab in place")
    markup = task_page.split("<script>")[0] + "".join(
        part.split("</script>", 1)[-1] for part in task_page.split("<script>")[1:])
    check(re.search(r"<a[^>]+(\n[^>]*)*?/log", markup) is None,
          "the task page markup carries no anchor to the log route")
    check(re.search(r"/log['\"]", markup) is None,
          "no log URL appears in any attribute (the tab is the only door)")

    print("== B4: both slots publish from the task entrypoint ==")
    # The embedded shell binds the page's task_log slot; it must name
    # the same task the standalone route's slot does.
    embedded = slot_public(auth, "/tasks/%d" % task_id, "task_log")
    standalone = slot_public(auth, "/tasks/%d/log" % task_id, "task_log")
    check(embedded.get("exists") is True
          and int(embedded["task_id"]) == task_id,
          "the task page entrypoint publishes the embedded log slot")
    for field in ("exists", "task_id", "task_title", "task_status", "task_url"):
        check(standalone.get(field) == embedded.get(field),
              "the %s field is identical on both surfaces" % field)
    check(embedded.get("task_title") == "T7 unified log task",
          "the embedded slot names the task")
    check(embedded.get("task_status") == "working",
          "the embedded slot carries the live status the badge reads")

    print("== B5: the standalone page keeps its context strip ==")
    # The forensic route renders the T5 context header (back chevron +
    # task title + live badge) and the not-found guard still works.
    check('class="log-header"' in log_page
          and 'data-log-title' in log_page
          and 'data-log-live' in log_page,
          "the standalone page keeps its one-row context header")
    check('stm-attribute.href="task_log.task_url"' in log_page,
          "the back chevron binds the task page URL from the slot")
    missing = slot_public(auth, "/tasks/99999999/log", "task_log")
    check(missing.get("exists") is False,
          "a missing task's log slot says exists=false (both surfaces render their own note)")

    print("== B6: the feed is reachable through both paths ==")
    # The JSON page + SSE stream are task-scoped, not route-scoped:
    # the embedded tab reads the same endpoints the standalone page
    # does (one engine, one wire).
    status, page_json = http("GET", "/tasks/%d/events" % task_id, timeout=15)
    rows = json.loads(page_json)
    check(status == 200
          and any(e.get("payload", {}).get("body") == "T7 probe comment"
                  for e in rows.get("events", [])),
          "the events endpoint serves the fixture event")
    request = urllib.request.Request(WEB + "/tasks/%d/events/stream" % task_id)
    with urllib.request.urlopen(request, timeout=15) as stream:
        probe = None
        deadline = time.monotonic() + 8
        while time.monotonic() < deadline and probe is None:
            line = stream.readline().decode().strip()
            if not line.startswith("data: ") or line == "data: ping":
                continue
            event = json.loads(line[6:])
            if (event.get("kind") == "comment"
                    and "T7 probe comment" in
                    (event.get("payload", {}).get("body") or "")):
                probe = event
        check(probe is not None,
              "the SSE stream carries the fixture event to either surface")

    print("== B7: the pulse refresh rebuilds BOTH slots ==")
    # The refresh action (the page's sealed pulse) rebuilds the task
    # slot AND the embedded task_log slot (the badge's status words
    # stay truthful after a transition without its own poll). Drive
    # it through the real action wire and read both back.
    task_slot = task_state(auth, task_id)
    refresh = task_slot.get("refresh")
    check(refresh is not None, "the task slot carries its sealed refresh action")
    if refresh:
        act(auth, refresh, {})
    # Flip the round's state through the push wire, refresh again,
    # and the embedded slot must name the NEW status.
    status, _ = saddle_api("POST", "/api/saddle/results", fake_token,
                           {"session_id": session_id, "state": "done",
                            "summary": "T7 round settled"})
    check(status == 200, "the round settles through the results push")
    act(auth, task_state(auth, task_id)["refresh"], {})
    settled = task_state(auth, task_id)
    refreshed_log = slot_public(auth, "/tasks/%d" % task_id, "task_log")
    check(refreshed_log.get("task_status") == settled.get("status"),
          "the embedded log slot's status tracks the task slot after a pulse refresh")

    print("== B8: the goal surface's run links navigate same-tab ==")
    # goal_detail's run rows and header link dropped target=_blank
    # (the goal page is T6's; only the log-link shape is T7's concern).
    goal_page_shape = open(os.path.join(os.path.dirname(os.path.abspath(__file__)),
                                        os.pardir, os.pardir,
                                        "src", "pages", "goal_detail.html"),
                           encoding="utf-8").read()
    check('run.terminal_url" target="_blank"' not in goal_page_shape,
          "the goal page's run links no longer open a new tab")