diff --git a/.github/workflows/v3-cleanup-diagnostic.yml b/.github/workflows/v3-cleanup-diagnostic.yml new file mode 100644 index 000000000..71c8d8007 --- /dev/null +++ b/.github/workflows/v3-cleanup-diagnostic.yml @@ -0,0 +1,271 @@ +name: v3 cleanup diagnostic +on: + pull_request: + branches: [v3] + paths: ['.github/workflows/v3-cleanup-diagnostic.yml', 'scripts/lib/package_commands.py', 'scripts/test-v3-installed-consumer.py'] +permissions: + contents: read +jobs: + macos-cleanup: + runs-on: macos-14 + timeout-minutes: 5 + steps: + - uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 + - name: Match Python prerequisite + run: brew install python + - name: Run native command contracts + timeout-minutes: 2 + shell: bash + run: | + mkdir -p "$RUNNER_TEMP/v3-cleanup-contracts" + cat > "$RUNNER_TEMP/v3-cleanup-command-contracts.py" <<'PYTHON' + import hashlib, importlib.util, json, os, subprocess, sys, unittest + from pathlib import Path + source = Path.cwd() + output = Path(os.environ['RUNNER_TEMP'], 'v3-cleanup-contracts') + paths = ('scripts/lib/package_commands.py', 'scripts/test-v3-installed-consumer.py') + receipt = dict(source_sha=subprocess.check_output(['git', 'rev-parse', 'HEAD'], text=True).strip(), + python=sys.version, hashes={p: hashlib.sha256((source / p).read_bytes()).hexdigest() for p in paths}) + (output / 'source.json').write_text(json.dumps(receipt, indent=2) + '\n') + spec = importlib.util.spec_from_file_location('native_contracts', source / paths[1]) + contracts = importlib.util.module_from_spec(spec) + spec.loader.exec_module(contracts) + snapshot = contracts.commands._process_snapshot + failures = [] + current_test = None + def observed_snapshot(deadline): + try: + return snapshot(deadline) + except Exception as error: + item = dict(test=current_test, error=str(error)) + frame = error.__traceback__ + while frame: + if frame.tb_frame.f_code.co_name == '_process_snapshot': + values = frame.tb_frame.f_locals + item.update(line=values.get('line'), fields=values.get('fields')) + buffer = values.get('output') + if buffer is not None: + data = buffer.getvalue() + path = output / ('snapshot-%04d.log' % len(failures)) + path.write_bytes(data) + item.update(log=path.name, bytes=len(data), sha256=hashlib.sha256(data).hexdigest()) + frame = frame.tb_next + failures.append(item) + (output / 'snapshot-failures.json').write_text(json.dumps(failures, indent=2) + '\n') + raise + # Retain parser evidence without replacing discovery, signals, or errors. + contracts.commands._process_snapshot = observed_snapshot + class Result(unittest.TextTestResult): + def startTest(self, test): + global current_test + current_test = test.id() + super().startTest(test) + suite = unittest.defaultTestLoader.loadTestsFromTestCase(contracts.CommandsTest) + expected = [test.id() for test in suite] + result = unittest.TextTestRunner(verbosity=2, resultclass=Result).run(suite) + report = dict(expected=expected, tests_run=result.testsRun, successful=result.wasSuccessful(), + failures=[test.id() for test, _ in result.failures], + errors=[test.id() for test, _ in result.errors], + skipped=[dict(test=test.id(), reason=reason) for test, reason in result.skipped]) + (output / 'result.json').write_text(json.dumps(report, indent=2) + '\n') + raise SystemExit(not result.wasSuccessful()) + PYTHON + set +e + python3 "$RUNNER_TEMP/v3-cleanup-command-contracts.py" 2>&1 | tee "$RUNNER_TEMP/v3-cleanup-contracts/commands-test.log" + status=${PIPESTATUS[0]} + printf '%s\n' "$status" > "$RUNNER_TEMP/v3-cleanup-contracts/exit-status" + exit "$status" + - name: Observe actual helper and native zombie signal semantics + if: always() + timeout-minutes: 2 + shell: bash + run: | + cat > "$RUNNER_TEMP/v3-cleanup-native-probe.py" <<'PY' + #!/usr/bin/env python3 + """Bounded observational probe: execute the checked-out helper unchanged.""" + import argparse + import importlib.util + import hashlib + import json + import os + from pathlib import Path + import platform + import select + import signal + import subprocess + import sys + import time + import traceback + + parser = argparse.ArgumentParser() + parser.add_argument('--source', type=Path, default=Path.cwd()) + parser.add_argument('--output', type=Path, required=True) + args = parser.parse_args() + source = args.source.resolve() + args.output.mkdir(parents=True, exist_ok=False) + sys.path.insert(0, str(source / 'scripts/lib')) + import package_commands as commands + spec = importlib.util.spec_from_file_location('audit', source / 'scripts/audit-v3-installed-package.py') + audit = importlib.util.module_from_spec(spec) + spec.loader.exec_module(audit) + + receipt = dict(source_sha=subprocess.check_output(['git', 'rev-parse', 'HEAD'], cwd=source, text=True).strip(), + hashes={p: hashlib.sha256((source / p).read_bytes()).hexdigest() for p in + ('scripts/lib/package_commands.py', 'scripts/test-v3-installed-consumer.py')}, + platform=platform.platform(), uname=list(platform.uname()), python=sys.version, + waitid=callable(getattr(os, 'waitid', None)), wnowait=hasattr(os, 'WNOWAIT'), + source=str(source), uid=os.getuid(), observations=[], commands=[], controls=[]) + + def state(tree, deadline): + members = commands._process_snapshot(deadline) + tree._check_leader(members.get(tree.child.pid)) + return dict(pid=tree.child.pid, parent=os.getpid(), returncode=tree.child.returncode, + anchor_exited=tree._anchor(deadline), + group={str(pid): value for pid, value in members.items() if value[1] == tree.child.pid}) + + original_kill_group = commands._CommandTree._kill_group + + def observed_kill_group(tree, deadline): + item = state(tree, deadline) + receipt['observations'].append(item) + try: + result = original_kill_group(tree, deadline) + item['signal'] = 'success' + return result + except BaseException as error: + item.update(signal=type(error).__name__, errno=getattr(error, 'errno', None), error=str(error)) + item['after'] = state(tree, deadline) + raise + + # Observation only; the original helper performs every signal and cleanup. + commands._CommandTree._kill_group = observed_kill_group + runner = commands.Commands(args.output) + for argv in (['git', 'branch', '--show-current'], [sys.executable, '-c', 'pass'], + [sys.executable, '-c', 'raise SystemExit(7)']): + item = dict(argv=argv) + try: + item['output'] = runner(argv, source, audit.clean_environment(), timeout=3, allow_failure=True) + item['result'] = 'success' + except Exception as error: + item.update(result=type(error).__name__, error=str(error), traceback=traceback.format_exc()) + receipt['commands'].append(item) + + # A fresh owned live leader is the contrasting permission control. + tree = commands._CommandTree([sys.executable, '-u', '-c', "import time; print('ready', flush=True); time.sleep(30)"], source, os.environ) + deadline = time.monotonic() + 5 + control = dict(kind='live-to-zombie') + try: + if not select.select([tree.child.stdout], [], [], 3)[0]: + raise RuntimeError('live control readiness deadline exceeded') + if tree.child.stdout.readline() != b'ready\n': + raise RuntimeError('live control did not become ready') + control['live'] = state(tree, deadline) + original_kill_group(tree, deadline) + control['live_sigkill'] = 'success' + while not tree.exited(deadline): + commands._remaining(deadline) + time.sleep(.001) + control['zombie'] = state(tree, deadline) + for label, operation in ( + ('zombie_pid_signal_zero', lambda: os.kill(tree.child.pid, 0)), + ('zombie_group_signal_zero', lambda: os.killpg(tree.child.pid, 0)), + ('zombie_helper_sigkill', lambda: original_kill_group(tree, deadline))): + tree._anchor(deadline) + try: + operation() + control[label] = dict(result='success') + except Exception as error: + control[label] = dict(result=type(error).__name__, errno=getattr(error, 'errno', None), error=str(error)) + control['after'] = state(tree, deadline) + except Exception as error: + control.update(error=str(error), traceback=traceback.format_exc()) + finally: + # Only our unreaped direct child; never signal after wait/reap. + if tree.child.returncode is None: + tree.child.kill() + tree.child.wait(timeout=2) + tree.child.stdout.close() + receipt['controls'].append(control) + (args.output / 'probe.json').write_text(json.dumps(receipt, indent=2) + '\n') + print(json.dumps(receipt, indent=2), flush=True) + PY + python3 "$RUNNER_TEMP/v3-cleanup-native-probe.py" --source "$GITHUB_WORKSPACE" --output "$RUNNER_TEMP/v3-cleanup-diagnostic" + - name: Require source-bound native cleanup proof + if: always() + shell: bash + run: | + python3 - <<'PY' + import hashlib, json, os, subprocess + from pathlib import Path + root = Path(os.environ['RUNNER_TEMP']) + output = root / 'v3-cleanup-diagnostic' + receipt = json.loads((output / 'probe.json').read_text()) + contracts = root / 'v3-cleanup-contracts' + binding = json.loads((contracts / 'source.json').read_text()) + source_sha = subprocess.check_output(['git', 'rev-parse', 'HEAD'], text=True).strip() + hashes = {p: hashlib.sha256(Path(p).read_bytes()).hexdigest() for p in + ('scripts/lib/package_commands.py', 'scripts/test-v3-installed-consumer.py')} + assert receipt['source_sha'] == binding['source_sha'] == source_sha == os.environ['GITHUB_SHA'] + assert receipt['hashes'] == binding['hashes'] == hashes + assert receipt['python'] == binding['python'] + assert (contracts / 'exit-status').read_text().strip() == '0' + result = json.loads((contracts / 'result.json').read_text()) + assert result['successful'] and not result['failures'] and not result['errors'] + assert result['tests_run'] == len(result['expected']) == 26 + assert len(set(result['expected'])) == 26 + assert { + 'native_contracts.CommandsTest.test_ps_unknown_state_preserves_complete_numeric_membership', + 'native_contracts.CommandsTest.test_ps_unknown_owned_member_cannot_justify_permission_denial', + 'native_contracts.CommandsTest.test_ps_unknown_state_does_not_relax_snapshot_validation', + 'native_contracts.CommandsTest.test_ps_unknown_state_cannot_hide_missing_or_changed_leader' + } <= set(result['expected']) + expected_skips = { + 'native_contracts.CommandsTest.test_cleanup_discovery_failure_retains_original_error', + 'native_contracts.CommandsTest.test_new_live_member_delays_retirement_and_zombies_do_not', + 'native_contracts.CommandsTest.test_discovery_consumes_cleanup_deadline_without_restarting_wait'} + assert len(result['skipped']) == 3 + assert {item['test'] for item in result['skipped']} == expected_skips + assert receipt['uname'][0] == 'Darwin' + assert len(receipt['commands']) == 3 + assert all(item['result'] == 'success' for item in receipt['commands']) + records = json.loads((output / 'commands.json').read_text()) + assert len(records) == 3 and [row['exit'] for row in records] == [0, 0, 7] + for item, record in zip(receipt['commands'], records): + assert 'error' not in record and item['argv'] == record['argv'] + data = Path(record['log']).read_bytes() + assert len(data) == record['bytes'] + assert hashlib.sha256(data).hexdigest() == record['sha256'] + assert item['output'] == data.decode(errors='replace') + def owned_zombie(item): + assert item['returncode'] is None and item['anchor_exited'] is True + assert item['group'] == {str(item['pid']): [item['parent'], item['pid'], 'Z']} + assert len(receipt['observations']) == 3 + for item in receipt['observations']: + owned_zombie(item) + assert item['signal'] == 'success' + assert len(receipt['controls']) == 1 + control = receipt['controls'][0] + assert 'error' not in control and control['kind'] == 'live-to-zombie' + live = control['live'] + assert live['returncode'] is None and live['anchor_exited'] is False + assert live['group'][str(live['pid'])][:2] == [live['parent'], live['pid']] + assert live['group'][str(live['pid'])][2] != 'Z' + assert control['live_sigkill'] == 'success' + owned_zombie(control['zombie']) + owned_zombie(control['after']) + assert live['pid'] == control['zombie']['pid'] == control['after']['pid'] + assert control['zombie_pid_signal_zero']['result'] == 'success' + assert control['zombie_group_signal_zero']['result'] == 'PermissionError' + assert control['zombie_group_signal_zero']['errno'] == 1 + assert control['zombie_helper_sigkill']['result'] == 'success' + (output / 'semantic-proof.json').write_text(json.dumps(dict(status='passed', source_sha=source_sha, + hashes=hashes, command_exits=[row['exit'] for row in records], native_contracts='passed'), indent=2) + '\n') + PY + - if: always() + uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02 + with: + name: v3-cleanup-diagnostic-${{ github.sha }} + path: | + ${{ runner.temp }}/v3-cleanup-diagnostic + ${{ runner.temp }}/v3-cleanup-contracts + if-no-files-found: error diff --git a/scripts/lib/package_commands.py b/scripts/lib/package_commands.py index 552f4cea0..e75901ded 100644 --- a/scripts/lib/package_commands.py +++ b/scripts/lib/package_commands.py @@ -158,6 +158,16 @@ def _kill_group(self, deadline): os.killpg(self.child.pid, signal.SIGKILL) except ProcessLookupError: pass # The snapshot must still prove that no live member remains. + except PermissionError: + # Darwin excludes zombies from group signals and can return EPERM. + self._anchor(deadline) + members = _process_snapshot(deadline) + self._check_leader(members.get(self.child.pid)) + self._anchor(deadline) + _remaining(deadline) + if any(group == self.child.pid and state != 'Z' + for _, group, state in members.values()): + raise def terminate(self, deadline=None): if self.job: @@ -251,8 +261,9 @@ def _process_snapshot(deadline): for line in output.getvalue().decode('ascii').splitlines(): _remaining(deadline) fields = line.split() + # Darwin can report unknown Mach state '?'; retain it as non-retired. if (len(fields) != 4 or any(not item.isdigit() for item in fields[:3]) - or not fields[3] or fields[3][0] not in 'RSDZTWtXxKIPLU' + or not fields[3] or fields[3][0] not in 'RSDZTWtXxKIPLU?' or any(flag not in '+<>ACEJLNSsVWX' for flag in fields[3][1:])): raise RuntimeError('malformed command process snapshot') pid, parent, group = map(int, fields[:3]) diff --git a/scripts/test-v3-installed-consumer.py b/scripts/test-v3-installed-consumer.py index f39c91c1e..21601def9 100644 --- a/scripts/test-v3-installed-consumer.py +++ b/scripts/test-v3-installed-consumer.py @@ -285,6 +285,114 @@ def test_success_and_allowed_failure_keep_output_and_hash_receipts(self): self.assertEqual(len(data), record['bytes']) self.assertEqual(hashlib.sha256(data).hexdigest(), record['sha256']) + @unittest.skipUnless(os.name == 'posix', 'POSIX zombie group permissions') + def test_owned_zombie_permission_denial_preserves_completed_receipts(self): + snapshot = commands._process_snapshot + scans = [] + denied = [] + def observe(deadline): + members = snapshot(deadline) + scans.append(members) + return members + def signal_group(pid, sig): + members = snapshot(time.monotonic() + 2) + leader = members.get(pid) + self.assertEqual((os.getpid(), pid, 'Z'), leader) + self.assertTrue(all(state == 'Z' for _, group, state in members.values() if group == pid)) + denied.append(pid) + raise PermissionError(1, 'zombie group has no signal targets') + for fallback in ('native', 'no-waitid', 'ps'): + for status in (0, 7): + with self.subTest(fallback=fallback, status=status): + scans.clear() + previous_denials = len(denied) + waitid = getattr(os, 'waitid', None) if fallback == 'native' else None + platform = 'darwin' if fallback == 'ps' else sys.platform + with patch.object(commands, '_process_snapshot', observe), patch.object(os, 'killpg', signal_group): + with patch.object(os, 'waitid', waitid, create=True), patch.object(sys, 'platform', platform): + output = self.runner(self.command(f"print('completed'); raise SystemExit({status})"), + self.root, os.environ, timeout=2, allow_failure=True) + self.assertEqual('completed\n', output) + record = json.loads((self.root / 'commands.json').read_text())[-1] + data = Path(record['log']).read_bytes() + self.assertEqual(status, record['exit']) + self.assertEqual(len(data), record['bytes']) + self.assertEqual(hashlib.sha256(data).hexdigest(), record['sha256']) + self.assertNotIn('error', record) + self.assertGreater(len(denied), previous_denials, 'each case must reach owned-zombie EPERM') + self.assertGreaterEqual(len(scans), 3, 'permission proof must retain stable retirement scans') + + def zombie_tree(self): + tree = commands._CommandTree(self.command('pass'), self.root, os.environ) + self.addCleanup(tree.child.stdout.close) + self.addCleanup(lambda: tree.child.wait(timeout=2)) + deadline = time.monotonic() + 2 + while not tree.exited(deadline): + self.assertLess(time.monotonic(), deadline) + time.sleep(.01) + self.assertIsNone(tree.child.returncode) + return tree + + @unittest.skipUnless(os.name == 'posix', 'POSIX live group permissions') + def test_live_group_permission_denial_fails_closed(self): + tree = commands._CommandTree(self.command('import time; time.sleep(30)'), + self.root, os.environ) + try: + self.assertFalse(tree.exited(time.monotonic() + 2)) + with patch.object(os, 'killpg', side_effect=PermissionError(1, 'live group denied')): + with self.assertRaisesRegex(PermissionError, 'live group denied'): + tree._kill_group(time.monotonic() + 2) + self.assertFalse(tree.exited(time.monotonic() + 2)) + finally: + tree.child.kill(); tree.child.wait(timeout=2); tree.child.stdout.close() + + @unittest.skipUnless(os.name == 'posix', 'POSIX permission evidence boundary') + def test_zombie_permission_denial_requires_complete_owned_retirement_evidence(self): + failures = dict(unavailable='unavailable', malformed='malformed', unknown='unverified group denied', + live='unverified group denied', missing='ownership', changed='ownership', + reaped='ownership', deadline='cleanup deadline exceeded') + for failure, message in failures.items(): + with self.subTest(failure=failure): + tree = self.zombie_tree() + snapshot = commands._process_snapshot + pid = tree.child.pid + elapsed = 0 + def evidence(deadline): + nonlocal elapsed + members = snapshot(deadline) + if failure in ('unavailable', 'malformed'): + raise RuntimeError(failure + ' command process snapshot') + if failure in ('unknown', 'live'): + members[pid + 1] = (pid, pid, '?' if failure == 'unknown' else 'S') + if failure == 'missing': del members[pid] + if failure == 'changed': members[pid] = (os.getpid() + 1, pid, 'Z') + if failure == 'reaped': tree.child.wait(timeout=2) + if failure == 'deadline': elapsed = 2 + return members + # The initial ownership check is real; corrupt only the post-denial proof. + def deny(pid, sig): + commands._process_snapshot = evidence + raise PermissionError(1, 'unverified group denied') + clock = (lambda: elapsed) if failure == 'deadline' else time.monotonic + with patch.object(commands, '_process_snapshot', snapshot), patch.object(os, 'killpg', side_effect=deny) as kill: + with patch.object(commands.time, 'monotonic', clock): + with self.assertRaisesRegex((PermissionError, RuntimeError), message): + tree._kill_group(1 if failure == 'deadline' else time.monotonic() + 2) + if failure == 'reaped': + self.assertEqual(0, tree.child.returncode) + with self.assertRaisesRegex(RuntimeError, 'ownership'): + tree._kill_group(time.monotonic() + 2) + self.assertEqual(1, kill.call_count, 'lost ownership must prevent another signal') + + @unittest.skipUnless(os.name == 'posix', 'POSIX cleanup error context') + def test_live_permission_failure_retains_original_command_timeout(self): + with patch.object(os, 'killpg', side_effect=PermissionError(1, 'live group denied')): + with self.assertRaisesRegex(RuntimeError, 'cleanup.*live group denied.*original command failure:.*deadline/output'): + self.runner(self.command('import time; time.sleep(.5)'), self.root, os.environ, timeout=.01) + receipt = json.loads((self.root / 'commands.json').read_text())[-1] + self.assertIn('original command failure:', receipt['error']) + self.assertNotIn('sha256', receipt) + def assert_success_retires_socket_owner(self): ready = self.root / 'descendant.json' worker = ("import json,os,socket,time; from pathlib import Path; " @@ -399,6 +507,61 @@ def launch(argv, *args, **kwargs): with self.assertRaisesRegex(RuntimeError, 'cleanup.*snapshot'): self.runner(self.command('pass'), self.root, os.environ, timeout=2) + def ps_output(self, data): + popen = subprocess.Popen + def launch(argv, *args, **kwargs): + if argv[0] in ('/bin/ps', '/usr/bin/ps'): + argv = self.command(f'import os; os.write(1, {data.encode("ascii")!r})') + return popen(argv, *args, **kwargs) + return patch.object(subprocess, 'Popen', launch) + + @unittest.skipUnless(os.name == 'posix', 'Darwin unknown scheduling state') + def test_ps_unknown_state_preserves_complete_numeric_membership(self): + # Captured native rows: the unknown process belongs to a different group. + data = ' 2255 2178 2255 Z< \n 2258 910 892 ?< \n' + with patch.object(sys, 'platform', 'darwin'), self.ps_output(data): + members = commands._process_snapshot(time.monotonic() + 2) + self.assertEqual({2255: (2178, 2255, 'Z'), 2258: (910, 892, '?')}, members) + + @unittest.skipUnless(os.name == 'posix', 'unknown owned process cannot prove retirement') + def test_ps_unknown_owned_member_cannot_justify_permission_denial(self): + for unknown in ('leader', 'descendant'): + with self.subTest(unknown=unknown): + tree = self.zombie_tree() + pid = tree.child.pid + data = f'{pid} {os.getpid()} {pid} ' + ('?<' if unknown == 'leader' else 'Z') + '\n' + if unknown == 'descendant': + data += f'{pid + 1} {pid} {pid} ?<\n' + with patch.object(sys, 'platform', 'darwin'), patch.object(os, 'waitid', None, create=True), self.ps_output(data): + with patch.object(os, 'killpg', side_effect=PermissionError(1, 'unknown group denied')): + with self.assertRaisesRegex(PermissionError, 'unknown group denied'): + tree._kill_group(time.monotonic() + 2) + self.assertIsNone(tree.child.returncode, 'unknown state cannot reap the ownership anchor') + + @unittest.skipUnless(os.name == 'posix', 'strict ps snapshot structure') + def test_ps_unknown_state_does_not_relax_snapshot_validation(self): + for data in ('pid 910 892 ?<', '2258 parent 892 ?<', '2258 910 group ?<', + '-1 910 892 ?<', '0 910 892 ?<', '2258 -1 892 ?<', '2258 910 -1 ?<', + '2258 910 ?<', '2258 910 892 ?< command', '2258 910 892 ?Q', + '2258 910 892 !<', '2258 910 892 ?<\n2258 910 892 Z'): + with self.subTest(data=data): + with patch.object(sys, 'platform', 'darwin'), self.ps_output(data + '\n'): + with self.assertRaisesRegex(RuntimeError, 'snapshot'): + commands._process_snapshot(time.monotonic() + 2) + + @unittest.skipUnless(os.name == 'posix', 'strict ps ownership anchor') + def test_ps_unknown_state_cannot_hide_missing_or_changed_leader(self): + tree = self.zombie_tree() + pid = tree.child.pid + for data in (f'{pid + 1} {pid} {pid} ?<', + f'{pid} {os.getpid() + 1} {pid} ?<', f'{pid} {os.getpid()} 0 ?<'): + with self.subTest(data=data): + with patch.object(sys, 'platform', 'darwin'), patch.object(os, 'waitid', None, create=True), self.ps_output(data + '\n'): + with patch.object(os, 'killpg') as kill: + with self.assertRaisesRegex(RuntimeError, 'ownership'): + tree._kill_group(time.monotonic() + 2) + kill.assert_not_called() + @unittest.skipUnless(os.name == 'posix', 'documented BSD/macOS state modifiers') def test_ps_fallback_accepts_documented_state_modifiers(self): popen = subprocess.Popen @@ -406,7 +569,7 @@ def launch(argv, *args, **kwargs): if argv[0] in ('/bin/ps', '/usr/bin/ps'): code = ("import subprocess; " f"data=subprocess.check_output({argv!r}).decode('ascii'); " - "print('\\n'.join(line + '>ACJ' for line in data.splitlines()))") + "print('\\n'.join(line.rstrip() + '>ACJ' for line in data.splitlines()))") argv = self.command(code) return popen(argv, *args, **kwargs) with patch.object(sys, 'platform', 'darwin'), patch.object(os, 'waitid', None, create=True), patch.object(subprocess, 'Popen', launch):