scripts.observe_pytest

Reference › scripts › observe_pytest

scripts/observe_pytest.py attaches to a live pytest run and streams its per-test events in real time, so a long run can be watched instead of waited on. A human runs it in another terminal (see the pytest live observer page). It reads the session registry and the per-session live log through orchestrata_ultimate.core.threading.pytest_live_reader; the producing side is the auto-attach shim orchestrata_pytest_observer.

There are two subcommands. list (cmd_list) prints every known session, newest first, one stable key=value line per session (_format_session_line, with progress as done over selected tests from _format_progress, and _pid_alive telling a live session from a dead one). attach <session_id> (cmd_attach) takes a session id or latest (the most recently started running session) and options --once (print the current snapshot of the log and exit, _render_snapshot) and --timeout (maximum seconds to wait for the session to finish while following). Following is done by _follow, which polls the log (default every 0.1 seconds), parses each line with _parse_line and prints it with _render_event. It only reads; it writes nothing and never touches the run it observes.

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

API

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/scripts/observe_pytest.py

#!/usr/bin/env python3
"""Attach to a LIVE pytest run and stream its per-test events in real time.

A running pytest broadcasts a discoverable reference (see
``orchestrata_ultimate.core.threading.pytest_live_observer``): a small JSON
entry per session written to a well-known registry directory
(``session_registry_dir()``), naming that session's live NDJSON event log.
This tool is the OBSERVER side -- it never talks to the pytest process
itself, only to files that process (best-effort) keeps up to date, so it
can attach to any session started elsewhere (a batched-executor subprocess,
or a plain ``pytest -p orchestrata_ultimate.core.threading.pytest_live_observer``
run) without any prior coordination.

Subcommands
-----------
list                       List active/finished sessions from the registry.
attach <session_id|latest> Follow a session's live log, rendering each
                           event (``PASS``/``FAIL``/``SKIP``/``ERROR`` lines
                           plus a final summary) until it finishes.

Examples
--------
    uv run python scripts/observe_pytest.py list
    uv run python scripts/observe_pytest.py attach latest
    uv run python scripts/observe_pytest.py attach 12345-1720000000000 --once
    uv run python scripts/observe_pytest.py attach latest --timeout 60
"""

from __future__ import annotations

import argparse
import json
import os
import sys
import time
from collections.abc import Mapping
from typing import Any

# The READER module, not the plugin. This script is deployed to
# /opt/orchestrata-tools, whose venv ships neo4j + watchfiles and no pytest --
# pytest-live runs the TARGET project's interpreter, so the tools venv has no
# reason to carry a test framework. Importing pytest_live_observer here reached
# `import pytest` through core.threading.__init__ -> test_executor and failed
# install-tools.sh, and with it the whole ship (2026-08-19).
from orchestrata_ultimate.core.threading.pytest_live_reader import (
    list_active_sessions,
)

_OUTCOME_LABELS = {"passed": "PASS", "failed": "FAIL", "skipped": "SKIP"}


def _fmt_ago(started_at: float, *, now: float | None = None) -> str:
    """Render a human-friendly "time since" duration (e.g. ``"12s"``)."""
    delta = (now if now is not None else time.time()) - started_at
    if delta < 0:
        delta = 0.0
    if delta < 60:
        return f"{delta:.0f}s"
    if delta < 3600:
        return f"{delta / 60:.0f}m"
    return f"{delta / 3600:.1f}h"


def _format_progress(session: dict[str, Any]) -> str:
    """``done/selected (left N)`` -- against the session's SELECTION, never
    the suite total (RUL-038). ``?`` until the session finished collecting."""
    counts = session.get("counts") or {}
    done = sum(
        int(counts.get(k) or 0) for k in ("passed", "failed", "errors", "skipped")
    )
    selected = session.get("selected")
    if selected is None:
        return f"{done}/? (left ?)"
    return f"{done}/{int(selected)} (left {max(0, int(selected) - done)})"


