orchestrata_ultimate.gateway.test_runner

Reference › orchestrata_ultimate › gateway › test_runner

test_runner provides TestRunner, which runs pytest as a subprocess on behalf of the gateway and streams progress to clients (SPEC-016 REQ-016-167). server constructs it with a broadcast function and a graph repository, and the test_metrics handler exposes it through RPC. See gateway.

run accepts a run id, an optional test path and extra arguments. Only one run is allowed at a time; a second call returns an error dictionary. It builds a verbose pytest command, broadcasts a started event, and reads the subprocess output line by line. _parse_line, _record_test_result and _record_summary_counts update running counts (collected, completed, passed, failed, errors, skipped) and the current test, and every line is emitted as a progress event through _emit. When the process ends the status becomes complete for a zero exit code and failed otherwise, a final event with the totals and failed tests is emitted, and _store_result saves the result to the graph. If the process outlives the wait timeout, the runner kills and reaps it. is_running reports whether a run is active.

At a glance

Class diagram

Dependency diagram

API

Classes

TestRunner

Decorators: traces(spec='016', story='US-007', reqs=['REQ-016-166']) · Line 40

Runs pytest asynchronously and broadcasts progress.

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/gateway/test_runner.py

"""Async pytest runner with real-time progress broadcasting."""

import asyncio
import json
import logging
import os
import re
import sys
from datetime import UTC, datetime
from typing import Any

from orchestrata_ultimate.core.base.traces import traces

logger = logging.getLogger(__name__)

# Resolve the interpreter running THIS process (i.e. the project's venv
# python) rather than trusting a bare "python" on PATH -- PATH need not
# include the venv's bin/ (the venv is not "activated" just because this
# process was launched with .venv/bin/python), so a bare "python" can
# silently resolve to an unrelated system interpreter lacking the project's
# pytest plugins (e.g. pytest-timeout), making every subprocess run fail.
_PYTEST_PYTHON = sys.executable or "python"

# Pytest output line parsers
_PYTEST_COLLECTED = re.compile(r"collected\s+(\d+)\s+item")
_PYTEST_PASSED = re.compile(r"(\d+)\s+passed")
_PYTEST_FAILED = re.compile(r"(\d+)\s+failed")
_PYTEST_ERROR = re.compile(r"(\d+)\s+error")
_PYTEST_SKIPPED = re.compile(r"(\d+)\s+skipped")
_PYTEST_TEST_NAME = re.compile(r"([\w/._-]+::[\w_]+)\s+(PASSED|FAILED|ERROR|SKIPPED)")

# Default ceiling on how long we wait for the pytest subprocess to exit after
# its stdout has been fully drained. Configurable per-instance (see __init__)
# so deployments can tune it and tests can exercise the timeout path without
# a real 10-minute wait.
DEFAULT_WAIT_TIMEOUT_SECONDS = 600.0


