{{ message }}
-
Notifications
You must be signed in to change notification settings - Fork 49.8k
Expand file tree
/
Copy pathrun_tests_parallel.py
More file actions
executable file
·1262 lines (1127 loc) · 51.5 KB
/
Copy pathrun_tests_parallel.py
File metadata and controls
executable file
·1262 lines (1127 loc) · 51.5 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
#!/usr/bin/env python3
"""Per-file parallel test runner.
The minimum-viable replacement for pytest-xdist + a subprocess-isolation
plugin. Discovers test files under ``tests/`` (excluding integration/e2e
unless explicitly requested), then runs one ``python -m pytest <file>``
subprocess per file, with bounded parallelism (default: ``os.cpu_count()``).
Why per-file rather than per-test?
Per-test spawn overhead (~250ms × 17k tests = 70min CPU minimum)
swamped the actual work. Per-file spawn (~250ms × ~850 files = ~3.5min)
fits in the budget while still giving every file a fresh Python
interpreter — the only isolation boundary that actually matters
(cross-file module-level state leakage was the original flake source;
intra-file state is the test author's responsibility).
Why drop xdist entirely?
xdist's persistent workers accumulate state across files, which is
exactly the leakage we wanted to fix. xdist also adds complexity
(loadfile vs loadscope, --max-worker-restart, internal control plane)
that we don't need when the unit of work is "run pytest on one file".
A subprocess.Popen pool gated by a semaphore is ~60 lines and does
the job.
Usage:
python scripts/run_tests_parallel.py [pytest_args...]
Common pytest args pass through to each per-file pytest invocation
(e.g. ``-q``, ``-v``, ``-x``, ``--tb=long``, ``-k 'pattern'``, ``--lf``)
with no special separator — a bare ``-q`` "just works". Anything after
a literal ``--`` is also passed through, and stacks with bare flags.
Environment:
HERMES_TEST_WORKERS Override worker count (default: os.cpu_count())
HERMES_TEST_PATHS Override discovery roots (colon-sep; on Windows
';' also works and drive letters are handled;
default: 'tests')
Exit code: 0 if every file's pytest exited 0; 1 otherwise.
"""
from __future__ import annotations
import argparse
import json
import os
import re
import shutil
import subprocess
import sys
import tempfile
import threading
import time
from concurrent.futures import ThreadPoolExecutor, Future
from pathlib import Path
from typing import Dict, List, Tuple
# Default test discovery roots.
_DEFAULT_ROOTS = ["tests"]
# Directories to skip during discovery — these suites require real
# external services (a model gateway, a docker daemon with a prebuilt
# image, etc.) and are run in their own dedicated CI jobs:
#
# tests/e2e/ — .github/workflows/tests.yml :: e2e job
# tests/integration/ — historical; legacy --ignore flags
# tests/docker/ — .github/workflows/docker.yml ::
# build-amd64 job (runs against the freshly-loaded
# nousresearch/hermes-agent:test image, via
# ``HERMES_TEST_IMAGE`` so the fixture skips
# rebuild). The full pytest-shard runner can't
# host these because the session-scoped
# ``built_image`` fixture would do a 3-7min
# ``docker build``,
# so the build is guaranteed to die in fixture
# setup. The dedicated job sidesteps both costs.
_SKIP_PARTS = {"integration", "e2e", "docker"}
# Per-file wall-clock cap. Override
# via --file-timeout or HERMES_TEST_FILE_TIMEOUT.
#
# Set to 300s (5 min) deliberately generous: the per-test subprocess
# isolation plugin spawns a fresh Python process per test, so a
# large-collection file pays N × (interpreter startup + import) of
# overhead before any test logic runs — and that overhead dilates under
# load on shared CI runners, producing false "no tests ran" timeouts on
# files that finish in ~100s on a quiet box. The Docker build matrix jobs
# take 7-10 min anyway, so this headroom costs nothing on total CI wall
# time while keeping a genuinely hung file bounded.
_DEFAULT_FILE_TIMEOUT_SECONDS = 300.0
# One-shot retry of failing test FILES. A file that exits non-zero is re-run
# once in a fresh subprocess; if the re-run passes, the file counts as passed
# but is loudly reported as FLAKY so it gets fixed rather than hidden.
# Deterministic failures fail both attempts — a real regression can never be
# laundered into green by this (it would have to flake in our favor twice in
# a row on the same runner, which is exactly the definition of a flake).
# Set to 0 to disable (env: HERMES_TEST_FILE_RETRIES).
_DEFAULT_FILE_RETRIES = 1
# Duration cache: maps relative file paths to last-observed subprocess
# wall-clock seconds. Used by ``--slice`` to distribute files across
# CI jobs by estimated total time, so no one job gets all the slow files.
_DURATIONS_FILE = "test_durations.json"
def _split_pathspec(value: str) -> List[str]:
"""Split a separator-joined path list (``--paths``/``--files``/
``HERMES_TEST_PATHS``) into individual paths.
POSIX: ``:``-separated, as documented.
Windows: ``;`` (``os.pathsep``) and ``:`` are both accepted as
separators, but a ``:`` that forms a drive letter (``C:\\...`` or
``C:/...``) stays glued to its path — a naive ``split(":")`` turns
``C:\\repo\\tests`` into ``['C', '\\repo\\tests']``, where the bogus
``C`` becomes a phantom discovery root and the rooted remainder only
resolves by accident of ``Path.__truediv__`` re-anchoring it onto
``repo_root``'s drive.
"""
if sys.platform != "win32":
return [p for p in value.split(":") if p.strip()]
parts: List[str] = []
for chunk in value.split(";"):
raw = chunk.split(":")
i = 0
while i < len(raw):
part = raw[i]
if (
len(part) == 1
and part.isalpha()
and i + 1 < len(raw)
and raw[i + 1][:1] in ("\\", "/")
):
part = f"{part}:{raw[i + 1]}"
i += 1
parts.append(part)
i += 1
return [p for p in parts if p.strip()]
# Host-OS gating (see the ``_OS_MARKS`` block in tests/conftest.py): tests
# marked for another host are collected and SKIPPED by the conftest hook —
# this runner never executes them, by construction. The summary calls that
# out explicitly so a local run isn't misread as covering macOS/Windows
# behaviour, and names the CI lane where those tests actually execute.
_OS_MARKERS = {
"linux_only": ("linux", "the main Linux CI lane"),
"macos_only": ("darwin", "the tests-os CI lane (macos-latest)"),
"windows_only": ("win32", "the tests-os CI lane (windows-latest)"),
}
def _off_host_marker_files(files: List[Path]) -> dict[str, int]:
"""Count discovered files referencing each marker for an OS we are not on.
Whole-word text match, same approach as scripts/ci/list_os_marked_tests.py:
over-counting a prose mention is harmless here (the note is informational);
what matters is never reporting 0 while gated tests exist.
"""
off_host = {
marker: re.compile(rf"\b{marker}\b")
for marker, (host_prefix, _) in _OS_MARKERS.items()
if not sys.platform.startswith(host_prefix)
}
counts = {marker: 0 for marker in off_host}
for path in files:
try:
text = path.read_text(encoding="utf-8", errors="replace")
except OSError:
continue
for marker, pattern in off_host.items():
if pattern.search(text):
counts[marker] += 1
return {marker: n for marker, n in counts.items() if n}
def _approximately_count_tests(
files: List[Path], repo_root: Path
) -> dict[Path, int]:
"""
Make a decent estimate at individual tests per file.
Running ``pytest --co -q`` is WAY too slow because it actually imports everything.
Returns a mapping ``{file_path: test_count}``. Files with zero
collected tests are omitted from the dict (not an error — e.g. the
file only defines fixtures / conftest helpers).
"""
results = {}
for path in files:
with open(path, "r", encoding="utf-8") as f:
contents = f.read()
results[path] = contents.count("def test_")
return results
def _discover_files(roots: List[Path]) -> List[Path]:
"""Return every ``test_*.py`` under the given roots (sorted).
Roots may be directories (recursed for ``test_*.py``) or explicit
``.py`` files (included as-is, even if they don't match the
``test_*`` prefix — caller knows what they want).
Exclude any file whose path contains a component in ``_SKIP_PARTS``,
UNLESS the user explicitly named it as a root (in which case the
user's intent overrides the skip filter). This makes
``scripts/run_tests.sh tests/docker/`` work locally the same way
``pytest tests/docker/`` does — the CI-level skip exists to keep
the sharded matrix from blowing up, not to block targeted runs.
"""
seen: set[Path] = set()
out: List[Path] = []
for root in roots:
if not root.exists():
continue
if root.is_file():
# Explicit file: include it as-is, skip the _SKIP_PARTS filter
# since the user named it directly.
real = root.resolve()
if real not in seen:
seen.add(real)
out.append(root)
continue
# If the explicit root itself sits inside a skipped dir (e.g.
# the user said ``tests/docker``), the user has overridden the
# skip for that subtree. Compute the set of skip-parts the user
# opted into, and only filter files whose path crosses a
# skip-part *outside* that opt-in.
root_skip_overrides = {
part for part in root.parts if part in _SKIP_PARTS
}
effective_skips = _SKIP_PARTS - root_skip_overrides
for path in root.rglob("test_*.py"):
if any(part in effective_skips for part in path.parts):
continue
real = path.resolve()
if real in seen:
continue
seen.add(real)
out.append(path)
return sorted(out)
def _kill_tree(proc: "subprocess.Popen", pgid: int | None = None) -> None:
"""Kill the pytest subprocess and every descendant it spawned.
A test run can spin up uvicorn servers, async runtimes, or other
long-running grandchildren that survive the pytest subprocess exit
if we don't kill the whole tree. ``subprocess.Popen.kill()`` only
targets the immediate child; grandchildren reparent to PID 1
(Linux) / get adopted by services.exe (Windows) and leak.
POSIX: the caller must pass ``pgid`` — the process group id captured
immediately after Popen (via ``os.getpgid(proc.pid)``). We can't
look it up here in the happy path because by the time we get
called the leader process has already been reaped and its pid is
gone from the kernel's process table, even though descendants in
the group are still alive. SIGKILL'ing the captured pgid takes out
everything in that group atomically.
Windows: ``taskkill /F /T /PID`` walks the recorded ppid chain and
terminates the whole tree, even when the root has already exited.
Why not psutil: psutil walks the parent-child tree, but in the
happy path the root has already been reaped so ``psutil.Process(pid)``
can't find it; grandchildren reparented to PID 1 are also
unreachable by tree walk at that point. The platform-native
primitives (process groups / taskkill) handle both cases correctly
without an extra abstraction layer.
"""
if proc.pid is None:
return
if sys.platform == "win32":
try:
subprocess.run(
["taskkill", "/F", "/T", "/PID", str(proc.pid)],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
timeout=10,
) # windows-footgun: ok
except (subprocess.TimeoutExpired, FileNotFoundError, OSError):
pass
else:
# POSIX: kill the captured pgid. Local-import signal so the
# SIGKILL attribute is never referenced on Windows.
if pgid is not None:
try:
import signal as _signal
os.killpg(pgid, _signal.SIGKILL) # windows-footgun: ok
except (ProcessLookupError, PermissionError, OSError):
pass
# Belt-and-suspenders: ensure subprocess.communicate() sees the exit.
try:
proc.kill()
except (ProcessLookupError, OSError):
pass
def _run_one_file(
file: Path,
pytest_args: List[str],
repo_root: Path,
file_timeout: float,
retries: int = 0,
) -> Tuple[Path, int, str, dict[str, int], float]:
"""Run ``python -m pytest <file> <pytest_args>`` in a fresh subprocess.
Returns (file, returncode, captured_combined_output, summary_counts, subprocess_wall_seconds).
``retries`` > 0 enables the one-shot flake retry: a non-zero exit is
re-run in a fresh subprocess; if the re-run passes, the file counts as
passed but the output is prefixed with a FLAKY banner and the file/output
are recorded in ``_FLAKY_RESULTS`` so the summary can call it out. A
deterministic failure fails every attempt, so real regressions cannot
be laundered green.
``summary_counts`` is the result of ``_parse_pytest_summary(output)`` —
pytest exit codes (https://docs.pytest.org/en/stable/reference/exit-codes.html):
0 = all tests passed
1 = some tests failed
2 = test execution interrupted
3 = internal error
4 = pytest CLI usage error
5 = no tests collected
We treat exit 5 as a pass: it just means every test in the file was
skipped or filtered by a marker (e.g. ``-m 'not integration'`` skips
files where every test is marked integration). That's intentional and
not a failure mode.
On per-file timeout (``file_timeout`` seconds) or any other exception
during ``communicate()``, we kill the whole process group / process
tree so grandchildren (uvicorn servers, async runtimes, etc.) do not
orphan onto PID 1. This outer timeout exists only to
bound a pathologically slow or hung file as a whole.
"""
file, rc, output, summary, subproc_wall = _run_one_file_once(
file, pytest_args, repo_root, file_timeout
)
attempt = 0
while rc != 0 and attempt < retries:
attempt += 1
first_output = output
file, rc, output, summary, subproc_wall2 = _run_one_file_once(
file, pytest_args, repo_root, file_timeout
)
subproc_wall += subproc_wall2
if rc == 0:
output = (
f"⚠ FLAKY: failed on attempt 1, passed on retry "
f"(attempt {attempt + 1}). Fix the flake — do not ignore this.\n"
f"--- first-attempt output ---\n{first_output}\n"
f"--- retry output ---\n{output}"
)
with _flaky_lock:
_FLAKY_RESULTS.append((file, output))
return file, rc, output, summary, subproc_wall
# Files that failed once and passed on retry, with both attempts' output.
# Keeping the traceback is load-bearing: a self-healed flake without its
# failing assertion is only a filename, which forces another expensive full
# run to rediscover the race.
_FLAKY_RESULTS: List[Tuple[Path, str]] = []
_flaky_lock = threading.Lock()
def _run_one_file_once(
file: Path,
pytest_args: List[str],
repo_root: Path,
file_timeout: float,
) -> Tuple[Path, int, str, dict[str, int], float]:
"""Single attempt of a per-file pytest subprocess (see _run_one_file)."""
cmd = [sys.executable, "-m", "pytest", str(file), *pytest_args]
# Give this subprocess its own pytest temp root.
#
# pytest builds its tmp_path root as <temproot>/pytest-of-<user>/. At the
# end of a session it walks that directory with cleanup_dead_symlinks().
# The walk lists the directory. Then it asks whether the `pytest-current`
# symlink resolves. Then it unlinks the symlink.
#
# Every file shared one root. A second process replaced that symlink
# between the question and the unlink. The first process then died with
# FileNotFoundError after all of its tests passed.
#
# The risk grows with the number of processes that finish together. At 8
# workers it never occurred. At 144 workers it occurs.
#
# One root for each subprocess removes the shared directory that the race
# needs. The parent deletes the root after the attempt.
env = os.environ.copy()
temproot = tempfile.mkdtemp(prefix="hermes-pytest-tmproot-")
env["PYTEST_DEBUG_TEMPROOT"] = temproot
subproc_start = time.monotonic()
# launch the pytest process
proc = subprocess.Popen(
cmd,
cwd=repo_root,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
text=True, encoding="utf-8", errors="replace",
env=env,
# POSIX: place the child at the head of its own process group so
# _kill_tree can SIGKILL the group atomically.
# Windows: this maps to CREATE_NEW_PROCESS_GROUP in CPython 3.12+;
# _kill_tree handles the Windows path via taskkill /F /T.
start_new_session=True,
)
# Capture the pgid NOW, before the leader can exit and be reaped. Once
# the leader is reaped, os.getpgid(proc.pid) raises ProcessLookupError
# even though grandchildren in that group are still alive — defeating
# the whole cleanup. None on Windows where the pgid concept doesn't apply.
pgid: int | None = None
if sys.platform != "win32":
try:
pgid = os.getpgid(proc.pid)
except (ProcessLookupError, PermissionError):
pgid = None
try:
output, _ = proc.communicate(timeout=file_timeout)
rc = proc.returncode
except subprocess.TimeoutExpired:
_kill_tree(proc, pgid=pgid)
try:
output, _ = proc.communicate(timeout=10)
except subprocess.TimeoutExpired:
output = "(file timeout exceeded; output unavailable)"
rc = 124 # de facto convention for "killed by timeout".
output = (
f"({file_timeout:.0f}s exceeded; "
f"process tree SIGKILL'd)\n{output}"
)
except BaseException:
# KeyboardInterrupt / runner crash — make sure no zombie
# grandchildren outlive us.
_kill_tree(proc, pgid=pgid)
raise
else:
# Happy path: pytest exited on its own. Kill the group anyway in
# case it left grandchildren behind; already-dead is a no-op.
_kill_tree(proc, pgid=pgid)
output += "\n"
finally:
# Delete the temp root for this attempt. Nothing reads it after the
# subprocess exits. More than 3000 of them fill the disk of the
# runner over one suite.
shutil.rmtree(temproot, ignore_errors=True)
if rc == 5:
# No tests collected in THIS file — legitimate per-file: a
# platform-gated or fully-marker-filtered file (e.g. a win32-only
# suite on Linux) collects nothing and must not fail the suite.
# Tolerated here; the RUN-level guard in main() still fails when
# NOTHING was collected across every file, so a broken invocation
# (venv without pytest, -k that matches nothing) can't report green.
rc = 0
summary = _parse_pytest_summary(output)
subproc_wall = time.monotonic() - subproc_start
return file, rc, output, summary, subproc_wall
def _parse_pytest_summary(output: str) -> dict[str, int]:
"""Extract per-file test pass/fail/skip counts from pytest output.
pytest prints a summary line like ``12 passed, 3 skipped, 1 failed in 2.1s``
as the last non-empty line before the short test summary. We scrape that
line for the individual counts so the progress display can show test-level
granularity instead of just file-level pass/fail.
Returns a dict with keys ``passed``, ``failed``, ``skipped``, ``errors``,
``xfailed``, ``xpassed`` (only keys found in the output are present).
"""
result: dict[str, int] = {}
# Walk backwards from the end — the summary line is always near the tail.
for line in reversed(output.splitlines()):
line = line.strip()
if not line:
continue
# Match "N passed", "N failed", "N skipped", "N errors", "N xfailed", "N xpassed"
for m in re.finditer(r"(\d+)\s+(passed|failed|skipped|errors|xfailed|xpassed)", line):
result[m.group(2)] = int(m.group(1))
# Also match "N error" (singular — pytest uses this sometimes).
for m in re.finditer(r"(\d+)\s+error\b", line):
result.setdefault("errors", result.get("errors", 0) + int(m.group(1)))
if result:
# Found the counts line — done.
break
# Stop at the short test summary header (if any) — everything above
# that is individual failure details, not the counts line.
if line.startswith("FAILED") or line.startswith("SHORT TEST SUMMARY"):
break
return result
def _format_file(file: Path, repo_root: Path) -> str:
"""Render a test-file path for display: strip the repo-root prefix
when possible so output reads ``tests/acp/test_auth.py`` instead of
``/home/runner/work/hermes-agent/hermes-agent/tests/acp/test_auth.py``.
Falls back to the absolute path for anything outside the repo root.
"""
try:
return str(file.resolve().relative_to(repo_root.resolve()))
except ValueError:
return str(file)
def _print_progress(
tests_done: int,
approx_total_tests: int,
file: Path,
rc: int,
dur: float,
repo_root: Path,
tests_passed: int,
tests_failed: int,
test_counts: dict[Path, int],
file_summary: dict[str, int] | None = None,
subproc_wall: float | None = None,
) -> None:
"""Single-line live progress.
When ``file_summary`` is provided (parsed from pytest output), the
per-file parenthetical shows individual test pass/fail counts instead
of just the total test count.
``subproc_wall`` is the actual subprocess wall-clock time (excluding
queue-wait). When available, the display shows both the subprocess
time and the queue-inclusive elapsed time.
"""
status = "✓" if rc == 0 else "✗"
pct = min((tests_done / approx_total_tests * 100), 100) if approx_total_tests else 0
# Digit width for left-side counter padding (derived from total file count).
fw = len(str(tests_passed + tests_failed))
# Build per-file test count string.
if file_summary:
parts = []
p = file_summary.get("passed", 0)
f = file_summary.get("failed", 0)
s = file_summary.get("skipped", 0)
e = file_summary.get("errors", 0)
if p:
parts.append(f"{p}✓")
if f:
parts.append(f"{f}✗")
if s:
parts.append(f"{s}s")
if e:
parts.append(f"{e}e")
# xfailed/xpassed are rare; include if present.
xf = file_summary.get("xfailed", 0)
xp = file_summary.get("xpassed", 0)
if xf:
parts.append(f"{xf}xf")
if xp:
parts.append(f"{xp}xp")
test_str = " ".join(parts) + ", " if parts else ""
else:
n_tests = test_counts.get(file, 0)
test_str = f"{n_tests} tests, " if n_tests else ""
# Show subprocess time when available; fall back to queue-inclusive dur.
if subproc_wall is not None:
time_str = f"{subproc_wall:.1f}s"
else:
time_str = f"{dur:.1f}s"
msg = (
f"[{pct:5.1f}% | {tests_done:>5}/~{approx_total_tests}"
f" | ✓{tests_passed:>{fw}} | ✗{tests_failed:>{fw}}] "
f"{status} {_format_file(file, repo_root)} ({test_str}{time_str})"
)
# Truncate to terminal width if available (no clobbering ANSI lines).
try:
cols = os.get_terminal_size().columns
if len(msg) > cols:
msg = msg[: cols - 1] + "…"
except OSError:
pass
print(msg, flush=True)
def _print_inline_failure(
file: Path, output: str, repo_root: Path, pytest_passthrough: List[str]
) -> None:
"""Print a compact failure summary immediately when a file fails.
Shows the tail of the pytest output (the failure section with stack
traces) and a ready-to-run repro command, so the developer doesn't
have to wait for the full run to finish before seeing what broke.
"""
rel = _format_file(file, repo_root)
# Build a repro command the developer can copy-paste.
passthrough_str = " ".join(pytest_passthrough) if pytest_passthrough else ""
repro = f"python -m pytest {rel}"
if passthrough_str:
repro += f" {passthrough_str}"
# Grab just the failure lines (last ~30 lines of pytest output —
# typically the FAILED summary + short test info).
lines = output.rstrip().splitlines()
tail = "\n".join(lines[-30:])
print(flush=True)
print(f" ╔╍ Failed: {rel} ╍╍╍╍╍╍╍╍╍╍╍╍╍╍╍╍╍╍╍╍╍╍╍╍╍╍╍╍╍╍╍╍╍╍╍╍╍╍╍╍╍╍╍╍╍╍╍╍╍", flush=True)
for line in tail.splitlines():
print(f" ║ {line}", flush=True)
print(" ║", flush=True)
print(f" ║ Repro: {repro}", flush=True)
print(" ╚╍╍╍╍╍╍╍╍╍╍╍╍╍╍╍╍╍╍╍╍╍╍╍╍╍╍╍╍╍╍╍╍╍╍╍╍╍╍╍╍╍╍╍╍╍╍╍╍╍╍╍╍╍╍╍╍╍╍╍╍╍╍╍╍╍╍╍╍╍╍╍", flush=True)
print(flush=True)
def _load_durations(repo_root: Path) -> dict[str, float]:
"""Read the duration cache from the repo root.
Returns a dict mapping relative file paths (e.g.
``tests/tools/test_code_execution.py``) to wall-clock seconds from
the last run. Missing or corrupt file → empty dict (safe fallback).
"""
path = repo_root / _DURATIONS_FILE
if not path.is_file():
return {}
try:
return json.loads(path.read_text(encoding="utf-8"))
except (json.JSONDecodeError, OSError) as e:
print("[ERROR] Failed to load json durations file! {e}")
return {}
def _save_durations(
file_times: List[Tuple[Path, float]],
repo_root: Path,
) -> None:
"""Write the duration cache so future ``--slice`` runs can use it.
Merges with any existing cache so entries from files not in the
current run (e.g. from a different slice) are preserved. Keys are
repo-relative paths so the cache is portable across checkouts
and CI runners.
"""
data: dict[str, float] = _load_durations(repo_root)
for f, t in file_times:
key = _format_file(f, repo_root)
data[key] = round(t, 3)
path = repo_root / _DURATIONS_FILE
path.write_text(json.dumps(data, indent=2, sort_keys=True) + "\n", encoding="utf-8")
def _compute_lpt_slices(
files: List[Path],
slice_count: int,
durations: dict[str, float],
repo_root: Path,
) -> List[List[Path]]:
"""Distribute files across N slices using LPT (Longest Processing Time first).
Sorts files by estimated duration descending, then greedily assigns each
file to the slice with the smallest accumulated time so far. This
minimizes the makespan (max slice duration) and keeps CI jobs balanced.
Files with no cached duration get a default estimate of 2.0s (roughly
the P50 from profiling). This means first-time runs (no cache) still
get reasonable distribution, and new files don't all land in one slice.
Returns a list of N file-lists, one per slice (0-indexed).
"""
if slice_count < 2:
return [files]
default_dur = 2.0
file_durs: List[Tuple[Path, float]] = []
for f in files:
rel = _format_file(f, repo_root)
dur = durations.get(rel, default_dur)
file_durs.append((f, dur))
# Sort longest first (LPT).
file_durs.sort(key=lambda x: x[1], reverse=True)
# Greedy assignment: for each file, add it to the slice with the
# smallest current total.
bucket_files: List[List[Path]] = [[] for _ in range(slice_count)]
bucket_totals: List[float] = [0.0] * slice_count
for f, dur in file_durs:
min_idx = min(range(slice_count), key=lambda i: bucket_totals[i])
bucket_files[min_idx].append(f)
bucket_totals[min_idx] += dur
return bucket_files
def _slice_files(
files: List[Path],
slice_index: int,
slice_count: int,
durations: dict[str, float],
repo_root: Path,
) -> List[Path]:
"""Return the subset of *files* belonging to slice *slice_index*.
Uses :func:`_compute_lpt_slices` for LPT distribution.
``slice_index`` is 1-indexed (1..slice_count) for ergonomics —
``--slice 1/4`` reads more naturally than ``--slice 0/4``.
"""
if slice_count < 2:
return files
if not (1 <= slice_index <= slice_count):
print(
f"error: --slice index must be 1..{slice_count}, got {slice_index}",
file=sys.stderr,
)
sys.exit(2)
bucket_files = _compute_lpt_slices(files, slice_count, durations, repo_root)
target = bucket_files[slice_index - 1]
target_dur = sum(
durations.get(_format_file(f, repo_root), 2.0) for f in target
)
total_dur = sum(
durations.get(_format_file(f, repo_root), 2.0)
for bucket in bucket_files
for f in bucket
)
print(
f"Slice {slice_index}/{slice_count}: {len(target)} files "
f"(~{target_dur:.0f}s estimated of {total_dur:.0f}s total)",
flush=True,
)
return target
def _make_stdio_glyph_safe() -> None:
"""Keep status glyphs from killing the runner on narrow console encodings.
On native Windows, piped or legacy-console stdio defaults to a locale
codec (usually cp1252) that cannot encode the ✓/✗ progress glyphs — the
first per-file status line then dies with UnicodeEncodeError before a
single test result is reported. Declare the runner's own output UTF-8
(what CI and every modern terminal already are), with errors="replace"
as the can't-crash backstop; where the encoding can't be changed, fall
back to errors="replace" alone so glyphs degrade to "?" instead of
killing the run. On already-UTF-8 stdio this is a no-op.
"""
for stream in (sys.stdout, sys.stderr):
reconfigure = getattr(stream, "reconfigure", None)
if reconfigure is None:
continue
try:
reconfigure(encoding="utf-8", errors="replace")
except Exception:
try:
reconfigure(errors="replace")
except Exception:
pass
def main() -> int:
_make_stdio_glyph_safe()
parser = argparse.ArgumentParser(
description=__doc__,
formatter_class=argparse.RawDescriptionHelpFormatter,
)
parser.add_argument(
"-j",
"--jobs",
type=int,
default=int(os.environ.get("HERMES_TEST_WORKERS") or (os.cpu_count() or 4) * 2),
help="Parallel worker count (default: $HERMES_TEST_WORKERS or cpu_count*2)",
)
parser.add_argument(
"--paths",
default=os.environ.get("HERMES_TEST_PATHS", ":".join(_DEFAULT_ROOTS)),
help=(
"Colon-separated discovery roots (default: 'tests'). On "
"Windows, ';' also separates and drive letters (C:\\...) are "
"kept intact."
),
)
parser.add_argument(
"--include-integration",
action="store_true",
help="Don't skip integration/ e2e/ during discovery",
)
parser.add_argument(
"--file-timeout",
type=float,
default=float(
os.environ.get("HERMES_TEST_FILE_TIMEOUT", _DEFAULT_FILE_TIMEOUT_SECONDS)
),
help=(
"Per-file wall-clock cap in seconds. On timeout, the pytest "
"subprocess and its full process tree are SIGKILL'd. "
f"Default: {_DEFAULT_FILE_TIMEOUT_SECONDS}s ({round(_DEFAULT_FILE_TIMEOUT_SECONDS/60)} min), env: HERMES_TEST_FILE_TIMEOUT."
),
)
parser.add_argument(
"--file-retries",
type=int,
default=int(
os.environ.get("HERMES_TEST_FILE_RETRIES", _DEFAULT_FILE_RETRIES)
),
help=(
"Re-run a failing test FILE this many times in a fresh subprocess "
"before declaring it failed. A pass-on-retry counts as passed but "
"is reported as FLAKY in the summary. 0 disables. "
f"Default: {_DEFAULT_FILE_RETRIES}, env: HERMES_TEST_FILE_RETRIES."
),
)
parser.add_argument(
"--slice",
metavar="I/N",
help=(
"Run only slice I of N (e.g. --slice 1/4). "
"Files are distributed across slices using cached durations "
"so each slice takes roughly equal wall time. "
"Without a duration cache, files are distributed by count. "
"Env: HERMES_TEST_SLICE (format: I/N)."
),
)
parser.add_argument(
"--generate-slices",
metavar="N",
type=int,
help=(
"Discover test files, distribute them across N slices using "
"LPT on cached durations, and print a JSON matrix to stdout "
"then exit (no tests run). The JSON has the shape "
"'{\"slices\": [{\"index\": 1, \"files\": [\"tests/foo.py\", ...]}, ...]}' "
"so the CI generate job can feed it directly into a matrix."
),
)
parser.add_argument(
"--files",
metavar="LIST",
help=(
"Explicit colon-separated list of test files to run (on "
"Windows, ';' also separates and drive letters are kept "
"intact). Bypasses discovery entirely — used by CI matrix "
"jobs that receive their file list from the generate job."
),
)
parser.add_argument(
"paths_positional",
nargs="*",
metavar="PATH",
help=(
"Restrict discovery to these paths (directories or .py files). "
"Mutually exclusive with --paths. Anything after a literal '--' "
"separator is passed through to each per-file pytest invocation."
),
)
# Split argv into "our flags + positional paths" vs "pytest passthrough".
#
# Two ways to pass args through to the per-file pytest invocation:
# 1. Explicit ``--`` separator: everything after it goes to pytest.
# 2. Bare pytest flags anywhere before ``--``: any token starting with
# ``-`` that isn't one of OUR options is routed to pytest, so a bare
# ``-q`` / ``-v`` / ``-x`` / ``--tb=long`` / ``-k expr`` "just works"
# without the developer remembering the ``--``. This matches the
# docstring's promise and pytest muscle-memory.
#
# The subtlety bare-flag routing must handle: value-taking pytest flags
# given in space-separated form (``-k expr``, ``-m mark``, ``-p plugin``,
# ``-o name=val``). Naively, ``expr`` would look like a positional path and
# clobber discovery. We peel the following token along with such flags so
# it never reaches our positional ``paths``. ``=``-joined forms
# (``-k=expr``, ``--tb=long``) are self-contained and need no lookahead.
OUR_FLAGS = {
"-j", "--jobs", "--paths", "--include-integration",
"--file-timeout", "--file-retries", "--slice", "--generate-slices", "--files",
}
# pytest short flags that consume the NEXT token as their value.
PYTEST_VALUE_FLAGS = {"-k", "-m", "-p", "-o", "-c", "-r", "-W"}
def _is_our_flag(tok: str) -> bool:
# Match exact (``-j``, ``--paths``), ``=``-joined (``--paths=x``),
# and attached short-value (``-j4``) forms of our own options.
if tok in OUR_FLAGS:
return True
head = tok.split("=", 1)[0]
if head in OUR_FLAGS:
return True
# Attached short value, e.g. ``-j4`` → ``-j``.
if len(tok) > 2 and tok[:2] in OUR_FLAGS and not tok[1] == "-":
return True
return False
argv = sys.argv[1:]
if "--" in argv:
sep = argv.index("--")
before, explicit_passthrough = argv[:sep], argv[sep + 1 :]
else:
before, explicit_passthrough = argv, []
our_args: List[str] = []
bare_passthrough: List[str] = []
i = 0
while i < len(before):
tok = before[i]
if tok.startswith("-") and not _is_our_flag(tok):
bare_passthrough.append(tok)
# Pull the value token for space-separated value flags.
if tok in PYTEST_VALUE_FLAGS and i + 1 < len(before):
bare_passthrough.append(before[i + 1])
i += 2
continue
else:
our_args.append(tok)
i += 1
args = parser.parse_args(our_args)
# ── Node-id selectors → file + ``-k`` filter ────────────────────────────
# This runner is FILE-granular: it spawns one ``pytest <file>`` per test
# file. A pytest node id (``tests/foo.py::TestBar::test_baz``) is not an
# existing path, so discovery silently dropped it and the run exited with
# "No test files to run" — the selector looked accepted but nothing ran.
# Translate instead: run the FILE and narrow with ``-k`` on the last
# segment, which is what the caller meant.
node_id_selectors: List[Tuple[str, str]] = []
if args.paths_positional:
translated: List[str] = []
for raw in args.paths_positional:
if "::" not in raw:
translated.append(raw)
continue
file_part, _, selector = raw.partition("::")
leaf = selector.rsplit("::", 1)[-1]
# Strip a parametrized id (``test_x[case]``) down to the function
# name; ``-k`` matches substrings, and brackets are -k syntax.
leaf = leaf.split("[", 1)[0]
node_id_selectors.append((raw, leaf))
translated.append(file_part)
if node_id_selectors:
args.paths_positional = translated
keys = [leaf for _, leaf in node_id_selectors]
expr = " or ".join(dict.fromkeys(keys))
for raw, leaf in node_id_selectors:
print(
f"note: '{raw}' is a pytest node id; this runner is "
f"file-granular. Running the file with -k {leaf!r}.",
file=sys.stderr,
)
# Only inject -k when the caller didn't pass one themselves; their
# explicit filter wins over our inferred one.
if not any(
t == "-k" or t.startswith("-k=") or (t.startswith("-k") and len(t) > 2)
for t in bare_passthrough + explicit_passthrough
):
bare_passthrough = bare_passthrough + ["-k", expr]
# Bare flags run before any explicit ``--`` passthrough so ordering is
# intuitive (``run_tests.sh tests/foo.py -q -- --tb=long`` → ``-q --tb=long``).
pytest_passthrough = bare_passthrough + explicit_passthrough
# Parse --slice (or HERMES_TEST_SLICE) early so we can exit on bad input
# before doing any expensive discovery.
slice_raw = args.slice or os.environ.get("HERMES_TEST_SLICE")
slice_index: int | None = None
slice_count: int = 1
if slice_raw:
try:
idx_s, count_s = slice_raw.split("/", 1)
slice_index = int(idx_s)
slice_count = int(count_s)
except (ValueError, AttributeError):
print(f"error: --slice must be I/N (e.g. 1/4), got: {slice_raw!r}", file=sys.stderr)
sys.exit(2)
repo_root = Path(__file__).resolve().parent.parent
# --files: explicit file list from the CI generate job — skip discovery.
if args.files:
files = [repo_root / f for f in _split_pathspec(args.files)]
roots = []
else:
# Resolve discovery roots: positional path args override --paths if any
# were supplied, otherwise --paths (which itself defaults to 'tests').
if args.paths_positional:
roots = [repo_root / p for p in args.paths_positional]
else:
roots = [repo_root / p for p in _split_pathspec(args.paths)]
You can’t perform that action at this time.