giles
fec5ecdfb1
- HTMX beforeSwap hook intercepts text/sexp responses and renders them client-side via sexp.js before HTMX swaps the result in - sexp_response() helper for returning text/sexp from route handlers - Test detail page (/test/<nodeid>) with clickable test names - HTMX navigation to detail returns sexp wire format (4x smaller than pre-rendered HTML), full page loads render server-side - ~test-detail component with back link, outcome badge, and error traceback display Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
254 lines
9.0 KiB
Python
254 lines
9.0 KiB
Python
"""Test service s-expression page components."""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
from datetime import datetime
|
|
|
|
from shared.sexp.jinja_bridge import render, load_service_components, client_components_tag
|
|
from shared.sexp.helpers import root_header_html, full_page
|
|
|
|
# Load test-specific .sexpr components at import time
|
|
load_service_components(os.path.dirname(os.path.dirname(__file__)))
|
|
|
|
|
|
def _format_time(ts: float | None) -> str:
|
|
"""Format a unix timestamp for display."""
|
|
if not ts:
|
|
return "never"
|
|
return datetime.fromtimestamp(ts).strftime("%-d %b %Y, %H:%M:%S")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Menu / header
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_FILTER_MAP = {
|
|
"passed": "passed",
|
|
"failed": "failed",
|
|
"errors": "error",
|
|
"skipped": "skipped",
|
|
}
|
|
|
|
|
|
def _test_header_html(ctx: dict, active_service: str | None = None) -> str:
|
|
"""Build the Tests menu-row (level 1) with service nav links."""
|
|
nav = _service_nav_html(ctx, active_service)
|
|
return render(
|
|
"menu-row",
|
|
id="test-row", level=1, colour="sky",
|
|
link_href="/", link_label="Tests", icon="fa fa-flask",
|
|
nav_html=nav,
|
|
child_id="test-header-child",
|
|
)
|
|
|
|
|
|
def _service_nav_html(ctx: dict, active_service: str | None = None) -> str:
|
|
"""Render service filter nav links using ~nav-link component."""
|
|
from runner import _SERVICE_ORDER
|
|
parts = []
|
|
# "All" link
|
|
parts.append(render(
|
|
"nav-link",
|
|
href="/",
|
|
label="all",
|
|
is_selected="true" if not active_service else None,
|
|
select_colours="aria-selected:bg-sky-200 aria-selected:text-sky-900",
|
|
))
|
|
for svc in _SERVICE_ORDER:
|
|
parts.append(render(
|
|
"nav-link",
|
|
href=f"/?service={svc}",
|
|
label=svc,
|
|
is_selected="true" if active_service == svc else None,
|
|
select_colours="aria-selected:bg-sky-200 aria-selected:text-sky-900",
|
|
))
|
|
return "".join(parts)
|
|
|
|
|
|
def _header_stack_html(ctx: dict, active_service: str | None = None) -> str:
|
|
"""Full header stack: root row + tests child row."""
|
|
hdr = root_header_html(ctx)
|
|
inner = _test_header_html(ctx, active_service)
|
|
hdr += render("header-child", inner_html=inner)
|
|
return hdr
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Test rows / grouping
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _test_rows_html(tests: list[dict]) -> str:
|
|
"""Render all test result rows."""
|
|
parts = []
|
|
for t in tests:
|
|
parts.append(render(
|
|
"test-row",
|
|
nodeid=t["nodeid"],
|
|
outcome=t["outcome"],
|
|
duration=str(t["duration"]),
|
|
longrepr=t.get("longrepr", ""),
|
|
))
|
|
return "".join(parts)
|
|
|
|
|
|
def _grouped_rows_html(tests: list[dict]) -> str:
|
|
"""Render test rows grouped by service with section headers."""
|
|
from runner import group_tests_by_service
|
|
sections = group_tests_by_service(tests)
|
|
parts = []
|
|
for sec in sections:
|
|
parts.append(render(
|
|
"test-service-header",
|
|
service=sec["service"],
|
|
total=str(sec["total"]),
|
|
passed=str(sec["passed"]),
|
|
failed=str(sec["failed"]),
|
|
))
|
|
parts.append(_test_rows_html(sec["tests"]))
|
|
return "".join(parts)
|
|
|
|
|
|
def _filter_tests(tests: list[dict], active_filter: str | None,
|
|
active_service: str | None) -> list[dict]:
|
|
"""Filter tests by outcome and/or service."""
|
|
from runner import _service_from_nodeid
|
|
filtered = tests
|
|
if active_filter and active_filter in _FILTER_MAP:
|
|
outcome = _FILTER_MAP[active_filter]
|
|
filtered = [t for t in filtered if t["outcome"] == outcome]
|
|
if active_service:
|
|
filtered = [t for t in filtered if _service_from_nodeid(t["nodeid"]) == active_service]
|
|
return filtered
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Results partial
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _results_partial_html(result: dict | None, running: bool, csrf: str,
|
|
active_filter: str | None = None,
|
|
active_service: str | None = None) -> str:
|
|
"""Render the results section (summary + table or running indicator)."""
|
|
if running and not result:
|
|
summary = render(
|
|
"test-summary",
|
|
status="running", passed="0", failed="0", errors="0",
|
|
skipped="0", total="0", duration="...",
|
|
last_run="in progress", running=True, csrf=csrf,
|
|
active_filter=active_filter,
|
|
)
|
|
return summary + render("test-running-indicator")
|
|
|
|
if not result:
|
|
summary = render(
|
|
"test-summary",
|
|
status=None, passed="0", failed="0", errors="0",
|
|
skipped="0", total="0", duration="0",
|
|
last_run="never", running=running, csrf=csrf,
|
|
active_filter=active_filter,
|
|
)
|
|
return summary + render("test-no-results")
|
|
|
|
status = "running" if running else result["status"]
|
|
summary = render(
|
|
"test-summary",
|
|
status=status,
|
|
passed=str(result["passed"]),
|
|
failed=str(result["failed"]),
|
|
errors=str(result["errors"]),
|
|
skipped=str(result.get("skipped", 0)),
|
|
total=str(result["total"]),
|
|
duration=str(result["duration"]),
|
|
last_run=_format_time(result["finished_at"]) if not running else "in progress",
|
|
running=running,
|
|
csrf=csrf,
|
|
active_filter=active_filter,
|
|
)
|
|
|
|
if running:
|
|
return summary + render("test-running-indicator")
|
|
|
|
tests = result.get("tests", [])
|
|
tests = _filter_tests(tests, active_filter, active_service)
|
|
if not tests:
|
|
return summary + render("test-no-results")
|
|
|
|
has_failures = result["failed"] > 0 or result["errors"] > 0
|
|
rows = _grouped_rows_html(tests)
|
|
table = render("test-results-table", rows_html=rows,
|
|
has_failures=str(has_failures).lower())
|
|
return summary + table
|
|
|
|
|
|
def _wrap_results_div(inner_html: str, running: bool) -> str:
|
|
"""Wrap results in a div with HTMX polling when running."""
|
|
attrs = 'id="test-results" class="space-y-6 p-4"'
|
|
if running:
|
|
attrs += ' hx-get="/results" hx-trigger="every 2s" hx-swap="outerHTML"'
|
|
return f'<div {attrs}>{inner_html}</div>'
|
|
|
|
|
|
async def render_dashboard_page(ctx: dict, result: dict | None,
|
|
running: bool, csrf: str,
|
|
active_filter: str | None = None,
|
|
active_service: str | None = None) -> str:
|
|
"""Full page: test dashboard."""
|
|
hdr = _header_stack_html(ctx, active_service)
|
|
inner = _results_partial_html(result, running, csrf, active_filter, active_service)
|
|
content = _wrap_results_div(inner, running)
|
|
body_end = client_components_tag()
|
|
return full_page(ctx, header_rows_html=hdr, content_html=content,
|
|
body_end_html=body_end)
|
|
|
|
|
|
async def render_results_partial(result: dict | None, running: bool,
|
|
csrf: str,
|
|
active_filter: str | None = None,
|
|
active_service: str | None = None) -> str:
|
|
"""HTMX partial: just the results section (wrapped in polling div)."""
|
|
inner = _results_partial_html(result, running, csrf, active_filter, active_service)
|
|
return _wrap_results_div(inner, running)
|
|
|
|
|
|
async def render_test_detail_page(ctx: dict, test: dict) -> str:
|
|
"""Full page: test detail view."""
|
|
hdr = _header_stack_html(ctx)
|
|
detail_row = render(
|
|
"menu-row",
|
|
id="test-detail-row", level=2, colour="sky",
|
|
link_href=f"/test/{test['nodeid']}",
|
|
link_label=test["nodeid"].rsplit("::", 1)[-1],
|
|
)
|
|
hdr += render("header-child", id="test-header-child",
|
|
inner_html=detail_row)
|
|
content = render(
|
|
"test-detail",
|
|
nodeid=test["nodeid"],
|
|
outcome=test["outcome"],
|
|
duration=str(test["duration"]),
|
|
longrepr=test.get("longrepr", ""),
|
|
)
|
|
body_end = client_components_tag()
|
|
return full_page(ctx, header_rows_html=hdr, content_html=content,
|
|
body_end_html=body_end)
|
|
|
|
|
|
def test_detail_sexp(test: dict) -> str:
|
|
"""Return s-expression wire format for a test detail view.
|
|
|
|
When a client has sexp.js loaded (HTMX navigation from the dashboard),
|
|
we can send the raw s-expression instead of pre-rendered HTML.
|
|
sexp.js will render it client-side.
|
|
"""
|
|
from shared.sexp.parser import serialize
|
|
nodeid = serialize(test["nodeid"])
|
|
outcome = serialize(test["outcome"])
|
|
duration = serialize(str(test["duration"]))
|
|
longrepr = serialize(test.get("longrepr", ""))
|
|
return (
|
|
f"(section :id \"main-panel\""
|
|
f" :class \"flex-1 md:h-full md:min-h-0 overflow-y-auto overscroll-contain js-grid-viewport\""
|
|
f" (~test-detail :nodeid {nodeid} :outcome {outcome}"
|
|
f" :duration {duration} :longrepr {longrepr}))"
|
|
)
|