clanker-saddle/test/bundles/t10views.py

"""The T10 views bundle (goal 41 task 10): the two phase-D structural
bets — saved views / URL state on /tasks (ux-notes bet 4) and the
cross-entity work stream on the dashboard (bet 5).

Bet 4 is a CLIENT-side engine (the statum entrypoint fetch carries no
query string, so filter state lives in location.hash): this bundle
verifies the engine's round-trip as pure logic (a python port of the
serialise/parse pair driving the exact page functions) plus the
markup hooks the page ships. The prefill namespace (#new?project=)
must keep coexisting with the view namespace (#view?...).

Bet 5 is server data + markup: the dashboard slot's work_stream rows
(joined, time-ordered, capped, linked) are cross-checked against
seeded ledger events, and the page renders the stream feed with the
old activity table gone.
"""
from harness import *  # noqa: F401,F403 — the shared wire helpers

# The open-work status set (window.TASK_OPEN_STATUSES on the page).
OPEN = ("queued", "assigned", "working", "blocked", "suspended")


def view_hash(filters):
    """window.taskViewHash ported verbatim: only non-default axes
    serialise ('open' status, all projects, all goals stay clean)."""
    parts = []
    for key in ("status", "project", "goal"):
        value = filters.get(key)
        if key == "status":
            value = value or "open"
            if value == "open":
                continue
        elif value in (None, "", 0, "0", "all"):
            continue
        from urllib.parse import quote
        parts.append("%s=%s" % (key, quote(str(value), safe="")))
    return "#view?" + "&".join(parts) if parts else ""


