rose-ash/test/sxc/pages/__init__.py

"""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, render_to_sx_with_env, _ctx_to_env, full_page_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")


_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


def _service_list() -> list[str]:
    from runner import _SERVICE_ORDER
    return list(_SERVICE_ORDER)


def _build_summary_data(result: dict | None, running: bool, csrf: str,
                        active_filter: str | None) -> dict:
    """Prepare summary data dict for the ~test-results-partial defcomp."""
    if running and not result:
        return dict(state="running", 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)
    if not result:
        return dict(state="no-results", 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)
    status = "running" if running else result["status"]
    return dict(
        state="running" if running else "has-results",
        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,
    )


async def test_detail_sx(test: dict) -> str:
    """Return s-expression wire format for a test detail view."""
    return await render_to_sx("test-detail-section", test=test)


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)."""
    from runner import group_tests_by_service

    summary_data = _build_summary_data(result, running, csrf, active_filter)
    sections = []
    has_failures = "false"
    if result and not running:
        tests = _filter_tests(result.get("tests", []), active_filter, active_service)
        if tests:
            sections = group_tests_by_service(tests)
            has_failures = str(result["failed"] > 0 or result["errors"] > 0).lower()
        else:
            summary_data["state"] = "empty-filtered"

    inner = await render_to_sx("test-results-partial",
        summary_data=summary_data, sections=sections, has_failures=has_failures)
    content = await render_to_sx("test-results-wrap", running=running, inner=SxExpr(inner))
    hdr = await render_to_sx_with_env("test-layout-full", _ctx_to_env(ctx),
        services=_service_list(),
        active_service=active_service,
    )
    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)."""
    from runner import group_tests_by_service

    summary_data = _build_summary_data(result, running, csrf, active_filter)
    sections = []
    has_failures = "false"
    if result and not running:
        tests = _filter_tests(result.get("tests", []), active_filter, active_service)
        if tests:
            sections = group_tests_by_service(tests)
            has_failures = str(result["failed"] > 0 or result["errors"] > 0).lower()
        else:
            summary_data["state"] = "empty-filtered"

    inner = await render_to_sx("test-results-partial",
        summary_data=summary_data, sections=sections, has_failures=has_failures)
    return await render_to_sx("test-results-wrap", running=running, inner=SxExpr(inner))


async def render_test_detail_page_sx(ctx: dict, test: dict) -> str:
    """Full page: test detail (sx wire format)."""
    hdr = await render_to_sx_with_env("test-detail-layout-full", _ctx_to_env(ctx),
        services=_service_list(),
        test_nodeid=test["nodeid"],
        test_label=test["nodeid"].rsplit("::", 1)[-1],
    )
    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)