more plans

test-sx-web/runner.py

@@ -0,0 +1,213 @@
+"""Pytest subprocess runner + in-memory result storage."""
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+import os
+import time
+from collections import OrderedDict
+from pathlib import Path
+
+log = logging.getLogger(__name__)
+
+# In-memory state
+_last_result: dict | None = None
+_running: bool = False
+
+# Each service group runs in its own pytest subprocess with its own PYTHONPATH
+_SERVICE_GROUPS: list[dict] = [
+    {"name": "shared", "dirs": ["shared/tests/", "shared/sx/tests/"],
+     "pythonpath": None},
+    {"name": "sx_docs", "dirs": ["sx_docs/tests/"],
+     "pythonpath": "/app/sx_docs"},
+]
+
+_SERVICE_ORDER = [g["name"] for g in _SERVICE_GROUPS]
+_REPORT_PATH = "/tmp/test-report-{}.json"
+
+
+def _parse_report(path: str) -> tuple[list[dict], dict]:
+    """Parse a pytest-json-report file."""
+    rp = Path(path)
+    if not rp.exists():
+        return [], {}
+    try:
+        report = json.loads(rp.read_text())
+    except (json.JSONDecodeError, OSError):
+        return [], {}
+
+    summary = report.get("summary", {})
+    tests_raw = report.get("tests", [])
+
+    tests = []
+    for t in tests_raw:
+        tests.append({
+            "nodeid": t.get("nodeid", ""),
+            "outcome": t.get("outcome", "unknown"),
+            "duration": round(t.get("duration", 0), 4),
+            "longrepr": (t.get("call", {}) or {}).get("longrepr", ""),
+        })
+    return tests, summary
+
+
+async def _run_group(group: dict) -> tuple[list[dict], dict, str]:
+    """Run pytest for a single service group."""
+    existing = [d for d in group["dirs"] if Path(f"/app/{d}").is_dir()]
+    if not existing:
+        return [], {}, ""
+
+    report_file = _REPORT_PATH.format(group["name"])
+    cmd = [
+        "python3", "-m", "pytest",
+        *existing,
+        "--json-report",
+        f"--json-report-file={report_file}",
+        "-q",
+        "--tb=short",
+    ]
+    env = {**os.environ}
+    if group["pythonpath"]:
+        env["PYTHONPATH"] = group["pythonpath"] + ":" + env.get("PYTHONPATH", "")
+
+    proc = await asyncio.create_subprocess_exec(
+        *cmd,
+        stdout=asyncio.subprocess.PIPE,
+        stderr=asyncio.subprocess.STDOUT,
+        cwd="/app",
+        env=env,
+    )
+    stdout, _ = await proc.communicate()
+    stdout_str = (stdout or b"").decode("utf-8", errors="replace")
+    tests, summary = _parse_report(report_file)
+    return tests, summary, stdout_str
+
+
+async def run_tests() -> dict:
+    """Run pytest in subprocess, parse JSON report, store results."""
+    global _last_result, _running
+
+    if _running:
+        return {"status": "already_running"}
+
+    _running = True
+    started_at = time.time()
+
+    try:
+        tasks = [_run_group(g) for g in _SERVICE_GROUPS]
+        results = await asyncio.gather(*tasks, return_exceptions=True)
+
+        all_tests: list[dict] = []
+        total_passed = total_failed = total_errors = total_skipped = total_count = 0
+        all_stdout: list[str] = []
+
+        for i, res in enumerate(results):
+            if isinstance(res, Exception):
+                log.error("Group %s failed: %s", _SERVICE_GROUPS[i]["name"], res)
+                continue
+            tests, summary, stdout_str = res
+            all_tests.extend(tests)
+            total_passed += summary.get("passed", 0)
+            total_failed += summary.get("failed", 0)
+            total_errors += summary.get("error", 0)
+            total_skipped += summary.get("skipped", 0)
+            total_count += summary.get("total", len(tests))
+            if stdout_str.strip():
+                all_stdout.append(stdout_str)
+
+        finished_at = time.time()
+        status = "failed" if total_failed > 0 or total_errors > 0 else "passed"
+
+        _last_result = {
+            "status": status,
+            "started_at": started_at,
+            "finished_at": finished_at,
+            "duration": round(finished_at - started_at, 2),
+            "passed": total_passed,
+            "failed": total_failed,
+            "errors": total_errors,
+            "skipped": total_skipped,
+            "total": total_count,
+            "tests": all_tests,
+            "stdout": "\n".join(all_stdout)[-5000:],
+        }
+
+        log.info(
+            "Test run complete: %s (%d passed, %d failed, %d errors, %.1fs)",
+            status, total_passed, total_failed, total_errors,
+            _last_result["duration"],
+        )
+        return _last_result
+
+    except Exception:
+        log.exception("Test run failed")
+        finished_at = time.time()
+        _last_result = {
+            "status": "error",
+            "started_at": started_at,
+            "finished_at": finished_at,
+            "duration": round(finished_at - started_at, 2),
+            "passed": 0,
+            "failed": 0,
+            "errors": 1,
+            "skipped": 0,
+            "total": 0,
+            "tests": [],
+            "stdout": "",
+        }
+        return _last_result
+    finally:
+        _running = False
+
+
+def get_results() -> dict | None:
+    """Return last run results."""
+    return _last_result
+
+
+def get_test(nodeid: str) -> dict | None:
+    """Look up a single test by nodeid."""
+    if not _last_result:
+        return None
+    for t in _last_result["tests"]:
+        if t["nodeid"] == nodeid:
+            return t
+    return None
+
+
+def is_running() -> bool:
+    """Check if tests are currently running."""
+    return _running
+
+
+def _service_from_nodeid(nodeid: str) -> str:
+    """Extract service name from a test nodeid."""
+    parts = nodeid.split("/")
+    return parts[0] if len(parts) >= 2 else "other"
+
+
+def group_tests_by_service(tests: list[dict]) -> list[dict]:
+    """Group tests into ordered sections by service."""
+    buckets: dict[str, list[dict]] = OrderedDict()
+    for svc in _SERVICE_ORDER:
+        buckets[svc] = []
+    for t in tests:
+        svc = _service_from_nodeid(t["nodeid"])
+        if svc not in buckets:
+            buckets[svc] = []
+        buckets[svc].append(t)
+
+    sections = []
+    for svc, svc_tests in buckets.items():
+        if not svc_tests:
+            continue
+        sections.append({
+            "service": svc,
+            "tests": svc_tests,
+            "total": len(svc_tests),
+            "passed": sum(1 for t in svc_tests if t["outcome"] == "passed"),
+            "failed": sum(1 for t in svc_tests if t["outcome"] == "failed"),
+            "errors": sum(1 for t in svc_tests if t["outcome"] == "error"),
+            "skipped": sum(1 for t in svc_tests if t["outcome"] == "skipped"),
+        })
+    return sections