feat: unit_test_timing leg — off-tick slow-test regression tracking (#63)

heart/checks/unit_test_timing.py

@@ -0,0 +1,217 @@
+"""heart/checks/unit_test_timing.py — track library unit-test durations, flag
+slow-test regressions.
+
+OFF-TICK by design. Running the library suites costs minutes (test_autofit is
+~1500 tests), which does not fit the <30s watch-loop budget (``docs/internals.md``
+rule 3). Run on a slower cadence — a daily/weekly cron or on demand — and do
+**not** wire it into ``heart/tick.sh``:
+
+    python -m heart.checks.unit_test_timing
+    HYGIENE_PYTHON=~/venv/PyAuto/bin/python python -m heart.checks.unit_test_timing
+
+The suites run in a SUBPROCESS (``pytest --durations``) — Heart's own process
+never imports the test/science stack (rule 4) — and the runner is injected so
+the stdlib-only tests feed fixtures and never run pytest-in-pytest. Advisory:
+surfaced on the board but never a release gate (test speed is not release-blocking).
+
+State: ~/.pyauto-heart/timings-unit/<slug>.json — rolling window per tracked test.
+Output: ~/.pyauto-heart/unit_test_timing.json — the latest regression summary.
+Classification (mirrors script_timing): ratio = latest / median(prior window);
+green <= yellow_factor (1.5), yellow <= red_factor (3.0), else red.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import re
+import statistics
+import subprocess
+import sys
+from pathlib import Path
+from typing import Any, Callable
+
+import yaml
+
+HEART_STATE_DIR = Path(
+    os.environ.get("HEART_STATE_DIR")
+    or Path.home() / ".pyauto-heart"
+)
+HEART_UNIT_TIMINGS_DIR = HEART_STATE_DIR / "timings-unit"
+HEART_HOME = Path(__file__).resolve().parents[2]
+CONFIG_PATH = HEART_HOME / "config" / "repos.yaml"
+
+# Each library's suite is discovered via its own pytest config (testpaths), so we
+# just invoke pytest from the repo root. Import-name maps to repo dir name.
+DEFAULT_REPOS = ["PyAutoConf", "PyAutoFit", "PyAutoArray", "PyAutoGalaxy", "PyAutoLens"]
+
+# A runner maps a repo checkout dir to {test_nodeid: call_seconds} for the
+# slowest tests, or None when the suite could not be run. Injected in tests.
+Runner = Callable[[Path], "dict[str, float] | None"]
+
+# pytest's "slowest durations" report lines: "1.23s call path::test_x"
+_DURATION_RE = re.compile(r"^\s*([0-9.]+)s\s+call\s+(\S+)\s*$")
+
+
+def load_thresholds() -> tuple[float, float, int, int]:
+    """Return (yellow_factor, red_factor, baseline_window, top_n) from config."""
+    yellow, red, window, top_n = 1.5, 3.0, 7, 15
+    if CONFIG_PATH.is_file():
+        cfg = yaml.safe_load(CONFIG_PATH.read_text()) or {}
+        t = cfg.get("thresholds", {}).get("unit_test_timing", {})
+        yellow = float(t.get("yellow_factor", yellow))
+        red = float(t.get("red_factor", red))
+        window = int(t.get("baseline_window", window))
+        top_n = int(t.get("top_n", top_n))
+    return yellow, red, window, top_n
+
+
+def parse_durations(stdout: str) -> dict[str, float]:
+    """Extract {nodeid: call_seconds} from a pytest --durations report."""
+    out: dict[str, float] = {}
+    for line in stdout.splitlines():
+        m = _DURATION_RE.match(line)
+        if m:
+            out[m.group(2)] = float(m.group(1))
+    return out
+
+
+def default_runner(python: str, timeout: float, top_n: int) -> Runner:
+    """Run ``pytest --durations`` per repo in a subprocess; parse the slowest."""
+    env = {
+        **os.environ,
+        "NUMBA_CACHE_DIR": os.environ.get("NUMBA_CACHE_DIR", "/tmp/numba_cache"),
+        "MPLCONFIGDIR": os.environ.get("MPLCONFIGDIR", "/tmp/matplotlib"),
+    }
+
+    def run_repo(repo_dir: Path) -> dict[str, float] | None:
+        if not repo_dir.is_dir():
+            return None
+        cmd = [python, "-m", "pytest", "--durations", str(top_n),
+               "-q", "-p", "no:cacheprovider"]
+        try:
+            proc = subprocess.run(
+                cmd, cwd=str(repo_dir), capture_output=True, text=True,
+                timeout=timeout, env=env,
+            )
+        except (subprocess.TimeoutExpired, FileNotFoundError):
+            return None
+        # A non-zero exit (test failures) still prints the durations report; parse
+        # regardless. Only a total absence of parseable lines means "no data".
+        parsed = parse_durations(proc.stdout)
+        return parsed or None
+
+    return run_repo
+
+
+def _slug(repo: str, nodeid: str) -> str:
+    safe = re.sub(r"[^A-Za-z0-9]+", "_", f"{repo}__{nodeid}").strip("_")
+    return f"{safe}.json"
+
+
+def update_history(slug: str, duration: float, window: int) -> list[float]:
+    HEART_UNIT_TIMINGS_DIR.mkdir(parents=True, exist_ok=True)
+    history_path = HEART_UNIT_TIMINGS_DIR / slug
+    if history_path.is_file():
+        try:
+            history = json.loads(history_path.read_text())
+        except json.JSONDecodeError:
+            history = []
+    else:
+        history = []
+    history.append(duration)
+    history = history[-window:]
+    history_path.write_text(json.dumps(history))
+    return history
+
+
+def classify(ratio: float, yellow: float, red: float) -> str:
+    if ratio > red:
+        return "red"
+    if ratio > yellow:
+        return "yellow"
+    return "green"
+
+
+def run(runner: Runner | None = None, repos: list[str] | None = None) -> dict[str, Any]:
+    """Run each repo's suite, update rolling per-test baselines, classify."""
+    yellow_factor, red_factor, window, top_n = load_thresholds()
+    repos = repos if repos is not None else DEFAULT_REPOS
+    python = os.environ.get("HYGIENE_PYTHON", "python3")
+    timeout = float(os.environ.get("HEART_UNIT_TEST_TIMEOUT", "1800"))
+    root = Path(os.environ.get("PYAUTO_ROOT", str(Path.home() / "Code" / "PyAutoLabs")))
+    runner = runner or default_runner(python, timeout, top_n)
+
+    findings: dict[str, list[dict[str, Any]]] = {"red": [], "yellow": [], "green": []}
+    repos_measured = 0
+    unavailable: list[str] = []
+    new_tests = 0
+
+    for repo in repos:
+        durations = runner(root / repo)
+        if durations is None:
+            unavailable.append(repo)
+            continue
+        repos_measured += 1
+        for nodeid, duration in durations.items():
+            history = update_history(_slug(repo, nodeid), duration, window)
+            if len(history) <= 1:
+                new_tests += 1
+                continue
+            baseline = statistics.median(history[:-1])
+            if baseline <= 0:
+                continue
+            ratio = duration / baseline
+            findings[classify(ratio, yellow_factor, red_factor)].append({
+                "repo": repo,
+                "test": nodeid,
+                "latest_seconds": round(duration, 3),
+                "baseline_seconds": round(baseline, 3),
+                "ratio": round(ratio, 2),
+                "samples": len(history) - 1,
+            })
+
+    summary = {
+        "python": python,
+        "repos_measured": repos_measured,
+        "repos_unavailable": unavailable,
+        "new_tests_no_baseline": new_tests,
+        "red_count": len(findings["red"]),
+        "yellow_count": len(findings["yellow"]),
+        "green_count": len(findings["green"]),
+        "red": sorted(findings["red"], key=lambda x: -x["ratio"]),
+        "yellow": sorted(findings["yellow"], key=lambda x: -x["ratio"]),
+    }
+
+    sys.path.insert(0, str(HEART_HOME))
+    from heart import state
+
+    state.atomic_write_json(HEART_STATE_DIR / "unit_test_timing.json", summary)
+    return summary
+
+
+def main(argv: list[str]) -> int:
+    summary = run()
+
+    from heart_color import c_ok, c_warn, c_fail, c_info, c_meta, glyph_ok, glyph_warn, glyph_fail
+
+    if summary["red_count"]:
+        glyph = glyph_fail()
+        label = c_fail(f"{summary['red_count']} red") + " " + c_warn(f"{summary['yellow_count']} yellow")
+    elif summary["yellow_count"]:
+        glyph = glyph_warn()
+        label = c_warn(f"{summary['yellow_count']} slower than baseline")
+    elif not summary["repos_measured"]:
+        glyph = glyph_warn()
+        label = c_warn("no suite runnable (set HYGIENE_PYTHON / PYAUTO_ROOT)")
+    else:
+        glyph = glyph_ok()
+        label = c_ok(f"{summary['green_count']} tracked tests within baseline")
+    extra = c_meta(f" ({summary['new_tests_no_baseline']} new, no baseline)")
+    print(f"{glyph} {c_info('unit_test_timing')} {label}{extra}")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+    sys.exit(main(sys.argv))

heart/dashboard.py

@@ -57,6 +57,7 @@
     "worktree_drift",
     "script_timing",
     "import_time",
+    "unit_test_timing",
     "profiling_drift",
     "test_run",
     "version_skew",
@@ -398,6 +399,30 @@ def build_board(
             ]
             sections.append(Section("import_time", "Import timing", st, summary, details))
 
+    # Unit-test timing (advisory; off-tick) -----------------------------------
+    if "unit_test_timing" in unobserved:
+        sections.append(_unobs_section("unit_test_timing", "Unit-test timing"))
+    else:
+        ut = snapshot.get("unit_test_timing") or {}
+        if ut:
+            r = _as_int(ut.get("red_count"))
+            y = _as_int(ut.get("yellow_count"))
+            g = _as_int(ut.get("green_count"))
+            if r:
+                st, summary = FAIL, f"{r} test regressions (>3× baseline), {y} slow (>1.5×)"
+            elif y:
+                st, summary = WARN, f"{y} tests >1.5× baseline, {g} within"
+            elif not _as_int(ut.get("repos_measured")):
+                st, summary = WARN, "no suite runnable (set HYGIENE_PYTHON / PYAUTO_ROOT)"
+            else:
+                st, summary = OK, f"{g} tracked tests within baseline"
+            details = [
+                f"✗ {e['repo']}  {e['test'].split('::')[-1]}  "
+                f"{e['latest_seconds']:.2f}s vs {e['baseline_seconds']:.2f}s ({e['ratio']}×)"
+                for e in (ut.get("red") or [])[:5]
+            ]
+            sections.append(Section("unit_test_timing", "Unit-test timing", st, summary, details))
+
     # Profiling pinned-value drift -------------------------------------------
     if "profiling_drift" in unobserved:
         sections.append(_unobs_section("profiling_drift", "Profiling drift"))

heart/state.py

@@ -75,6 +75,7 @@ def aggregate() -> dict[str, Any]:
         "worktree_drift": _read_json_or_default(HEART_STATE_DIR / "worktree_drift.json", {}),
         "script_timing": _read_json_or_default(HEART_STATE_DIR / "script_timing.json", {}),
         "import_time": _read_json_or_default(HEART_STATE_DIR / "import_time.json", {}),
+        "unit_test_timing": _read_json_or_default(HEART_STATE_DIR / "unit_test_timing.json", {}),
         "profiling_drift": _read_json_or_default(HEART_STATE_DIR / "profiling_drift.json", {}),
         "test_run": _read_json_or_default(HEART_STATE_DIR / "test_run.json", {}),
         "version_skew": _read_json_or_default(HEART_STATE_DIR / "version_skew.json", {}),

tests/test_unit_test_timing.py

@@ -0,0 +1,86 @@
+"""tests/test_unit_test_timing.py — slow-test regression classifier + parser.
+
+Stdlib-only (internals rule 4): the pytest runner is injected, so the suite
+never runs pytest-in-pytest or imports the science stack.
+"""
+
+from __future__ import annotations
+
+import importlib
+import json
+
+import pytest
+
+
+@pytest.fixture
+def tmp_state(tmp_path, monkeypatch):
+    monkeypatch.setenv("HEART_STATE_DIR", str(tmp_path))
+    import heart.state as state_mod
+    importlib.reload(state_mod)
+    import heart.checks.unit_test_timing as ut
+    importlib.reload(ut)
+    ut.HEART_STATE_DIR = tmp_path
+    ut.HEART_UNIT_TIMINGS_DIR = tmp_path / "timings-unit"
+    ut.HEART_UNIT_TIMINGS_DIR.mkdir(parents=True, exist_ok=True)
+    return tmp_path, ut
+
+
+def _runner(mapping):
+    """A runner that returns `mapping` (or None) for any repo."""
+    return lambda repo_dir: mapping
+
+
+def test_first_observation_has_no_baseline(tmp_state):
+    _, ut = tmp_state
+    s = ut.run(runner=_runner({"t.py::test_a": 5.0}), repos=["PyAutoFit"])
+    assert s["new_tests_no_baseline"] == 1
+    assert s["red_count"] == 0 and s["yellow_count"] == 0
+    assert s["repos_measured"] == 1
+
+
+def test_within_baseline_green(tmp_state):
+    _, ut = tmp_state
+    ut.run(runner=_runner({"t.py::test_a": 5.0}), repos=["PyAutoFit"])
+    s = ut.run(runner=_runner({"t.py::test_a": 5.0}), repos=["PyAutoFit"])
+    assert s["green_count"] == 1 and s["red_count"] == 0 and s["yellow_count"] == 0
+
+
+def test_yellow_and_red_thresholds(tmp_state):
+    _, ut = tmp_state
+    ut.run(runner=_runner({"t.py::test_a": 5.0}), repos=["PyAutoFit"])
+    s = ut.run(runner=_runner({"t.py::test_a": 10.0}), repos=["PyAutoFit"])  # 2.0×
+    assert s["yellow_count"] == 1 and s["red_count"] == 0
+    ut.run(runner=_runner({"t.py::test_b": 5.0}), repos=["PyAutoFit"])
+    s = ut.run(runner=_runner({"t.py::test_b": 20.0}), repos=["PyAutoFit"])  # 4.0×
+    assert s["red_count"] == 1
+
+
+def test_repo_unavailable(tmp_state):
+    _, ut = tmp_state
+    s = ut.run(runner=_runner(None), repos=["PyAutoFit"])
+    assert s["repos_measured"] == 0 and s["repos_unavailable"] == ["PyAutoFit"]
+
+
+def test_rolling_window_caps_history(tmp_state):
+    _, ut = tmp_state
+    for d in [1.0, 1.1, 1.2, 1.3, 1.4, 1.5, 1.6, 1.7, 1.8, 1.9]:
+        ut.run(runner=_runner({"t.py::test_a": d}), repos=["PyAutoFit"])
+    files = list(ut.HEART_UNIT_TIMINGS_DIR.glob("*.json"))
+    assert len(files) == 1
+    assert len(json.loads(files[0].read_text())) == 7
+
+
+def test_parse_durations_reads_call_lines_only(tmp_state):
+    _, ut = tmp_state
+    stdout = (
+        "===== slowest 3 durations =====\n"
+        "12.34s call test_autofit/test_x.py::test_slow\n"
+        "1.20s setup test_autofit/test_x.py::test_slow\n"
+        "0.98s call test_autofit/test_y.py::test_other\n"
+        "\n===== short test summary =====\n"
+    )
+    parsed = ut.parse_durations(stdout)
+    assert parsed == {
+        "test_autofit/test_x.py::test_slow": 12.34,
+        "test_autofit/test_y.py::test_other": 0.98,
+    }