program blocking program copying input to output methods must raise processlookuperror if the process exited feed data get output and exitcode start the new process in a new session expect 1 but sometimes get 0 on windows create a new process group so that killing process kills the process and all its children on windows it is an alias of terminate which sets the return code expect 1 but sometimes get 0 expect 1 but sometimes get 0 bpo31034 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 basic synchronization to wait until the program is sleeping buffer large enough to feed the whole pipe buffer from subprocess import startupinfo startupinfo startupinfo startupinfo lpattributelist handlelist handle kwargs dictstartupinfostartupinfo else code f import os fd rfd os readfd 1 kwargs dictpassfdsrfd the program ends before the stdin can be fed proc self loop rununtilcomplete asyncio createsubprocessexec sys executable c code stdinsubprocess pipe kwargs async def writestdinproc data proc stdin writedata only exit the child process once the write buffer is filled os writewfd b go await proc stdin drain coro writestdinproc largedata drain must raise brokenpipeerror or connectionreseterror with testutils disablelogger self assertraisesbrokenpipeerror connectionreseterror self loop rununtilcomplete coro self loop rununtilcompleteproc wait def testcommunicateignorebrokenpipeself buffer large enough to feed the whole pipe buffer largedata b x support pipemaxsize the program ends before the stdin can be fed proc self loop rununtilcomplete asyncio createsubprocessexec sys executable c pass stdinsubprocess pipe communicate must ignore brokenpipeerror when feeding stdin self loop setexceptionhandlerlambda loop msg none self loop rununtilcompleteproc communicatelargedata self loop rununtilcompleteproc wait def testpausereadingself limit 10 size limit 2 1 async def testpausereading code n join import sys sys stdout writex s size sys stdout flush connectreadpipe self loop connectreadpipe async def connectreadpipemockargs kw transport protocol await connectreadpipeargs kw transport pausereading mock mock transport resumereading mock mock return transport protocol self loop connectreadpipe connectreadpipemock proc await asyncio createsubprocessexec sys executable c code stdinasyncio subprocess pipe stdoutasyncio subprocess pipe limitlimit stdouttransport proc transport getpipetransport1 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 stdouttransport issue 22685 ensure that the stream reader pauses the protocol when the child process produces too much data stdout transport self loop rununtilcompletetestpausereading self assertequalstdout b x size self asserttruetransport pausereading called self asserttruetransport resumereading called def teststdinnotinheritableself asyncio issue 209 stdin must not be inheritable otherwise the process communicate hangs async def lenmessagemessage code import sys data sys stdin read printlendata proc await asyncio createsubprocessexec sys executable c code stdinasyncio subprocess pipe stdoutasyncio subprocess pipe stderrasyncio subprocess pipe closefdsfalse stdout stderr await proc communicatemessage exitcode await proc wait return stdout exitcode output exitcode self loop rununtilcompletelenmessageb abc self assertequaloutput rstrip b 3 self assertequalexitcode 0 def testemptyinputself async def emptyinput code import sys data sys stdin read printlendata proc await asyncio createsubprocessexec sys executable c code stdinasyncio subprocess pipe stdoutasyncio subprocess pipe stderrasyncio subprocess pipe closefdsfalse stdout stderr await proc communicateb exitcode await proc wait return stdout exitcode output exitcode self loop rununtilcompleteemptyinput self assertequaloutput rstrip b 0 self assertequalexitcode 0 def testdevnullinputself async def emptyinput code import sys data sys stdin read printlendata proc await asyncio createsubprocessexec sys executable c code stdinasyncio subprocess devnull stdoutasyncio subprocess pipe stderrasyncio subprocess pipe closefdsfalse stdout stderr await proc communicate exitcode await proc wait return stdout exitcode output exitcode self loop rununtilcompleteemptyinput self assertequaloutput rstrip b 0 self assertequalexitcode 0 def testdevnulloutputself async def emptyoutput code import sys data sys stdin read printlendata proc await asyncio createsubprocessexec sys executable c code stdinasyncio subprocess pipe stdoutasyncio subprocess devnull stderrasyncio subprocess pipe closefdsfalse stdout stderr await proc communicatebabc exitcode await proc wait return stdout exitcode output exitcode self loop rununtilcompleteemptyoutput self assertequaloutput none self assertequalexitcode 0 def testdevnullerrorself async def emptyerror code import sys data sys stdin read printlendata proc await asyncio createsubprocessexec sys executable c code stdinasyncio subprocess pipe stdoutasyncio subprocess pipe stderrasyncio subprocess devnull closefdsfalse stdout stderr await proc communicatebabc exitcode await proc wait return stderr exitcode output exitcode self loop rununtilcompleteemptyerror self assertequaloutput none self assertequalexitcode 0 unittest skipifsys platform linux don t have devstdin def testdevstdininputself async def devstdininputmessage code file opendevstdin data file read printlendata proc await asyncio createsubprocessexec sys executable c code stdinasyncio subprocess pipe stdoutasyncio subprocess pipe stderrasyncio subprocess pipe closefdsfalse stdout stderr await proc communicatemessage exitcode await proc wait return stdout exitcode output exitcode self loop rununtilcompletedevstdininputb abc self assertequaloutput rstrip b 3 self assertequalexitcode 0 def testcancelprocesswaitself issue 23140 cancel process wait async def cancelwait proc await asyncio createsubprocessexecprogramblocked create an internal future waiting on the process exit task self loop createtaskproc wait self loop callsoontask 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 rununtilcompletecancelwait def testcancelmakesubprocesstransportexecself async def cancelmaketransport coro asyncio createsubprocessexecprogramblocked task self loop createtaskcoro self loop callsoontask cancel try await task except asyncio cancellederror pass ignore the log exception during subprocess creation kill the subprocess with testutils disablelogger self loop rununtilcompletecancelmaketransport def testcancelpostinitself async def cancelmaketransport coro self loop subprocessexecasyncio subprocessprotocol programblocked task self loop createtaskcoro self loop callsoontask cancel try await task except asyncio cancellederror pass ignore the log exception during subprocess creation kill the subprocess with testutils disablelogger self loop rununtilcompletecancelmaketransport testutils runbrieflyself loop def testclosekillrunningself async def killrunning create self loop subprocessexecasyncio subprocessprotocol programblocked transport protocol await create killcalled false def kill nonlocal killcalled killcalled true origkill proc transport getextrainfo subprocess origkill proc kill proc kill kill returncode transport getreturncode transport close await asyncio waitfortransport wait 5 return returncode killcalled ignore close running child process kill log with testutils disablelogger try returncode killed self loop rununtilcomplete killrunning except asyncio timeouterror self skiptest timeout failure on waiting for subprocess stopping self assertisnonereturncode transport close must kill the process if it is still running self asserttruekilled testutils runbrieflyself loop def testclosedontkillfinishedself async def killrunning create self loop subprocessexecasyncio subprocessprotocol programblocked transport protocol await create proc transport getextrainfo subprocess kill the process but asyncio is not notified immediately proc kill proc wait proc kill mock mock procreturncode proc poll transportreturncode transport getreturncode transport close return procreturncode transportreturncode proc kill called ignore unknown child process pid log of safechildwatcher emitted because the test already consumes the exit status proc wait with testutils disablelogger result self loop rununtilcompletekillrunning testutils runbrieflyself loop procreturncode transportreturncode killed result self assertisnotnoneprocreturncode self assertisnonetransportreturncode transport close must not kill the process if it finished even if the transport was not notified yet self assertfalsekilled 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 isinstanceself subprocessfastwatchertests with warnings catchwarnings warnings simplefilter ignore deprecationwarning asyncio getchildwatcher callbacks clear async def testpopenerrorself stdin if sys platform win32 target asyncio windowsutils popen else target subprocess popen with mock patchtarget as popen exc zerodivisionerror popen sideeffect exc with warnings catchwarningsrecordtrue as warns with self assertraisesexc await asyncio createsubprocessexec sys executable c pass stdinstdin self assertequalwarns def testpopenerrorself issue 24763 check that the subprocess transport is closed when basesubprocesstransport fails self loop rununtilcompleteself testpopenerrorstdinnone def testpopenerrorwithstdinpipeself issue 35721 check that newly created socket pair is closed when popen fails self loop rununtilcomplete self testpopenerrorstdinsubprocess pipe def testreadstdoutafterprocessexitself async def execute code n join import sys for in range64 sys stdout writex 4096 sys stdout flush sys exit1 process await asyncio createsubprocessexec sys executable c code stdoutasyncio subprocess pipe while true data await process stdout read65536 if data await asyncio sleep0 3 else break self loop rununtilcompleteexecute def testcreatesubprocessexectextmodefailsself async def execute with self assertraisesvalueerror await subprocess createsubprocessexecsys executable texttrue with self assertraisesvalueerror await subprocess createsubprocessexecsys executable encodingutf8 with self assertraisesvalueerror await subprocess createsubprocessexecsys executable errorsstrict self loop rununtilcompleteexecute def testcreatesubprocessshelltextmodefailsself async def execute with self assertraisesvalueerror await subprocess createsubprocessshellsys executable texttrue with self assertraisesvalueerror await subprocess createsubprocessshellsys executable encodingutf8 with self assertraisesvalueerror await subprocess createsubprocessshellsys executable errorsstrict self loop rununtilcompleteexecute def testcreatesubprocessexecwithpathself async def execute p await subprocess createsubprocessexec oshelper fakepathsys executable c pass await p wait p await subprocess createsubprocessexec sys executable c pass oshelper fakepath await p wait self assertisnoneself loop rununtilcompleteexecute async def checkstdoutoutputself coro output proc await coro stdout await proc communicate self assertequalstdout output self assertequalproc returncode 0 task asyncio createtaskproc wait await asyncio sleep0 self assertequaltask result proc returncode def testcreatesubprocessenvshellself none async def main none cmd f sys executable c import os sys sys stdout writeos getenv foo env os environ copy envfoo bar proc await asyncio createsubprocessshell cmd envenv stdoutsubprocess pipe return proc self loop rununtilcompleteself checkstdoutoutputmain b bar def testcreatesubprocessenvexecself none async def main none cmd sys executable c import os sys sys stdout writeos getenv foo env os environ copy envfoo baz proc await asyncio createsubprocessexec cmd envenv stdoutsubprocess pipe return proc self loop rununtilcompleteself checkstdoutoutputmain b baz def testsubprocessconcurrentwaitself none async def main none proc await asyncio createsubprocessexec programcat stdinsubprocess pipe stdoutsubprocess pipe stdout await proc communicateb some data self assertequalstdout bsome data self assertequalproc returncode 0 self assertequalawait asyncio gatherproc wait for in range10 proc returncode 10 self loop rununtilcompletemain def testsubprocessprotocoleventsself gh108973 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 pipedatareceived 1 b stdout pipedatareceived 2 b stderr pipeconnectionlost 1 pipeconnectionlost 2 processexited perfdexpected pipedatareceived pipeconnectionlost class myprotocolasyncio subprocessprotocol def initself exitfuture asyncio future none self exitfuture exitfuture def pipedatareceivedself fd data none events append pipedatareceived fd data self exitmaybe def pipeconnectionlostself fd exc none events append pipeconnectionlost fd self exitmaybe def processexitedself none events append processexited self exitmaybe def exitmaybeself only exit when we got all expected events if lenevents lenexpected self exitfuture setresulttrue async def main none loop asyncio getrunningloop exitfuture asyncio future code import sys sys stdout writestdout sys stderr writestderr transport await loop subprocessexeclambda myprotocolexitfuture sys executable c code stdinnone await exitfuture transport close return events events self loop rununtilcompletemain first make sure that we received all events self assertsetequalsetevents setexpected second check order of pipe events per file descriptor perfdevents fd for fd in fds for event in events if event processexited continue name fd event 2 perfdeventsfd appendname for fd in fds self assertequalperfdeventsfd perfdexpected fd events def testsubprocesscommunicatestdoutself see https github compythoncpythonissues100133 async def getcommandstdoutcmd args proc await asyncio createsubprocessexec cmd args stdoutasyncio subprocess pipe stdout await proc communicate return stdout decode strip async def main outputs f fooi for i in range10 res await asyncio gathergetcommandstdoutsys executable c f printout r for out in outputs self assertequalres outputs self loop rununtilcompletemain if sys platform win32 unix class subprocesswatchermixinsubprocessmixin watcher none def setupself super setup policy asyncio geteventlooppolicy self loop policy neweventloop self seteventloopself loop watcher self getwatcher watcher attachloopself loop with warnings catchwarnings warnings simplefilter ignore deprecationwarning policy setchildwatcherwatcher def teardownself super teardown policy asyncio geteventlooppolicy with warnings catchwarnings warnings simplefilter ignore deprecationwarning watcher policy getchildwatcher policy setchildwatchernone watcher attachloopnone watcher close class subprocessthreadedwatchertestssubprocesswatchermixin testutils testcase def getwatcherself return unixevents threadedchildwatcher class subprocesssafewatchertestssubprocesswatchermixin testutils testcase def getwatcherself with self assertwarnsdeprecationwarning return unixevents safechildwatcher class multiloopchildwatcherteststestutils testcase def testwarnsself with self assertwarnsdeprecationwarning unixevents multiloopchildwatcher class subprocessfastwatchertestssubprocesswatchermixin testutils testcase def getwatcherself with self assertwarnsdeprecationwarning return unixevents fastchildwatcher unittest skipunless unixevents canusepidfd operating system does not support pidfds class subprocesspidfdwatchertestssubprocesswatchermixin testutils testcase def getwatcherself return unixevents pidfdchildwatcher class genericwatcherteststestutils testcase def testcreatesubprocessfailswithinactivewatcherself watcher mock createautospecasyncio abstractchildwatcher watcher isactive returnvalue false async def execute asyncio setchildwatcherwatcher with self assertraisesruntimeerror await subprocess createsubprocessexec oshelper fakepathsys executable c pass watcher addchildhandler assertnotcalled with asyncio runnerloopfactoryasyncio neweventloop as runner with warnings catchwarnings warnings simplefilter ignore deprecationwarning self assertisnonerunner runexecute self assertlistequalwatcher mockcalls mock call enter mock call isactive mock call exitruntimeerror mock any mock any watcher mockcalls unittest skipunless unixevents canusepidfd operating system does not support pidfds def testcreatesubprocesswithpidfdself async def inthread proc await asyncio createsubprocessexec programcat stdinsubprocess pipe stdoutsubprocess pipe stdout stderr await proc communicatebsome data return proc returncode stdout async def main asyncio runner did not call asyncio seteventloop with self assertraisesruntimeerror asyncio geteventlooppolicy geteventloop return await asyncio tothreadasyncio run inthread with self assertwarnsdeprecationwarning asyncio setchildwatcherasyncio pidfdchildwatcher try with asyncio runnerloopfactoryasyncio neweventloop as runner returncode stdout runner runmain self assertequalreturncode 0 self assertequalstdout b some data finally with self assertwarnsdeprecationwarning asyncio setchildwatchernone else windows class subprocessproactortestssubprocessmixin testutils testcase def setupself super setup self loop asyncio proactoreventloop self seteventloopself loop if name main unittest main program blocking program copying input to output methods must raise processlookuperror if the process exited feed data get output and exitcode start the new process in a new session expect 1 but sometimes get 0 on windows create a new process group so that killing process kills the process and all its children on windows it is an alias of terminate which sets the return code expect 1 but sometimes get 0 expect 1 but sometimes get 0 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 basic synchronization to wait until the program is sleeping buffer large enough to feed the whole pipe buffer import os msvcrt handle handle fd msvcrt open_osfhandle handle os o_rdonly os read fd 1 the program ends before the stdin can be fed only exit the child process once the write buffer is filled drain must raise brokenpipeerror or connectionreseterror buffer large enough to feed the whole pipe buffer the program ends before the stdin can be fed communicate must ignore brokenpipeerror when feeding stdin the child process produced more than limit bytes of output the stream reader transport should pause the protocol to not allocate too much memory issue 22685 ensure that the stream reader pauses the protocol when the child process produces too much data asyncio issue 209 stdin must not be inheritable otherwise the process communicate hangs issue 23140 cancel process wait create an internal future waiting on the process exit cancel the future kill the process and wait until it is done ignore the log exception during subprocess creation kill the subprocess ignore the log exception during subprocess creation kill the subprocess ignore close running child process kill log transport close must kill the process if it is still running kill the process but asyncio is not notified immediately ignore unknown child process pid log of safechildwatcher emitted because the test already consumes the exit status proc wait transport close must not kill the process if it finished even if the transport was not notified yet 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 issue 24763 check that the subprocess transport is closed when basesubprocesstransport fails issue 35721 check that newly created socket pair is closed when popen fails sys executable c import os sys sys stdout write os getenv foo 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 only exit when we got all expected events first make sure that we received all events second check order of pipe events per file descriptor see https github com python cpython issues 100133 unix asyncio runner did not call asyncio set_event_loop windows
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_BLOCKED = [sys.executable, '-c', 'import time; time.sleep(3600)']


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


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)

        
        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()


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,
            )

            
            proc.stdin.write(data)
            await proc.stdin.drain()
            proc.stdin.close()

            
            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):
        
        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)
            
        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
            
            
            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)
        
        proc.kill()
        returncode = self.loop.run_until_complete(proc.wait())
        if sys.platform == 'win32':
            self.assertIsInstance(returncode, int)
            
        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)
            
        else:
            self.assertEqual(-signal.SIGTERM, returncode)

    @unittest.skipIf(sys.platform == 'win32', "Don't have SIGHUP")
    def test_send_signal(self):
        
        
        
        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):
                
                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):
        
        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)
            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,))

        
        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)
            
            os.write(wfd, b'go')
            await proc.stdin.drain()

        coro = write_stdin(proc, large_data)
        
        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):
        
        large_data = b'x' * support.PIPE_MAX_SIZE

        
        proc = self.loop.run_until_complete(
            asyncio.create_subprocess_exec(
                sys.executable, '-c', 'pass',
                stdin=subprocess.PIPE,
            )
        )

        
        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()

            
            
            
            return (stdout, stdout_transport)

        
        
        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):
        
        
        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):
        

        async def cancel_wait():
            proc = await asyncio.create_subprocess_exec(*PROGRAM_BLOCKED)

            
            task = self.loop.create_task(proc.wait())
            self.loop.call_soon(task.cancel)
            try:
                await task
            except asyncio.CancelledError:
                pass

            
            task.cancel()

            
            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

        
        
        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

        
        
        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)

        
        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)

        
        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')

            
            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)

        
        
        
        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)

        
        
        self.assertFalse(killed)

        
        
        
        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):
        
        
        self.loop.run_until_complete(self._test_popen_error(stdin=None))

    def test_popen_error_with_stdin_pipe(self):
        
        
        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:
            cmd = f
            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):
        
        
        
        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):
                
                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())

        
        self.assertSetEqual(set(events), set(expected))

        
        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):
        
        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':
    
    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():
                
                with self.assertRaises(RuntimeError):
                    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:
    
    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()