Vito Sansevero
72a769d752
PM_TEST_SLOWEST=N makes tests/run_suite.py list the N slowest tests. Three history tests restored while their own read held a snapshot, so each publish sat out the 30 s WAL-truncate wait; they now use a short wait. Temporary diagnostics for the Windows timings ride along in this commit.
337 lines
12 KiB
Python
337 lines
12 KiB
Python
"""Tests for tests/run_suite.py, the CI test runner and its hang watchdog (#350).
|
|
|
|
Each test writes a temporary folder of tiny test modules and runs
|
|
``python -m tests.run_suite -s <folder> -v`` in a subprocess from the repo
|
|
root. Every subprocess call has a timeout, so a broken watchdog fails this
|
|
file instead of hanging the suite. Probes that should hang wait a bounded
|
|
time, so a watchdog that never fires shows up as a plain assertion failure.
|
|
"""
|
|
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import textwrap
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parent.parent
|
|
SUBPROCESS_TIMEOUT = 60
|
|
WATCHDOG_ENV = ("PM_TEST_HANG_LIMIT", "PM_TEST_SHUTDOWN_LIMIT")
|
|
# Integer settings the runner reads; a bad value is a usage error.
|
|
RUNNER_ENV = (*WATCHDOG_ENV, "PM_TEST_SLOWEST")
|
|
|
|
|
|
def run_suite(folder, extra_env=None, *args, verbosity=("-v",)):
|
|
"""Run the runner on ``folder`` and return the completed process."""
|
|
env = {k: v for k, v in os.environ.items() if k not in RUNNER_ENV}
|
|
# Windows writes a piped stderr in the ANSI codepage; ask for the UTF-8
|
|
# this helper decodes, so non-ASCII values and paths survive.
|
|
env["PYTHONIOENCODING"] = "utf-8"
|
|
env.update(extra_env or {})
|
|
command = [sys.executable, "-m", "tests.run_suite", "-s", str(folder)]
|
|
return subprocess.run(
|
|
[*command, *verbosity, *args],
|
|
cwd=REPO_ROOT,
|
|
env=env,
|
|
capture_output=True,
|
|
text=True,
|
|
encoding="utf-8",
|
|
errors="replace",
|
|
timeout=SUBPROCESS_TIMEOUT,
|
|
)
|
|
|
|
|
|
class RunSuiteTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self._tmp = tempfile.TemporaryDirectory(prefix="pm-run-suite-")
|
|
self.addCleanup(self._tmp.cleanup)
|
|
self.folder = Path(self._tmp.name)
|
|
|
|
def write(self, name, source):
|
|
(self.folder / name).write_text(textwrap.dedent(source), encoding="utf-8")
|
|
|
|
def assertDumped(self, result, test_id):
|
|
detail = result.stderr[-4000:]
|
|
self.assertNotEqual(result.returncode, 0, detail)
|
|
self.assertIn(f"start: {test_id}", result.stderr, detail)
|
|
self.assertIn("Timeout (", result.stderr, detail)
|
|
self.assertIn("Thread 0x", result.stderr, detail)
|
|
|
|
def test_passing_test_exits_zero_and_names_each_start(self):
|
|
self.write(
|
|
"test_probe_pass.py",
|
|
"""
|
|
import unittest
|
|
|
|
class Probe(unittest.TestCase):
|
|
def test_ok(self):
|
|
pass
|
|
""",
|
|
)
|
|
result = run_suite(self.folder)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertIn("start: test_probe_pass.Probe.test_ok", result.stderr)
|
|
|
|
def test_start_line_is_written_at_quiet_verbosity_too(self):
|
|
self.write(
|
|
"test_probe_quiet.py",
|
|
"""
|
|
import unittest
|
|
|
|
class Probe(unittest.TestCase):
|
|
def test_ok(self):
|
|
pass
|
|
""",
|
|
)
|
|
result = run_suite(self.folder, None, "-q")
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertIn("start: test_probe_quiet.Probe.test_ok", result.stderr)
|
|
|
|
def test_start_line_begins_its_own_line_at_default_verbosity(self):
|
|
self.write(
|
|
"test_probe_dots.py",
|
|
"""
|
|
import unittest
|
|
|
|
class Probe(unittest.TestCase):
|
|
def test_a(self):
|
|
pass
|
|
|
|
def test_b(self):
|
|
pass
|
|
""",
|
|
)
|
|
result = run_suite(self.folder, verbosity=())
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
lines = result.stderr.splitlines()
|
|
for name in ("test_a", "test_b"):
|
|
self.assertIn(f"start: test_probe_dots.Probe.{name}", lines)
|
|
|
|
def test_warnings_are_shown_as_with_unittest_discover(self):
|
|
self.write(
|
|
"test_probe_warn.py",
|
|
"""
|
|
import unittest
|
|
import warnings
|
|
|
|
class Probe(unittest.TestCase):
|
|
def test_warns(self):
|
|
warnings.warn("probe-deprecated", DeprecationWarning)
|
|
""",
|
|
)
|
|
env = {"PYTHONWARNINGS": ""}
|
|
result = run_suite(self.folder, env)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertIn("probe-deprecated", result.stderr)
|
|
|
|
def test_failing_test_exits_one(self):
|
|
self.write(
|
|
"test_probe_fail.py",
|
|
"""
|
|
import unittest
|
|
|
|
class Probe(unittest.TestCase):
|
|
def test_fails(self):
|
|
self.fail("expected")
|
|
""",
|
|
)
|
|
result = run_suite(self.folder)
|
|
self.assertEqual(result.returncode, 1, result.stderr)
|
|
|
|
def test_pattern_option_selects_modules(self):
|
|
self.write(
|
|
"check_probe.py",
|
|
"""
|
|
import unittest
|
|
|
|
class Probe(unittest.TestCase):
|
|
def test_fails(self):
|
|
self.fail("selected by -p")
|
|
""",
|
|
)
|
|
self.assertEqual(run_suite(self.folder).returncode, 5)
|
|
result = run_suite(self.folder, None, "-p", "check_*.py")
|
|
self.assertEqual(result.returncode, 1, result.stderr)
|
|
|
|
def test_lora_manager_settings_read_a_disposable_folder(self):
|
|
# #355: no test reads the developer's real LoraManager settings.
|
|
record = self.folder / "seen.txt"
|
|
self.write(
|
|
"test_probe_lora.py",
|
|
f"""
|
|
import os
|
|
import unittest
|
|
|
|
class Probe(unittest.TestCase):
|
|
def test_isolated(self):
|
|
folder = os.environ["LORA_MANAGER_SETTINGS_DIR"]
|
|
with open({str(record)!r}, "w", encoding="utf-8") as handle:
|
|
handle.write(folder)
|
|
self.assertTrue(os.path.isdir(folder))
|
|
self.assertEqual(os.listdir(folder), [])
|
|
""",
|
|
)
|
|
result = run_suite(self.folder, {"LORA_MANAGER_SETTINGS_DIR": "/real/one"})
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
folder = record.read_text(encoding="utf-8")
|
|
self.assertNotEqual(folder, "/real/one")
|
|
self.assertFalse(os.path.exists(folder), "removed when the run ends")
|
|
|
|
def test_empty_folder_exits_five(self):
|
|
result = run_suite(self.folder)
|
|
self.assertEqual(result.returncode, 5, result.stderr)
|
|
|
|
def test_repo_root_modules_import_as_with_unittest_discover(self):
|
|
self.write(
|
|
"test_probe_import.py",
|
|
"""
|
|
import unittest
|
|
|
|
from utils.hashing import generate_prompt_hash
|
|
|
|
class Probe(unittest.TestCase):
|
|
def test_import(self):
|
|
same = generate_prompt_hash(" A ") == generate_prompt_hash("a")
|
|
self.assertTrue(same)
|
|
""",
|
|
)
|
|
result = run_suite(self.folder)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
|
|
def test_hung_test_dumps_every_thread_and_exits(self):
|
|
self.write(
|
|
"test_probe_hang.py",
|
|
"""
|
|
import time
|
|
import unittest
|
|
|
|
class Probe(unittest.TestCase):
|
|
def test_hangs(self):
|
|
time.sleep(30)
|
|
""",
|
|
)
|
|
result = run_suite(self.folder, {"PM_TEST_HANG_LIMIT": "2"})
|
|
self.assertDumped(result, "test_probe_hang.Probe.test_hangs")
|
|
|
|
def test_hang_in_async_teardown_dumps_and_exits(self):
|
|
self.write(
|
|
"test_probe_teardown.py",
|
|
"""
|
|
import threading
|
|
import unittest
|
|
|
|
class Probe(unittest.IsolatedAsyncioTestCase):
|
|
async def asyncTearDown(self):
|
|
threading.Event().wait(30)
|
|
|
|
async def test_body_passes(self):
|
|
pass
|
|
""",
|
|
)
|
|
result = run_suite(self.folder, {"PM_TEST_HANG_LIMIT": "2"})
|
|
self.assertDumped(result, "test_probe_teardown.Probe.test_body_passes")
|
|
|
|
def test_hang_in_event_loop_shutdown_after_ok_dumps_and_exits(self):
|
|
# The #349 shape: the test passes ("ok" is printed), then the loop's
|
|
# shutdown waits for a default-executor thread that never finishes.
|
|
self.write(
|
|
"test_probe_loop.py",
|
|
"""
|
|
import asyncio
|
|
import threading
|
|
import unittest
|
|
|
|
class Probe(unittest.IsolatedAsyncioTestCase):
|
|
async def test_leaves_executor_busy(self):
|
|
loop = asyncio.get_running_loop()
|
|
loop.run_in_executor(None, threading.Event().wait, 30)
|
|
""",
|
|
)
|
|
result = run_suite(self.folder, {"PM_TEST_HANG_LIMIT": "2"})
|
|
self.assertDumped(result, "test_probe_loop.Probe.test_leaves_executor_busy")
|
|
start = result.stderr.index("start: test_probe_loop")
|
|
self.assertIn("ok", result.stderr[start:], result.stderr[-4000:])
|
|
|
|
def test_hang_at_interpreter_exit_dumps_and_exits(self):
|
|
self.write(
|
|
"test_probe_exit.py",
|
|
"""
|
|
import threading
|
|
import unittest
|
|
|
|
class Probe(unittest.TestCase):
|
|
def test_leaves_thread(self):
|
|
threading.Thread(target=threading.Event().wait, args=(30,)).start()
|
|
""",
|
|
)
|
|
result = run_suite(self.folder, {"PM_TEST_SHUTDOWN_LIMIT": "2"})
|
|
detail = result.stderr[-4000:]
|
|
self.assertNotEqual(result.returncode, 0, detail)
|
|
self.assertIn("ok", result.stderr, detail)
|
|
self.assertIn("Timeout (", result.stderr, detail)
|
|
|
|
def test_slowest_tests_are_listed_when_asked(self):
|
|
self.write(
|
|
"test_probe_slowest.py",
|
|
"""
|
|
import time
|
|
import unittest
|
|
|
|
class Probe(unittest.TestCase):
|
|
def test_fast(self):
|
|
pass
|
|
|
|
def test_slow(self):
|
|
time.sleep(0.3)
|
|
|
|
def test_slower(self):
|
|
time.sleep(0.6)
|
|
""",
|
|
)
|
|
result = run_suite(self.folder, {"PM_TEST_SLOWEST": "2"})
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
report = result.stderr.split("slowest 2 of 3 tests:\n", 1)
|
|
self.assertEqual(len(report), 2, result.stderr)
|
|
lines = [line.split() for line in report[1].strip().splitlines()[:2]]
|
|
self.assertEqual(
|
|
[line[1] for line in lines],
|
|
[
|
|
"test_probe_slowest.Probe.test_slower",
|
|
"test_probe_slowest.Probe.test_slow",
|
|
],
|
|
result.stderr,
|
|
)
|
|
self.assertGreaterEqual(float(lines[0][0].rstrip("s")), 0.6)
|
|
self.assertNotIn("test_fast", report[1])
|
|
|
|
def test_slowest_tests_are_not_listed_by_default(self):
|
|
self.write(
|
|
"test_probe_quiet.py",
|
|
"""
|
|
import unittest
|
|
|
|
class Probe(unittest.TestCase):
|
|
def test_ok(self):
|
|
pass
|
|
""",
|
|
)
|
|
result = run_suite(self.folder)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertNotIn("slowest", result.stderr)
|
|
|
|
def test_invalid_limit_is_a_usage_error(self):
|
|
for name in RUNNER_ENV:
|
|
for value in ("abc", "0", "-3", "1.5", "\u00b2"):
|
|
with self.subTest(name=name, value=value):
|
|
result = run_suite(self.folder, {name: value})
|
|
self.assertEqual(result.returncode, 2, result.stderr)
|
|
lines = result.stderr.strip().splitlines()
|
|
self.assertEqual(len(lines), 1, result.stderr)
|
|
self.assertIn(name, lines[0])
|
|
self.assertIn(repr(value), lines[0])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|