| import os |
| import signal |
| import sys |
| import textwrap |
| import unittest |
| import warnings |
| from unittest import mock |
| |
| import asyncio |
| from asyncio import base_subprocess |
| from asyncio import subprocess |
| from test.test_asyncio import utils as test_utils |
| from test import support |
| from test.support import os_helper |
| |
| |
| if support.MS_WINDOWS: |
| import msvcrt |
| else: |
| from asyncio import unix_events |
| |
| |
| if support.check_sanitizer(address=True): |
| raise unittest.SkipTest("Exposes ASAN flakiness in GitHub CI") |
| |
| # Program blocking |
| PROGRAM_BLOCKED = [sys.executable, '-c', 'import time; time.sleep(3600)'] |
| |
| # Program copying input to output |
| PROGRAM_CAT = [ |
| sys.executable, '-c', |
| ';'.join(('import sys', |
| 'data = sys.stdin.buffer.read()', |
| 'sys.stdout.buffer.write(data)'))] |
| |
| |
| def tearDownModule(): |
| asyncio.set_event_loop_policy(None) |
| |
| |
| class TestSubprocessTransport(base_subprocess.BaseSubprocessTransport): |
| def _start(self, *args, **kwargs): |
| self._proc = mock.Mock() |
| self._proc.stdin = None |
| self._proc.stdout = None |
| self._proc.stderr = None |
| self._proc.pid = -1 |
| |
| |
| @support.requires_subprocess() |
| class SubprocessTransportTests(test_utils.TestCase): |
| def setUp(self): |
| super().setUp() |
| self.loop = self.new_test_loop() |
| self.set_event_loop(self.loop) |
| |
| def create_transport(self, waiter=None): |
| protocol = mock.Mock() |
| transport = TestSubprocessTransport( |
| self.loop, protocol, ['test'], False, |
| None, None, None, 0, waiter=waiter) |
| return (transport, protocol) |
| |
| def test_proc_exited(self): |
| waiter = self.loop.create_future() |
| transport, protocol = self.create_transport(waiter) |
| transport._process_exited(6) |
| self.loop.run_until_complete(waiter) |
| |
| self.assertEqual(transport.get_returncode(), 6) |
| |
| self.assertTrue(protocol.connection_made.called) |
| self.assertTrue(protocol.process_exited.called) |
| self.assertTrue(protocol.connection_lost.called) |
| self.assertEqual(protocol.connection_lost.call_args[0], (None,)) |
| |
| self.assertFalse(transport.is_closing()) |
| self.assertIsNone(transport._loop) |
| self.assertIsNone(transport._proc) |
| self.assertIsNone(transport._protocol) |
| |
| # methods must raise ProcessLookupError if the process exited |
| self.assertRaises(ProcessLookupError, |
| transport.send_signal, signal.SIGTERM) |
| self.assertRaises(ProcessLookupError, transport.terminate) |
| self.assertRaises(ProcessLookupError, transport.kill) |
| |
| transport.close() |
| |
| def test_subprocess_repr(self): |
| waiter = self.loop.create_future() |
| transport, protocol = self.create_transport(waiter) |
| transport._process_exited(6) |
| self.loop.run_until_complete(waiter) |
| |
| self.assertEqual( |
| repr(transport), |
| "<TestSubprocessTransport pid=-1 returncode=6>" |
| ) |
| transport._returncode = None |
| self.assertEqual( |
| repr(transport), |
| "<TestSubprocessTransport pid=-1 running>" |
| ) |
| transport._pid = None |
| transport._returncode = None |
| self.assertEqual( |
| repr(transport), |
| "<TestSubprocessTransport not started>" |
| ) |
| transport.close() |
| |
| |
| @support.requires_subprocess() |
| class SubprocessMixin: |
| |
| def test_stdin_stdout(self): |
| args = PROGRAM_CAT |
| |
| async def run(data): |
| proc = await asyncio.create_subprocess_exec( |
| *args, |
| stdin=subprocess.PIPE, |
| stdout=subprocess.PIPE, |
| ) |
| |
| # feed data |
| proc.stdin.write(data) |
| await proc.stdin.drain() |
| proc.stdin.close() |
| |
| # get output and exitcode |
| data = await proc.stdout.read() |
| exitcode = await proc.wait() |
| return (exitcode, data) |
| |
| task = run(b'some data') |
| task = asyncio.wait_for(task, 60.0) |
| exitcode, stdout = self.loop.run_until_complete(task) |
| self.assertEqual(exitcode, 0) |
| self.assertEqual(stdout, b'some data') |
| |
| def test_communicate(self): |
| args = PROGRAM_CAT |
| |
| async def run(data): |
| proc = await asyncio.create_subprocess_exec( |
| *args, |
| stdin=subprocess.PIPE, |
| stdout=subprocess.PIPE, |
| ) |
| stdout, stderr = await proc.communicate(data) |
| return proc.returncode, stdout |
| |
| task = run(b'some data') |
| task = asyncio.wait_for(task, support.LONG_TIMEOUT) |
| exitcode, stdout = self.loop.run_until_complete(task) |
| self.assertEqual(exitcode, 0) |
| self.assertEqual(stdout, b'some data') |
| |
| def test_communicate_none_input(self): |
| args = PROGRAM_CAT |
| |
| async def run(): |
| proc = await asyncio.create_subprocess_exec( |
| *args, |
| stdin=subprocess.PIPE, |
| stdout=subprocess.PIPE, |
| ) |
| stdout, stderr = await proc.communicate() |
| return proc.returncode, stdout |
| |
| task = run() |
| task = asyncio.wait_for(task, support.LONG_TIMEOUT) |
| exitcode, stdout = self.loop.run_until_complete(task) |
| self.assertEqual(exitcode, 0) |
| self.assertEqual(stdout, b'') |
| |
| def test_shell(self): |
| proc = self.loop.run_until_complete( |
| asyncio.create_subprocess_shell('exit 7') |
| ) |
| exitcode = self.loop.run_until_complete(proc.wait()) |
| self.assertEqual(exitcode, 7) |
| |
| def test_start_new_session(self): |
| # start the new process in a new session |
| proc = self.loop.run_until_complete( |
| asyncio.create_subprocess_shell( |
| 'exit 8', |
| start_new_session=True, |
| ) |
| ) |
| exitcode = self.loop.run_until_complete(proc.wait()) |
| self.assertEqual(exitcode, 8) |
| |
| def test_kill(self): |
| args = PROGRAM_BLOCKED |
| proc = self.loop.run_until_complete( |
| asyncio.create_subprocess_exec(*args) |
| ) |
| proc.kill() |
| returncode = self.loop.run_until_complete(proc.wait()) |
| if sys.platform == 'win32': |
| self.assertIsInstance(returncode, int) |
| # expect 1 but sometimes get 0 |
| else: |
| self.assertEqual(-signal.SIGKILL, returncode) |
| |
| def test_kill_issue43884(self): |
| if sys.platform == 'win32': |
| blocking_shell_command = f'"{sys.executable}" -c "import time; time.sleep(2)"' |
| else: |
| blocking_shell_command = 'sleep 1; sleep 1' |
| creationflags = 0 |
| if sys.platform == 'win32': |
| from subprocess import CREATE_NEW_PROCESS_GROUP |
| # On windows create a new process group so that killing process |
| # kills the process and all its children. |
| creationflags = CREATE_NEW_PROCESS_GROUP |
| proc = self.loop.run_until_complete( |
| asyncio.create_subprocess_shell(blocking_shell_command, stdout=asyncio.subprocess.PIPE, |
| creationflags=creationflags) |
| ) |
| self.loop.run_until_complete(asyncio.sleep(1)) |
| if sys.platform == 'win32': |
| proc.send_signal(signal.CTRL_BREAK_EVENT) |
| # On windows it is an alias of terminate which sets the return code |
| proc.kill() |
| returncode = self.loop.run_until_complete(proc.wait()) |
| if sys.platform == 'win32': |
| self.assertIsInstance(returncode, int) |
| # expect 1 but sometimes get 0 |
| else: |
| self.assertEqual(-signal.SIGKILL, returncode) |
| |
| def test_terminate(self): |
| args = PROGRAM_BLOCKED |
| proc = self.loop.run_until_complete( |
| asyncio.create_subprocess_exec(*args) |
| ) |
| proc.terminate() |
| returncode = self.loop.run_until_complete(proc.wait()) |
| if sys.platform == 'win32': |
| self.assertIsInstance(returncode, int) |
| # expect 1 but sometimes get 0 |
| else: |
| self.assertEqual(-signal.SIGTERM, returncode) |
| |
| @unittest.skipIf(sys.platform == 'win32', "Don't have SIGHUP") |
| def test_send_signal(self): |
| # bpo-31034: Make sure that we get the default signal handler (killing |
| # the process). The parent process may have decided to ignore SIGHUP, |
| # and signal handlers are inherited. |
| old_handler = signal.signal(signal.SIGHUP, signal.SIG_DFL) |
| try: |
| code = 'import time; print("sleeping", flush=True); time.sleep(3600)' |
| args = [sys.executable, '-c', code] |
| proc = self.loop.run_until_complete( |
| asyncio.create_subprocess_exec( |
| *args, |
| stdout=subprocess.PIPE, |
| ) |
| ) |
| |
| async def send_signal(proc): |
| # basic synchronization to wait until the program is sleeping |
| line = await proc.stdout.readline() |
| self.assertEqual(line, b'sleeping\n') |
| |
| proc.send_signal(signal.SIGHUP) |
| returncode = await proc.wait() |
| return returncode |
| |
| returncode = self.loop.run_until_complete(send_signal(proc)) |
| self.assertEqual(-signal.SIGHUP, returncode) |
| finally: |
| signal.signal(signal.SIGHUP, old_handler) |
| |
| def test_stdin_broken_pipe(self): |
| # buffer large enough to feed the whole pipe buffer |
| large_data = b'x' * support.PIPE_MAX_SIZE |
| |
| rfd, wfd = os.pipe() |
| self.addCleanup(os.close, rfd) |
| self.addCleanup(os.close, wfd) |
| if support.MS_WINDOWS: |
| handle = msvcrt.get_osfhandle(rfd) |
| os.set_handle_inheritable(handle, True) |
| code = textwrap.dedent(f''' |
| import os, msvcrt |
| handle = {handle} |
| fd = msvcrt.open_osfhandle(handle, os.O_RDONLY) |
| os.read(fd, 1) |
| ''') |
| from subprocess import STARTUPINFO |
| startupinfo = STARTUPINFO() |
| startupinfo.lpAttributeList = {"handle_list": [handle]} |
| kwargs = dict(startupinfo=startupinfo) |
| else: |
| code = f'import os; fd = {rfd}; os.read(fd, 1)' |
| kwargs = dict(pass_fds=(rfd,)) |
| |
| # the program ends before the stdin can be fed |
| proc = self.loop.run_until_complete( |
| asyncio.create_subprocess_exec( |
| sys.executable, '-c', code, |
| stdin=subprocess.PIPE, |
| **kwargs |
| ) |
| ) |
| |
| async def write_stdin(proc, data): |
| proc.stdin.write(data) |
| # Only exit the child process once the write buffer is filled |
| os.write(wfd, b'go') |
| await proc.stdin.drain() |
| |
| coro = write_stdin(proc, large_data) |
| # drain() must raise BrokenPipeError or ConnectionResetError |
| with test_utils.disable_logger(): |
| self.assertRaises((BrokenPipeError, ConnectionResetError), |
| self.loop.run_until_complete, coro) |
| self.loop.run_until_complete(proc.wait()) |
| |
| def test_communicate_ignore_broken_pipe(self): |
| # buffer large enough to feed the whole pipe buffer |
| large_data = b'x' * support.PIPE_MAX_SIZE |
| |
| # the program ends before the stdin can be fed |
| proc = self.loop.run_until_complete( |
| asyncio.create_subprocess_exec( |
| sys.executable, '-c', 'pass', |
| stdin=subprocess.PIPE, |
| ) |
| ) |
| |
| # communicate() must ignore BrokenPipeError when feeding stdin |
| self.loop.set_exception_handler(lambda loop, msg: None) |
| self.loop.run_until_complete(proc.communicate(large_data)) |
| self.loop.run_until_complete(proc.wait()) |
| |
| def test_pause_reading(self): |
| limit = 10 |
| size = (limit * 2 + 1) |
| |
| async def test_pause_reading(): |
| code = '\n'.join(( |
| 'import sys', |
| 'sys.stdout.write("x" * %s)' % size, |
| 'sys.stdout.flush()', |
| )) |
| |
| connect_read_pipe = self.loop.connect_read_pipe |
| |
| async def connect_read_pipe_mock(*args, **kw): |
| transport, protocol = await connect_read_pipe(*args, **kw) |
| transport.pause_reading = mock.Mock() |
| transport.resume_reading = mock.Mock() |
| return (transport, protocol) |
| |
| self.loop.connect_read_pipe = connect_read_pipe_mock |
| |
| proc = await asyncio.create_subprocess_exec( |
| sys.executable, '-c', code, |
| stdin=asyncio.subprocess.PIPE, |
| stdout=asyncio.subprocess.PIPE, |
| limit=limit, |
| ) |
| stdout_transport = proc._transport.get_pipe_transport(1) |
| |
| stdout, stderr = await proc.communicate() |
| |
| # The child process produced more than limit bytes of output, |
| # the stream reader transport should pause the protocol to not |
| # allocate too much memory. |
| return (stdout, stdout_transport) |
| |
| # Issue #22685: Ensure that the stream reader pauses the protocol |
| # when the child process produces too much data |
| stdout, transport = self.loop.run_until_complete(test_pause_reading()) |
| |
| self.assertEqual(stdout, b'x' * size) |
| self.assertTrue(transport.pause_reading.called) |
| self.assertTrue(transport.resume_reading.called) |
| |
| def test_stdin_not_inheritable(self): |
| # asyncio issue #209: stdin must not be inheritable, otherwise |
| # the Process.communicate() hangs |
| async def len_message(message): |
| code = 'import sys; data = sys.stdin.read(); print(len(data))' |
| proc = await asyncio.create_subprocess_exec( |
| sys.executable, '-c', code, |
| stdin=asyncio.subprocess.PIPE, |
| stdout=asyncio.subprocess.PIPE, |
| stderr=asyncio.subprocess.PIPE, |
| close_fds=False, |
| ) |
| stdout, stderr = await proc.communicate(message) |
| exitcode = await proc.wait() |
| return (stdout, exitcode) |
| |
| output, exitcode = self.loop.run_until_complete(len_message(b'abc')) |
| self.assertEqual(output.rstrip(), b'3') |
| self.assertEqual(exitcode, 0) |
| |
| def test_empty_input(self): |
| |
| async def empty_input(): |
| code = 'import sys; data = sys.stdin.read(); print(len(data))' |
| proc = await asyncio.create_subprocess_exec( |
| sys.executable, '-c', code, |
| stdin=asyncio.subprocess.PIPE, |
| stdout=asyncio.subprocess.PIPE, |
| stderr=asyncio.subprocess.PIPE, |
| close_fds=False, |
| ) |
| stdout, stderr = await proc.communicate(b'') |
| exitcode = await proc.wait() |
| return (stdout, exitcode) |
| |
| output, exitcode = self.loop.run_until_complete(empty_input()) |
| self.assertEqual(output.rstrip(), b'0') |
| self.assertEqual(exitcode, 0) |
| |
| def test_devnull_input(self): |
| |
| async def empty_input(): |
| code = 'import sys; data = sys.stdin.read(); print(len(data))' |
| proc = await asyncio.create_subprocess_exec( |
| sys.executable, '-c', code, |
| stdin=asyncio.subprocess.DEVNULL, |
| stdout=asyncio.subprocess.PIPE, |
| stderr=asyncio.subprocess.PIPE, |
| close_fds=False, |
| ) |
| stdout, stderr = await proc.communicate() |
| exitcode = await proc.wait() |
| return (stdout, exitcode) |
| |
| output, exitcode = self.loop.run_until_complete(empty_input()) |
| self.assertEqual(output.rstrip(), b'0') |
| self.assertEqual(exitcode, 0) |
| |
| def test_devnull_output(self): |
| |
| async def empty_output(): |
| code = 'import sys; data = sys.stdin.read(); print(len(data))' |
| proc = await asyncio.create_subprocess_exec( |
| sys.executable, '-c', code, |
| stdin=asyncio.subprocess.PIPE, |
| stdout=asyncio.subprocess.DEVNULL, |
| stderr=asyncio.subprocess.PIPE, |
| close_fds=False, |
| ) |
| stdout, stderr = await proc.communicate(b"abc") |
| exitcode = await proc.wait() |
| return (stdout, exitcode) |
| |
| output, exitcode = self.loop.run_until_complete(empty_output()) |
| self.assertEqual(output, None) |
| self.assertEqual(exitcode, 0) |
| |
| def test_devnull_error(self): |
| |
| async def empty_error(): |
| code = 'import sys; data = sys.stdin.read(); print(len(data))' |
| proc = await asyncio.create_subprocess_exec( |
| sys.executable, '-c', code, |
| stdin=asyncio.subprocess.PIPE, |
| stdout=asyncio.subprocess.PIPE, |
| stderr=asyncio.subprocess.DEVNULL, |
| close_fds=False, |
| ) |
| stdout, stderr = await proc.communicate(b"abc") |
| exitcode = await proc.wait() |
| return (stderr, exitcode) |
| |
| output, exitcode = self.loop.run_until_complete(empty_error()) |
| self.assertEqual(output, None) |
| self.assertEqual(exitcode, 0) |
| |
| @unittest.skipIf(sys.platform != 'linux', "Don't have /dev/stdin") |
| def test_devstdin_input(self): |
| |
| async def devstdin_input(message): |
| code = 'file = open("/dev/stdin"); data = file.read(); print(len(data))' |
| proc = await asyncio.create_subprocess_exec( |
| sys.executable, '-c', code, |
| stdin=asyncio.subprocess.PIPE, |
| stdout=asyncio.subprocess.PIPE, |
| stderr=asyncio.subprocess.PIPE, |
| close_fds=False, |
| ) |
| stdout, stderr = await proc.communicate(message) |
| exitcode = await proc.wait() |
| return (stdout, exitcode) |
| |
| output, exitcode = self.loop.run_until_complete(devstdin_input(b'abc')) |
| self.assertEqual(output.rstrip(), b'3') |
| self.assertEqual(exitcode, 0) |
| |
| def test_cancel_process_wait(self): |
| # Issue #23140: cancel Process.wait() |
| |
| async def cancel_wait(): |
| proc = await asyncio.create_subprocess_exec(*PROGRAM_BLOCKED) |
| |
| # Create an internal future waiting on the process exit |
| task = self.loop.create_task(proc.wait()) |
| self.loop.call_soon(task.cancel) |
| try: |
| await task |
| except asyncio.CancelledError: |
| pass |
| |
| # Cancel the future |
| task.cancel() |
| |
| # Kill the process and wait until it is done |
| proc.kill() |
| await proc.wait() |
| |
| self.loop.run_until_complete(cancel_wait()) |
| |
| def test_cancel_make_subprocess_transport_exec(self): |
| |
| async def cancel_make_transport(): |
| coro = asyncio.create_subprocess_exec(*PROGRAM_BLOCKED) |
| task = self.loop.create_task(coro) |
| |
| self.loop.call_soon(task.cancel) |
| try: |
| await task |
| except asyncio.CancelledError: |
| pass |
| |
| # ignore the log: |
| # "Exception during subprocess creation, kill the subprocess" |
| with test_utils.disable_logger(): |
| self.loop.run_until_complete(cancel_make_transport()) |
| |
| def test_cancel_post_init(self): |
| |
| async def cancel_make_transport(): |
| coro = self.loop.subprocess_exec(asyncio.SubprocessProtocol, |
| *PROGRAM_BLOCKED) |
| task = self.loop.create_task(coro) |
| |
| self.loop.call_soon(task.cancel) |
| try: |
| await task |
| except asyncio.CancelledError: |
| pass |
| |
| # ignore the log: |
| # "Exception during subprocess creation, kill the subprocess" |
| with test_utils.disable_logger(): |
| self.loop.run_until_complete(cancel_make_transport()) |
| test_utils.run_briefly(self.loop) |
| |
| def test_close_kill_running(self): |
| |
| async def kill_running(): |
| create = self.loop.subprocess_exec(asyncio.SubprocessProtocol, |
| *PROGRAM_BLOCKED) |
| transport, protocol = await create |
| |
| kill_called = False |
| def kill(): |
| nonlocal kill_called |
| kill_called = True |
| orig_kill() |
| |
| proc = transport.get_extra_info('subprocess') |
| orig_kill = proc.kill |
| proc.kill = kill |
| returncode = transport.get_returncode() |
| transport.close() |
| await asyncio.wait_for(transport._wait(), 5) |
| return (returncode, kill_called) |
| |
| # Ignore "Close running child process: kill ..." log |
| with test_utils.disable_logger(): |
| try: |
| returncode, killed = self.loop.run_until_complete( |
| kill_running() |
| ) |
| except asyncio.TimeoutError: |
| self.skipTest( |
| "Timeout failure on waiting for subprocess stopping" |
| ) |
| self.assertIsNone(returncode) |
| |
| # transport.close() must kill the process if it is still running |
| self.assertTrue(killed) |
| test_utils.run_briefly(self.loop) |
| |
| def test_close_dont_kill_finished(self): |
| |
| async def kill_running(): |
| create = self.loop.subprocess_exec(asyncio.SubprocessProtocol, |
| *PROGRAM_BLOCKED) |
| transport, protocol = await create |
| proc = transport.get_extra_info('subprocess') |
| |
| # kill the process (but asyncio is not notified immediately) |
| proc.kill() |
| proc.wait() |
| |
| proc.kill = mock.Mock() |
| proc_returncode = proc.poll() |
| transport_returncode = transport.get_returncode() |
| transport.close() |
| return (proc_returncode, transport_returncode, proc.kill.called) |
| |
| # Ignore "Unknown child process pid ..." log of SafeChildWatcher, |
| # emitted because the test already consumes the exit status: |
| # proc.wait() |
| with test_utils.disable_logger(): |
| result = self.loop.run_until_complete(kill_running()) |
| test_utils.run_briefly(self.loop) |
| |
| proc_returncode, transport_return_code, killed = result |
| |
| self.assertIsNotNone(proc_returncode) |
| self.assertIsNone(transport_return_code) |
| |
| # transport.close() must not kill the process if it finished, even if |
| # the transport was not notified yet |
| self.assertFalse(killed) |
| |
| # Unlike SafeChildWatcher, FastChildWatcher does not pop the |
| # callbacks if waitpid() is called elsewhere. Let's clear them |
| # manually to avoid a warning when the watcher is detached. |
| if (sys.platform != 'win32' and |
| isinstance(self, SubprocessFastWatcherTests)): |
| with warnings.catch_warnings(): |
| warnings.simplefilter('ignore', DeprecationWarning) |
| asyncio.get_child_watcher()._callbacks.clear() |
| |
| async def _test_popen_error(self, stdin): |
| if sys.platform == 'win32': |
| target = 'asyncio.windows_utils.Popen' |
| else: |
| target = 'subprocess.Popen' |
| with mock.patch(target) as popen: |
| exc = ZeroDivisionError |
| popen.side_effect = exc |
| |
| with warnings.catch_warnings(record=True) as warns: |
| with self.assertRaises(exc): |
| await asyncio.create_subprocess_exec( |
| sys.executable, |
| '-c', |
| 'pass', |
| stdin=stdin |
| ) |
| self.assertEqual(warns, []) |
| |
| def test_popen_error(self): |
| # Issue #24763: check that the subprocess transport is closed |
| # when BaseSubprocessTransport fails |
| self.loop.run_until_complete(self._test_popen_error(stdin=None)) |
| |
| def test_popen_error_with_stdin_pipe(self): |
| # Issue #35721: check that newly created socket pair is closed when |
| # Popen fails |
| self.loop.run_until_complete( |
| self._test_popen_error(stdin=subprocess.PIPE)) |
| |
| def test_read_stdout_after_process_exit(self): |
| |
| async def execute(): |
| code = '\n'.join(['import sys', |
| 'for _ in range(64):', |
| ' sys.stdout.write("x" * 4096)', |
| 'sys.stdout.flush()', |
| 'sys.exit(1)']) |
| |
| process = await asyncio.create_subprocess_exec( |
| sys.executable, '-c', code, |
| stdout=asyncio.subprocess.PIPE, |
| ) |
| |
| while True: |
| data = await process.stdout.read(65536) |
| if data: |
| await asyncio.sleep(0.3) |
| else: |
| break |
| |
| self.loop.run_until_complete(execute()) |
| |
| def test_create_subprocess_exec_text_mode_fails(self): |
| async def execute(): |
| with self.assertRaises(ValueError): |
| await subprocess.create_subprocess_exec(sys.executable, |
| text=True) |
| |
| with self.assertRaises(ValueError): |
| await subprocess.create_subprocess_exec(sys.executable, |
| encoding="utf-8") |
| |
| with self.assertRaises(ValueError): |
| await subprocess.create_subprocess_exec(sys.executable, |
| errors="strict") |
| |
| self.loop.run_until_complete(execute()) |
| |
| def test_create_subprocess_shell_text_mode_fails(self): |
| |
| async def execute(): |
| with self.assertRaises(ValueError): |
| await subprocess.create_subprocess_shell(sys.executable, |
| text=True) |
| |
| with self.assertRaises(ValueError): |
| await subprocess.create_subprocess_shell(sys.executable, |
| encoding="utf-8") |
| |
| with self.assertRaises(ValueError): |
| await subprocess.create_subprocess_shell(sys.executable, |
| errors="strict") |
| |
| self.loop.run_until_complete(execute()) |
| |
| def test_create_subprocess_exec_with_path(self): |
| async def execute(): |
| p = await subprocess.create_subprocess_exec( |
| os_helper.FakePath(sys.executable), '-c', 'pass') |
| await p.wait() |
| p = await subprocess.create_subprocess_exec( |
| sys.executable, '-c', 'pass', os_helper.FakePath('.')) |
| await p.wait() |
| |
| self.assertIsNone(self.loop.run_until_complete(execute())) |
| |
| async def check_stdout_output(self, coro, output): |
| proc = await coro |
| stdout, _ = await proc.communicate() |
| self.assertEqual(stdout, output) |
| self.assertEqual(proc.returncode, 0) |
| task = asyncio.create_task(proc.wait()) |
| await asyncio.sleep(0) |
| self.assertEqual(task.result(), proc.returncode) |
| |
| def test_create_subprocess_env_shell(self) -> None: |
| async def main() -> None: |
| executable = sys.executable |
| if sys.platform == "win32": |
| executable = f'"{executable}"' |
| cmd = f'''{executable} -c "import os, sys; sys.stdout.write(os.getenv('FOO'))"''' |
| env = os.environ.copy() |
| env["FOO"] = "bar" |
| proc = await asyncio.create_subprocess_shell( |
| cmd, env=env, stdout=subprocess.PIPE |
| ) |
| return proc |
| |
| self.loop.run_until_complete(self.check_stdout_output(main(), b'bar')) |
| |
| def test_create_subprocess_env_exec(self) -> None: |
| async def main() -> None: |
| cmd = [sys.executable, "-c", |
| "import os, sys; sys.stdout.write(os.getenv('FOO'))"] |
| env = os.environ.copy() |
| env["FOO"] = "baz" |
| proc = await asyncio.create_subprocess_exec( |
| *cmd, env=env, stdout=subprocess.PIPE |
| ) |
| return proc |
| |
| self.loop.run_until_complete(self.check_stdout_output(main(), b'baz')) |
| |
| |
| def test_subprocess_concurrent_wait(self) -> None: |
| async def main() -> None: |
| proc = await asyncio.create_subprocess_exec( |
| *PROGRAM_CAT, |
| stdin=subprocess.PIPE, |
| stdout=subprocess.PIPE, |
| ) |
| stdout, _ = await proc.communicate(b'some data') |
| self.assertEqual(stdout, b"some data") |
| self.assertEqual(proc.returncode, 0) |
| self.assertEqual(await asyncio.gather(*[proc.wait() for _ in range(10)]), |
| [proc.returncode] * 10) |
| |
| self.loop.run_until_complete(main()) |
| |
| def test_subprocess_protocol_events(self): |
| # gh-108973: Test that all subprocess protocol methods are called. |
| # The protocol methods are not called in a determistic order. |
| # The order depends on the event loop and the operating system. |
| events = [] |
| fds = [1, 2] |
| expected = [ |
| ('pipe_data_received', 1, b'stdout'), |
| ('pipe_data_received', 2, b'stderr'), |
| ('pipe_connection_lost', 1), |
| ('pipe_connection_lost', 2), |
| 'process_exited', |
| ] |
| per_fd_expected = [ |
| 'pipe_data_received', |
| 'pipe_connection_lost', |
| ] |
| |
| class MyProtocol(asyncio.SubprocessProtocol): |
| def __init__(self, exit_future: asyncio.Future) -> None: |
| self.exit_future = exit_future |
| |
| def pipe_data_received(self, fd, data) -> None: |
| events.append(('pipe_data_received', fd, data)) |
| self.exit_maybe() |
| |
| def pipe_connection_lost(self, fd, exc) -> None: |
| events.append(('pipe_connection_lost', fd)) |
| self.exit_maybe() |
| |
| def process_exited(self) -> None: |
| events.append('process_exited') |
| self.exit_maybe() |
| |
| def exit_maybe(self): |
| # Only exit when we got all expected events |
| if len(events) >= len(expected): |
| self.exit_future.set_result(True) |
| |
| async def main() -> None: |
| loop = asyncio.get_running_loop() |
| exit_future = asyncio.Future() |
| code = 'import sys; sys.stdout.write("stdout"); sys.stderr.write("stderr")' |
| transport, _ = await loop.subprocess_exec(lambda: MyProtocol(exit_future), |
| sys.executable, '-c', code, stdin=None) |
| await exit_future |
| transport.close() |
| |
| return events |
| |
| events = self.loop.run_until_complete(main()) |
| |
| # First, make sure that we received all events |
| self.assertSetEqual(set(events), set(expected)) |
| |
| # Second, check order of pipe events per file descriptor |
| per_fd_events = {fd: [] for fd in fds} |
| for event in events: |
| if event == 'process_exited': |
| continue |
| name, fd = event[:2] |
| per_fd_events[fd].append(name) |
| |
| for fd in fds: |
| self.assertEqual(per_fd_events[fd], per_fd_expected, (fd, events)) |
| |
| def test_subprocess_communicate_stdout(self): |
| # See https://github.com/python/cpython/issues/100133 |
| async def get_command_stdout(cmd, *args): |
| proc = await asyncio.create_subprocess_exec( |
| cmd, *args, stdout=asyncio.subprocess.PIPE, |
| ) |
| stdout, _ = await proc.communicate() |
| return stdout.decode().strip() |
| |
| async def main(): |
| outputs = [f'foo{i}' for i in range(10)] |
| res = await asyncio.gather(*[get_command_stdout(sys.executable, '-c', |
| f'print({out!r})') for out in outputs]) |
| self.assertEqual(res, outputs) |
| |
| self.loop.run_until_complete(main()) |
| |
| |
| if sys.platform != 'win32': |
| # Unix |
| class SubprocessWatcherMixin(SubprocessMixin): |
| |
| Watcher = None |
| |
| def setUp(self): |
| super().setUp() |
| policy = asyncio.get_event_loop_policy() |
| self.loop = policy.new_event_loop() |
| self.set_event_loop(self.loop) |
| |
| watcher = self._get_watcher() |
| watcher.attach_loop(self.loop) |
| with warnings.catch_warnings(): |
| warnings.simplefilter('ignore', DeprecationWarning) |
| policy.set_child_watcher(watcher) |
| |
| def tearDown(self): |
| super().tearDown() |
| policy = asyncio.get_event_loop_policy() |
| with warnings.catch_warnings(): |
| warnings.simplefilter('ignore', DeprecationWarning) |
| watcher = policy.get_child_watcher() |
| policy.set_child_watcher(None) |
| watcher.attach_loop(None) |
| watcher.close() |
| |
| class SubprocessThreadedWatcherTests(SubprocessWatcherMixin, |
| test_utils.TestCase): |
| |
| def _get_watcher(self): |
| return unix_events.ThreadedChildWatcher() |
| |
| class SubprocessSafeWatcherTests(SubprocessWatcherMixin, |
| test_utils.TestCase): |
| |
| def _get_watcher(self): |
| with self.assertWarns(DeprecationWarning): |
| return unix_events.SafeChildWatcher() |
| |
| class MultiLoopChildWatcherTests(test_utils.TestCase): |
| |
| def test_warns(self): |
| with self.assertWarns(DeprecationWarning): |
| unix_events.MultiLoopChildWatcher() |
| |
| class SubprocessFastWatcherTests(SubprocessWatcherMixin, |
| test_utils.TestCase): |
| |
| def _get_watcher(self): |
| with self.assertWarns(DeprecationWarning): |
| return unix_events.FastChildWatcher() |
| |
| @unittest.skipUnless( |
| unix_events.can_use_pidfd(), |
| "operating system does not support pidfds", |
| ) |
| class SubprocessPidfdWatcherTests(SubprocessWatcherMixin, |
| test_utils.TestCase): |
| |
| def _get_watcher(self): |
| return unix_events.PidfdChildWatcher() |
| |
| |
| class GenericWatcherTests(test_utils.TestCase): |
| |
| def test_create_subprocess_fails_with_inactive_watcher(self): |
| watcher = mock.create_autospec(asyncio.AbstractChildWatcher) |
| watcher.is_active.return_value = False |
| |
| async def execute(): |
| asyncio.set_child_watcher(watcher) |
| |
| with self.assertRaises(RuntimeError): |
| await subprocess.create_subprocess_exec( |
| os_helper.FakePath(sys.executable), '-c', 'pass') |
| |
| watcher.add_child_handler.assert_not_called() |
| |
| with asyncio.Runner(loop_factory=asyncio.new_event_loop) as runner: |
| with warnings.catch_warnings(): |
| warnings.simplefilter('ignore', DeprecationWarning) |
| self.assertIsNone(runner.run(execute())) |
| self.assertListEqual(watcher.mock_calls, [ |
| mock.call.__enter__(), |
| mock.call.is_active(), |
| mock.call.__exit__(RuntimeError, mock.ANY, mock.ANY), |
| ], watcher.mock_calls) |
| |
| |
| @unittest.skipUnless( |
| unix_events.can_use_pidfd(), |
| "operating system does not support pidfds", |
| ) |
| def test_create_subprocess_with_pidfd(self): |
| async def in_thread(): |
| proc = await asyncio.create_subprocess_exec( |
| *PROGRAM_CAT, |
| stdin=subprocess.PIPE, |
| stdout=subprocess.PIPE, |
| ) |
| stdout, stderr = await proc.communicate(b"some data") |
| return proc.returncode, stdout |
| |
| async def main(): |
| # asyncio.Runner did not call asyncio.set_event_loop() |
| with warnings.catch_warnings(): |
| warnings.simplefilter('error', DeprecationWarning) |
| # get_event_loop() raises DeprecationWarning if |
| # set_event_loop() was never called and RuntimeError if |
| # it was called at least once. |
| with self.assertRaises((RuntimeError, DeprecationWarning)): |
| asyncio.get_event_loop_policy().get_event_loop() |
| return await asyncio.to_thread(asyncio.run, in_thread()) |
| with self.assertWarns(DeprecationWarning): |
| asyncio.set_child_watcher(asyncio.PidfdChildWatcher()) |
| try: |
| with asyncio.Runner(loop_factory=asyncio.new_event_loop) as runner: |
| returncode, stdout = runner.run(main()) |
| self.assertEqual(returncode, 0) |
| self.assertEqual(stdout, b'some data') |
| finally: |
| with self.assertWarns(DeprecationWarning): |
| asyncio.set_child_watcher(None) |
| else: |
| # Windows |
| class SubprocessProactorTests(SubprocessMixin, test_utils.TestCase): |
| |
| def setUp(self): |
| super().setUp() |
| self.loop = asyncio.ProactorEventLoop() |
| self.set_event_loop(self.loop) |
| |
| |
| if __name__ == '__main__': |
| unittest.main() |
