mirror of
https://github.com/openswarm-ai/openswarm.git
synced 2026-10-01 05:54:56 +02:00
[haik]: ckpt, added in the test runner skelton i made in another repo -> still gotta make tweaks to fold it onto the current repo
This commit is contained in:
haikdc
@@ -0,0 +1,58 @@
|
||||
# Test runner
|
||||
|
||||
An interactive test runner: discover tests via pytest's own collector, cherry-pick
|
||||
them in a Textual tree, and run them while a live Rich dashboard streams pass/fail.
|
||||
|
||||
```
|
||||
python -m tests.runner # discover → picker → run
|
||||
python -m tests.runner -k ingest # skip the picker, run a -k selection
|
||||
python -m tests.runner tests/api # run specific paths
|
||||
```
|
||||
|
||||
## Two-venv design
|
||||
|
||||
The runner is split across two interpreters so it can be dropped into any repo
|
||||
without polluting that repo's test environment:
|
||||
|
||||
| Process | venv | Imports | Needs |
|
||||
| --- | --- | --- | --- |
|
||||
| Runner (parent) | runner venv | `typer`, `rich`, `textual` | the UI libs only — **never** pytest |
|
||||
| Worker (subprocess) | test venv | `pytest`, `coverage` | pytest, pytest-asyncio, coverage + the project's own deps |
|
||||
|
||||
Discovery and execution both run in the **test venv** via subprocess. The worker
|
||||
([`_worker.py`](./_worker.py)) registers a thin pytest plugin that frames every
|
||||
collection/run event (and, in `-s` mode, each line of streamed output) as JSON
|
||||
onto a pipe. The parent ([`run.py`](./run.py)) reads those events and owns all
|
||||
Rich rendering. See [`events.py`](./events.py) for the wire protocol.
|
||||
|
||||
```
|
||||
parent (rich) <—— JSON events over a pipe FD —— worker (pytest, test venv)
|
||||
```
|
||||
|
||||
## Setup
|
||||
|
||||
1. Runner venv (the UI):
|
||||
|
||||
```
|
||||
python -m venv .runner-venv
|
||||
.runner-venv/bin/pip install -e tests/runner
|
||||
```
|
||||
|
||||
2. Test venv (where tests actually run) — your project's existing venv with
|
||||
`pytest`, `pytest-asyncio`, and `coverage` installed.
|
||||
|
||||
3. Point the runner at the test venv in [`config.json`](./config.json).
|
||||
|
||||
## config.json
|
||||
|
||||
All path/venv coupling lives here so no code edits are needed per repo:
|
||||
|
||||
| key | meaning | default |
|
||||
| --- | --- | --- |
|
||||
| `repo_root` | cwd for pytest; resolved relative to this folder | `../..` (repo root) |
|
||||
| `test_paths` | default search roots when no paths are given | `["tests/unit", "tests/api"]` |
|
||||
| `venv_python` | interpreter the **tests** run in; relative to `repo_root` | the runner's own interpreter |
|
||||
| `coverage_source` | packages measured under `--cov` and the report filter | `["backend"]` |
|
||||
|
||||
If `venv_python` is unset or missing on disk, the runner falls back to the
|
||||
current interpreter (single-venv mode), reproducing the original behavior.
|
||||
@@ -0,0 +1,8 @@
|
||||
"""Crisp interactive test runner for product-analytics-v1.
|
||||
|
||||
A small Typer CLI that discovers tests via pytest's own collector, lets you
|
||||
cherry-pick them in a Textual tree, and runs them in-process while a Rich
|
||||
dashboard streams pass/fail live.
|
||||
|
||||
Entry point: python -m tests.runner
|
||||
"""
|
||||
@@ -0,0 +1,88 @@
|
||||
"""Typer entrypoint: python -m tests.runner
|
||||
|
||||
Default (no args): discover tests → interactive Textual picker → run selection.
|
||||
With paths / -k / --no-pick: skip the picker and run directly.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import List, Optional
|
||||
|
||||
import typer
|
||||
from rich.console import Console
|
||||
|
||||
from tests.runner.discovery import discover
|
||||
from tests.runner.picker import run_picker
|
||||
from tests.runner.run import RunOptions, run_tests
|
||||
|
||||
console = Console()
|
||||
|
||||
|
||||
def main(
|
||||
paths: Optional[List[str]] = typer.Argument(
|
||||
None, help="pytest paths or node IDs to target (skips the picker)."
|
||||
),
|
||||
keyword: Optional[str] = typer.Option(
|
||||
None, "-k", help="Only tests matching this keyword expression (skips the picker)."
|
||||
),
|
||||
cov: bool = typer.Option(False, "--cov", help="Measure and report coverage of backend/."),
|
||||
exitfirst: bool = typer.Option(
|
||||
False, "-x", "--exitfirst", help="Stop after the first failure."
|
||||
),
|
||||
last_failed: bool = typer.Option(
|
||||
False, "--lf", "--last-failed", help="Run only the tests that failed last time."
|
||||
),
|
||||
failed_first: bool = typer.Option(
|
||||
False, "--ff", "--failed-first", help="Run last-failed tests first, then the rest."
|
||||
),
|
||||
verbose: bool = typer.Option(
|
||||
False, "-v", "--verbose", help="Richer failure output (untruncated assertions)."
|
||||
),
|
||||
no_capture: bool = typer.Option(
|
||||
False, "-s", "--no-capture", help="Show test stdout / print() output live (pytest -s)."
|
||||
),
|
||||
show_output: bool = typer.Option(
|
||||
False, "-O", "--show-output", help="Show captured output for passing tests too (not just failures)."
|
||||
),
|
||||
pick: bool = typer.Option(
|
||||
True, "--pick/--no-pick", help="Open the interactive picker (default on)."
|
||||
),
|
||||
) -> None:
|
||||
try:
|
||||
node_ids = discover(paths, keyword)
|
||||
except RuntimeError as exc:
|
||||
console.print(f"[red]discovery failed:[/]\n{exc}")
|
||||
raise typer.Exit(2)
|
||||
|
||||
if not node_ids:
|
||||
console.print("[yellow]No tests found.[/]")
|
||||
raise typer.Exit(5)
|
||||
|
||||
# CLI flags seed the picker's toggles and drive the non-interactive run.
|
||||
opts = RunOptions(
|
||||
cov=cov,
|
||||
exitfirst=exitfirst,
|
||||
last_failed=last_failed,
|
||||
failed_first=failed_first,
|
||||
verbose=verbose,
|
||||
no_capture=no_capture,
|
||||
show_output=show_output,
|
||||
)
|
||||
|
||||
interactive = pick and not paths and not keyword
|
||||
if interactive:
|
||||
picked = run_picker(node_ids, opts)
|
||||
if picked is None:
|
||||
console.print("[dim]cancelled[/]")
|
||||
raise typer.Exit(0)
|
||||
node_ids, opts = picked
|
||||
if not node_ids:
|
||||
console.print("[yellow]Nothing selected.[/]")
|
||||
raise typer.Exit(0)
|
||||
|
||||
code = run_tests(node_ids, opts)
|
||||
raise typer.Exit(code)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
typer.run(main)
|
||||
@@ -0,0 +1,171 @@
|
||||
"""Subprocess entry point that runs pytest in the *test* venv.
|
||||
|
||||
Spawned by the parent runner as::
|
||||
|
||||
<venv_python> -m tests.runner._worker <event_fd> <options_json>
|
||||
|
||||
with ``cwd`` set to the repo root (PEP 420 makes ``tests.runner`` importable
|
||||
without an ``__init__.py``, the same trick discovery uses). It registers a thin
|
||||
pytest plugin that frames every collection/run event as JSON onto ``event_fd``,
|
||||
optionally streams ``-s`` output through a gutter shim onto the same FD, and (for
|
||||
``--cov``) measures coverage and emits the computed rows. The parent owns all
|
||||
Rich rendering — this process imports neither rich nor textual.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import io
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
from io import StringIO
|
||||
|
||||
import pytest
|
||||
|
||||
from tests.runner import events
|
||||
|
||||
|
||||
class _GutterStream:
|
||||
"""stdout/stderr shim used in ``-s`` mode.
|
||||
|
||||
Buffers writes and forwards each completed line as an OUTPUT event so the
|
||||
parent can rail it under the current test's header. Mirrors the old
|
||||
in-process shim, but emits framed events instead of printing.
|
||||
"""
|
||||
|
||||
def __init__(self, fd: int) -> None:
|
||||
self._fd = fd
|
||||
self._buf = ""
|
||||
|
||||
def write(self, s) -> int:
|
||||
if isinstance(s, bytes):
|
||||
s = s.decode("utf-8", "replace")
|
||||
self._buf += s
|
||||
while "\n" in self._buf:
|
||||
line, self._buf = self._buf.split("\n", 1)
|
||||
events.emit(self._fd, {"type": events.OUTPUT, "line": line})
|
||||
return len(s)
|
||||
|
||||
def flush(self) -> None:
|
||||
if self._buf:
|
||||
events.emit(self._fd, {"type": events.OUTPUT, "line": self._buf})
|
||||
self._buf = ""
|
||||
|
||||
def isatty(self) -> bool:
|
||||
return False
|
||||
|
||||
def writable(self) -> bool:
|
||||
return True
|
||||
|
||||
def fileno(self): # some libraries probe this; signal "not a real fd"
|
||||
raise io.UnsupportedOperation("fileno")
|
||||
|
||||
|
||||
class _EventEmitter:
|
||||
"""pytest plugin: forward collection/run hooks as framed events."""
|
||||
|
||||
def __init__(self, fd: int) -> None:
|
||||
self._fd = fd
|
||||
|
||||
def pytest_collection_finish(self, session) -> None:
|
||||
events.emit(
|
||||
self._fd,
|
||||
{"type": events.COLLECTION, "items": [item.nodeid for item in session.items]},
|
||||
)
|
||||
|
||||
def pytest_runtest_logstart(self, nodeid, location) -> None:
|
||||
events.emit(self._fd, {"type": events.LOGSTART, "nodeid": nodeid})
|
||||
|
||||
def pytest_runtest_logreport(self, report) -> None:
|
||||
events.emit(
|
||||
self._fd,
|
||||
{
|
||||
"type": events.LOGREPORT,
|
||||
"nodeid": report.nodeid,
|
||||
"when": report.when,
|
||||
"passed": bool(report.passed),
|
||||
"failed": bool(report.failed),
|
||||
"skipped": bool(report.skipped),
|
||||
"duration": float(getattr(report, "duration", 0.0) or 0.0),
|
||||
"capstdout": report.capstdout or "",
|
||||
"capstderr": report.capstderr or "",
|
||||
"longreprtext": report.longreprtext or "",
|
||||
},
|
||||
)
|
||||
|
||||
def pytest_runtest_logfinish(self, nodeid, location) -> None:
|
||||
events.emit(self._fd, {"type": events.LOGFINISH, "nodeid": nodeid})
|
||||
|
||||
|
||||
def _emit_coverage(fd: int, cov, coverage_source: list[str], repo_root: str) -> None:
|
||||
"""Compute the per-file coverage table and emit it (rendering is the parent's job)."""
|
||||
try:
|
||||
cov.stop()
|
||||
cov.save()
|
||||
total = cov.report(file=StringIO())
|
||||
except Exception as exc: # pragma: no cover - defensive
|
||||
events.emit(fd, {"type": events.COVERAGE, "error": str(exc)})
|
||||
return
|
||||
|
||||
needles = [f"{os.sep}{name}{os.sep}" for name in coverage_source]
|
||||
data = cov.get_data()
|
||||
rows = []
|
||||
for path in data.measured_files():
|
||||
if not any(n in path for n in needles):
|
||||
continue
|
||||
try:
|
||||
_, statements, _, missing, _ = cov.analysis2(path)
|
||||
except Exception:
|
||||
continue
|
||||
n = len(statements)
|
||||
if n == 0:
|
||||
continue
|
||||
miss = len(missing)
|
||||
pct = (n - miss) / n * 100
|
||||
rel = os.path.relpath(path, repo_root)
|
||||
rows.append([rel, n, miss, pct])
|
||||
|
||||
events.emit(fd, {"type": events.COVERAGE, "rows": rows, "total": float(total)})
|
||||
|
||||
|
||||
def main(argv: list[str]) -> int:
|
||||
fd = int(argv[0])
|
||||
opts = json.loads(argv[1])
|
||||
|
||||
pytest_args = opts.get("pytest_args", [])
|
||||
no_capture = bool(opts.get("no_capture"))
|
||||
cov = bool(opts.get("cov"))
|
||||
coverage_source = opts.get("coverage_source") or []
|
||||
repo_root = opts.get("repo_root") or os.getcwd()
|
||||
|
||||
cov_obj = None
|
||||
if cov:
|
||||
import coverage
|
||||
|
||||
cov_obj = coverage.Coverage(source=coverage_source)
|
||||
cov_obj.start()
|
||||
|
||||
emitter = _EventEmitter(fd)
|
||||
args = ["-o", "addopts=", "-p", "no:terminal", *pytest_args]
|
||||
|
||||
saved_out, saved_err = sys.stdout, sys.stderr
|
||||
if no_capture:
|
||||
sys.stdout = _GutterStream(fd)
|
||||
sys.stderr = _GutterStream(fd)
|
||||
try:
|
||||
code = int(pytest.main(args, plugins=[emitter]))
|
||||
finally:
|
||||
if no_capture:
|
||||
sys.stdout.flush()
|
||||
sys.stderr.flush()
|
||||
sys.stdout, sys.stderr = saved_out, saved_err
|
||||
|
||||
if cov_obj is not None:
|
||||
_emit_coverage(fd, cov_obj, coverage_source, repo_root)
|
||||
|
||||
events.emit(fd, {"type": events.DONE, "code": code})
|
||||
return code
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main(sys.argv[1:]))
|
||||
@@ -0,0 +1,6 @@
|
||||
{
|
||||
"repo_root": "../..",
|
||||
"test_paths": ["tests/unit", "tests/api"],
|
||||
"venv_python": "backend/.venv/bin/python",
|
||||
"coverage_source": ["backend"]
|
||||
}
|
||||
@@ -0,0 +1,80 @@
|
||||
"""Runner configuration: paths + the test venv interpreter.
|
||||
|
||||
A single ``config.json`` next to this module centralizes everything that used to
|
||||
be hard-coded in ``discovery.py`` / ``run.py`` so the runner can be dropped into
|
||||
another repo by editing data instead of code:
|
||||
|
||||
repo_root where pytest runs (cwd); test_paths are relative to it
|
||||
test_paths default search roots when no paths are given
|
||||
venv_python the interpreter the *tests* run in (separate from the
|
||||
runner's own venv); falls back to the current interpreter
|
||||
coverage_source packages measured under --cov, and the report path filter
|
||||
|
||||
All paths in the JSON are resolved relative to this directory (the runner
|
||||
folder) unless absolute. Missing keys fall back to today's behavior.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import sys
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
|
||||
RUNNER_DIR = Path(__file__).resolve().parent
|
||||
DEFAULT_CONFIG_PATH = RUNNER_DIR / "config.json"
|
||||
|
||||
# Defaults that reproduce the pre-config behavior of the runner.
|
||||
_DEFAULT_REPO_ROOT = RUNNER_DIR.parents[1] # <repo>/tests/runner -> <repo>
|
||||
_DEFAULT_TEST_PATHS = ["tests/unit", "tests/api"]
|
||||
_DEFAULT_COVERAGE_SOURCE = ["backend"]
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Config:
|
||||
repo_root: Path
|
||||
test_paths: list[str]
|
||||
venv_python: str
|
||||
coverage_source: list[str]
|
||||
|
||||
|
||||
def _resolve(base: Path, value: str) -> Path:
|
||||
p = Path(value).expanduser()
|
||||
return p if p.is_absolute() else (base / p).resolve()
|
||||
|
||||
|
||||
def load_config(path: Path | None = None) -> Config:
|
||||
"""Load and resolve the runner config. Unknown/missing keys use defaults."""
|
||||
cfg_path = path or DEFAULT_CONFIG_PATH
|
||||
raw: dict = {}
|
||||
if cfg_path.is_file():
|
||||
raw = json.loads(cfg_path.read_text())
|
||||
|
||||
repo_root = (
|
||||
_resolve(RUNNER_DIR, raw["repo_root"])
|
||||
if raw.get("repo_root")
|
||||
else _DEFAULT_REPO_ROOT
|
||||
)
|
||||
|
||||
test_paths = raw.get("test_paths") or list(_DEFAULT_TEST_PATHS)
|
||||
coverage_source = raw.get("coverage_source") or list(_DEFAULT_COVERAGE_SOURCE)
|
||||
|
||||
# The test interpreter is resolved against repo_root. If it is unset or does
|
||||
# not exist, fall back to the interpreter running the runner so the tool
|
||||
# still works in a single-venv setup. We deliberately do NOT call
|
||||
# ``.resolve()`` here: a venv's ``python`` is a symlink to the base
|
||||
# interpreter, and following it would dereference away the venv.
|
||||
venv_python = sys.executable
|
||||
if raw.get("venv_python"):
|
||||
candidate = Path(raw["venv_python"]).expanduser()
|
||||
if not candidate.is_absolute():
|
||||
candidate = repo_root / candidate
|
||||
if candidate.exists():
|
||||
venv_python = str(candidate)
|
||||
|
||||
return Config(
|
||||
repo_root=repo_root,
|
||||
test_paths=test_paths,
|
||||
venv_python=venv_python,
|
||||
coverage_source=coverage_source,
|
||||
)
|
||||
@@ -0,0 +1,56 @@
|
||||
"""Test discovery by delegating to pytest's own collector.
|
||||
|
||||
We never parse test files ourselves — we ask pytest to collect and emit node
|
||||
IDs. This matches pytest exactly (handles asyncio_mode=auto, parametrization,
|
||||
classes, markers) instead of guessing from decorators like the old runner did.
|
||||
|
||||
Collection runs in the configured *test* venv (``config.venv_python``), because
|
||||
collecting imports the test modules and therefore needs the project's full test
|
||||
dependencies — not the runner's own venv.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import subprocess
|
||||
|
||||
from tests.runner.config import load_config
|
||||
|
||||
_CONFIG = load_config()
|
||||
|
||||
# Kept for backwards-compatible imports (e.g. run.py uses it for relpath).
|
||||
REPO_ROOT = _CONFIG.repo_root
|
||||
|
||||
# Default search roots come from config; we avoid scanning tests/runner itself.
|
||||
DEFAULT_PATHS: list[str] = _CONFIG.test_paths
|
||||
|
||||
|
||||
def discover(paths: list[str] | None = None, keyword: str | None = None) -> list[str]:
|
||||
"""Return pytest node IDs for the given paths (optionally -k filtered).
|
||||
|
||||
Raises RuntimeError if pytest collection itself errored.
|
||||
"""
|
||||
search = paths or DEFAULT_PATHS
|
||||
cmd = [
|
||||
_CONFIG.venv_python,
|
||||
"-m",
|
||||
"pytest",
|
||||
"-o",
|
||||
"addopts=", # drop the global -q so node IDs print one per line
|
||||
"--collect-only",
|
||||
"-q",
|
||||
*search,
|
||||
]
|
||||
if keyword:
|
||||
cmd += ["-k", keyword]
|
||||
|
||||
proc = subprocess.run(
|
||||
cmd, capture_output=True, text=True, cwd=str(REPO_ROOT)
|
||||
)
|
||||
# Collection errors (import errors, bad -k) → surface stderr/stdout.
|
||||
if proc.returncode not in (0, 5): # 5 = "no tests collected"
|
||||
raise RuntimeError(
|
||||
f"pytest collection failed (exit {proc.returncode}):\n"
|
||||
f"{proc.stdout}\n{proc.stderr}".strip()
|
||||
)
|
||||
|
||||
return [line.strip() for line in proc.stdout.splitlines() if "::" in line]
|
||||
@@ -0,0 +1,31 @@
|
||||
"""Wire protocol shared by the parent runner and the subprocess worker.
|
||||
|
||||
The worker (running in the *test* venv) writes one JSON object per line to an
|
||||
inherited pipe FD; the parent (running in the *runner* venv) reads them and
|
||||
drives the Rich dashboard. This module is deliberately dependency-free (only the
|
||||
stdlib) so it imports cleanly in *both* venvs — no rich, textual, or pytest.
|
||||
|
||||
Ordering guarantee: both control events and streamed ``-s`` output are written
|
||||
through :func:`emit` to the *same* FD from the *same* process, so an OUTPUT
|
||||
event always lands between its test's LOGSTART and LOGFINISH.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
|
||||
# Event types (the "type" field of every framed message).
|
||||
COLLECTION = "collection" # {"items": [nodeid, ...]}
|
||||
LOGSTART = "logstart" # {"nodeid": ...}
|
||||
LOGREPORT = "logreport" # {nodeid, when, passed, failed, skipped, duration, capstdout, capstderr, longreprtext}
|
||||
LOGFINISH = "logfinish" # {"nodeid": ...}
|
||||
OUTPUT = "output" # {"line": ...} (streamed -s test output)
|
||||
COVERAGE = "coverage" # {"rows": [[rel, stmts, miss, pct], ...], "total": float}
|
||||
DONE = "done" # {"code": int}
|
||||
|
||||
|
||||
def emit(fd: int, obj: dict) -> None:
|
||||
"""Write one framed event to ``fd``. Unbuffered to preserve ordering."""
|
||||
data = (json.dumps(obj) + "\n").encode("utf-8")
|
||||
os.write(fd, data)
|
||||
@@ -0,0 +1,333 @@
|
||||
"""Interactive Textual tree picker.
|
||||
|
||||
Renders the discovered test tree with checkboxes and a row of run-option
|
||||
toggles (coverage, stop-on-first-fail, last-failed, failed-first, verbose, and a
|
||||
``-k`` keyword expression). Returns ``(node_ids, RunOptions)`` for the selection
|
||||
or ``None`` if cancelled. Selecting a branch selects all descendant leaves.
|
||||
Every row always shows a box: empty when unselected, checked when selected, and
|
||||
a half-filled box when only some descendants are selected.
|
||||
|
||||
The tree keybindings live on a ``Screen`` (not the ``App``) so they go quiet
|
||||
while the keyword modal is open and you can freely type expressions like
|
||||
``ingest and not batch`` without the single-key toggles firing.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from rich.text import Text
|
||||
from textual.app import App, ComposeResult
|
||||
from textual.binding import Binding
|
||||
from textual.containers import Vertical
|
||||
from textual.screen import ModalScreen, Screen
|
||||
from textual.widgets import Footer, Header, Input, Label, Static, Tree
|
||||
from textual.widgets.tree import TreeNode
|
||||
|
||||
from tests.runner.run import RunOptions
|
||||
from tests.runner.tree import TNode, build_tree
|
||||
|
||||
# Always-visible ballot-box glyphs (matched empty/checked pair) + a partial box.
|
||||
_BOX_EMPTY = "\u2610" # ☐
|
||||
_BOX_FULL = "\u2611" # ☑
|
||||
_BOX_PARTIAL = "\u25a3" # ▣
|
||||
|
||||
|
||||
class KeywordScreen(ModalScreen[str | None]):
|
||||
"""Modal text prompt for the pytest ``-k`` expression."""
|
||||
|
||||
CSS = """
|
||||
KeywordScreen { align: center middle; }
|
||||
#kw-box {
|
||||
width: 64;
|
||||
height: auto;
|
||||
padding: 1 2;
|
||||
background: $panel;
|
||||
border: round $accent;
|
||||
}
|
||||
#kw-box Label { margin-bottom: 1; }
|
||||
"""
|
||||
|
||||
BINDINGS = [Binding("escape", "cancel", "Cancel")]
|
||||
|
||||
def __init__(self, current: str) -> None:
|
||||
super().__init__()
|
||||
self._current = current
|
||||
|
||||
def compose(self) -> ComposeResult:
|
||||
with Vertical(id="kw-box"):
|
||||
yield Label("pytest -k expression (enter to apply, blank to clear)")
|
||||
yield Input(
|
||||
value=self._current,
|
||||
placeholder="e.g. ingest and not batch",
|
||||
id="kw-input",
|
||||
)
|
||||
|
||||
def on_mount(self) -> None:
|
||||
self.query_one(Input).focus()
|
||||
|
||||
def on_input_submitted(self, event: Input.Submitted) -> None:
|
||||
self.dismiss(event.value.strip())
|
||||
|
||||
def action_cancel(self) -> None:
|
||||
self.dismiss(None)
|
||||
|
||||
|
||||
class PickerScreen(Screen):
|
||||
CSS = """
|
||||
Tree { padding: 0 1; }
|
||||
#status { padding: 0 1; }
|
||||
"""
|
||||
|
||||
BINDINGS = [
|
||||
Binding("space", "toggle_select", "Select", priority=True),
|
||||
Binding("right", "expand_node", "Expand", priority=True),
|
||||
Binding("left", "collapse_node", "Collapse", priority=True),
|
||||
Binding("a", "toggle_all", "All", priority=True),
|
||||
Binding("enter", "run", "Run", priority=True),
|
||||
Binding("c", "toggle_cov", "cov"),
|
||||
Binding("x", "toggle_x", "stop"),
|
||||
Binding("l", "toggle_lf", "last-fail"),
|
||||
Binding("f", "toggle_ff", "fail-first"),
|
||||
Binding("v", "toggle_v", "verbose"),
|
||||
Binding("s", "toggle_s", "stdout"),
|
||||
Binding("o", "toggle_o", "output"),
|
||||
Binding("k", "keyword", "keyword"),
|
||||
Binding("q", "cancel", "Quit", priority=True),
|
||||
Binding("escape", "cancel", "Quit", priority=True),
|
||||
]
|
||||
|
||||
def __init__(self, root: TNode, opts: RunOptions) -> None:
|
||||
super().__init__()
|
||||
self._root = root
|
||||
self.selected: set[str] = set()
|
||||
self._leaf_id: dict[TreeNode, str] = {}
|
||||
self._base_label: dict[TreeNode, str] = {}
|
||||
self._kind: dict[TreeNode, str] = {}
|
||||
# Run-option state, seeded from the CLI flags.
|
||||
self.cov = opts.cov
|
||||
self.exitfirst = opts.exitfirst
|
||||
self.last_failed = opts.last_failed
|
||||
self.failed_first = opts.failed_first
|
||||
self.verbose = opts.verbose
|
||||
self.no_capture = opts.no_capture
|
||||
self.show_output = opts.show_output
|
||||
self.keyword = opts.keyword
|
||||
|
||||
def compose(self) -> ComposeResult:
|
||||
yield Header(show_clock=False)
|
||||
yield Static(id="status")
|
||||
tree: Tree = Tree(self._root.label, id="tree")
|
||||
tree.root.expand()
|
||||
self._base_label[tree.root] = self._root.label
|
||||
self._kind[tree.root] = self._root.kind
|
||||
self._build(tree.root, self._root)
|
||||
self._refresh_labels(tree)
|
||||
yield tree
|
||||
yield Footer()
|
||||
|
||||
def on_mount(self) -> None:
|
||||
self.app.title = "test picker"
|
||||
self.app.sub_title = (
|
||||
f"{_BOX_EMPTY} empty {_BOX_FULL} selected {_BOX_PARTIAL} partial"
|
||||
)
|
||||
self._refresh_status()
|
||||
|
||||
# --- tree construction -------------------------------------------------
|
||||
def _build(self, widget: TreeNode, tnode: TNode) -> None:
|
||||
for child in tnode.children:
|
||||
if child.is_leaf:
|
||||
leaf = widget.add_leaf(child.label)
|
||||
self._base_label[leaf] = child.label
|
||||
self._kind[leaf] = child.kind
|
||||
if child.node_id:
|
||||
self._leaf_id[leaf] = child.node_id
|
||||
else:
|
||||
branch = widget.add(child.label, expand=True)
|
||||
self._base_label[branch] = child.label
|
||||
self._kind[branch] = child.kind
|
||||
self._build(branch, child)
|
||||
|
||||
# --- selection helpers -------------------------------------------------
|
||||
def _leaves_under(self, node: TreeNode) -> set[str]:
|
||||
if node in self._leaf_id:
|
||||
return {self._leaf_id[node]}
|
||||
out: set[str] = set()
|
||||
for child in node.children:
|
||||
out |= self._leaves_under(child)
|
||||
return out
|
||||
|
||||
def _all_nodes(self, node: TreeNode) -> list[TreeNode]:
|
||||
nodes = [node]
|
||||
for child in node.children:
|
||||
nodes.extend(self._all_nodes(child))
|
||||
return nodes
|
||||
|
||||
def _refresh_labels(self, tree: Tree) -> None:
|
||||
for node in self._all_nodes(tree.root):
|
||||
base = self._base_label.get(node, str(node.label))
|
||||
ids = self._leaves_under(node)
|
||||
sel = ids & self.selected
|
||||
if ids and sel == ids:
|
||||
glyph, style = _BOX_FULL, "bold green"
|
||||
elif sel:
|
||||
glyph, style = _BOX_PARTIAL, "bold yellow"
|
||||
else:
|
||||
glyph, style = _BOX_EMPTY, "grey62"
|
||||
label = Text.assemble((f"{glyph} ", style), (base, ""))
|
||||
node.set_label(label)
|
||||
|
||||
# --- run-option status line -------------------------------------------
|
||||
def _refresh_status(self) -> None:
|
||||
def mark(on: bool) -> str:
|
||||
return "[bold green]on[/]" if on else "[dim]off[/]"
|
||||
|
||||
kw = self.keyword or "[dim]—[/]"
|
||||
status = " ".join(
|
||||
[
|
||||
f"[bold]\\[c][/] cov {mark(self.cov)}",
|
||||
f"[bold]\\[x][/] stop {mark(self.exitfirst)}",
|
||||
f"[bold]\\[l][/] --lf {mark(self.last_failed)}",
|
||||
f"[bold]\\[f][/] --ff {mark(self.failed_first)}",
|
||||
f"[bold]\\[v][/] -v {mark(self.verbose)}",
|
||||
f"[bold]\\[s][/] -s {mark(self.no_capture)}",
|
||||
f"[bold]\\[o][/] out {mark(self.show_output)}",
|
||||
f"[bold]\\[k][/] -k {kw}",
|
||||
]
|
||||
)
|
||||
self.query_one("#status", Static).update(status)
|
||||
|
||||
def options(self) -> RunOptions:
|
||||
return RunOptions(
|
||||
cov=self.cov,
|
||||
exitfirst=self.exitfirst,
|
||||
last_failed=self.last_failed,
|
||||
failed_first=self.failed_first,
|
||||
verbose=self.verbose,
|
||||
no_capture=self.no_capture,
|
||||
show_output=self.show_output,
|
||||
keyword=self.keyword,
|
||||
)
|
||||
|
||||
# --- selection actions -------------------------------------------------
|
||||
def action_toggle_select(self) -> None:
|
||||
tree = self.query_one(Tree)
|
||||
node = tree.cursor_node
|
||||
if node is None:
|
||||
return
|
||||
ids = self._leaves_under(node)
|
||||
if ids and ids <= self.selected:
|
||||
self.selected -= ids
|
||||
else:
|
||||
self.selected |= ids
|
||||
self._refresh_labels(tree)
|
||||
|
||||
def action_expand_node(self) -> None:
|
||||
tree = self.query_one(Tree)
|
||||
node = tree.cursor_node
|
||||
if node is None or not node.allow_expand:
|
||||
return
|
||||
if not node.is_expanded:
|
||||
node.expand()
|
||||
elif node.children:
|
||||
# Already expanded → step into the first child (like pressing down).
|
||||
tree.move_cursor(node.children[0])
|
||||
|
||||
def action_collapse_node(self) -> None:
|
||||
tree = self.query_one(Tree)
|
||||
node = tree.cursor_node
|
||||
if node is None:
|
||||
return
|
||||
# On an expanded directory, collapse the directory itself.
|
||||
if (
|
||||
node is not tree.root
|
||||
and self._kind.get(node) == "dir"
|
||||
and node.allow_expand
|
||||
and node.is_expanded
|
||||
):
|
||||
node.collapse()
|
||||
return
|
||||
# Anywhere else (file, function, case, or a collapsed dir): collapse the
|
||||
# section that contains this node and move the cursor onto it.
|
||||
parent = node.parent
|
||||
if parent is None:
|
||||
return
|
||||
if parent is not tree.root and parent.allow_expand and parent.is_expanded:
|
||||
parent.collapse()
|
||||
tree.move_cursor(parent)
|
||||
|
||||
def action_toggle_all(self) -> None:
|
||||
tree = self.query_one(Tree)
|
||||
everything = self._leaves_under(tree.root)
|
||||
self.selected = set() if self.selected >= everything else set(everything)
|
||||
self._refresh_labels(tree)
|
||||
|
||||
# --- run-option actions ------------------------------------------------
|
||||
def action_toggle_cov(self) -> None:
|
||||
self.cov = not self.cov
|
||||
self._refresh_status()
|
||||
|
||||
def action_toggle_x(self) -> None:
|
||||
self.exitfirst = not self.exitfirst
|
||||
self._refresh_status()
|
||||
|
||||
def action_toggle_lf(self) -> None:
|
||||
self.last_failed = not self.last_failed
|
||||
if self.last_failed:
|
||||
self.failed_first = False # --lf and --ff are mutually exclusive
|
||||
self._refresh_status()
|
||||
|
||||
def action_toggle_ff(self) -> None:
|
||||
self.failed_first = not self.failed_first
|
||||
if self.failed_first:
|
||||
self.last_failed = False
|
||||
self._refresh_status()
|
||||
|
||||
def action_toggle_v(self) -> None:
|
||||
self.verbose = not self.verbose
|
||||
self._refresh_status()
|
||||
|
||||
def action_toggle_s(self) -> None:
|
||||
self.no_capture = not self.no_capture
|
||||
self._refresh_status()
|
||||
|
||||
def action_toggle_o(self) -> None:
|
||||
self.show_output = not self.show_output
|
||||
self._refresh_status()
|
||||
|
||||
def action_keyword(self) -> None:
|
||||
def apply(value: str | None) -> None:
|
||||
if value is not None: # None = cancelled, leave keyword untouched
|
||||
self.keyword = value or None
|
||||
self._refresh_status()
|
||||
|
||||
self.app.push_screen(KeywordScreen(self.keyword or ""), apply)
|
||||
|
||||
# --- finish ------------------------------------------------------------
|
||||
def action_run(self) -> None:
|
||||
chosen = set(self.selected)
|
||||
if not chosen:
|
||||
node = self.query_one(Tree).cursor_node
|
||||
if node is not None:
|
||||
chosen = self._leaves_under(node)
|
||||
self.app.exit((sorted(chosen), self.options()))
|
||||
|
||||
def action_cancel(self) -> None:
|
||||
self.app.exit(None)
|
||||
|
||||
|
||||
class TestPicker(App):
|
||||
def __init__(self, root: TNode, opts: RunOptions) -> None:
|
||||
super().__init__()
|
||||
self._root = root
|
||||
self._opts = opts
|
||||
|
||||
def on_mount(self) -> None:
|
||||
self.push_screen(PickerScreen(self._root, self._opts))
|
||||
|
||||
|
||||
def run_picker(
|
||||
node_ids: list[str], opts: RunOptions
|
||||
) -> tuple[list[str], RunOptions] | None:
|
||||
"""Launch the picker. Returns ``(node_ids, options)`` or None if cancelled."""
|
||||
root = build_tree(node_ids)
|
||||
app = TestPicker(root, opts)
|
||||
return app.run()
|
||||
@@ -0,0 +1,27 @@
|
||||
# Dependency manifest for the test runner's OWN venv.
|
||||
#
|
||||
# These are the UI libraries the runner process needs. They are intentionally
|
||||
# separate from the test venv: the runner imports typer/rich/textual but never
|
||||
# pytest, and the worker subprocess imports pytest but never rich/textual.
|
||||
#
|
||||
# Install into a dedicated runner venv, e.g.:
|
||||
# python -m venv .runner-venv
|
||||
# .runner-venv/bin/pip install -e tests/runner
|
||||
#
|
||||
# The *test* venv (pointed at by config.json -> "venv_python") is a separate
|
||||
# environment that must provide: pytest, pytest-asyncio, coverage, plus the
|
||||
# project's own runtime/test dependencies.
|
||||
|
||||
[project]
|
||||
name = "test-runner"
|
||||
version = "0.1.0"
|
||||
description = "Interactive subprocess test runner with a live Rich dashboard."
|
||||
requires-python = ">=3.9"
|
||||
dependencies = [
|
||||
"typer", # CLI entrypoint (__main__.py)
|
||||
"rich", # live dashboard (run.py)
|
||||
"textual", # interactive tree picker (picker.py)
|
||||
]
|
||||
|
||||
[tool.setuptools]
|
||||
packages = ["tests.runner"]
|
||||
@@ -0,0 +1,581 @@
|
||||
"""Subprocess pytest execution with a live Rich dashboard.
|
||||
|
||||
pytest runs in a separate process (the configured *test* venv) via
|
||||
``tests.runner._worker``. That worker streams framed events over a pipe; this
|
||||
module consumes them, mutates a :class:`Dashboard`, and renders a live Rich
|
||||
panel plus a final summary and (optionally) a coverage table. Keeping execution
|
||||
out-of-process is what lets the runner's own venv stay free of pytest and the
|
||||
project's test dependencies.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
import time
|
||||
from collections import defaultdict
|
||||
from dataclasses import dataclass
|
||||
|
||||
from rich import box
|
||||
from rich.console import Console, Group
|
||||
from rich.live import Live
|
||||
from rich.panel import Panel
|
||||
from rich.progress_bar import ProgressBar
|
||||
from rich.spinner import Spinner
|
||||
from rich.table import Table
|
||||
from rich.text import Text
|
||||
|
||||
from tests.runner import events
|
||||
from tests.runner.config import load_config
|
||||
|
||||
_CONFIG = load_config()
|
||||
|
||||
# Cap per-test captured output so one chatty test can't flood the scrollback.
|
||||
_MAX_OUTPUT_LINES = 40
|
||||
|
||||
_OUTCOME_GLYPH: dict[str, tuple[str, str]] = {
|
||||
"passed": ("\u2713", "green"), # ✓
|
||||
"failed": ("\u2717", "bold red"), # ✗
|
||||
"skipped": ("\u21b7", "yellow"), # ↷
|
||||
"error": ("\u2717", "bold magenta"), # ✗ (setup/teardown)
|
||||
}
|
||||
|
||||
|
||||
def _short(nodeid: str, width: int = 80) -> str:
|
||||
"""Keep the spinner's current-test label on a single line."""
|
||||
return nodeid if len(nodeid) <= width else "…" + nodeid[-(width - 1):]
|
||||
|
||||
|
||||
@dataclass
|
||||
class RunOptions:
|
||||
"""Execution options shared between the CLI, the picker, and run_tests."""
|
||||
|
||||
cov: bool = False # measure backend/ coverage
|
||||
exitfirst: bool = False # -x : stop after the first failure
|
||||
last_failed: bool = False # --lf: run only last-failed tests
|
||||
failed_first: bool = False # --ff: run last-failed first, then the rest
|
||||
verbose: bool = False # -v : richer failure output (untruncated asserts)
|
||||
no_capture: bool = False # -s : stream test stdout / print() output live
|
||||
show_output: bool = False # -O : show captured output for passing tests too
|
||||
keyword: str | None = None # -k : only tests matching this expression
|
||||
|
||||
|
||||
class Dashboard:
|
||||
"""Event consumer + Rich renderer (runs in the parent / runner venv).
|
||||
|
||||
Equivalent to the old in-process collector, but instead of pytest hooks it
|
||||
is driven by :meth:`handle_event` over events streamed from the worker. All
|
||||
counters, per-file tallies, and the two live layouts are unchanged:
|
||||
|
||||
* ``-s`` ON: output streams live, so each test gets a header rule + a
|
||||
persistent ``✓/✗ nodeid duration`` scrollback line bracketing its prints.
|
||||
* ``-s`` OFF: a per-file table with live ``passed/failed/total`` counters is
|
||||
pinned at the bottom; captured output for failures (or all tests under
|
||||
``-O``) still prints above it.
|
||||
|
||||
The dashboard renders itself (``__rich__``) so Live's auto-refresh animates
|
||||
the spinner and ticks the elapsed clock without us pushing snapshots.
|
||||
"""
|
||||
|
||||
def __init__(self, opts: RunOptions) -> None:
|
||||
self.opts = opts
|
||||
self.total = 0
|
||||
self.passed = 0
|
||||
self.failed = 0
|
||||
self.skipped = 0
|
||||
self.errors = 0
|
||||
self.current = ""
|
||||
self.failures: list[tuple[str, str]] = []
|
||||
self.live: Live | None = None
|
||||
self._start = time.time()
|
||||
self._current_start = self._start
|
||||
# Per-test accumulator across setup/call/teardown phases, keyed by nodeid.
|
||||
self._pending: dict[str, dict] = {}
|
||||
self._spinner = Spinner("dots", style="cyan")
|
||||
# Per-file tallies for the -s-off view (files in collection order).
|
||||
self._file_order: list[str] = []
|
||||
self._file_total: dict[str, int] = defaultdict(int)
|
||||
self._file_pass: dict[str, int] = defaultdict(int)
|
||||
self._file_fail: dict[str, int] = defaultdict(int)
|
||||
self._file_skip: dict[str, int] = defaultdict(int)
|
||||
self._current_file = ""
|
||||
# -s streaming: track the open file bracket so we can rail + close it.
|
||||
self._streamed_file: str | None = None
|
||||
self._file_open = False
|
||||
|
||||
@staticmethod
|
||||
def _file_of(nodeid: str) -> str:
|
||||
return nodeid.split("::", 1)[0]
|
||||
|
||||
# --- event dispatch ----------------------------------------------------
|
||||
def handle_event(self, evt: dict) -> None:
|
||||
etype = evt.get("type")
|
||||
if etype == events.COLLECTION:
|
||||
self._on_collection(evt["items"])
|
||||
elif etype == events.LOGSTART:
|
||||
self._on_logstart(evt["nodeid"])
|
||||
elif etype == events.LOGREPORT:
|
||||
self._on_logreport(evt)
|
||||
elif etype == events.LOGFINISH:
|
||||
self._on_logfinish(evt["nodeid"])
|
||||
elif etype == events.OUTPUT:
|
||||
self._emit_stream_line(evt["line"])
|
||||
|
||||
def _on_collection(self, items: list[str]) -> None:
|
||||
self.total = len(items)
|
||||
for nodeid in items:
|
||||
f = self._file_of(nodeid)
|
||||
if f not in self._file_total:
|
||||
self._file_order.append(f)
|
||||
self._file_total[f] += 1
|
||||
self._refresh()
|
||||
|
||||
def _on_logstart(self, nodeid: str) -> None:
|
||||
self.current = nodeid
|
||||
self._current_file = self._file_of(nodeid)
|
||||
self._current_start = time.time()
|
||||
# In -s mode, output streams live; print a file header box (once per
|
||||
# file) and a per-test header box so the prints that follow are
|
||||
# attributable. The streamed output itself is left unboxed.
|
||||
if self.opts.no_capture and self.live is not None:
|
||||
self._emit_stream_headers(nodeid)
|
||||
self._refresh()
|
||||
|
||||
def _on_logreport(self, evt: dict) -> None:
|
||||
nodeid = evt["nodeid"]
|
||||
when = evt["when"]
|
||||
entry = self._pending.setdefault(
|
||||
nodeid, {"outcome": "passed", "duration": 0.0, "out": ""}
|
||||
)
|
||||
entry["duration"] += evt.get("duration", 0.0) or 0.0
|
||||
# Take captured output from the authoritative phase only: the call, or
|
||||
# setup when setup fails/skips (no call then). pytest's teardown report
|
||||
# re-includes the call's capstdout, so summing every phase would
|
||||
# duplicate the test's output.
|
||||
if when == "call" or (when == "setup" and (evt["failed"] or evt["skipped"])):
|
||||
out = (evt.get("capstdout") or "") + (evt.get("capstderr") or "")
|
||||
if out:
|
||||
entry["out"] = out
|
||||
|
||||
longrepr = evt.get("longreprtext") or ""
|
||||
if when == "setup":
|
||||
if evt["skipped"]:
|
||||
entry["outcome"] = "skipped"
|
||||
self.skipped += 1
|
||||
elif evt["failed"]:
|
||||
entry["outcome"] = "error"
|
||||
self.errors += 1
|
||||
self.failures.append((nodeid + " (setup)", longrepr))
|
||||
elif when == "call":
|
||||
if evt["passed"]:
|
||||
self.passed += 1
|
||||
elif evt["failed"]:
|
||||
entry["outcome"] = "failed"
|
||||
self.failed += 1
|
||||
self.failures.append((nodeid, longrepr))
|
||||
elif evt["skipped"]:
|
||||
entry["outcome"] = "skipped"
|
||||
self.skipped += 1
|
||||
elif when == "teardown" and evt["failed"]:
|
||||
if entry["outcome"] == "passed":
|
||||
entry["outcome"] = "error"
|
||||
self.errors += 1
|
||||
self.failures.append((nodeid + " (teardown)", longrepr))
|
||||
self._refresh()
|
||||
|
||||
def _on_logfinish(self, nodeid: str) -> None:
|
||||
entry = self._pending.pop(nodeid, None)
|
||||
self.current = ""
|
||||
if entry is None:
|
||||
self._refresh()
|
||||
return
|
||||
|
||||
# Update per-file tallies for the -s-off table.
|
||||
f = self._file_of(nodeid)
|
||||
outcome = entry["outcome"]
|
||||
if outcome == "passed":
|
||||
self._file_pass[f] += 1
|
||||
elif outcome in ("failed", "error"):
|
||||
self._file_fail[f] += 1
|
||||
elif outcome == "skipped":
|
||||
self._file_skip[f] += 1
|
||||
|
||||
if self.live is not None:
|
||||
if self.opts.no_capture:
|
||||
# Streamed mode: persistent per-test line bracketing the output.
|
||||
self._emit_test_line(nodeid, entry)
|
||||
else:
|
||||
# File-counter mode: no per-test line, but still surface
|
||||
# captured output for failures (or all tests under -O).
|
||||
self._emit_output_panel(nodeid, entry)
|
||||
self._refresh()
|
||||
|
||||
# --- streaming (-s) nested brackets -----------------------------------
|
||||
# Two levels: a blue outer bracket per .py file wraps the grey inner
|
||||
# bracket of each test, so tests chunk visually by their parent file.
|
||||
#
|
||||
# ╔═ test_persist.py ═╗ (file header box)
|
||||
# ┃ ╭ test_one (┃ = blue file rail; ╭ opens the test)
|
||||
# ┃ │ ...streamed output (railed + gutter-wrapped)
|
||||
# ┃ ╰ ✓ 11ms (closes the test)
|
||||
# ┃
|
||||
# ┗━━━ (blue foot closes the file)
|
||||
_OUTER = "\u2503 " # "┃ "
|
||||
_FOOT = "\u2517" + "\u2501" * 3 # "┗━━━"
|
||||
_OPEN = "\u256d " # "╭ "
|
||||
_RAIL = "\u2502 " # "│ "
|
||||
_CLOSE = "\u2570 " # "╰ "
|
||||
_OUTER_STYLE = "blue"
|
||||
|
||||
def _outer(self) -> Text:
|
||||
return Text(self._OUTER, style=self._OUTER_STYLE)
|
||||
|
||||
def _emit_stream_headers(self, nodeid: str) -> None:
|
||||
f = self._file_of(nodeid)
|
||||
if f != self._streamed_file:
|
||||
self._close_file_bracket() # close the previous file's blue bracket
|
||||
self._streamed_file = f
|
||||
self.live.console.print(self._file_header(f))
|
||||
self._file_open = True
|
||||
name = nodeid.split("::", 1)[1] if "::" in nodeid else nodeid
|
||||
self.live.console.print(
|
||||
self._outer() + Text.assemble((self._OPEN, "grey50"), (name, "bold"))
|
||||
)
|
||||
|
||||
def _close_file_bracket(self) -> None:
|
||||
if self.live is not None and self._file_open:
|
||||
self.live.console.print(Text(self._FOOT, style=self._OUTER_STYLE))
|
||||
self.live.console.print()
|
||||
self._file_open = False
|
||||
|
||||
@staticmethod
|
||||
def _file_header(f: str):
|
||||
return Panel(
|
||||
Text(f, style="bold blue"),
|
||||
box=box.DOUBLE,
|
||||
border_style="blue",
|
||||
expand=False,
|
||||
padding=(0, 1),
|
||||
)
|
||||
|
||||
def _emit_stream_line(self, line: str) -> None:
|
||||
"""Forward one line of test output, railed inside both brackets."""
|
||||
if self.live is None:
|
||||
return
|
||||
if line == "":
|
||||
self.live.console.print(self._outer() + Text("\u2502", style="grey50"))
|
||||
return
|
||||
inner = Text(self._RAIL, style="grey50")
|
||||
text = Text.from_ansi(line)
|
||||
prefix_w = len(self._OUTER) + len(self._RAIL)
|
||||
width = max(self.live.console.size.width - prefix_w, 1)
|
||||
for seg in text.wrap(self.live.console, width):
|
||||
self.live.console.print(self._outer() + inner + seg)
|
||||
|
||||
# --- scrollback emitters ----------------------------------------------
|
||||
def _emit_test_line(self, nodeid: str, entry: dict) -> None:
|
||||
outcome = entry["outcome"]
|
||||
glyph, style = _OUTCOME_GLYPH.get(outcome, ("\u2022", "white"))
|
||||
dur = entry["duration"]
|
||||
timing = f"{dur * 1000:.0f}ms" if dur < 1 else f"{dur:.2f}s"
|
||||
# Close the test bracket with the result, then a rail-only spacer line.
|
||||
self.live.console.print(
|
||||
self._outer()
|
||||
+ Text.assemble((self._CLOSE, style), (f"{glyph} ", style), (timing, "dim"))
|
||||
)
|
||||
self.live.console.print(self._outer())
|
||||
|
||||
def _emit_output_panel(self, nodeid: str, entry: dict) -> None:
|
||||
outcome = entry["outcome"]
|
||||
_, style = _OUTCOME_GLYPH.get(outcome, ("\u2022", "white"))
|
||||
out = entry["out"].rstrip()
|
||||
show = self.opts.show_output or outcome in ("failed", "error")
|
||||
if out and show:
|
||||
self.live.console.print(self._output_panel(nodeid, out, style))
|
||||
|
||||
def _output_panel(self, nodeid: str, out: str, style: str) -> Panel:
|
||||
lines = out.splitlines()
|
||||
hidden = len(lines) - _MAX_OUTPUT_LINES
|
||||
if hidden > 0:
|
||||
body = Text("\n".join(lines[-_MAX_OUTPUT_LINES:]))
|
||||
body.append(f"\n… {hidden} earlier line(s) hidden", style="dim italic")
|
||||
else:
|
||||
body = Text(out)
|
||||
return Panel(
|
||||
body,
|
||||
title=f"output · {nodeid}",
|
||||
title_align="left",
|
||||
border_style="grey42",
|
||||
padding=(0, 1),
|
||||
)
|
||||
|
||||
# --- rendering ---------------------------------------------------------
|
||||
@property
|
||||
def done(self) -> int:
|
||||
return self.passed + self.failed + self.skipped + self.errors
|
||||
|
||||
def _refresh(self) -> None:
|
||||
if self.live is not None:
|
||||
try:
|
||||
self.live.refresh()
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
def __rich__(self):
|
||||
return self.render_progress()
|
||||
|
||||
def render_progress(self):
|
||||
# -s off → per-file counter table; -s on → overall bar + streaming spinner.
|
||||
if self.opts.no_capture:
|
||||
return self._render_streaming()
|
||||
return self._render_by_file()
|
||||
|
||||
def _current_line(self):
|
||||
"""The animated 'currently running' spinner line (shared by both views)."""
|
||||
if self.current:
|
||||
elapsed = time.time() - self._current_start
|
||||
self._spinner.update(
|
||||
text=Text.assemble(
|
||||
(_short(self.current), "cyan"), (f" {elapsed:0.1f}s", "cyan dim")
|
||||
)
|
||||
)
|
||||
return self._spinner
|
||||
if self.total and self.done >= self.total:
|
||||
return Text("finished", style="green dim")
|
||||
return Text("collecting…", style="cyan dim")
|
||||
|
||||
def _render_streaming(self):
|
||||
bar = ProgressBar(total=max(self.total, 1), completed=self.done, width=46)
|
||||
counts = Text.assemble(
|
||||
(" passed ", "dim"), (f"{self.passed}", "bold green"),
|
||||
(" failed ", "dim"), (f"{self.failed}", "bold red"),
|
||||
(" skipped ", "dim"), (f"{self.skipped}", "bold yellow"),
|
||||
(" errors ", "dim"), (f"{self.errors}", "bold magenta"),
|
||||
)
|
||||
progress_line = Text.assemble(
|
||||
(f"{self.done}", "bold"), (f"/{self.total} ", "dim"),
|
||||
)
|
||||
return Panel(
|
||||
Group(Group(progress_line, bar), counts, self._current_line()),
|
||||
title="[bold]running tests[/bold]",
|
||||
border_style="cyan",
|
||||
)
|
||||
|
||||
def _render_by_file(self):
|
||||
if not self._file_order:
|
||||
return Panel(
|
||||
Text("collecting…", style="cyan dim"),
|
||||
title="[bold]running tests[/bold]",
|
||||
border_style="cyan",
|
||||
)
|
||||
table = Table(show_header=False, box=None, pad_edge=False, expand=False)
|
||||
table.add_column(width=1) # running marker
|
||||
table.add_column(no_wrap=True) # file
|
||||
table.add_column(justify="right") # passed
|
||||
table.add_column(justify="right") # failed
|
||||
table.add_column(justify="right") # done/total
|
||||
table.add_column(width=18) # progress bar
|
||||
for f in self._file_order:
|
||||
total = self._file_total.get(f, 0)
|
||||
passed = self._file_pass.get(f, 0)
|
||||
failed = self._file_fail.get(f, 0)
|
||||
skipped = self._file_skip.get(f, 0)
|
||||
done = passed + failed + skipped
|
||||
complete = total > 0 and done >= total
|
||||
running = f == self._current_file and not complete and bool(self.current)
|
||||
|
||||
if running:
|
||||
marker = Text("\u25b6", style="cyan") # ▶
|
||||
elif complete:
|
||||
marker = Text("\u2713" if failed == 0 else "\u2717",
|
||||
style="green" if failed == 0 else "bold red")
|
||||
else:
|
||||
marker = Text(" ")
|
||||
|
||||
name_style = "bold red" if failed else ("green" if complete else
|
||||
"cyan" if running else "")
|
||||
bar_style = "red" if failed else "green"
|
||||
table.add_row(
|
||||
marker,
|
||||
Text(f, style=name_style),
|
||||
Text(f"{passed}\u2713", style="green" if passed else "dim"),
|
||||
Text(f"{failed}\u2717", style="bold red" if failed else "dim"),
|
||||
Text(f"{done}/{total}", style="dim"),
|
||||
ProgressBar(
|
||||
total=max(total, 1), completed=done, width=18,
|
||||
complete_style=bar_style, finished_style=bar_style,
|
||||
),
|
||||
)
|
||||
return Panel(
|
||||
Group(table, self._current_line()),
|
||||
title="[bold]running tests[/bold]",
|
||||
border_style="cyan",
|
||||
)
|
||||
|
||||
def render_summary(self, console: Console) -> None:
|
||||
elapsed = time.time() - self._start
|
||||
ok = self.failed == 0 and self.errors == 0
|
||||
table = Table(show_header=False, box=None, pad_edge=False)
|
||||
table.add_row(Text("passed", style="green"), Text(str(self.passed), style="bold green"))
|
||||
if self.failed:
|
||||
table.add_row(Text("failed", style="red"), Text(str(self.failed), style="bold red"))
|
||||
if self.skipped:
|
||||
table.add_row(Text("skipped", style="yellow"), Text(str(self.skipped), style="bold yellow"))
|
||||
if self.errors:
|
||||
table.add_row(Text("errors", style="magenta"), Text(str(self.errors), style="bold magenta"))
|
||||
table.add_row(Text("time", style="dim"), Text(f"{elapsed:.2f}s", style="dim"))
|
||||
|
||||
verdict = "[bold green]PASSED[/]" if ok else "[bold red]FAILED[/]"
|
||||
console.print(
|
||||
Panel(table, title=verdict, border_style="green" if ok else "red")
|
||||
)
|
||||
|
||||
for nodeid, longrepr in self.failures:
|
||||
console.print(
|
||||
Panel(
|
||||
Text(longrepr or "(no traceback captured)"),
|
||||
title=f"[red]{nodeid}[/]",
|
||||
border_style="red",
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
def render_coverage(console: Console, rows: list, total: float) -> None:
|
||||
"""Render the coverage table from rows computed by the worker."""
|
||||
table = Table(title="coverage", title_style="bold", header_style="dim")
|
||||
table.add_column("file")
|
||||
table.add_column("stmts", justify="right")
|
||||
table.add_column("miss", justify="right")
|
||||
table.add_column("cover", justify="right")
|
||||
|
||||
for rel, n, miss, pct in sorted(rows, key=lambda r: r[3]):
|
||||
colour = "green" if pct >= 90 else "yellow" if pct >= 70 else "red"
|
||||
table.add_row(rel, str(n), str(miss), Text(f"{pct:.0f}%", style=colour))
|
||||
|
||||
console.print(table)
|
||||
if total is not None:
|
||||
tcolour = "green" if total >= 90 else "yellow" if total >= 70 else "red"
|
||||
console.print(Text.assemble(("TOTAL ", "bold"), (f"{total:.0f}%", f"bold {tcolour}")))
|
||||
|
||||
|
||||
def _build_pytest_args(node_ids: list[str], opts: RunOptions) -> list[str]:
|
||||
"""Translate RunOptions into the pytest flags handed to the worker."""
|
||||
args: list[str] = []
|
||||
if opts.exitfirst:
|
||||
args.append("-x")
|
||||
if opts.last_failed:
|
||||
args.append("--lf")
|
||||
if opts.failed_first:
|
||||
args.append("--ff")
|
||||
if opts.verbose:
|
||||
# `-v` is owned by pytest's terminal plugin, which the worker disables
|
||||
# via `-p no:terminal`. The part of verbosity that still matters for our
|
||||
# custom dashboard is untruncated assertion diffs in the failure panels,
|
||||
# which this core ini option controls independently.
|
||||
args += ["-o", "verbosity_assertions=2"]
|
||||
if opts.no_capture:
|
||||
# Disable pytest's output capture so test stdout / print() reaches the
|
||||
# worker's gutter shim, which frames each line back to us as an event.
|
||||
args.append("-s")
|
||||
if opts.keyword:
|
||||
args += ["-k", opts.keyword]
|
||||
args += node_ids
|
||||
return args
|
||||
|
||||
|
||||
def run_tests(node_ids: list[str], opts: RunOptions | None = None) -> int:
|
||||
"""Run the given node IDs in the test venv (subprocess). Returns the exit code."""
|
||||
if opts is None:
|
||||
opts = RunOptions()
|
||||
if not node_ids:
|
||||
node_ids = list(_CONFIG.test_paths)
|
||||
|
||||
console = Console(file=sys.stdout)
|
||||
dashboard = Dashboard(opts)
|
||||
|
||||
worker_opts = {
|
||||
"pytest_args": _build_pytest_args(node_ids, opts),
|
||||
"no_capture": opts.no_capture,
|
||||
"cov": opts.cov,
|
||||
"coverage_source": _CONFIG.coverage_source,
|
||||
"repo_root": str(_CONFIG.repo_root),
|
||||
}
|
||||
|
||||
# The worker writes framed events to `w`; we read them from `r`. Its stderr
|
||||
# goes to a temp file (not a pipe) so a chatty/crashing worker can never
|
||||
# deadlock against our event read.
|
||||
r, w = os.pipe()
|
||||
err_file = tempfile.TemporaryFile(mode="w+")
|
||||
proc = subprocess.Popen(
|
||||
[_CONFIG.venv_python, "-m", "tests.runner._worker", str(w), json.dumps(worker_opts)],
|
||||
cwd=str(_CONFIG.repo_root),
|
||||
pass_fds=(w,),
|
||||
stdout=subprocess.DEVNULL,
|
||||
stderr=err_file,
|
||||
)
|
||||
os.close(w) # parent keeps only the read end
|
||||
|
||||
exit_code: int | None = None
|
||||
saw_collection = False
|
||||
cov_rows: list | None = None
|
||||
cov_total: float | None = None
|
||||
cov_error: str | None = None
|
||||
|
||||
with Live(
|
||||
dashboard,
|
||||
console=console,
|
||||
refresh_per_second=12,
|
||||
redirect_stdout=False,
|
||||
redirect_stderr=False,
|
||||
) as live:
|
||||
dashboard.live = live
|
||||
with os.fdopen(r, "r", buffering=1) as stream:
|
||||
for line in stream:
|
||||
line = line.strip()
|
||||
if not line:
|
||||
continue
|
||||
try:
|
||||
evt = json.loads(line)
|
||||
except json.JSONDecodeError:
|
||||
continue
|
||||
etype = evt.get("type")
|
||||
if etype == events.COLLECTION:
|
||||
saw_collection = True
|
||||
elif etype == events.COVERAGE:
|
||||
cov_rows = evt.get("rows")
|
||||
cov_total = evt.get("total")
|
||||
cov_error = evt.get("error")
|
||||
continue
|
||||
elif etype == events.DONE:
|
||||
exit_code = int(evt.get("code", 1))
|
||||
continue
|
||||
dashboard.handle_event(evt)
|
||||
if opts.no_capture:
|
||||
dashboard._close_file_bracket() # close the last file bracket
|
||||
live.refresh()
|
||||
|
||||
proc.wait()
|
||||
if exit_code is None:
|
||||
exit_code = proc.returncode
|
||||
|
||||
dashboard.render_summary(console)
|
||||
if cov_error:
|
||||
console.print(f"[yellow]coverage unavailable: {cov_error}[/]")
|
||||
elif cov_rows is not None:
|
||||
render_coverage(console, cov_rows, cov_total)
|
||||
|
||||
# If the worker died before collecting anything (e.g. an import error in a
|
||||
# test module, or a missing test venv), surface its stderr.
|
||||
if not saw_collection and exit_code != 0:
|
||||
err_file.seek(0)
|
||||
stderr_text = err_file.read().strip()
|
||||
if stderr_text:
|
||||
console.print(
|
||||
Panel(Text(stderr_text), title="[red]worker stderr[/]", border_style="red")
|
||||
)
|
||||
err_file.close()
|
||||
|
||||
return int(exit_code)
|
||||
@@ -0,0 +1,79 @@
|
||||
"""Turn a flat list of pytest node IDs into a selectable tree.
|
||||
|
||||
Node ID shapes handled:
|
||||
tests/unit/test_tokens.py::test_roundtrip
|
||||
tests/unit/test_route_classifier.py::test_models[gemini-3-flash] (parametrized)
|
||||
|
||||
Directories and files become inner nodes; a parametrized function becomes an
|
||||
inner node whose leaves are its individual cases; a plain function is a leaf.
|
||||
Only leaves carry a runnable node_id.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass, field
|
||||
|
||||
|
||||
@dataclass
|
||||
class TNode:
|
||||
label: str
|
||||
kind: str # 'dir' | 'file' | 'func' | 'case'
|
||||
node_id: str | None = None # set only on leaves (func without params, or case)
|
||||
children: list["TNode"] = field(default_factory=list)
|
||||
|
||||
@property
|
||||
def is_leaf(self) -> bool:
|
||||
return not self.children
|
||||
|
||||
def _find_or_add(self, label: str, kind: str) -> "TNode":
|
||||
for child in self.children:
|
||||
if child.label == label and child.kind == kind:
|
||||
return child
|
||||
child = TNode(label=label, kind=kind)
|
||||
self.children.append(child)
|
||||
return child
|
||||
|
||||
def leaf_ids(self) -> list[str]:
|
||||
if self.is_leaf:
|
||||
return [self.node_id] if self.node_id else []
|
||||
out: list[str] = []
|
||||
for child in self.children:
|
||||
out.extend(child.leaf_ids())
|
||||
return out
|
||||
|
||||
|
||||
def build_tree(node_ids: list[str], root_label: str = "tests") -> TNode:
|
||||
root = TNode(label=root_label, kind="dir")
|
||||
for nid in node_ids:
|
||||
file_part, _, test_part = nid.partition("::")
|
||||
segments = [s for s in file_part.split("/") if s]
|
||||
# Drop a leading "tests" segment so it nests under the single root.
|
||||
if segments and segments[0] == root_label:
|
||||
segments = segments[1:]
|
||||
|
||||
cursor = root
|
||||
for seg in segments[:-1]:
|
||||
cursor = cursor._find_or_add(seg, "dir")
|
||||
if segments:
|
||||
cursor = cursor._find_or_add(segments[-1], "file")
|
||||
|
||||
if "[" in test_part:
|
||||
func_name = test_part.split("[", 1)[0]
|
||||
func_node = cursor._find_or_add(func_name, "func")
|
||||
func_node.children.append(
|
||||
TNode(label=test_part, kind="case", node_id=nid)
|
||||
)
|
||||
else:
|
||||
cursor.children.append(
|
||||
TNode(label=test_part, kind="func", node_id=nid)
|
||||
)
|
||||
_sort(root)
|
||||
return root
|
||||
|
||||
|
||||
def _sort(node: TNode) -> None:
|
||||
# Dirs first, then files, then funcs/cases; alphabetical within a kind.
|
||||
order = {"dir": 0, "file": 1, "func": 2, "case": 3}
|
||||
node.children.sort(key=lambda c: (order.get(c.kind, 9), c.label))
|
||||
for child in node.children:
|
||||
_sort(child)
|
||||
Reference in New Issue
Block a user