diff options
| -rw-r--r-- | test/hil/helper/hil_util.py | 33 | ||||
| -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 | 1 |
4 files changed, 16 insertions, 31 deletions
diff --git a/test/hil/helper/hil_util.py b/test/hil/helper/hil_util.py index dfd37467a..a6074d1c2 100644 --- a/test/hil/helper/hil_util.py +++ b/test/hil/helper/hil_util.py @@ -516,26 +516,22 @@ def run_cmd(cmd: str, 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: @@ -571,11 +567,10 @@ def run_cmd(cmd: str, 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 + try: + os.killpg(p.pid, signal.SIGKILL) + except OSError: + pass else: p.kill() _close_pipes(p) diff --git a/test/hil/hil_test.py b/test/hil/hil_test.py index 140860677..15857aa6f 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 @@ -1242,9 +1242,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 2230ef422..5d6296400 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..4d8b6119f 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') |