giles
b51b050dda
Move relations component loading into app.py. Move test rendering functions to test/sxc/pages/__init__.py, update route imports, and delete both sx_components.py files. Zero sx_components imports remain. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
241 lines
9.0 KiB
Python
241 lines
9.0 KiB
Python
"""Test service s-expression page components."""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
from datetime import datetime
|
|
|
|
from shared.sx.jinja_bridge import load_service_components
|
|
from shared.sx.helpers import (
|
|
render_to_sx, SxExpr,
|
|
root_header_sx, full_page_sx, header_child_sx,
|
|
)
|
|
|
|
# Load test-specific .sx components at import time
|
|
load_service_components(os.path.dirname(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 _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
|
|
# ---------------------------------------------------------------------------
|
|
|
|
async def test_detail_sx(test: dict) -> str:
|
|
"""Return s-expression wire format for a test detail view."""
|
|
inner = await render_to_sx(
|
|
"test-detail",
|
|
nodeid=test["nodeid"],
|
|
outcome=test["outcome"],
|
|
duration=str(test["duration"]),
|
|
longrepr=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' {inner})'
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Sx-native versions — return sx source (not HTML)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
async def _test_header_sx(ctx: dict, active_service: str | None = None) -> str:
|
|
"""Build the Tests menu-row as sx call."""
|
|
nav = await _service_nav_sx(ctx, active_service)
|
|
return await render_to_sx("menu-row-sx",
|
|
id="test-row", level=1, colour="sky",
|
|
link_href="/", link_label="Tests", icon="fa fa-flask",
|
|
nav=SxExpr(nav),
|
|
child_id="test-header-child",
|
|
)
|
|
|
|
|
|
async def _service_nav_sx(ctx: dict, active_service: str | None = None) -> str:
|
|
"""Service filter nav as sx."""
|
|
from runner import _SERVICE_ORDER
|
|
parts = []
|
|
parts.append(await render_to_sx("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(await render_to_sx("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) + ")"
|
|
|
|
|
|
async def _header_stack_sx(ctx: dict, active_service: str | None = None) -> str:
|
|
"""Full header stack as sx."""
|
|
hdr = await root_header_sx(ctx)
|
|
inner = await _test_header_sx(ctx, active_service)
|
|
child = await header_child_sx(inner)
|
|
return "(<> " + hdr + " " + child + ")"
|
|
|
|
|
|
async def _test_rows_sx(tests: list[dict]) -> str:
|
|
"""Render all test result rows as sx."""
|
|
parts = []
|
|
for t in tests:
|
|
parts.append(await render_to_sx("test-row",
|
|
nodeid=t["nodeid"],
|
|
outcome=t["outcome"],
|
|
duration=str(t["duration"]),
|
|
longrepr=t.get("longrepr", ""),
|
|
))
|
|
return "(<> " + " ".join(parts) + ")"
|
|
|
|
|
|
async def _grouped_rows_sx(tests: list[dict]) -> str:
|
|
"""Test rows grouped by service as sx."""
|
|
from runner import group_tests_by_service
|
|
sections = group_tests_by_service(tests)
|
|
parts = []
|
|
for sec in sections:
|
|
parts.append(await render_to_sx("test-service-header",
|
|
service=sec["service"],
|
|
total=str(sec["total"]),
|
|
passed=str(sec["passed"]),
|
|
failed=str(sec["failed"]),
|
|
))
|
|
parts.append(await _test_rows_sx(sec["tests"]))
|
|
return "(<> " + " ".join(parts) + ")"
|
|
|
|
|
|
async def _results_partial_sx(result: dict | None, running: bool, csrf: str,
|
|
active_filter: str | None = None,
|
|
active_service: str | None = None) -> str:
|
|
"""Results section as sx."""
|
|
if running and not result:
|
|
summary = await render_to_sx("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 + " " + await render_to_sx("test-running-indicator") + ")"
|
|
|
|
if not result:
|
|
summary = await render_to_sx("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 + " " + await render_to_sx("test-no-results") + ")"
|
|
|
|
status = "running" if running else result["status"]
|
|
summary = await render_to_sx("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 + " " + await render_to_sx("test-running-indicator") + ")"
|
|
|
|
tests = result.get("tests", [])
|
|
tests = _filter_tests(tests, active_filter, active_service)
|
|
if not tests:
|
|
return "(<> " + summary + " " + await render_to_sx("test-no-results") + ")"
|
|
|
|
has_failures = result["failed"] > 0 or result["errors"] > 0
|
|
rows = await _grouped_rows_sx(tests)
|
|
table = await render_to_sx("test-results-table",
|
|
rows=SxExpr(rows),
|
|
has_failures=str(has_failures).lower(),
|
|
)
|
|
return "(<> " + summary + " " + table + ")"
|
|
|
|
|
|
def _wrap_results_div_sx(inner: str, running: bool) -> str:
|
|
"""Wrap results in a div with HTMX polling (sx)."""
|
|
attrs = ':id "test-results" :class "space-y-6 p-4"'
|
|
if running:
|
|
attrs += ' :sx-get "/results" :sx-trigger "every 2s" :sx-swap "outerHTML"'
|
|
return f'(div {attrs} {inner})'
|
|
|
|
|
|
async def render_dashboard_page_sx(ctx: dict, result: dict | None,
|
|
running: bool, csrf: str,
|
|
active_filter: str | None = None,
|
|
active_service: str | None = None) -> str:
|
|
"""Full page: test dashboard (sx wire format)."""
|
|
hdr = await _header_stack_sx(ctx, active_service)
|
|
inner = await _results_partial_sx(result, running, csrf, active_filter, active_service)
|
|
content = _wrap_results_div_sx(inner, running)
|
|
return await full_page_sx(ctx, header_rows=hdr, content=content)
|
|
|
|
|
|
async def render_results_partial_sx(result: dict | None, running: bool,
|
|
csrf: str,
|
|
active_filter: str | None = None,
|
|
active_service: str | None = None) -> str:
|
|
"""HTMX partial: results section (sx wire format)."""
|
|
inner = await _results_partial_sx(result, running, csrf, active_filter, active_service)
|
|
return _wrap_results_div_sx(inner, running)
|
|
|
|
|
|
async def render_test_detail_page_sx(ctx: dict, test: dict) -> str:
|
|
"""Full page: test detail (sx wire format)."""
|
|
root_hdr = await root_header_sx(ctx)
|
|
test_row = await _test_header_sx(ctx)
|
|
detail_row = await render_to_sx("menu-row-sx",
|
|
id="test-detail-row", level=2, colour="sky",
|
|
link_href=f"/test/{test['nodeid']}",
|
|
link_label=test["nodeid"].rsplit("::", 1)[-1],
|
|
)
|
|
hdr_child_detail = await header_child_sx(detail_row, id="test-header-child")
|
|
inner = "(<> " + test_row + " " + hdr_child_detail + ")"
|
|
hdr_child_inner = await header_child_sx(inner)
|
|
hdr = "(<> " + root_hdr + " " + hdr_child_inner + ")"
|
|
content = await render_to_sx("test-detail",
|
|
nodeid=test["nodeid"],
|
|
outcome=test["outcome"],
|
|
duration=str(test["duration"]),
|
|
longrepr=test.get("longrepr", ""),
|
|
)
|
|
return await full_page_sx(ctx, header_rows=hdr, content=content)
|