def _format_session_line(session: dict[str, Any]) -> str:
    """One stable, greppable ``key=value ...`` line per session."""
    counts = session.get("counts") or {}
    return (
        f"session_id={session.get('session_id', '?')} "
        f"pid={session.get('pid', '?')} "
        f"worker={session.get('worker', '')} "
        f"status={session.get('status', '?')} "
        f"progress={_format_progress(session)} "
        f"passed={counts.get('passed', 0)} "
        f"failed={counts.get('failed', 0)} "
        f"errors={counts.get('errors', 0)} "
        f"skipped={counts.get('skipped', 0)} "
        f"started_ago={_fmt_ago(session.get('started_at', time.time()))} "
        f"live_log={session.get('live_log_path', '')}"
    )


def cmd_list(_args: argparse.Namespace, env: Mapping[str, str]) -> int:
    """``observe_pytest.py list`` -- print every known session, newest first."""
    sessions = list_active_sessions(env=env)
    if not sessions:
        print("No pytest sessions found.")
        return 0
    for session in sessions:
        print(_format_session_line(session))
    return 0


def _pid_alive(pid: Any) -> bool:
    """True if ``pid`` is an int naming a live process."""
    if not isinstance(pid, int):
        return False
    try:
        os.kill(pid, 0)
    except ProcessLookupError:
        return False
    except PermissionError:
        # The pid exists but belongs to another user.
        return True
    return True


def _find_session(session_id: str, env: Mapping[str, str]) -> dict[str, Any] | None:
    """Resolve ``session_id`` (or the literal ``"latest"``) against the
    registry. ``"latest"`` means the most recently started RUNNING session
    whose pid is still alive — a SIGKILLed run leaves a permanently
    "running" registry entry (and its live log may be gone), which must
    never shadow a genuinely live session.
    """
    sessions = list_active_sessions(env=env)
    if session_id == "latest":
        running = [
            s
            for s in sessions
            if s.get("status") == "running" and _pid_alive(s.get("pid"))
        ]
        return running[0] if running else None
    for session in sessions:
        if session.get("session_id") == session_id:
            return session
    return None


def _parse_line(line: str) -> dict[str, Any] | None:
    line = line.strip()
    if not line:
        return None
    try:
        parsed = json.loads(line)
    except ValueError:
        return None
    return parsed if isinstance(parsed, dict) else None


def _render_event(event: dict[str, Any], counts: dict[str, int]) -> None:
    """Print one live-rendering line for ``event``, updating running
    ``counts`` in place (used for the final summary).
    """
    kind = event.get("event")
    if kind == "report":
        when = event.get("when")
        outcome: str = event.get("outcome") or ""
        terminal = when == "call" or (when == "setup" and outcome == "failed")
        if not terminal:
            return
        bucket = outcome if outcome in counts else "errors"
        counts[bucket] += 1
        label = _OUTCOME_LABELS.get(outcome, "ERROR")
        duration = event.get("duration", 0.0)
        print(f"{label} {event.get('nodeid', '?')} ({duration:.2f}s)")
    elif kind == "collect_error":
        counts["errors"] += 1
        print(f"ERROR {event.get('nodeid', '?')} (collection failed)")
    elif kind == "session_start":
        print("== session start ==")
    elif kind == "collection_finished":
        print(
            f"== selected {event.get('selected', 0)} test(s) in "
            f"{event.get('files', 0)} file(s) =="
        )
    elif kind == "capped":
        print("== CAPPED: the run's hard cap stopped this session (RUL-038) ==")
    elif kind == "session_finish":
        print(
            f"== session finished exitstatus={event.get('exitstatus')} "
            f"passed={counts['passed']} failed={counts['failed']} "
            f"errors={counts['errors']} skipped={counts['skipped']} =="
        )


