diff options
| author | hathach <[email protected]> | 2026-08-27 01:01:30 +0700 |
|---|---|---|
| committer | hathach <[email protected]> | 2026-08-27 08:54:07 +0700 |
| commit | 9baa97a8c6cc671aefe26168b3d8281da070ebf9 (patch) | |
| tree | 4adb5cea77de2001fdcfd2e0479abdfa967d7a83 | |
| parent | d799b6f572e4b39ebebcf72126d3120f7829c034 (diff) | |
test/hil: drop the Windows accommodations, which accommodate nothing
hil_test.py cannot run on Windows and never could: it imports helper.hil_lock,
whose module-level `import fcntl` is POSIX-only, so the harness fails at import
before a line of it executes. Past that it reads /sys/bus/usb, /dev/bus/usb,
/dev/serial/by-id and /proc, kills by process group, and takes flock board
locks -- none of which Windows has.
So the guards were protecting a platform the code cannot reach:
- run_cmd branched three ways on os.name to decide whether to set
start_new_session and whether to killpg. The non-POSIX arm called p.kill()
instead, which kills only the direct child -- exactly the semantics the whole
containment design rejects, since a flasher run through a shell reparents out
of reach. Dead code that documented the wrong answer.
- hil_test picked multiprocessing's default context on Windows "so it still
IMPORTS there". It does not import there.
- test_device_audio_test_freertos returned 'skipped' on nt before touching
ALSA, in a function only ever reached from a worker that cannot start there.
- Seven @unittest.skipIf(os.name == 'nt') decorators across the two suites.
These were the only ones with a real effect -- the unit tests DO import and
run on Windows, because they stub pyserial and mostly exercise pure logic --
but what they buy is a partially-green suite for a harness that cannot run,
and nothing verifies the set is correct: the hil-test hook only ever runs on
ubuntu-latest, so a missing guard fails silently until someone tries.
Removing them makes the POSIX assumption single and explicit rather than
scattered and half-honoured. Nothing changes on Linux: every removed branch was
the one already taken there.
Removing the run_cmd guards also removes their `else: p.kill()` arms. Those were
the Windows branches, and p.kill() reaches only the direct child -- a flasher run
through a shell keeps grandchildren it cannot touch, which is the semantics this
containment design rejects. RunCmdCleanupShape pins what is left: both cleanup
paths killpg, no try carries an else whose body would run when the kill
SUCCEEDED, and the BaseException path still re-raises. Structural rather than
behavioural because driving a real SIGINT into a blocked communicate() is
timing-dependent, and what actually breaks this block is an edit that rebinds a
branch -- which is a shape.
| -rw-r--r-- | test/hil/helper/hil_util.py | 35 | ||||
| -rwxr-xr-x | test/hil/hil_test.py | 7 | ||||
| -rw-r--r-- | test/hil/test/test_hil_bounded.py | 6 | ||||
| -rw-r--r-- | test/hil/test/test_hil_util.py | 54 |
4 files changed, 69 insertions, 33 deletions
diff --git a/test/hil/helper/hil_util.py b/test/hil/helper/hil_util.py index a0279fd4c..03d01270f 100644 --- a/test/hil/helper/hil_util.py +++ b/test/hil/helper/hil_util.py @@ -534,26 +534,22 @@ def run_cmd(cmd: str | list, cwd: str | None = None, timeout: int | None = None, } if not binary: popen_kwargs.update({'text': True, 'encoding': 'utf-8', 'errors': 'replace'}) - if os.name != 'nt': - # C-level setsid, same process-group semantics as preexec_fn=os.setsid but - # safe when called from threads (pool_check runs flashes from a thread pool) - popen_kwargs['start_new_session'] = True + # C-level setsid, same process-group semantics as preexec_fn=os.setsid but safe when + # called from threads (pool_check runs flashes from a thread pool) + popen_kwargs['start_new_session'] = True p = subprocess.Popen(cmd, **popen_kwargs) try: out, err = p.communicate(timeout=timeout) r = subprocess.CompletedProcess(args=cmd, returncode=p.returncode, stdout=out, stderr=err) except subprocess.TimeoutExpired as ex: - if os.name != 'nt': - try: - os.killpg(p.pid, signal.SIGKILL) - except OSError: - # ProcessLookupError: already gone. PermissionError: an all-root group - # refuses the group kill -- letting either escape would skip the bounded - # reap, the pipe close and the rc-124 return this handler exists for. - pass - else: - p.kill() + try: + os.killpg(p.pid, signal.SIGKILL) + except OSError: + # ProcessLookupError: already gone. PermissionError: an all-root group refuses + # the group kill -- letting either escape would skip the bounded reap, the pipe + # close and the rc-124 return this handler exists for. + pass try: out, err = p.communicate(timeout=10) except subprocess.TimeoutExpired: @@ -589,13 +585,10 @@ def run_cmd(cmd: str | list, cwd: str | None = None, timeout: int | None = None, # its OWN group, so it never got the terminal's SIGINT -- without this, Ctrl-C # leaves the flasher or testusb holding the probe and its usbfs node. Kill and # close, never wait: this path must not add a hang of its own. - if os.name != 'nt': - try: - os.killpg(p.pid, signal.SIGKILL) - except OSError: - pass - else: - p.kill() + try: + os.killpg(p.pid, signal.SIGKILL) + except OSError: + pass _close_pipes(p) raise diff --git a/test/hil/hil_test.py b/test/hil/hil_test.py index 253fb1b98..e32998420 100755 --- a/test/hil/hil_test.py +++ b/test/hil/hil_test.py @@ -69,9 +69,9 @@ from helper.hil_util import device_tests, dual_tests, host_test # Raw Lock/Semaphore objects in Pool initargs are inheritable only under fork # (spawn/forkserver pickle them and fail at Pool creation), so pin it against an -# interpreter default change. Windows has no fork: fall back so it still IMPORTS there. +# interpreter default change. -_mp = multiprocessing.get_context('fork') if os.name != 'nt' else multiprocessing.get_context() +_mp = multiprocessing.get_context('fork') Pool, Lock, Semaphore, Manager = _mp.Pool, _mp.Lock, _mp.Semaphore, _mp.Manager import string @@ -1306,9 +1306,6 @@ def test_device_midi_test(board): def test_device_audio_test_freertos(board): uid = board['uid'] - if os.name == 'nt': - return 'skipped' - pcm = None timeout = enum_timeout() while timeout > 0: diff --git a/test/hil/test/test_hil_bounded.py b/test/hil/test/test_hil_bounded.py index 9e60a19e2..6da65543f 100644 --- a/test/hil/test/test_hil_bounded.py +++ b/test/hil/test/test_hil_bounded.py @@ -75,7 +75,6 @@ def run_bounded(fn, timeout: float): return not t.is_alive(), exc[0] if exc else None [email protected](os.name == 'nt', 'POSIX shell fakes') class ReadDiskFile(unittest.TestCase): def setUp(self): self.tmp = TemporaryDirectory() @@ -342,7 +341,6 @@ class _MtpFakeRig: os.environ[k] = v [email protected](os.name == 'nt', 'POSIX shell fakes') @unittest.skipIf(sys.version_info < (3, 11), 'fake-pymtp steering needs PYTHONSAFEPATH') class DeviceMtp(_MtpFakeRig, unittest.TestCase): """test_device_mtp end to end: the real mtp_test.py subprocess under run_cmd, @@ -442,7 +440,6 @@ class BoundedOpen(unittest.TestCase): self.assertIsNone(self.hil_util.bounded_open( str(Path(self.tmp.name) / 'nope'), os.O_RDONLY, 5)) - @unittest.skipIf(os.name == 'nt', 'POSIX fifo') def test_blocking_open_gives_up_and_does_not_leak_fds(self): """A reader-less FIFO blocks open(O_WRONLY) forever -- the closest portable stand-in for a wedged usblp node.""" @@ -457,7 +454,6 @@ class BoundedOpen(unittest.TestCase): self.assertLessEqual(len(os.listdir('/proc/self/fd')) - before, 1, 'bounded_open leaked fds on the blocking path') - @unittest.skipIf(os.name == 'nt', 'POSIX fifo') def test_open_completing_during_the_abandon_does_not_leak(self): """The window the handoff lock exists for: the worker is at its store-or-close decision when the caller gives up and drains the box. @@ -528,7 +524,6 @@ class SysfsUnknownIsNotAbsent(unittest.TestCase): def test_missing_attribute_is_none(self): self.assertIsNone(self.hil_util.read_sysfs(str(Path(self.tmp.name) / 'nope'))) - @unittest.skipIf(os.name == 'nt', 'POSIX fifo') def test_blocking_read_is_unknown_not_absent(self): """A reader-less FIFO stands in for the wedged device whose sysfs read never returns; None here would read as "the board is gone".""" @@ -822,7 +817,6 @@ class WedgedPidsFailsClosed(unittest.TestCase): self.assertFalse(complete, 'a hidden holder was reported as absent') [email protected](os.name == 'nt', 'POSIX shell fakes') @unittest.skipIf(sys.version_info < (3, 11), 'fake-pymtp steering needs PYTHONSAFEPATH') class StrandMemoRemembersUnstattablePaths(unittest.TestCase): """A stranded path whose inode could not be read is stored as None -- which dict.get() diff --git a/test/hil/test/test_hil_util.py b/test/hil/test/test_hil_util.py index c95e20b6d..e06d2ba8b 100644 --- a/test/hil/test/test_hil_util.py +++ b/test/hil/test/test_hil_util.py @@ -19,7 +19,6 @@ sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from helper import hil_util [email protected](os.name == 'nt', 'POSIX shell commands') class RunCmdModes(unittest.TestCase): def test_default_mode_unchanged(self): r = hil_util.run_cmd('printf out; printf err >&2') @@ -227,5 +226,58 @@ class RunAlongsideKeepsStderrOffThePayload(unittest.TestCase): 'child stderr leaked into the payload stream') +class RunCmdCleanupShape(unittest.TestCase): + """run_cmd's two cleanup paths, asserted structurally. + + Both must kill the process GROUP: start_new_session puts the child in its own group, so + a flasher run through a shell keeps children a p.kill() cannot reach, and on the + BaseException path the child never receives the terminal's SIGINT either. + + Structural rather than behavioural on purpose. Driving a real SIGINT into a blocked + communicate() from a unit test is timing-dependent, and a flaky guard on this block is + worse than none -- while what actually breaks it is an edit that rebinds a branch. Both + times this block has been mis-edited, an `else:` ended up attached to the `try` instead + of the `if` it belonged to, so `p.kill()` ran when killpg had SUCCEEDED and its + ProcessLookupError masked the caller's exception. That is a shape, and shapes are + exactly what an AST can pin. + """ + + def _run_cmd_ast(self): + import ast + src = Path(hil_util.__file__).read_text() + return next(n for n in ast.walk(ast.parse(src)) + if isinstance(n, ast.FunctionDef) and n.name == 'run_cmd') + + def test_no_cleanup_try_has_an_else(self): + import ast + for n in ast.walk(self._run_cmd_ast()): + if isinstance(n, ast.Try) and n.orelse: + self.fail(f'try/else at line {n.lineno}: an else here runs when the kill ' + f'SUCCEEDED, and its ProcessLookupError masks the caller\'s ' + f'exception -- this block has been mis-edited that way twice') + + def test_both_cleanup_paths_kill_the_group(self): + import ast + fn = self._run_cmd_ast() + killers = [getattr(c.func, 'attr', '') for c in ast.walk(fn) + if isinstance(c, ast.Call) and getattr(c.func, 'attr', '') in + ('killpg', 'kill')] + self.assertEqual(killers.count('killpg'), 2, + 'both the timeout and the BaseException path must killpg') + self.assertEqual(killers.count('kill'), 0, + 'p.kill() reaches only the direct child; a flasher run through a ' + 'shell keeps grandchildren it cannot touch') + + def test_the_interrupt_path_reraises(self): + import ast + fn = self._run_cmd_ast() + base = [h for n in ast.walk(fn) if isinstance(n, ast.Try) for h in n.handlers + if isinstance(h.type, ast.Name) and h.type.id == 'BaseException'] + self.assertTrue(base, 'the BaseException cleanup path is gone') + for h in base: + self.assertTrue(any(isinstance(x, ast.Raise) for x in ast.walk(h)), + 'the interrupt path must re-raise, or Ctrl-C is swallowed') + + if __name__ == '__main__': unittest.main() |