def filters_from_hash(hash_text):
    """window.taskFiltersFromHash ported verbatim (None = not a view
    hash: the #new prefill namespace and bare hashes fall through)."""
    if not hash_text or not hash_text.startswith("#view?"):
        return None
    filters = {}
    for part in hash_text[len("#view?"):].split("&"):
        if "=" not in part:
            continue
        key, value = part.split("=", 1)
        from urllib.parse import unquote
        key, value = unquote(key), unquote(value)
        if key == "status" and value:
            filters["status"] = value
        elif key == "project" and value and value != "0":
            filters["project"] = value
        elif key == "goal" and value and value != "all":
            filters["goal"] = value
    return filters


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

    now = "strftime('%s','now')"

    # ------------------------------------------------------------------
    print("== bet 4: the hash engine (round-trip as pure logic) ==")
    cases = [
        ({}, ""),  # the default view keeps the URL clean
        ({"status": "open"}, ""),
        ({"status": "working"}, "#view?status=working"),
        ({"status": "blocked"}, "#view?status=blocked"),
        ({"status": "all"}, "#view?status=all"),
        ({"project": 3}, "#view?project=3"),
        ({"project": "3"}, "#view?project=3"),
        ({"project": 0}, ""),  # "all projects" is the default
        ({"goal": "standalone"}, "#view?goal=standalone"),
        ({"goal": "all"}, ""),
        ({"goal": 12}, "#view?goal=12"),
        ({"status": "working", "project": 3, "goal": "standalone"},
         "#view?status=working&project=3&goal=standalone"),
        # A title with spaces/ampersands survives the round-trip
        # encoded (the goal axis carries ids in practice, but the
        # engine must not corrupt any value).
        ({"goal": "a & b=c"}, "#view?goal=a%20%26%20b%3Dc"),
    ]
    for filters, expected in cases:
        got = view_hash(filters)
        check(got == expected,
              "serialise %r -> %r" % (filters, expected))
        if expected:
            back = filters_from_hash(expected)
            wanted = {}
            for key, value in filters.items():
                text = str(value)
                if key == "status":
                    if text not in ("open", ""):
                        wanted[key] = text
                elif text not in ("0", "all", ""):
                    wanted[key] = text
            check(back == wanted,
                  "parse %r restores %r (got %r)" % (expected, wanted, back))
    # The prefill namespace never parses as a view.
    check(filters_from_hash("#new?project=7") is None,
          "the #new prefill hash is not a view hash")
    check(filters_from_hash("") is None and filters_from_hash("#") is None,
          "bare hashes are not view hashes")

    print("== bet 4: the page ships the engine + the view row ==")
    tasks_page = http_get_page(auth, "/tasks")
    for needle, label in (
            ("window.taskViewHash", "the serialiser"),
            ("window.taskFiltersFromHash", "the parser"),
            ("window.applyTaskViewHash", "the load-time seed"),
            ("window.copyTaskViewLink", "the copy-link affordance"),
            ("syncTaskHash", "the filter-change hash write"),
    ):
        check(needle in tasks_page, "the tasks page ships %s" % label)
    check("syncTaskHash();" in tasks_page,
          "every filter change lands in the hash (setTaskFilter)")
    check("window.history.replaceState" in tasks_page,
          "hash updates use replaceState (no Back-button spam)")
    check("hashchange" in tasks_page,
          "a pasted #view hash re-seeds the filters in-place")
    check("window.applyTaskViewHash();" in tasks_page,
          "the hash seeds taskFilters before the first render")
    check("#view?" in tasks_page,
          "the view namespace is distinct from the #new prefill")
    # The view row: presets + copy button.
    check('data-task-view-row' in tasks_page,
          "the view row is marked for the bundle/harness")
    check("Copy view link" in tasks_page
          and "copyTaskViewLink" in tasks_page,
          "the one-click copy-link button ships")
    for preset in ("Open work", "All", "Blocked"):
        check(preset in tasks_page,
              "the %r preset chip ships in the view row" % preset)
    check("navigator.clipboard" in tasks_page,
          "the copy uses the async clipboard API")
    # The presets ARE the chips' statuses (no second filter state).
    check(tasks_page.count("window.taskChipClass('open')") >= 2,
          "the presets reuse the chip engine's status vocabulary")

    print("== bet 4: prefill coexistence ==")
    # T9's #new?project= deep-link still parses + drops itself, and
    # the view engine ignores it (namespace check above; markup).
    check("indexOf('#new')" in tasks_page,
          "the #new prefill contract still ships (T9)")
    check("indexOf('#view?')" in tasks_page,
          "the view parser gates on its own namespace")
    # A prefill hash and a view hash in one URL: the prefill wins on
    # load (it consumes and drops the whole hash); the view engine
    # then reads nothing - the ported parser mirrors that by gating.
    check(filters_from_hash("#new?project=7") is None,
          "a load-time prefill never double-applies as a view")

    # ------------------------------------------------------------------
    print("== bet 5: the work stream (joined, ordered, capped, linked) ==")
    # Seed the three row grammars through the ledger the timeline
    # reads: a task progress line, a goal-scoped narrative event and
    # a plain task comment (log highlight).
    conn = db()
    conn.execute(
        "insert into tasks (project_id, title, status, priority, created_at)"
        " values (1, 'T10 stream task', 'working', 5, %s)" % now)
    stream_task = conn.execute(
        "select last_insert_rowid()").fetchone()[0]
    conn.execute(
        "insert into goals (project_id, title, brief, branch_name, status,"
        " created_at) values (1, 'T10 stream goal', 'The stream probe.',"
        " 'goal/t10', 'running', %s)" % now)
    stream_goal = conn.execute(
        "select last_insert_rowid()").fetchone()[0]
    # Goal runs are internal - their task rows never stream.
    conn.execute(
        "insert into tasks (project_id, goal_id, kind, title, status,"
        " priority, created_at) values (1, ?, 'goal_run',"
        " 'T10 internal run', 'working', 5, %s)" % now, (stream_goal,))
    run_task = conn.execute(
        "select last_insert_rowid()").fetchone()[0]

    def event(kind, scope_type, scope_id, payload, offset):
        conn.execute(
            "insert into events (kind, actor, scope_type, scope_id,"
            " payload, created_at) values (?, 'model', ?, ?, ?,"
            " strftime('%s','now') - ?)",
            (kind, scope_type, scope_id, payload, offset))

    event("progress", "task", stream_task,
          '{"text": "T10 progress line"}', 30)
    event("state_change", "goal", stream_goal,
          '{"to": "running", "text": "T10 goal narrative"}', 20)
    event("comment", "task", stream_task,
          '{"author": "human", "body": "T10 comment highlight"}', 10)
    event("progress", "task", run_task,
          '{"text": "T10 internal run progress"}', 5)
    conn.commit()
    conn.close()

    dash = slot_public(auth, "/", "dashboard")
    stream = dash.get("work_stream", [])
    check(isinstance(stream, list),
          "the dashboard slot carries the work_stream rows")
    check(0 < len(stream) <= 15,
          "the stream is capped at 15 rows (got %d)" % len(stream))
    epochs = [int(row["at_epoch"]) for row in stream]
    check(epochs == sorted(epochs, reverse=True),
          "the stream is newest-first")

    progress_rows = [row for row in stream
                     if row["kind"] == "task_progress"]
    goal_rows = [row for row in stream if row["kind"] == "goal_event"]
    highlight_rows = [row for row in stream
                      if row["kind"] == "log_highlight"]
    check(any("T10 progress line" in row["text"]
              and int(row["task_id"]) == stream_task
              and row["href"] == "/tasks/%d" % stream_task
              for row in progress_rows),
          "task progress rows join with the task link")
    check(any("T10 goal narrative" in row["text"]
              and int(row["goal_id"]) == stream_goal
              and row["href"] == "/goals/%d" % stream_goal
              for row in goal_rows),
          "goal narrative rows join with the goal link")
    check(any("T10 comment highlight" in row["text"]
              and int(row["task_id"]) == stream_task
              for row in highlight_rows),
          "comment rows stream as log highlights")
    check(all(not ("internal run" in row["text"]
                   or int(row.get("task_id", 0)) == run_task)
              for row in stream),
          "goal-run internals never stream (the goal carries them)")

    # The three grammars agree with the activity projection they
    # derive from (one window, two projections - same order).
    activity = dash.get("activity", [])
    check(len(stream) <= len(activity),
          "the stream is the window's newest slice (activity keeps the full window)")
    check(len(activity) > 15 or len(stream) == len(activity),
          "under the cap the stream covers the whole window")

    # Every row links somewhere or is a bare highlight.
    for row in stream:
        if row["href"]:
            check(row["href"].startswith(("/tasks/", "/goals/")),
                  "row hrefs stay in-app (%s)" % row["href"])
            break

    print("== bet 5: the dashboard renders the stream as its feed ==")
    home_page = http_get_page(auth, "/")
    check('data-work-stream' in home_page,
          "the dashboard ships the work-stream feed")
    check("work-stream" in home_page and "dashboard.work_stream" in home_page,
          "the feed renders the slot's work_stream rows")
    check("Work stream" in home_page,
          "the feed is titled as the work stream")
    check("badge-stream-" in home_page,
          "the rows carry the stream's badge vocabulary")
    # The old three-grammar surfaces it replaces on THIS page: the
    # activity table is gone from the dashboard.
    check("dash-activity-table" not in home_page,
          "the old activity table is gone from the dashboard")
    check("data-activity-toggle" not in home_page,
          "the activity Show-all toggle is gone (the server caps)")
    check("Show all" not in home_page,
          "no stale Show-all control remains")
    # The new-since badge keeps working off the stream rows.
    check("state.work_stream" in home_page,
          "the new-since badge counts stream rows")
    # The underlying surfaces live elsewhere (not deleted): the goal
    # page narrative + the task log stay their pages' own.
    check("data-activity-new" in home_page,
          "the ambient expander frame (with the new badge) keeps the feed quiet")

    print("== bet 5: the stream survives the pulse refresh ==")
    refresh = dash.get("refresh")
    check(refresh is not None, "the dashboard refresh action is authored")
    act(auth, refresh, {})
    dash2 = slot_public(auth, "/", "dashboard")
    check(len(dash2.get("work_stream", [])) > 0,
          "the sealed refresh rebuilds the stream too")