def _session_status(session_id: str, env: Mapping[str, str]) -> str | None:
    """Re-read the registry for ``session_id``'s current ``status``."""
    for session in list_active_sessions(env=env):
        if session.get("session_id") == session_id:
            return session.get("status")
    return None


def _render_snapshot(log_path: str, counts: dict[str, int]) -> int:
    """``--once``: render every event currently in the log and exit."""
    try:
        with open(log_path, encoding="utf-8") as fh:
            lines = fh.readlines()
    except OSError as exc:
        print(f"Could not open live log {log_path!r}: {exc}", file=sys.stderr)
        return 1
    for line in lines:
        event = _parse_line(line)
        if event is not None:
            _render_event(event, counts)
    return 0


def _follow(
    log_path: str,
    session_id: str,
    counts: dict[str, int],
    timeout: float | None,
    *,
    env: Mapping[str, str],
    poll_interval: float = 0.1,
) -> int:
    """``tail -f`` the live log: render new events as they're appended,
    exiting cleanly on ``session_finish`` or a registry flip to
    ``"finished"``, and non-zero if ``timeout`` elapses first.
    """
    start = time.monotonic()
    try:
        fh = open(log_path, encoding="utf-8")
    except OSError as exc:
        print(f"Could not open live log {log_path!r}: {exc}", file=sys.stderr)
        return 1
    with fh:
        while True:
            line = fh.readline()
            if line:
                event = _parse_line(line)
                if event is not None:
                    _render_event(event, counts)
                    if event.get("event") == "session_finish":
                        return 0
                    if event.get("event") == "capped":
                        return 1  # killed by the cap: there is no session_finish
                continue
            status = _session_status(session_id, env=env)
            if status == "finished":
                return 0
            if status == "capped":
                return 1
            if timeout is not None and (time.monotonic() - start) > timeout:
                print(
                    f"Timed out after {timeout}s waiting for session "
                    f"{session_id!r} to finish.",
                    file=sys.stderr,
                )
                return 1
            time.sleep(poll_interval)


def cmd_attach(args: argparse.Namespace, env: Mapping[str, str]) -> int:
    """``observe_pytest.py attach <session_id|latest>``."""
    session = _find_session(args.session_id, env=env)
    if session is None:
        print(f"No such session: {args.session_id!r}", file=sys.stderr)
        return 1
    log_path = session.get("live_log_path") or ""
    if not log_path:
        print(
            f"Session {session.get('session_id')!r} has no live_log_path.",
            file=sys.stderr,
        )
        return 1

    counts = {"passed": 0, "failed": 0, "errors": 0, "skipped": 0}
    if args.once:
        return _render_snapshot(log_path, counts)
    return _follow(log_path, str(session["session_id"]), counts, args.timeout, env=env)


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        description="Observe a live pytest run via the session registry broadcast."
    )
    sub = parser.add_subparsers(dest="command", required=True)

    sub.add_parser("list", help="List active/finished pytest sessions.")

    attach = sub.add_parser(
        "attach", help="Attach to a pytest session and stream its live events."
    )
    attach.add_argument(
        "session_id",
        help="Session id to attach to, or 'latest' for the most recently "
        "started RUNNING session.",
    )
    attach.add_argument(
        "--once",
        action="store_true",
        help="Print the current snapshot of the live log and exit (don't follow).",
    )
    attach.add_argument(
        "--timeout",
        type=float,
        default=None,
        help="Max seconds to wait for the session to finish while following.",
    )
    return parser


def main(
    argv: list[str] | None = None,
    *,
    env: Mapping[str, str] | None = None,
) -> int:
    parser = build_parser()
    args = parser.parse_args(argv)
    env = os.environ if env is None else env
    if args.command == "list":
        return cmd_list(args, env=env)
    if args.command == "attach":
        return cmd_attach(args, env=env)
    parser.error(f"unknown command: {args.command}")  # pragma: no cover
    return 2  # pragma: no cover


if __name__ == "__main__":
    raise SystemExit(main())