scripts.test_durations

Reference › scripts › test_durations

scripts/test_durations.py is a pure reader and report over reports/test-durations.ndjson, the file the live pytest observer appends per-test durations to (workflow-improvements-plan Phase 3.3). A human runs it to find slow tests; tests/scripts/test_test_durations.py and tests/scripts/test_test_durations_residual_coverage.py cover it. The same records feed the scheduler (see schedule_tests) and are written by runs of run_tests.

The file has two line kinds: per-test duration rows (legacy and run-identified) and run-identity header lines (kind: "run"). _iter_raw is a tolerant line reader that skips malformed lines, and _read_all is shape-aware. read_records returns the duration rows and read_headers the headers. slowest returns the N slowest tests using each test's latest duration only, over_threshold lists every test at or above a number of seconds, and summarize_runs groups identified rows by run id and joins them to their headers. Options: --path, --slow N (default 20), --threshold (seconds; lists defect candidates), --run RUN_ID (scope to one run). The subcommand runs (with --last N) lists identified runs, newest first. It only reads.

At a glance

Class diagram

Dependency diagram

No internal imports and no importers.

API

Classes

DurationRecord (dataclass)

Decorators: dataclass(frozen=True) · Line 54

RunHeader (dataclass)

Decorators: dataclass(frozen=True) · Line 64

RunSummary (dataclass)

Decorators: dataclass(frozen=True) · Line 73

Functions

Source

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

#!/usr/bin/env python3
"""Pure reader over ``reports/test-durations.ndjson`` (plans/workflow-
improvements-plan.md Phase 3, Step 3.3).

The write side lives entirely in
``orchestrata_ultimate.core.threading.pytest_live_observer`` -- every
observed pytest run (RUL-009 mandates the observer plugin on every run)
flushes its terminal-phase records to this file at ``pytest_sessionfinish``,
automatically, with zero manual steps. This script never writes to that
file; it only aggregates and reports.

Three record shapes coexist in the sink (see
``tests/core/threading/test_durations_identity.py`` for the writer-side
contract this reader must never disturb):

1. LEGACY rows: ``{"test", "duration", "ts"}`` -- no run identity at all.
2. IDENTIFIED rows: legacy fields + ``"outcome", "run_id", "batch_id"`` --
   rows deliberately carry ONLY the join keys, not the run-level constants.
3. HEADER lines: ``{"kind": "run", "ts", "run_id", "head_commit", "batch_id",
   "lane"}`` -- one per (run_id, batch_id) flush; ``head_commit`` is constant
   across a run's headers, ``lane`` varies per batch. No "test"/"duration".

CLI:
    python scripts/test_durations.py --slow 20
    python scripts/test_durations.py --slow 5 --threshold 60
    python scripts/test_durations.py --run 1786700000-4242 --slow 10
    python scripts/test_durations.py runs --last 10
"""

from __future__ import annotations

import argparse
import json
import sys
from collections.abc import Iterator
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path

REPO_ROOT = Path(__file__).resolve().parent.parent
DEFAULT_PATH = REPO_ROOT / "reports" / "test-durations.ndjson"

#: A test slower than this is a defect candidate (plan-mandated).
DEFAULT_SLOW_THRESHOLD_S = 60.0

#: How many runs the `runs` subcommand shows by default.
DEFAULT_RUNS_LAST = 10

#: Display width for a joined ``head_commit`` (full SHAs are 40 chars).
HEAD_COMMIT_DISPLAY_LEN = 12


@dataclass(frozen=True)
class DurationRecord:
    test: str
    duration: float
    ts: float
    run_id: str | None = None
    batch_id: str | None = None
    outcome: str | None = None


@dataclass(frozen=True)
class RunHeader:
    run_id: str
    batch_id: str
    ts: float
    head_commit: str | None
    lane: str | None


@dataclass(frozen=True)
class RunSummary:
    run_id: str
    head_commit: str | None
    batches: int
    lanes: list[str]
    tests: int
    passed: int
    failed: int
    other_outcomes: int
    wall_seconds: float
    duration_seconds: float
    min_ts: float


def _iter_raw(path: Path) -> Iterator[dict]:
    """Tolerant NDJSON line reader -> parsed dict objects, one per line.

    Malformed/truncated lines (a mid-write kill, same durability discipline
    as the live-observer log) and non-dict JSON are skipped, never fatal.
    Missing file yields nothing.
    """
    if not path.exists():
        return
    with path.open(encoding="utf-8") as fh:
        for line in fh:
            line = line.strip()
            if not line:
                continue
            try:
                obj = json.loads(line)
            except json.JSONDecodeError:
                continue
            if not isinstance(obj, dict):
                continue
            yield obj


