Slim sxc/pages/__init__.py for federation, test, cart, blog
Move render functions, layouts, helpers, and utils from __init__.py to sub-modules (renders.py, layouts.py, helpers.py, utils.py). Update all bp route imports to point at sub-modules directly. Each __init__.py is now ≤20 lines of setup + registration. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -14,7 +14,7 @@ def register(url_prefix: str = "/") -> Blueprint:
|
||||
"""Full page dashboard with last results."""
|
||||
from shared.sx.page import get_template_context
|
||||
from shared.browser.app.csrf import generate_csrf_token
|
||||
from sxc.pages import render_dashboard_page_sx
|
||||
from sxc.pages.renders import render_dashboard_page_sx
|
||||
import runner
|
||||
|
||||
ctx = await get_template_context()
|
||||
@@ -63,12 +63,12 @@ def register(url_prefix: str = "/") -> Blueprint:
|
||||
if is_htmx:
|
||||
# S-expression wire format — sx.js renders client-side
|
||||
from shared.sx.helpers import sx_response
|
||||
from sxc.pages import test_detail_sx
|
||||
from sxc.pages.renders import test_detail_sx
|
||||
return sx_response(await test_detail_sx(test))
|
||||
|
||||
# Full page render (direct navigation / refresh)
|
||||
from shared.sx.page import get_template_context
|
||||
from sxc.pages import render_test_detail_page_sx
|
||||
from sxc.pages.renders import render_test_detail_page_sx
|
||||
|
||||
ctx = await get_template_context()
|
||||
html = await render_test_detail_page_sx(ctx, test)
|
||||
@@ -78,7 +78,7 @@ def register(url_prefix: str = "/") -> Blueprint:
|
||||
async def results():
|
||||
"""HTMX partial — poll target for results table."""
|
||||
from shared.browser.app.csrf import generate_csrf_token
|
||||
from sxc.pages import render_results_partial_sx
|
||||
from sxc.pages.renders import render_results_partial_sx
|
||||
import runner
|
||||
|
||||
result = runner.get_results()
|
||||
|
||||
@@ -1,145 +1,13 @@
|
||||
"""Test service s-expression page components."""
|
||||
"""Test service defpage setup."""
|
||||
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 setup_test_pages() -> None:
|
||||
"""Load test page definitions."""
|
||||
_load_test_page_files()
|
||||
|
||||
|
||||
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)
|
||||
def _load_test_page_files() -> None:
|
||||
import os
|
||||
from shared.sx.pages import load_page_dir
|
||||
load_page_dir(os.path.dirname(__file__), "test")
|
||||
|
||||
145
test/sxc/pages/renders.py
Normal file
145
test/sxc/pages/renders.py
Normal file
@@ -0,0 +1,145 @@
|
||||
"""Test service render functions — called from bp routes."""
|
||||
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)
|
||||
Reference in New Issue
Block a user