orchestrata_ultimate.knowledge_graph.test_results_store

Reference › orchestrata_ultimate › knowledge_graph › test_results_store

Stores pytest results in the operational graph (orchkg on orchgraph) so that "which tests cover function X" can be answered when X changes (Spec 008, REQ-008-021). TestResultRecord describes one test (name, status, duration, error message, the requirement ids it traces and the source functions it covers), and TestRunRecord describes a whole run (project, timestamp, tag, branch, commit, pass, fail and error counts, duration and the results). TestResultsStore takes a synchronous graph repository, and save_run first validates its input, raising ValueError for a None run, a wrong type, an empty project or a non-list results. It then writes inside one unit of work: a TestRun node merged on its id (with a date property so the metrics ordering is not shadowed), TestResult nodes linked to the run with RAN_IN, and COVERS edges from results to project-qualified Function nodes. The module docstring records that, as of July 2026, the store has no production caller, so it is exercised by tests only.

At a glance

Class diagram

Dependency diagram

API

Classes

TestResultRecord (dataclass)

Decorators: dataclass, traces(spec='008', story='US-005', reqs=['REQ-008-021']) · Line 37

A single test result.

TestRunRecord (dataclass)

Decorators: dataclass, traces(spec='008', story='US-005', reqs=['REQ-008-021']) · Line 59

A complete test run.

TestResultsStore

Decorators: traces(spec='008', story='US-005', reqs=['REQ-008-021']) · Line 94

Persist test results to the graph for impact analysis.

Source

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

# mypy: ignore-errors
"""Store pytest results in the graph for impact analysis.

After each test run, saves:
- TestRun node: timestamp, tag, branch, pass_count, fail_count, duration
- TestResult nodes: test name, status, duration, error message
- COVERS relationships: TestResult → SourceFunction (from @traces)
- RAN_IN relationships: TestResult → TestRun

Query: "what tests cover function X?" → find affected tests when X changes.

Wiring gap (tracked separately, not addressed by this module): as of
2026-07-22, ``TestResultsStore`` has no production caller anywhere in
``src/`` -- only tests construct it and call ``save_run``. The write path
itself is complete and canonical-schema conformant; it is simply not yet
wired into any real test-run reporter/CI hook.
"""

from __future__ import annotations

import logging
from dataclasses import dataclass, field
from datetime import UTC, datetime
from typing import TYPE_CHECKING

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.knowledge_graph import canonical_schema as cs

if TYPE_CHECKING:
    from orchestrata_ultimate.core.graph import SyncGraphRepository

_logger = logging.getLogger(__name__)


@dataclass
@traces(spec="008", story="US-005", reqs=["REQ-008-021"])
class TestResultRecord:
    """A single test result.

    Attributes:
        name: Fully qualified test name (file::class::method).
        status: passed, failed, error, skipped.
        duration_s: Execution time in seconds.
        error_msg: Error message if failed/error.
        traces_reqs: Requirement IDs from @traces decorator.
        source_functions: Source functions this test covers.
    """

    name: str
    status: str = "passed"
    duration_s: float = 0.0
    error_msg: str = ""
    traces_reqs: list[str] = field(default_factory=list)
    source_functions: list[str] = field(default_factory=list)


@dataclass
@traces(spec="008", story="US-005", reqs=["REQ-008-021"])
class TestRunRecord:
    """A complete test run.

    Attributes:
        project: Project slug/id this run belongs to. REQUIRED -- no
            default -- an unset/optional project would silently recreate
            the MERGE-key/project-scoping mismatch this field exists to
            close (2026-07-22 Function/COVERS canonical-id fix). Used to
            build the project-qualified Function id for COVERS edges via
            ``canonical_schema.project_qualified_id`` (see
            ``TestResultsStore.save_run``).
        timestamp: When the run started.
        tag: Git tag or session tag.
        branch: Git branch.
        commit: Git commit hash.
        pass_count: Number of passed tests.
        fail_count: Number of failed tests.
        error_count: Number of error tests.
        duration_s: Total run duration.
        results: Individual test results.
    """

    project: str
    timestamp: datetime = field(default_factory=lambda: datetime.now(UTC))
    tag: str = ""
    branch: str = ""
    commit: str = ""
    pass_count: int = 0
    fail_count: int = 0
    error_count: int = 0
    duration_s: float = 0.0
    results: list[TestResultRecord] = field(default_factory=list)