def _read_all(path: Path) -> tuple[list[DurationRecord], list[RunHeader]]:
    """One pass over the sink, shape-aware: a ``kind: "run"`` line is a
    run-identity HEADER and is never coerced into a per-test duration row
    (it carries no "test"/"duration" at all) -- split explicitly here
    rather than relying on the accident of a missing "test" key falling
    through the per-row isinstance check.
    """
    records: list[DurationRecord] = []
    headers: list[RunHeader] = []
    for obj in _iter_raw(path):
        if obj.get("kind") == "run":
            run_id = obj.get("run_id")
            batch_id = obj.get("batch_id")
            if not isinstance(run_id, str) or not isinstance(batch_id, str):
                continue
            ts = obj.get("ts")
            head_commit = obj.get("head_commit")
            lane = obj.get("lane")
            headers.append(
                RunHeader(
                    run_id=run_id,
                    batch_id=batch_id,
                    ts=float(ts) if isinstance(ts, (int, float)) else 0.0,
                    head_commit=head_commit if isinstance(head_commit, str) else None,
                    lane=lane if isinstance(lane, str) else None,
                )
            )
            continue

        test = obj.get("test")
        duration = obj.get("duration")
        if not isinstance(test, str) or not isinstance(duration, (int, float)):
            continue
        ts = obj.get("ts")
        run_id = obj.get("run_id")
        batch_id = obj.get("batch_id")
        outcome = obj.get("outcome")
        records.append(
            DurationRecord(
                test=test,
                duration=float(duration),
                ts=float(ts) if isinstance(ts, (int, float)) else 0.0,
                run_id=run_id if isinstance(run_id, str) else None,
                batch_id=batch_id if isinstance(batch_id, str) else None,
                outcome=outcome if isinstance(outcome, str) else None,
            )
        )
    return records, headers


def read_records(path: Path) -> list[DurationRecord]:
    """Every per-test duration row (LEGACY + IDENTIFIED), header lines
    excluded. Byte-compatible with the pre-identity reader for the fields
    it always exposed (test/duration/ts); run_id/batch_id/outcome are
    ``None`` on legacy rows.
    """
    records, _headers = _read_all(path)
    return records


def read_headers(path: Path) -> list[RunHeader]:
    """Every run-identity HEADER line (``kind: "run"``)."""
    _records, headers = _read_all(path)
    return headers


def slowest(records: list[DurationRecord], n: int) -> list[DurationRecord]:
    """The ``n`` slowest records, each test's LATEST recorded duration only
    (a test run repeatedly across sessions should reflect its most recent
    timing, not stack duplicate stale entries)."""
    latest: dict[str, DurationRecord] = {}
    for rec in records:
        prior = latest.get(rec.test)
        if prior is None or rec.ts >= prior.ts:
            latest[rec.test] = rec
    ordered = sorted(latest.values(), key=lambda r: r.duration, reverse=True)
    return ordered[:n]


def over_threshold(
    records: list[DurationRecord], threshold: float = DEFAULT_SLOW_THRESHOLD_S
) -> list[DurationRecord]:
    """Every test (latest duration) at or above ``threshold`` seconds --
    the run-summary defect-candidate list."""
    latest: dict[str, DurationRecord] = {}
    for rec in records:
        prior = latest.get(rec.test)
        if prior is None or rec.ts >= prior.ts:
            latest[rec.test] = rec
    hits = [r for r in latest.values() if r.duration >= threshold]
    return sorted(hits, key=lambda r: r.duration, reverse=True)


def summarize_runs(
    records: list[DurationRecord], headers: list[RunHeader]
) -> list[RunSummary]:
    """Group IDENTIFIED rows by run_id and join each run's header lines.

    A run whose headers are missing (identified rows but no header line --
    e.g. headers aged out of the retention window, or the writer died before
    flushing one) still produces a summary: head_commit is ``None`` (never
    invented) and the batch count falls back to the distinct batch_ids
    carried on the rows themselves.
    """
    by_run: dict[str, list[DurationRecord]] = {}
    for rec in records:
        if rec.run_id is None:
            continue
        by_run.setdefault(rec.run_id, []).append(rec)

    headers_by_run: dict[str, list[RunHeader]] = {}
    for hdr in headers:
        headers_by_run.setdefault(hdr.run_id, []).append(hdr)

    summaries: list[RunSummary] = []
    for run_id, rows in by_run.items():
        hdrs = headers_by_run.get(run_id, [])
        head_commit = next((h.head_commit for h in hdrs if h.head_commit), None)
        lanes = sorted({h.lane for h in hdrs if h.lane})
        batch_ids = {h.batch_id for h in hdrs}
        if not batch_ids:
            batch_ids = {r.batch_id for r in rows if r.batch_id}
        tss = [r.ts for r in rows]
        passed = sum(1 for r in rows if r.outcome == "passed")
        failed = sum(1 for r in rows if r.outcome == "failed")
        summaries.append(
            RunSummary(
                run_id=run_id,
                head_commit=head_commit,
                batches=len(batch_ids),
                lanes=lanes,
                tests=len(rows),
                passed=passed,
                failed=failed,
                other_outcomes=len(rows) - passed - failed,
                wall_seconds=(max(tss) - min(tss)) if tss else 0.0,
                duration_seconds=sum(r.duration for r in rows),
                min_ts=min(tss) if tss else 0.0,
            )
        )
    summaries.sort(key=lambda s: s.min_ts, reverse=True)
    return summaries