@traces(spec="016", story="US-007", reqs=["REQ-016-166"])
class TestRunner:
    """Runs pytest asynchronously and broadcasts progress.

    Emits ``tests.progress`` events with:
    - ``status``: "started" | "collecting" | "running" | "complete" | "error"
    - ``run_id``: unique identifier
    - ``collected``: total tests discovered
    - ``completed``: tests finished so far
    - ``passed`` / ``failed`` / ``errors`` / ``skipped``: counters
    - ``current_test``: name of test currently running
    - ``output``: last line of pytest output
    """

    def __init__(
        self,
        broadcast_fn: Any,
        graph_repository: Any = None,
        project_root: str = ".",
        wait_timeout_seconds: float = DEFAULT_WAIT_TIMEOUT_SECONDS,
    ) -> None:
        """Initialise the runner.

        Args:
            broadcast_fn: Async callable ``(topic, payload) -> int``
                for pushing events to subscribers.
            graph_repository: Optional operational-graph repository for
                storing results.
            project_root: Directory to run pytest from.
            wait_timeout_seconds: How long to wait for the pytest subprocess
                to exit after its stdout stream ends, before giving up and
                killing it. Defaults to 10 minutes.
        """
        self._broadcast = broadcast_fn
        self._graph_repository = graph_repository
        self._project_root = project_root
        self._wait_timeout_seconds = wait_timeout_seconds
        self._running = False
        self._current_run_id: str | None = None

    @property
    @traces(spec="016", story="US-007", reqs=["REQ-016-168"])
    def is_running(self) -> bool:
        """True if a test run is currently in progress."""
        return self._running

    @traces(spec="016", story="US-007", reqs=["REQ-016-167"])
    async def run(
        self,
        run_id: str,
        test_path: str | None = None,
        extra_args: list[str] | None = None,
    ) -> dict[str, Any]:
        """Start a pytest run and stream progress.

        Args:
            run_id: Unique identifier for this run.
            test_path: Optional path/filter to pass to pytest.
            extra_args: Additional pytest arguments.

        Returns:
            Final result dict.
        """
        if self._running:
            return {"error": "Test run already in progress"}

        self._running = True
        self._current_run_id = run_id

        start_time = datetime.now(UTC)
        cmd = [
            _PYTEST_PYTHON,
            "-m",
            "pytest",
            "-v",
            "-s",  # unbuffered output for real-time progress
            "--timeout=120",
            "--tb=short",
            "--color=no",
        ]
        if extra_args:
            cmd.extend(extra_args)
        if test_path:
            cmd.append(test_path)
        else:
            cmd.append("tests/")

        await self._emit(
            run_id,
            {
                "status": "started",
                "command": " ".join(cmd),
                "timestamp": start_time.isoformat(),
            },
        )

        state: dict[str, Any] = {
            "run_id": run_id,
            "status": "running",
            "collected": 0,
            "completed": 0,
            "passed": 0,
            "failed": 0,
            "errors": 0,
            "skipped": 0,
            "failed_tests": [],
            "output_lines": [],
        }

        proc: asyncio.subprocess.Process | None = None
        try:
            proc = await asyncio.create_subprocess_exec(
                *cmd,
                cwd=self._project_root,
                stdout=asyncio.subprocess.PIPE,
                stderr=asyncio.subprocess.STDOUT,
                env={**os.environ, "PYTHONUNBUFFERED": "1"},
            )

            assert proc.stdout is not None
            async for raw_line in proc.stdout:
                line = raw_line.decode("utf-8", errors="replace").rstrip()
                state["output_lines"].append(line)
                self._parse_line(line, state)
                await self._emit(
                    run_id,
                    {
                        "status": state["status"],
                        "collected": state["collected"],
                        "completed": state["completed"],
                        "passed": state["passed"],
                        "failed": state["failed"],
                        "errors": state["errors"],
                        "skipped": state["skipped"],
                        "current_test": state.get("current_test"),
                        "output": line,
                        "timestamp": datetime.now(UTC).isoformat(),
                    },
                )

            returncode = await asyncio.wait_for(
                proc.wait(), timeout=self._wait_timeout_seconds
            )
            duration = (datetime.now(UTC) - start_time).total_seconds()

            state["status"] = "complete" if returncode == 0 else "failed"
            state["duration_seconds"] = duration
            state["returncode"] = returncode

            await self._emit(
                run_id,
                {
                    "status": state["status"],
                    "collected": state["collected"],
                    "completed": state["completed"],
                    "passed": state["passed"],
                    "failed": state["failed"],
                    "errors": state["errors"],
                    "skipped": state["skipped"],
                    "failed_tests": state["failed_tests"],
                    "duration_seconds": duration,
                    "returncode": returncode,
                    "timestamp": datetime.now(UTC).isoformat(),
                },
            )

            await self._store_result(state)
            return state

        except TimeoutError:
            # The subprocess outlived our wait budget: kill and reap it so we
            # don't leak the process and its stdio file descriptors. proc is
            # always bound here -- the only timeout is on proc.wait() above,
            # which cannot run before proc is assigned -- but we guard
            # defensively since static analysis can't see that.
            if proc is not None:
                try:
                    proc.kill()
                except ProcessLookupError:
                    pass  # already exited
                await proc.wait()
            state["status"] = "error"
            state["error"] = f"Test run timed out after {self._wait_timeout_seconds:g}s"
            await self._emit(
                run_id, {**state, "timestamp": datetime.now(UTC).isoformat()}
            )
            return state
        except Exception as exc:
            logger.exception("Test run failed")
            state["status"] = "error"
            state["error"] = str(exc)
            await self._emit(
                run_id, {**state, "timestamp": datetime.now(UTC).isoformat()}
            )
            return state
        finally:
            self._running = False
            self._current_run_id = None

    @traces(spec="016", story="US-007", reqs=["REQ-016-166"])
    def _parse_line(self, line: str, state: dict[str, Any]) -> None:
        """Parse a single line of pytest output and update state."""
        # Detect collection count
        m = _PYTEST_COLLECTED.search(line)
        if m:
            state["collected"] = int(m.group(1))
            state["status"] = "collecting"
            return

        # Detect individual test result
        m = _PYTEST_TEST_NAME.search(line)
        if m:
            self._record_test_result(m.group(1), m.group(2), state)
            return

        # Detect summary counts (fallback)
        self._record_summary_counts(line, state)

    @staticmethod
    @traces(spec="016", story="US-007", reqs=["REQ-016-166"])
    def _record_test_result(
        test_name: str, outcome: str, state: dict[str, Any]
    ) -> None:
        """Update state for a single per-test result line."""
        state["current_test"] = test_name
        state["completed"] += 1
        state["status"] = "running"
        if outcome == "PASSED":
            state["passed"] += 1
        elif outcome == "FAILED":
            state["failed"] += 1
            state["failed_tests"].append(test_name)
        elif outcome == "ERROR":
            state["errors"] += 1
            state["failed_tests"].append(test_name)
        elif outcome == "SKIPPED":
            state["skipped"] += 1

    @staticmethod
    @traces(spec="016", story="US-007", reqs=["REQ-016-166"])
    def _record_summary_counts(line: str, state: dict[str, Any]) -> None:
        """Update state from a pytest summary line (passed/failed/error/skipped)."""
        m = _PYTEST_PASSED.search(line)
        if m:
            state["passed"] = int(m.group(1))
        m = _PYTEST_FAILED.search(line)
        if m:
            state["failed"] = int(m.group(1))
        m = _PYTEST_ERROR.search(line)
        if m:
            state["errors"] = int(m.group(1))
        m = _PYTEST_SKIPPED.search(line)
        if m:
            state["skipped"] = int(m.group(1))

    @traces(spec="016", story="US-007", reqs=["REQ-016-166"])
    async def _emit(self, run_id: str, payload: dict[str, Any]) -> None:
        """Broadcast a tests.progress event."""
        try:
            await self._broadcast("tests.progress", {"run_id": run_id, **payload})
        except Exception:
            logger.exception("Failed to broadcast test progress")

    async def _store_result(self, state: dict[str, Any]) -> bool:
        """Store the final test run result in the operational knowledge graph.

        Returns:
            True if the write landed, False if there is no graph repository
            configured (nothing was stored) or the write itself failed.
            Non-fatal either way — storing the result is a side-channel,
            never a reason to break the caller.
        """
        if not self._graph_repository:
            # No repository configured: nothing was stored, distinct from a
            # write that was attempted and failed below.
            return False
        try:
            # MERGE on `id`, TestRun's canonical merge key (canonical_schema
            # CANONICAL_MERGE_KEYS). This was CREATE-without-`id` until
            # 2026-08-07: every gateway run left a keyless node, so
            # `scripts/canonical_conformance.py` reported real drift
            # ("TestRun: N node(s) lacking their canonical key") and a retried
            # run_id created a duplicate instead of updating. The sibling
            # writer, knowledge_graph/test_results_store.save_run, already
            # MERGEs on `id`; this one is now consistent with it. `run_id` is
            # kept as a plain property — the metrics handler reads it.
            query = """
            MERGE (t:TestRun {id: $run_id})
            SET t += {
                run_id: $run_id,
                date: datetime(),
                test_count: $test_count,
                pass_rate: $pass_rate,
                coverage: $coverage,
                duration: $duration,
                failed_tests: $failed_tests,
                status: $status
            }
            """
            total = state.get("completed", 0)
            passed = state.get("passed", 0)
            pass_rate = (passed / total * 100) if total > 0 else 0.0
            await self._graph_repository.execute_write(
                query,
                {
                    "run_id": state["run_id"],
                    "test_count": total,
                    "pass_rate": pass_rate,
                    "coverage": 0.0,  # TODO: integrate coverage.py
                    "duration": state.get("duration_seconds", 0),
                    "failed_tests": json.dumps(state.get("failed_tests", [])),
                    "status": state.get("status", "unknown"),
                },
            )
            logger.info("Stored test run %s in the graph", state["run_id"])
            return True
        except Exception:
            logger.exception("Failed to store test result in the graph")
            return False