File size: 3,387 Bytes
50ad5a9 | 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 | import importlib.util
import os
import select
import signal
import subprocess
import sys
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path
def test_matrix_preview_uses_four_bf16_models_without_starting(tmp_path):
root = Path(__file__).resolve().parents[1]
output = tmp_path / "does-not-exist"
engine_python = tmp_path / "engine-venv-python"
engine_python.symlink_to(sys.executable)
result = subprocess.run(
[
sys.executable,
str(root / "benchmarks/run_matrix.py"),
"--output",
str(output),
"--include-external",
"--concurrency",
"8",
"--engine-python",
str(engine_python),
],
capture_output=True,
text=True,
check=True,
)
assert "Preview only" in result.stdout
assert "ministral3-3b-bf16" in result.stdout
assert "ministral3-3b-fp8" not in result.stdout
assert "qwen36-27b-bf16" in result.stdout
assert "qwen36-35b-a3b-bf16" in result.stdout
assert "qwen35-4b-bf16" in result.stdout
assert "scienthoon-v1" in result.stdout
assert "jevbench-hard/public.jsonl" in result.stdout
assert "--probe-vision" not in result.stdout
assert f"--engine-python {engine_python}" in result.stdout
assert not output.exists()
def test_stop_owned_cleans_runtime_that_inherits_ignored_sigint(tmp_path):
root = Path(__file__).resolve().parents[1]
spec = importlib.util.spec_from_file_location(
"benchmark_matrix_under_test", root / "benchmarks/run_matrix.py"
)
matrix = importlib.util.module_from_spec(spec)
spec.loader.exec_module(matrix)
cleanup_marker = tmp_path / "runtime-cleaned-up"
child_code = """
import signal
import sys
from pathlib import Path
# A background launcher can inherit this disposition from its parent shell.
signal.signal(signal.SIGINT, signal.SIG_IGN)
def terminate(signum, frame):
raise KeyboardInterrupt
signal.signal(signal.SIGTERM, terminate)
try:
print("ready", flush=True)
while True:
signal.pause()
except KeyboardInterrupt:
pass
finally:
Path(sys.argv[1]).write_text("graceful SIGTERM cleanup")
"""
process = subprocess.Popen(
[sys.executable, "-c", child_code, str(cleanup_marker)],
start_new_session=True,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
)
try:
readable, _, _ = select.select([process.stdout], [], [], 5)
assert readable, "child did not install its signal handlers"
assert process.stdout.readline().strip() == "ready"
# Bound regressions without replacing real OS signal delivery or wait().
with ThreadPoolExecutor(max_workers=1) as executor:
stopped = executor.submit(matrix.stop_owned, process)
try:
stopped.result(timeout=5)
finally:
if process.poll() is None:
os.killpg(process.pid, signal.SIGKILL)
process.wait(timeout=5)
assert process.returncode == 0
assert cleanup_marker.read_text() == "graceful SIGTERM cleanup"
finally:
if process.poll() is None:
os.killpg(process.pid, signal.SIGKILL)
process.wait(timeout=5)
process.stdout.close()
process.stderr.close()
|