def _build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        description="Report over reports/test-durations.ndjson (Phase 3.3)."
    )
    parser.add_argument(
        "--path", type=Path, default=DEFAULT_PATH, help="NDJSON durations file."
    )
    parser.add_argument(
        "--slow", type=int, default=20, metavar="N", help="Show the N slowest tests."
    )
    parser.add_argument(
        "--threshold",
        type=float,
        default=DEFAULT_SLOW_THRESHOLD_S,
        help="Also list every test at/above this many seconds (defect candidates).",
    )
    parser.add_argument(
        "--run",
        type=str,
        default=None,
        metavar="RUN_ID",
        help="Scope --slow/--threshold to one run's rows only.",
    )
    subparsers = parser.add_subparsers(dest="command")
    runs_parser = subparsers.add_parser(
        "runs", help="List identified runs, newest first, joined against headers."
    )
    runs_parser.add_argument(
        "--path", type=Path, default=DEFAULT_PATH, help="NDJSON durations file."
    )
    runs_parser.add_argument(
        "--last",
        type=int,
        default=DEFAULT_RUNS_LAST,
        metavar="N",
        help="Show the N most recent runs.",
    )
    return parser


def _run_report(args: argparse.Namespace) -> int:
    records, headers = _read_all(args.path)

    if args.run is not None:
        known_run_ids = {r.run_id for r in records if r.run_id is not None}
        known_run_ids |= {h.run_id for h in headers}
        if args.run not in known_run_ids:
            print(f"unknown run id: {args.run}", file=sys.stderr)
            return 1
        records = [r for r in records if r.run_id == args.run]

    if not records:
        print(f"no duration records found at {args.path}")
        return 0

    print(
        f"{len(records)} duration record(s) across {len({r.test for r in records})} test(s)"
    )
    print(f"\nslowest {args.slow}:")
    for rec in slowest(records, args.slow):
        print(f"  {rec.duration:8.3f}s  {rec.test}")

    over = over_threshold(records, args.threshold)
    if over:
        print(f"\ndefect candidates (>= {args.threshold:.0f}s):")
        for rec in over:
            print(f"  {rec.duration:8.3f}s  {rec.test}")

    return 0


def _run_runs(args: argparse.Namespace) -> int:
    records, headers = _read_all(args.path)
    summaries = summarize_runs(records, headers)
    legacy_count = sum(1 for r in records if r.run_id is None)

    if not summaries and not legacy_count:
        print(f"no duration records found at {args.path}")
        return 0

    if not summaries:
        print("no identified runs found")
    else:
        print(
            f"{'run_id':<20} {'head_commit':<14} {'batches':>7} "
            f"{'tests':>6} {'passed':>6} {'failed':>6} {'other':>6} "
            f"{'wall_s':>9} {'sum_s':>9}  lanes"
        )
        for summary in summaries[: args.last]:
            commit = (
                summary.head_commit[:HEAD_COMMIT_DISPLAY_LEN]
                if summary.head_commit
                else "(absent)"
            )
            lanes = ",".join(summary.lanes) if summary.lanes else "(none)"
            print(
                f"{summary.run_id:<20} {commit:<14} {summary.batches:>7} "
                f"{summary.tests:>6} {summary.passed:>6} {summary.failed:>6} "
                f"{summary.other_outcomes:>6} {summary.wall_seconds:>9.1f} "
                f"{summary.duration_seconds:>9.1f}  {lanes}"
            )

    if legacy_count:
        legacy_ts = max((r.ts for r in records if r.run_id is None), default=0.0)
        legacy_date = datetime.fromtimestamp(legacy_ts).strftime("%Y-%m-%d")
        print(
            f"\n{legacy_count} legacy records without run identity "
            f"(before {legacy_date})"
        )

    return 0


def main(argv: list[str] | None = None) -> int:
    parser = _build_parser()
    args = parser.parse_args(argv)
    if getattr(args, "command", None) == "runs":
        return _run_runs(args)
    return _run_report(args)


if __name__ == "__main__":
    sys.exit(main())