ci: bound unittest-leg reimport growth and add stall diagnostics (#1052)

pyproject.toml

@@ -278,6 +278,12 @@ test = [
     # test_xdist_run_reports_the_diagnostic_and_fails goes red rather than
     # skipped, since `importlib.util.find_spec('xdist')` still succeeds.
     "pytest-xdist>=3.6.1",
+    # A per-test backstop (#1052): a hung test fails by name, with every
+    # thread's stack, instead of running until the job's ``timeout-minutes``
+    # cancels it with nothing in the log. The value and method live in
+    # [tool.pytest.ini_options] below. Unpinned, like ``isort`` below, rather
+    # than inventing a floor (verified against 2.4.0).
+    "pytest-timeout",
     "typing-extensions",
     # Enough of the ``vendor`` extra for a test to import a vendor crawler.
     # Without these, ``tests/vendor/test_ipx_socket_unit.py`` reports "9 skipped"
@@ -354,6 +360,16 @@ testpaths = [
 python_files = [
     "test_*.py",
 ]
+# pytest-timeout (#1052). 900s per test is a hang backstop, not a speed limit:
+# the slowest test measured locally was 21s (in test_pyshark_encap_map.py), so
+# even the 10-15x slowdown seen on CI (315s) leaves it about 3x short, while a
+# hang still fails by name well inside the jobs' 45-minute cap. ``thread``, not
+# the POSIX default ``signal``: a pending SIGALRM fails
+# TimeLimitTests.test_nothing_is_re_armed_when_nothing_was_pending, measured.
+# ``thread`` leaks a pcapkit generation per test unless tests/conftest.py's
+# pytest_timeout_cancel_timer hook drops the retained timer.
+timeout = 900
+timeout_method = "thread"
 
 [tool.coverage.run]
 source = [

tests/conftest.py

@@ -249,6 +249,26 @@ def restore_module_table() -> 'Iterator[None]':
         restore_modules(_pinned_snapshot, ISOLATED_PREFIXES)
 
 
+@pytest.hookimpl(hookwrapper=True, optionalhook=True)
+def pytest_timeout_cancel_timer(item: 'pytest.Item') -> 'Iterator[None]':
+    """Drop the timer :mod:`pytest_timeout` leaves on ``item`` once it is cancelled.
+
+    Issue #1052. pytest-timeout's ``thread`` method keeps ``item.cancel_timeout``, a
+    closure over the test's :class:`threading.Timer`, for the whole session. A
+    :class:`~threading.Thread` captures :data:`sys.excepthook` and
+    :data:`threading.excepthook` when it is created, and every ``import pcapkit``
+    installs fresh tbtrim hooks there, so each retained timer pins the
+    :mod:`pcapkit` generation that was live when its test started. Measured on
+    ``tests/const/test_const_registry_protocol.py`` under ``thread``: past 9 GiB
+    and killed, against 283 MiB with this hook. That growth is what took every
+    ``test`` leg's runner down at about 13% on CI. ``optionalhook``: without
+    pytest-timeout installed the hookspec does not exist.
+
+    """
+    yield
+    item.__dict__.pop('cancel_timeout', None)
+
+
 def _collected_modules(items: 'Iterable[pytest.Item]') -> 'Iterator[pathlib.Path]':
     """Distinct module paths behind ``items``, in collection order.
 

tests/project/test_pytest_timeout_timer_release.py

@@ -0,0 +1,44 @@
+# -*- coding: utf-8 -*-
+"""Tests that ``tests/conftest.py`` drops pytest-timeout's retained timer (#1052).
+
+pytest-timeout's ``thread`` method leaves ``item.cancel_timeout`` -- a closure over
+the test's :class:`threading.Timer` -- on every item for the whole session, and the
+timer pins the :mod:`pcapkit` generation whose tbtrim excepthooks were live when it
+was created. Kept, that took the ``test`` legs' runners past their memory on CI.
+
+"""
+
+from __future__ import annotations
+
+import types
+import unittest
+
+from tests import conftest
+
+
+class TestCancelTimerHook(unittest.TestCase):
+
+    def _drive(self, item):
+        hook = conftest.pytest_timeout_cancel_timer(item)
+        next(hook)
+        with self.assertRaises(StopIteration):
+            next(hook)
+
+    def test_the_retained_timer_is_dropped_after_cancel(self):
+        item = types.SimpleNamespace(cancel_timeout=lambda: None)
+        self._drive(item)
+        self.assertFalse(hasattr(item, 'cancel_timeout'))
+
+    def test_an_item_without_a_timer_is_left_alone(self):
+        item = types.SimpleNamespace(nodeid='x')
+        self._drive(item)
+        self.assertEqual(vars(item), {'nodeid': 'x'})
+
+    def test_it_is_an_optional_wrapper(self):
+        opts = conftest.pytest_timeout_cancel_timer.pytest_impl
+        self.assertTrue(opts['hookwrapper'])
+        self.assertTrue(opts['optionalhook'])
+
+
+if __name__ == '__main__':
+    unittest.main()

tests/project/test_run_unittest_leg_stall_dump.py

@@ -0,0 +1,172 @@
+# -*- coding: utf-8 -*-
+"""Tests for ``util/run_unittest_leg.py``'s stall dump, periodic collect and no-restore contract (#1052).
+
+A leg that stalls is killed by the step cap with nothing in the log to say where
+it was, so the runner arms :func:`faulthandler.dump_traceback_later` before it
+builds the suite and cancels it on the way out. The in-process tests mock
+:mod:`faulthandler`, because this module itself runs inside the ``project`` leg,
+whose own dump is armed; the subprocess test lets a real one fire.
+
+The last class pins the decision #1052 recorded against restoring
+:data:`sys.modules` between tests: a sibling's write to the ``pcapkit`` region
+must survive into the next test, or the leg stops seeing #981's class of skew.
+
+"""
+
+from __future__ import annotations
+
+import os
+import pathlib
+import subprocess
+import sys
+import textwrap
+import unittest
+from unittest import mock
+
+ROOT = pathlib.Path(__file__).resolve().parents[2]
+
+if str(ROOT / 'util') not in sys.path:
+    sys.path.insert(0, str(ROOT / 'util'))
+
+import run_unittest_leg as leg  # noqa: E402  pylint: disable=wrong-import-position
+
+
+class TestStallDumpArming(unittest.TestCase):
+
+    def _main(self, argv, env=None, run=None):
+        environ = {k: v for k, v in os.environ.items() if k != leg.STALL_DUMP_ENV}
+        environ.update(env or {})
+        with mock.patch.dict(os.environ, environ, clear=True), \
+                mock.patch.object(leg.faulthandler, 'dump_traceback_later') as arm, \
+                mock.patch.object(leg.faulthandler, 'cancel_dump_traceback_later') as cancel, \
+                mock.patch.object(leg, '_run', run or mock.Mock(return_value=0)):
+            try:
+                code = leg.main(argv)
+            except (Exception, SystemExit) as exc:  # pylint: disable=broad-except
+                code = exc
+        return code, arm, cancel
+
+    def test_default_is_just_under_a_twenty_minute_cap(self):
+        code, arm, cancel = self._main(['cli'])
+        self.assertEqual(code, 0)
+        arm.assert_called_once()
+        self.assertEqual(arm.call_args.args, (1080.0,))
+        self.assertEqual(arm.call_args.kwargs['repeat'], False)
+        self.assertEqual(arm.call_args.kwargs['exit'], False)
+        self.assertLess(leg.DEFAULT_STALL_DUMP, 20 * 60)
+        cancel.assert_called_once_with()
+
+    def test_environment_overrides_the_default(self):
+        _, arm, _ = self._main(['cli'], env={leg.STALL_DUMP_ENV: '30'})
+        self.assertEqual(arm.call_args.args, (30.0,))
+
+    def test_flag_overrides_the_environment(self):
+        _, arm, _ = self._main(['cli', '--stall-dump', '7'], env={leg.STALL_DUMP_ENV: '30'})
+        self.assertEqual(arm.call_args.args, (7.0,))
+
+    def test_zero_disables_it(self):
+        _, arm, _ = self._main(['cli', '--stall-dump', '0'])
+        arm.assert_not_called()
+
+    def test_bad_environment_value_is_a_usage_error(self):
+        with mock.patch('sys.stderr'):
+            code, arm, _ = self._main(['cli'], env={leg.STALL_DUMP_ENV: 'soon'})
+        self.assertIsInstance(code, SystemExit)
+        self.assertEqual(code.code, 2)
+        arm.assert_not_called()
+
+    def test_cancelled_when_the_leg_raises(self):
+        code, arm, cancel = self._main(['cli'], run=mock.Mock(side_effect=KeyError('boom')))
+        self.assertIsInstance(code, KeyError)
+        arm.assert_called_once()
+        cancel.assert_called_once_with()
+
+
+class TestStallDumpFires(unittest.TestCase):
+
+    def test_a_stalled_leg_dumps_its_stack_and_keeps_running(self):
+        script = textwrap.dedent(f'''
+            import sys, time, unittest
+            sys.path.insert(0, {str(ROOT / 'util')!r})
+            import run_unittest_leg as leg
+
+            class Stall(unittest.TestCase):
+                def test_sleeps_past_the_dump(self):
+                    time.sleep(2)
+
+            leg.build_suite = lambda *a: unittest.TestLoader().loadTestsFromTestCase(Stall)
+            sys.exit(leg.main(['cli', '--stall-dump', '0.5']))
+        ''')
+        proc = subprocess.run([sys.executable, '-c', script], cwd=str(ROOT),
+                              capture_output=True, text=True, timeout=120, check=False)
+        self.assertEqual(proc.returncode, 0, proc.stderr)
+        self.assertIn('Timeout (0:00:00.500000)!', proc.stderr)
+        self.assertIn('test_sleeps_past_the_dump', proc.stderr)
+        self.assertEqual(proc.stderr.count('Timeout ('), 1, 'repeat=False dumps once')
+        self.assertIn('1 test(s), 0 failure(s), 0 error(s)', proc.stdout)
+
+
+class TestPeriodicCollect(unittest.TestCase):
+    """``--gc-every N`` frees unreachable re-import generations every N tests (#1052)."""
+
+    @staticmethod
+    def _suite(n):
+        class Trivial(unittest.TestCase):
+            pass
+        for i in range(n):
+            setattr(Trivial, f'test_{i}', lambda self: None)
+        return unittest.TestLoader().loadTestsFromTestCase(Trivial)
+
+    def _collects(self, argv, tests=6):
+        with mock.patch.object(leg, 'build_suite', return_value=self._suite(tests)), \
+                mock.patch.object(leg.faulthandler, 'dump_traceback_later'), \
+                mock.patch.object(leg.faulthandler, 'cancel_dump_traceback_later'), \
+                mock.patch.object(leg.gc, 'collect') as collect, \
+                mock.patch('sys.stderr'), mock.patch('sys.stdout'):
+            self.assertEqual(leg.main(argv), 0)
+        return collect.call_count
+
+    def test_default_collects_every_ten_tests(self):
+        self.assertEqual(leg.DEFAULT_GC_EVERY, 10)
+        self.assertEqual(self._collects(['cli'], tests=25), 2)
+
+    def test_every_n_tests(self):
+        self.assertEqual(self._collects(['cli', '--gc-every', '4'], tests=9), 2)
+        self.assertEqual(self._collects(['cli', '--gc-every', '1'], tests=3), 3)
+
+    def test_zero_disables_it(self):
+        self.assertEqual(self._collects(['cli', '--gc-every', '0']), 0)
+
+
+class TestNoModuleTableRestore(unittest.TestCase):
+    """The leg must not heal a sibling's ``sys.modules`` write (#981, #1052)."""
+
+    PROBE = 'pcapkit._run_unittest_leg_probe'
+
+    def tearDown(self):
+        sys.modules.pop(self.PROBE, None)
+
+    def test_a_write_survives_into_the_next_test(self):
+        probe, seen = object(), []
+
+        class Writer(unittest.TestCase):
+            def test_a(self_):  # pylint: disable=no-self-argument
+                sys.modules[TestNoModuleTableRestore.PROBE] = probe
+
+        class Reader(unittest.TestCase):
+            def test_b(self_):  # pylint: disable=no-self-argument
+                seen.append(sys.modules.get(TestNoModuleTableRestore.PROBE))
+
+        loader = unittest.TestLoader()
+        suite = unittest.TestSuite([loader.loadTestsFromTestCase(Writer),
+                                    loader.loadTestsFromTestCase(Reader)])
+        args = leg.argparse.Namespace(directory='cli', exclude=[], verbose=False,
+                                     gc_every=leg.DEFAULT_GC_EVERY)
+        with mock.patch.object(leg, 'build_suite', return_value=suite), \
+                mock.patch('sys.stderr'), mock.patch('sys.stdout'):
+            self.assertEqual(leg._run(args, 0.0), 0)  # pylint: disable=protected-access
+        self.assertEqual(seen, [probe])
+
+
+if __name__ == '__main__':
+    unittest.main()

util/run_unittest_leg.py

@@ -100,10 +100,37 @@
 because it is already run whole, under :program:`pytest`, by the
 ``integration`` job.
 
+GitHub issue #1052 -- a stalled leg: ``--stall-dump SECONDS`` (or
+``PCAPKIT_UNITTEST_STALL_DUMP``) arms :func:`faulthandler.dump_traceback_later`
+so that a leg still running after that long writes every thread's stack to
+stderr once, then carries on. The default, 1080s, is just under a 20-minute step
+cap, so a leg that is about to be killed says where it was first; ``0`` disables
+it.
+
+#1052 also found the memory growth behind the slow ``test_mh_unit`` tail. Every
+purge-then-import leaves the previous generation as cyclic garbage, and the
+interpreter's own collector intermittently falls behind: three plain runs of
+``protocols/internet`` peaked at 1008, 781 and 1650 MiB, the last with ``test_mh_unit``
+tests at up to 7.3s rather than 1.2-1.8s. ``--gc-every N`` (default 10) therefore
+runs :func:`gc.collect` after every N tests: 491 MiB peak at the same 420s wall.
+Every test cost 80s more for 320 MiB.
+
+It is a collect, deliberately *not* the :data:`sys.modules` restore #1052 first
+proposed. The loader imports every module before any test runs, so a module's
+import-time bindings belong to the generation live at load; restoring that
+snapshot hands it back, which is the very skew this leg exists to expose. On
+#981's own reproduction (``32bcfba15^``, ``test_http_unit`` then
+``test_base_class_contract``) no restore gives 1 failure and 3 errors; restoring
+after each module, or around each test, gives 0 and 0; a collect after each test
+keeps 1 and 3, since it frees only what no test can reach.
+
 """
 from __future__ import annotations
 
 import argparse
+import faulthandler
+import gc
+import os
 import pathlib
 import sys
 import time
@@ -137,6 +164,34 @@
 #: this script exists to catch.
 _EXCLUDED_ROOT_MODULES = frozenset({'tests.test_tier_guard_xdist'})
 
+#: Seconds before a still-running leg dumps every thread's stack (#1052): just
+#: under the 20-minute step cap, so the dump lands before the kill.
+DEFAULT_STALL_DUMP = 1080
+
+#: Environment override for :data:`DEFAULT_STALL_DUMP`; ``--stall-dump`` wins.
+STALL_DUMP_ENV = 'PCAPKIT_UNITTEST_STALL_DUMP'
+
+#: Run a full :func:`gc.collect` after every this many tests (#1052); see the
+#: module docstring for the measurements behind it. ``--gc-every 0`` disables.
+DEFAULT_GC_EVERY = 10
+
+
+class _CollectingResult(unittest.TextTestResult):
+    """A :class:`~unittest.TextTestResult` that runs :func:`gc.collect` every *n* tests.
+
+    Only unreachable objects are freed, so nothing a later test can reach -- the
+    :data:`sys.modules` table above all -- changes; when they would otherwise be
+    freed was already up to the interpreter's own thresholds.
+
+    """
+
+    gc_every = DEFAULT_GC_EVERY
+
+    def stopTest(self, test: 'unittest.TestCase') -> None:
+        super().stopTest(test)
+        if self.gc_every > 0 and self.testsRun % self.gc_every == 0:
+            gc.collect()
+
 
 def _dotted(path: 'pathlib.Path') -> 'str':
     """``path`` as the dotted module name :class:`unittest.TestLoader` wants."""
@@ -228,11 +283,42 @@ def main(argv: 'list[str] | None' = None) -> 'int':
         '-v', '--verbose', action='store_true',
         help='pass verbosity 2 to unittest.TextTestRunner instead of the default 1',
     )
+    parser.add_argument(
+        '--stall-dump', type=float, default=None, metavar='SECONDS',
+        help=f"dump every thread's stack to stderr once the leg has run this long "
+             f"(default ${STALL_DUMP_ENV}, else {DEFAULT_STALL_DUMP}); 0 disables",
+    )
+    parser.add_argument(
+        '--gc-every', type=int, default=DEFAULT_GC_EVERY, metavar='N',
+        help=f'run gc.collect() after every N tests (default {DEFAULT_GC_EVERY}); 0 disables',
+    )
     args = parser.parse_args(argv)
+    stall_dump = args.stall_dump
+    if stall_dump is None:
+        try:
+            stall_dump = float(os.environ.get(STALL_DUMP_ENV) or DEFAULT_STALL_DUMP)
+        except ValueError:
+            parser.error(f'${STALL_DUMP_ENV} is not a number of seconds: '
+                         f'{os.environ[STALL_DUMP_ENV]!r}')
 
     start = time.monotonic()
+    if stall_dump > 0:
+        # ``sys.__stderr__``, not ``sys.stderr``: faulthandler keeps the file
+        # descriptor it is given, and a test that swaps ``sys.stderr`` for a
+        # buffer must not be where the dump goes.
+        faulthandler.dump_traceback_later(stall_dump, repeat=False,
+                                          file=sys.__stderr__ or sys.stderr, exit=False)
+    try:
+        return _run(args, start)
+    finally:
+        faulthandler.cancel_dump_traceback_later()
+
+
+def _run(args: 'argparse.Namespace', start: 'float') -> 'int':
+    """Build and run the leg, then print its one-line tally whatever happened."""
     suite = build_suite(args.directory, args.exclude)
-    runner = unittest.TextTestRunner(verbosity=2 if args.verbose else 1)
+    result_class = type('_LegResult', (_CollectingResult,), {'gc_every': args.gc_every})
+    runner = unittest.TextTestRunner(verbosity=2 if args.verbose else 1, resultclass=result_class)
     result: 'unittest.TestResult | None' = None
     try:
         result = runner.run(suite)