diff --git a/heart/checks/unit_test_timing.py b/heart/checks/unit_test_timing.py new file mode 100644 index 0000000..db18381 --- /dev/null +++ b/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/.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)) diff --git a/heart/dashboard.py b/heart/dashboard.py index b62ed18..c6f60c2 100644 --- a/heart/dashboard.py +++ b/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")) diff --git a/heart/state.py b/heart/state.py index d2650c1..20dbef6 100644 --- a/heart/state.py +++ b/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", {}), diff --git a/tests/test_unit_test_timing.py b/tests/test_unit_test_timing.py new file mode 100644 index 0000000..12fc017 --- /dev/null +++ b/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, + }