orchestrata_ultimate.gateway.handlers.test_metrics

Reference › orchestrata_ultimate › gateway › handlers › test_metrics

TestMetricsHandler serves the tests.run and metrics.current RPCs. handle_tests_run starts a pytest run through the gateway's test_runner as a background task and returns at once with a timestamped run id and status started; if a run is already in progress it returns an error instead. Optional test_path and extra_args parameters are passed on. handle_metrics_current reads the latest test-run node from the graph (test count, pass rate, coverage, duration and failed tests) through _fetch_metrics_from_neo4j; if the graph is unavailable or holds no data it returns zeroed metrics with an explanatory error. rpc creates it. See gateway.

At a glance

Class diagram

Dependency diagram

API

Classes

TestMetricsHandler

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

Handles test execution and quality metrics RPCs.

Source

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

"""Test execution and metrics RPC handlers."""

import asyncio
import logging
from datetime import UTC, datetime
from typing import Any

from orchestrata_ultimate.core.base.traces import traces

logger = logging.getLogger(__name__)


@traces(spec="016", story="US-007", reqs=["REQ-016-167"])
class TestMetricsHandler:
    """Handles test execution and quality metrics RPCs."""

    def __init__(self, graph_repository: Any = None, runner: Any = None) -> None:
        self._graph_repository = graph_repository
        self._runner = runner
        self._test_run_in_progress = False

    def set_graph_repository(self, graph_repository: Any) -> None:
        """Inject the operational-graph repository for metrics queries.

        Args:
            graph_repository: An AsyncGraphRepository instance.
        """
        self._graph_repository = graph_repository

    @traces(spec="016", story="US-007", reqs=["REQ-016-167"])
    async def handle_tests_run(
        self,
        connection_id: str,
        params: dict[str, Any],
    ) -> dict[str, Any]:
        """Trigger pytest execution. Returns immediately with run_id.

        Args:
            connection_id: Caller connection ID.
            params: Optional test filters.

        Returns:
            Run status dict.
        """
        if self._runner and self._runner.is_running:
            return {"error": "Test run already in progress"}
        run_id = f"run_{datetime.now(UTC).strftime('%Y%m%d_%H%M%S')}"
        if self._runner:
            # Start the runner in a background task
            asyncio.create_task(
                self._runner.run(
                    run_id=run_id,
                    test_path=params.get("test_path"),
                    extra_args=params.get("extra_args"),
                )
            )
        else:
            self._test_run_in_progress = True
        return {
            "run_id": run_id,
            "status": "started",
            "message": "Test run initiated",
        }

    async def handle_metrics_current(
        self,
        connection_id: str,
        params: dict[str, Any],
    ) -> dict[str, Any]:
        """Return current test metrics.

        Args:
            connection_id: Caller connection ID.
            params: Ignored.

        Returns:
            Metrics dict or fallback error.
        """
        if self._graph_repository:
            try:
                metrics = await self._fetch_metrics_from_neo4j()
                if metrics:
                    return metrics
            except Exception as exc:
                logger.warning("graph metrics fetch failed: %s", exc)
        return {
            "test_count": 0,
            "pass_rate": 0.0,
            "coverage_percentage": 0.0,
            "duration_seconds": 0,
            "failed_tests": [],
            "error": "Metrics not available — no test data found",
        }

    async def _fetch_metrics_from_neo4j(self) -> dict[str, Any] | None:
        """Fetch latest metrics from the graph.

        Returns:
            Metrics dict or None if no data.
        """
        query = """
        MATCH (t:TestRun)
        WHERE t.test_count IS NOT NULL AND t.date IS NOT NULL
        RETURN t.test_count as test_count, t.pass_rate as pass_rate,
               t.coverage as coverage, t.duration as duration,
               t.failed_tests as failed_tests
        ORDER BY t.date DESC
        LIMIT 1
        """
        rows = await self._graph_repository.execute_read(query)
        for row in rows:
            return {
                "test_count": row.get("test_count", 0) or 0,
                "pass_rate": row.get("pass_rate", 0.0) or 0.0,
                "coverage_percentage": row.get("coverage", 0.0) or 0.0,
                "duration_seconds": row.get("duration", 0) or 0,
                "failed_tests": row.get("failed_tests", []) or [],
            }
        return None