@traces(spec="008", story="US-005", reqs=["REQ-008-021"])
class TestResultsStore:
    """Persist test results to the graph for impact analysis.

    Creates TestRun and TestResult nodes with COVERS relationships
    to source functions, enabling "what tests cover this function?"
    queries for selective test execution.

    Attributes:
        repo: The operational-graph repository seam (SPEC backlog #163).
    """

    def __init__(self, repo: SyncGraphRepository) -> None:
        """Initialize with a graph repository.

        Args:
            repo: Synchronous operational-graph repository.
        """
        self._repo = repo

    def save_run(self, run: TestRunRecord) -> None:
        """Save a complete test run to the graph.

        Creates a TestRun node and TestResult nodes with
        RAN_IN and COVERS relationships.

        Args:
            run: The test run record to persist.

        Raises:
            ValueError: If ``run`` is ``None``, has a non-list ``results``,
                or an empty ``project``.
        """
        if run is None:
            raise ValueError("run must be a TestRunRecord, got None")
        if not isinstance(run, TestRunRecord):
            raise ValueError(f"run must be a TestRunRecord, got {type(run).__name__}")
        if not run.project:
            raise ValueError(
                "TestRunRecord.project must be a non-empty string -- it is "
                "REQUIRED (no default) so the Function/COVERS writer below "
                "can never silently fall back to a non-project-qualified id"
            )
        if not isinstance(run.results, list):
            raise ValueError(
                "TestRunRecord.results must be a list, "
                f"got {type(run.results).__name__}"
            )
        for result in run.results:
            if not isinstance(result, TestResultRecord):
                raise ValueError(
                    "each item in TestRunRecord.results must be a "
                    f"TestResultRecord, got {type(result).__name__}"
                )

        with self._repo.unit_of_work() as uow:
            # Create TestRun node. TestRun's canonical merge key is `id`
            # (canonical_schema CANONICAL_MERGE_KEYS -- uncontested, no prior
            # writer divergence documented). MERGEing on the non-constrained
            # `run_id` property let two writers create distinct TestRun
            # nodes for the same logical run; `run_id` is kept as a plain
            # property (never drop data; the TestResult MATCH below still
            # keys on it since it is unique per run either way).
            run_id = f"run-{run.timestamp.isoformat()}"
            uow.run(
                "MERGE (r:TestRun {id: $run_id}) "
                "SET r.run_id = $run_id, "
                "    r.timestamp = $ts, "
                # `date` aligns this writer with gateway/test_runner.py's
                # TestRun schema — the metrics handler orders by t.date DESC,
                # and (Cypher) nulls sort FIRST under DESC, so a date-less
                # TestRun from this store would shadow every real run.
                # Schema convergence proper is D12/W8.3 (canonical writer).
                "    r.date = datetime($ts), "
                "    r.tag = $tag, "
                "    r.branch = $branch, "
                "    r.commit_hash = $commit, "
                "    r.pass_count = $pass_count, "
                "    r.fail_count = $fail_count, "
                "    r.error_count = $error_count, "
                "    r.duration_s = $duration_s",
                {
                    "run_id": run_id,
                    "ts": run.timestamp.isoformat(),
                    "tag": run.tag,
                    "branch": run.branch,
                    "commit": run.commit,
                    "pass_count": run.pass_count,
                    "fail_count": run.fail_count,
                    "error_count": run.error_count,
                    "duration_s": run.duration_s,
                },
            )

            # Batch create TestResult nodes. id is run-scoped
            # ("{run_id}:{name}"), not the bare test name: the same test name
            # recurs across runs (every re-run of the suite) AND across
            # projects, so a bare-name MERGE key collapses every run's
            # result for that test onto one node (2026-07-12 multi-project
            # identity migration). run_id is a per-run unique value, so this
            # is strictly finer-grained than project-scoping alone.
            if run.results:
                result_rows = [
                    {
                        "id": f"{run_id}:{r.name}",
                        "name": r.name,
                        "status": r.status,
                        "duration_s": r.duration_s,
                        "error_msg": r.error_msg,
                    }
                    for r in run.results
                ]
                uow.run(
                    "MATCH (tr:TestRun {id: $run_id}) "
                    "UNWIND $rows AS row "
                    "MERGE (t:TestResult {id: row.id}) "
                    "SET t.name = row.name, "
                    "    t.status = row.status, "
                    "    t.duration_s = row.duration_s, "
                    "    t.error_msg = row.error_msg, "
                    "    t.last_run = $ts "
                    "MERGE (t)-[:RAN_IN]->(tr)",
                    {
                        "run_id": run_id,
                        "rows": result_rows,
                        "ts": run.timestamp.isoformat(),
                    },
                )

                # Batch create COVERS relationships. Function's canonical
                # merge key is `id`, project-qualified via
                # canonical_schema.project_qualified_id -- the SAME shared
                # helper knowledge_graph.sync.GraphSyncer._sync_functions
                # calls for its own Function nodes
                # (f"{project}:{qualified_name}"), so the two writers'
                # schemes cannot drift apart (2026-07-22 MERGE-key audit
                # ruling: `fn` was previously used bare as both the merge
                # key AND the display name, which collapsed the same
                # function name from different projects onto one node).
                # `run.project` (REQUIRED, no default -- see TestRunRecord)
                # supplies the previously-missing project scope. Each
                # `source_functions` entry is already the qualified_name --
                # there is no AST dict here to re-derive one from the way
                # sync.py's `func.get("qualified_name") or
                # f"{mod_path}.{name}"` fallback does; the entry IS that
                # value.
                covers_rows = [
                    {
                        "test_id": f"{run_id}:{r.name}",
                        "func_id": cs.project_qualified_id(run.project, fn),
                        "func_name": fn,
                    }
                    for r in run.results
                    for fn in r.source_functions
                ]
                if covers_rows:
                    uow.run(
                        "UNWIND $rows AS row "
                        "MATCH (t:TestResult {id: row.test_id}) "
                        "MERGE (f:Function {id: row.func_id}) "
                        "SET f.name = row.func_name "
                        "MERGE (t)-[:COVERS]->(f)",
                        {"rows": covers_rows},
                    )

        _logger.info(
            "Saved test run: %d passed, %d failed, %.1fs",
            run.pass_count,
            run.fail_count,
            run.duration_s,
        )

    def get_tests_covering(self, function_name: str) -> list[str]:
        """Find all tests that cover a given function.

        Args:
            function_name: Fully qualified function name.

        Returns:
            List of test names that cover this function.
        """
        records = self._repo.execute_read(
            "MATCH (t:TestResult)-[:COVERS]->(f:Function) "
            "WHERE f.name = $func_name "
            "RETURN t.name AS test_name",
            {"func_name": function_name},
        )
        return [r["test_name"] for r in records]

    def get_affected_tests(
        self,
        changed_functions: list[str],
    ) -> list[str]:
        """Find all tests affected by a set of changed functions.

        Args:
            changed_functions: List of function names that changed.

        Returns:
            Deduplicated list of test names to re-run.
        """
        tests = {
            test for func in changed_functions for test in self.get_tests_covering(func)
        }
        return sorted(tests)