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>
103 lines
3.3 KiB
Python
103 lines
3.3 KiB
Python
"""Test dashboard routes."""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
|
|
from quart import Blueprint, Response, make_response, request
|
|
|
|
|
|
def register(url_prefix: str = "/") -> Blueprint:
|
|
bp = Blueprint("dashboard", __name__, url_prefix=url_prefix)
|
|
|
|
@bp.get("/")
|
|
async def index():
|
|
"""Full page dashboard with last results."""
|
|
from shared.sexp.page import get_template_context
|
|
from shared.browser.app.csrf import generate_csrf_token
|
|
from sexp.sexp_components import render_dashboard_page
|
|
import runner
|
|
|
|
ctx = await get_template_context()
|
|
result = runner.get_results()
|
|
running = runner.is_running()
|
|
csrf = generate_csrf_token()
|
|
active_filter = request.args.get("filter")
|
|
active_service = request.args.get("service")
|
|
|
|
html = await render_dashboard_page(
|
|
ctx, result, running, csrf,
|
|
active_filter=active_filter,
|
|
active_service=active_service,
|
|
)
|
|
return await make_response(html, 200)
|
|
|
|
@bp.post("/run")
|
|
async def run():
|
|
"""Trigger a test run, redirect to /."""
|
|
import runner
|
|
|
|
if not runner.is_running():
|
|
asyncio.create_task(runner.run_tests())
|
|
|
|
# HX-Redirect for HTMX, regular redirect for non-HTMX
|
|
if request.headers.get("HX-Request"):
|
|
resp = Response("", status=200)
|
|
resp.headers["HX-Redirect"] = "/"
|
|
return resp
|
|
|
|
from quart import redirect as qredirect
|
|
return qredirect("/")
|
|
|
|
@bp.get("/test/<path:nodeid>")
|
|
async def test_detail(nodeid: str):
|
|
"""Test detail view — full page or sexp wire format."""
|
|
import runner
|
|
|
|
test = runner.get_test(nodeid)
|
|
if not test:
|
|
from quart import abort
|
|
abort(404)
|
|
|
|
is_htmx = bool(request.headers.get("HX-Request"))
|
|
|
|
if is_htmx:
|
|
# S-expression wire format — sexp.js renders client-side
|
|
from shared.sexp.helpers import sexp_response
|
|
from sexp.sexp_components import test_detail_sexp
|
|
return sexp_response(test_detail_sexp(test))
|
|
|
|
# Full page render (direct navigation / refresh)
|
|
from shared.sexp.page import get_template_context
|
|
from sexp.sexp_components import render_test_detail_page
|
|
|
|
ctx = await get_template_context()
|
|
html = await render_test_detail_page(ctx, test)
|
|
return await make_response(html, 200)
|
|
|
|
@bp.get("/results")
|
|
async def results():
|
|
"""HTMX partial — poll target for results table."""
|
|
from shared.browser.app.csrf import generate_csrf_token
|
|
from sexp.sexp_components import render_results_partial
|
|
import runner
|
|
|
|
result = runner.get_results()
|
|
running = runner.is_running()
|
|
csrf = generate_csrf_token()
|
|
active_filter = request.args.get("filter")
|
|
active_service = request.args.get("service")
|
|
|
|
html = await render_results_partial(
|
|
result, running, csrf,
|
|
active_filter=active_filter,
|
|
active_service=active_service,
|
|
)
|
|
|
|
resp = Response(html, status=200, content_type="text/html")
|
|
# If still running, tell HTMX to keep polling
|
|
if running:
|
|
resp.headers["HX-Trigger-After-Swap"] = "test-still-running"
|
|
return resp
|
|
|
|
return bp
|