orchestrata_ultimate.session.controller._tests

Reference › orchestrata_ultimate › session › controller › _tests

TestStateMixin supplies the two test observations the SessionController needs. _get_test_state runs pytest --collect-only under the project's package manager with a 120-second limit and returns a string such as 7678 collected, or unknown if collection fails, times out or the tool is missing. _run_tests runs the test suite quietly with a 300-second limit and returns true only on exit code 0; a timeout or a missing tool counts as a failure. The first value goes into the project state when a session ends, and the second is one of the checks of the close checklist.

At a glance

Class diagram

Dependency diagram

API

Classes

TestStateMixin

Line 10

Test-state collection and test execution for sessions.

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/session/controller/_tests.py

# mypy: ignore-errors
"""TestStateMixin — pytest collection and run helpers for SessionController."""

import os
import subprocess
from collections.abc import Mapping
from pathlib import Path


class TestStateMixin:
    """Test-state collection and test execution for sessions."""

    # Provided by SessionController.__init__ (composed via multiple inheritance).
    root: Path

    def _get_test_state(self, env: Mapping[str, str] | None = None) -> str:
        """Collect current test counts via pytest --co -q.

        Args:
            env: Optional environment mapping for the spawned subprocess.
                Defaults to ``os.environ``.

        Returns:
            Formatted string like "7678 collected"
            or "unknown" if collection fails.
        """
        runtime_env = os.environ if env is None else dict(env)
        try:
            result = subprocess.run(
                ["uv", "run", "pytest", "tests/", "--co", "-q"],
                cwd=self.root,
                capture_output=True,
                text=True,
                timeout=120,
                env=runtime_env,
            )
            if result.returncode == 0:
                import re as _re

                for line in reversed(result.stdout.strip().splitlines()):
                    line = line.strip()
                    if "test" in line and "collected" in line:
                        m = _re.search(r"(\d+)\s+test", line)
                        if m:
                            return f"{m.group(1)} collected"
                        return line
            return "unknown"
        except subprocess.TimeoutExpired, FileNotFoundError, OSError:
            return "unknown"

    def _run_tests(self, env: Mapping[str, str] | None = None) -> bool:
        """Run pytest and return True if tests pass.

        Args:
            env: Optional environment mapping for the spawned subprocess.
                Defaults to ``os.environ``.
        """
        runtime_env = os.environ if env is None else dict(env)
        try:
            result = subprocess.run(
                ["uv", "run", "pytest", "tests/", "-q"],
                cwd=self.root,
                capture_output=True,
                timeout=300,
                env=runtime_env,
            )
            return result.returncode == 0
        except subprocess.TimeoutExpired, FileNotFoundError:
            return False