summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
authorhathach <[email protected]>2026-08-27 01:01:30 +0700
committerhathach <[email protected]>2026-08-27 08:54:07 +0700
commit9baa97a8c6cc671aefe26168b3d8281da070ebf9 (patch)
tree4adb5cea77de2001fdcfd2e0479abdfa967d7a83
parentd799b6f572e4b39ebebcf72126d3120f7829c034 (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.py35
-rwxr-xr-xtest/hil/hil_test.py7
-rw-r--r--test/hil/test/test_hil_bounded.py6
-rw-r--r--test/hil/test/test_hil_util.py54
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()