tests for unixevents py import contextlib import errno import io import multiprocessing from multiprocessing util import cleanuptests as multiprocessingcleanuptests import os import pathlib import signal import socket import stat import sys import threading import time import unittest from unittest import mock import warnings from test import support from test support import oshelper from test support import sockethelper from test support import waitprocess from test support import hashlibhelper if sys platform win32 raise unittest skiptest unix only import asyncio from asyncio import log from asyncio import unixevents from test testasyncio import utils as testutils def teardownmodule asyncio seteventlooppolicynone mockany mock any def exitcodeexitcode return 32768 exitcode def signalsignum if not 1 signum 68 raise assertionerrorf invalid signum signum return 32768 signum def closepipetransporttransport don t call transport close because the event loop and the selector are mocked if transport pipe is none return transport pipe close transport pipe none unittest skipunlesssignal signals are not supported class selectoreventloopsignalteststestutils testcase def setupself super setup self loop asyncio selectoreventloop self seteventloopself loop def testchecksignalself self assertraises typeerror self loop checksignal 1 self assertraises valueerror self loop checksignal signal nsig 1 def testhandlesignalnohandlerself self loop handlesignalsignal nsig 1 def testhandlesignalcancelledhandlerself h asyncio handlemock mock loopmock mock h cancel self loop signalhandlerssignal nsig 1 h self loop removesignalhandler mock mock self loop handlesignalsignal nsig 1 self loop removesignalhandler assertcalledwithsignal nsig 1 mock patch asyncio unixevents signal def testaddsignalhandlersetuperrorself msignal msignal nsig signal nsig msignal validsignals signal validsignals msignal setwakeupfd sideeffect valueerror self assertraises runtimeerror self loop addsignalhandler signal sigint lambda true mock patch asyncio unixevents signal def testaddsignalhandlercoroutineerrorself msignal msignal nsig signal nsig async def simplecoroutine pass callback must not be a coroutine function corofunc simplecoroutine coroobj corofunc self addcleanupcoroobj close for func in corofunc coroobj self assertraisesregex typeerror coroutines cannot be used with addsignalhandler self loop addsignalhandler signal sigint func mock patch asyncio unixevents signal def testaddsignalhandlerself msignal msignal nsig signal nsig msignal validsignals signal validsignals cb lambda true self loop addsignalhandlersignal sighup cb h self loop signalhandlers getsignal sighup self assertisinstanceh asyncio handle self assertequalh callback cb mock patch asyncio unixevents signal def testaddsignalhandlerinstallerrorself msignal msignal nsig signal nsig msignal validsignals signal validsignals def setwakeupfdfd if fd 1 raise valueerror msignal setwakeupfd setwakeupfd class erroserror errno errno efault msignal signal sideeffect err self assertraises err self loop addsignalhandler signal sigint lambda true mock patch asyncio unixevents signal mock patch asyncio baseevents logger def testaddsignalhandlerinstallerror2self mlogging msignal msignal nsig signal nsig msignal validsignals signal validsignals class erroserror errno errno einval msignal signal sideeffect err self loop signalhandlerssignal sighup lambda true self assertraises runtimeerror self loop addsignalhandler signal sigint lambda true self assertfalsemlogging info called self assertequal1 msignal setwakeupfd callcount mock patch asyncio unixevents signal mock patch asyncio baseevents logger def testaddsignalhandlerinstallerror3self mlogging msignal class erroserror errno errno einval msignal signal sideeffect err msignal nsig signal nsig msignal validsignals signal validsignals self assertraises runtimeerror self loop addsignalhandler signal sigint lambda true self assertfalsemlogging info called self assertequal2 msignal setwakeupfd callcount mock patch asyncio unixevents signal def testremovesignalhandlerself msignal msignal nsig signal nsig msignal validsignals signal validsignals self loop addsignalhandlersignal sighup lambda true self asserttrue self loop removesignalhandlersignal sighup self asserttruemsignal setwakeupfd called self asserttruemsignal signal called self assertequal signal sighup msignal sigdfl msignal signal callargs0 mock patch asyncio unixevents signal def testremovesignalhandler2self msignal msignal nsig signal nsig msignal sigint signal sigint msignal validsignals signal validsignals self loop addsignalhandlersignal sigint lambda true self loop signalhandlerssignal sighup object msignal setwakeupfd resetmock self asserttrue self loop removesignalhandlersignal sigint self assertfalsemsignal setwakeupfd called self asserttruemsignal signal called self assertequal signal sigint msignal defaultinthandler msignal signal callargs0 mock patch asyncio unixevents signal mock patch asyncio baseevents logger def testremovesignalhandlercleanuperrorself mlogging msignal msignal nsig signal nsig msignal validsignals signal validsignals self loop addsignalhandlersignal sighup lambda true msignal setwakeupfd sideeffect valueerror self loop removesignalhandlersignal sighup self asserttruemlogging info mock patch asyncio unixevents signal def testremovesignalhandlererrorself msignal msignal nsig signal nsig msignal validsignals signal validsignals self loop addsignalhandlersignal sighup lambda true msignal signal sideeffect oserror self assertraises oserror self loop removesignalhandler signal sighup mock patch asyncio unixevents signal def testremovesignalhandlererror2self msignal msignal nsig signal nsig msignal validsignals signal validsignals self loop addsignalhandlersignal sighup lambda true class erroserror errno errno einval msignal signal sideeffect err self assertraises runtimeerror self loop removesignalhandler signal sighup mock patch asyncio unixevents signal def testcloseself msignal msignal nsig signal nsig msignal validsignals signal validsignals self loop addsignalhandlersignal sighup lambda true self loop addsignalhandlersignal sigchld lambda true self assertequallenself loop signalhandlers 2 msignal setwakeupfd resetmock self loop close self assertequallenself loop signalhandlers 0 msignal setwakeupfd assertcalledoncewith1 mock patch asyncio unixevents sys mock patch asyncio unixevents signal def testcloseonfinalizingself msignal msys msignal nsig signal nsig msignal validsignals signal validsignals self loop addsignalhandlersignal sighup lambda true self assertequallenself loop signalhandlers 1 msys isfinalizing returnvalue true msignal signal resetmock with self assertwarnsregexresourcewarning skipping signal handlers removal self loop close self assertequallenself loop signalhandlers 0 self assertfalsemsignal signal called unittest skipunlesshasattrsocket afunix unix sockets are not supported class selectoreventloopunixsocketteststestutils testcase def setupself super setup self loop asyncio selectoreventloop self seteventloopself loop sockethelper skipunlessbindunixsocket def testcreateunixserverexistingpathsockself with testutils unixsocketpath as path sock socket socketsocket afunix sock bindpath sock listen1 sock close coro self loop createunixserverlambda none path srv self loop rununtilcompletecoro srv close self loop rununtilcompletesrv waitclosed sockethelper skipunlessbindunixsocket def testcreateunixserverpathlibself with testutils unixsocketpath as path path pathlib pathpath srvcoro self loop createunixserverlambda none path srv self loop rununtilcompletesrvcoro srv close self loop rununtilcompletesrv waitclosed def testcreateunixconnectionpathlibself with testutils unixsocketpath as path path pathlib pathpath coro self loop createunixconnectionlambda none path with self assertraisesfilenotfounderror if pathlib path wasn t supported the exception would be different self loop rununtilcompletecoro def testcreateunixserverexistingpathnonsockself path testutils genunixsocketpath self addcleanuposhelper unlink path create the file openpath wb close coro self loop createunixserverlambda none path with self assertraisesregexoserror address is already in use self loop rununtilcompletecoro def testcreateunixserversslboolself coro self loop createunixserverlambda none path spam ssltrue with self assertraisesregextypeerror ssl argument must be an sslcontext self loop rununtilcompletecoro def testcreateunixservernopathnosockself coro self loop createunixserverlambda none pathnone with self assertraisesregexvalueerror path was not specified and no sock self loop rununtilcompletecoro def testcreateunixserverpathinetsockself sock socket socket with sock coro self loop createunixserverlambda none pathnone socksock with self assertraisesregexvalueerror a unix domain stream was expected self loop rununtilcompletecoro def testcreateunixserverpathdgramself sock socket socketsocket afunix socket sockdgram with sock coro self loop createunixserverlambda none pathnone socksock with self assertraisesregexvalueerror a unix domain stream was expected self loop rununtilcompletecoro unittest skipunlesshasattrsocket socknonblock no socket socknonblock linux only sockethelper skipunlessbindunixsocket def testcreateunixserverpathstreambittypeself fn testutils genunixsocketpath self addcleanuposhelper unlink fn sock socket socketsocket afunix socket sockstream socket socknonblock with sock sock bindfn coro self loop createunixserverlambda none pathnone socksock srv self loop rununtilcompletecoro srv close self loop rununtilcompletesrv waitclosed def testcreateunixserverssltimeoutwithplainsockself coro self loop createunixserverlambda none path spam sslhandshaketimeout1 with self assertraisesregex valueerror sslhandshaketimeout is only meaningful with ssl self loop rununtilcompletecoro def testcreateunixconnectionpathinetsockself sock socket socket with sock coro self loop createunixconnectionlambda none socksock with self assertraisesregexvalueerror a unix domain stream was expected self loop rununtilcompletecoro mock patch asyncio unixevents socket def testcreateunixserverbinderrorself msocket ensure that the socket is closed on any bind error sock mock mock msocket socket returnvalue sock sock bind sideeffect oserror coro self loop createunixserverlambda none pathtest with self assertraisesoserror self loop rununtilcompletecoro self asserttruesock close called sock bind sideeffect memoryerror coro self loop createunixserverlambda none pathtest with self assertraisesmemoryerror self loop rununtilcompletecoro self asserttruesock close called def testcreateunixconnectionpathsockself coro self loop createunixconnection lambda none os devnull sockobject with self assertraisesregexvalueerror path and sock can not be self loop rununtilcompletecoro def testcreateunixconnectionnopathnosockself coro self loop createunixconnection lambda none none with self assertraisesregexvalueerror no path and sock were specified self loop rununtilcompletecoro def testcreateunixconnectionnosslserverhostself coro self loop createunixconnection lambda none os devnull serverhostname spam with self assertraisesregexvalueerror serverhostname is only meaningful self loop rununtilcompletecoro def testcreateunixconnectionsslnoserverhostself coro self loop createunixconnection lambda none os devnull ssltrue with self assertraisesregex valueerror you have to pass serverhostname when using ssl self loop rununtilcompletecoro def testcreateunixconnectionssltimeoutwithplainsockself coro self loop createunixconnectionlambda none path spam sslhandshaketimeout1 with self assertraisesregex valueerror sslhandshaketimeout is only meaningful with ssl self loop rununtilcompletecoro unittest skipunlesshasattros sendfile sendfile is not supported class selectoreventloopunixsocksendfileteststestutils testcase data b12345abcde 16 1024 160 kib class myprotoasyncio protocol def initself loop self started false self closed false self data bytearray self fut loop createfuture self transport none self ready loop createfuture def connectionmadeself transport self started true self transport transport self ready setresultnone def datareceivedself data self data extenddata def connectionlostself exc self closed true self fut setresultnone async def waitclosedself await self fut classmethod def setupclasscls with openoshelper testfn wb as fp fp writecls data super setupclass classmethod def teardownclasscls oshelper unlinkoshelper testfn super teardownclass def setupself self loop asyncio neweventloop self seteventloopself loop self file openoshelper testfn rb self addcleanupself file close super setup def makesocketself cleanuptrue sock socket socketsocket afinet socket sockstream sock setblockingfalse sock setsockoptsocket solsocket socket sosndbuf 1024 sock setsockoptsocket solsocket socket sorcvbuf 1024 if cleanup self addcleanupsock close return sock def runloopself coro return self loop rununtilcompletecoro def prepareself sock self makesocket proto self myprotoself loop port sockethelper findunusedport srvsock self makesocketcleanupfalse srvsock bindsockethelper host port server self runloopself loop createserver lambda proto socksrvsock self runloopself loop sockconnectsock sockethelper host port self runloopproto ready def cleanup proto transport close self runloopproto waitclosed server close self runloopserver waitclosed self addcleanupcleanup return sock proto def testsocksendfilenotavailableself sock proto self prepare with mock patch asyncio unixevents os spec with self assertraisesregexasyncio sendfilenotavailableerror os sendfile is not available self runloopself loop socksendfilenativesock self file 0 none self assertequalself file tell 0 def testsocksendfilenotafileself sock proto self prepare f object with self assertraisesregexasyncio sendfilenotavailableerror not a regular file self runloopself loop socksendfilenativesock f 0 none self assertequalself file tell 0 def testsocksendfileiobufferself sock proto self prepare f io bytesio with self assertraisesregexasyncio sendfilenotavailableerror not a regular file self runloopself loop socksendfilenativesock f 0 none self assertequalself file tell 0 def testsocksendfilenotregularfileself sock proto self prepare f mock mock f fileno returnvalue 1 with self assertraisesregexasyncio sendfilenotavailableerror not a regular file self runloopself loop socksendfilenativesock f 0 none self assertequalself file tell 0 def testsocksendfilecancel1self sock proto self prepare fut self loop createfuture fileno self file fileno self loop socksendfilenativeimplfut none sock fileno 0 none lenself data 0 fut cancel with contextlib suppressasyncio cancellederror self runloopfut with self assertraiseskeyerror self loop selector getkeysock def testsocksendfilecancel2self sock proto self prepare fut self loop createfuture fileno self file fileno self loop socksendfilenativeimplfut none sock fileno 0 none lenself data 0 fut cancel self loop socksendfilenativeimplfut sock fileno sock fileno 0 none lenself data 0 with self assertraiseskeyerror self loop selector getkeysock def testsocksendfileblockingerrorself sock proto self prepare fileno self file fileno fut mock mock fut cancelled returnvalue false with mock patch os sendfile sideeffectblockingioerror self loop socksendfilenativeimplfut none sock fileno 0 none lenself data 0 key self loop selector getkeysock self assertisnotnonekey fut adddonecallback assertcalledoncewithmock any def testsocksendfileoserrorfirstcallself sock proto self prepare fileno self file fileno fut self loop createfuture with mock patch os sendfile sideeffectoserror self loop socksendfilenativeimplfut none sock fileno 0 none lenself data 0 with self assertraiseskeyerror self loop selector getkeysock exc fut exception self assertisinstanceexc asyncio sendfilenotavailableerror self assertequal0 self file tell def testsocksendfileoserrornextcallself sock proto self prepare fileno self file fileno fut self loop createfuture err oserror with mock patch os sendfile sideeffecterr self loop socksendfilenativeimplfut sock fileno sock fileno 1000 none lenself data 1000 with self assertraiseskeyerror self loop selector getkeysock exc fut exception self assertisexc err self assertequal1000 self file tell def testsocksendfileexceptionself sock proto self prepare fileno self file fileno fut self loop createfuture err asyncio sendfilenotavailableerror with mock patch os sendfile sideeffecterr self loop socksendfilenativeimplfut sock fileno sock fileno 1000 none lenself data 1000 with self assertraiseskeyerror self loop selector getkeysock exc fut exception self assertisexc err self assertequal1000 self file tell class unixreadpipetransportteststestutils testcase def setupself super setup self loop self newtestloop self protocol testutils maketestprotocolasyncio protocol self pipe mock mockspecsetio rawiobase self pipe fileno returnvalue 5 blockingpatcher mock patch os setblocking blockingpatcher start self addcleanupblockingpatcher stop fstatpatcher mock patch os fstat mfstat fstatpatcher start st mock mock st stmode stat sififo mfstat returnvalue st self addcleanupfstatpatcher stop def readpipetransportself waiternone transport unixevents unixreadpipetransportself loop self pipe self protocol waiterwaiter self addcleanupclosepipetransport transport return transport def testctorself waiter self loop createfuture tr self readpipetransportwaiterwaiter self loop rununtilcompletewaiter self protocol connectionmade assertcalledwithtr self loop assertreader5 tr readready self assertisnonewaiter result mock patch os read def testreadreadyself mread tr self readpipetransport mread returnvalue b data tr readready mread assertcalledwith5 tr maxsize self protocol datareceived assertcalledwithb data mock patch os read def testreadreadyeofself mread tr self readpipetransport mread returnvalue b tr readready mread assertcalledwith5 tr maxsize self assertfalseself loop readers testutils runbrieflyself loop self protocol eofreceived assertcalledwith self protocol connectionlost assertcalledwithnone mock patch os read def testreadreadyblockedself mread tr self readpipetransport mread sideeffect blockingioerror tr readready mread assertcalledwith5 tr maxsize testutils runbrieflyself loop self assertfalseself protocol datareceived called mock patch asyncio log logger error mock patch os read def testreadreadyerrorself mread mlogexc tr self readpipetransport err oserror mread sideeffect err tr close mock mock tr readready mread assertcalledwith5 tr maxsize tr close assertcalledwitherr mlogexc assertcalledwith testutils mockpattern fatal read error on pipe transport nprotocol ntransport excinfooserror mockany mockany mock patch os read def testpausereadingself mread tr self readpipetransport m mock mock self loop addreader5 m tr pausereading self assertfalseself loop readers mock patch os read def testresumereadingself mread tr self readpipetransport tr pausereading tr resumereading self loop assertreader5 tr readready mock patch os read def testcloseself mread tr self readpipetransport tr close mock mock tr close tr close assertcalledwithnone mock patch os read def testclosealreadyclosingself mread tr self readpipetransport tr closing true tr close mock mock tr close self assertfalsetr close called mock patch os read def testcloseself mread tr self readpipetransport err object tr closeerr self asserttruetr isclosing self assertfalseself loop readers testutils runbrieflyself loop self protocol connectionlost assertcalledwitherr def testcallconnectionlostself tr self readpipetransport self assertisnotnonetr protocol self assertisnotnonetr loop err none tr callconnectionlosterr self protocol connectionlost assertcalledwitherr self pipe close assertcalledwith self assertisnonetr protocol self assertisnonetr loop def testcallconnectionlostwitherrself tr self readpipetransport self assertisnotnonetr protocol self assertisnotnonetr loop err oserror tr callconnectionlosterr self protocol connectionlost assertcalledwitherr self pipe close assertcalledwith self assertisnonetr protocol self assertisnonetr loop def testpausereadingonclosedpipeself tr self readpipetransport tr close testutils runbrieflyself loop self assertisnonetr loop tr pausereading def testpausereadingonpausedpipeself tr self readpipetransport tr pausereading the second call should do nothing tr pausereading def testresumereadingonclosedpipeself tr self readpipetransport tr close testutils runbrieflyself loop self assertisnonetr loop tr resumereading def testresumereadingonpausedpipeself tr self readpipetransport the pipe is not paused resuming should do nothing tr resumereading class unixwritepipetransportteststestutils testcase def setupself super setup self loop self newtestloop self protocol testutils maketestprotocolasyncio baseprotocol self pipe mock mockspecsetio rawiobase self pipe fileno returnvalue 5 blockingpatcher mock patch os setblocking blockingpatcher start self addcleanupblockingpatcher stop fstatpatcher mock patch os fstat mfstat fstatpatcher start st mock mock st stmode stat sifsock mfstat returnvalue st self addcleanupfstatpatcher stop def writepipetransportself waiternone transport unixevents unixwritepipetransportself loop self pipe self protocol waiterwaiter self addcleanupclosepipetransport transport return transport def testctorself waiter self loop createfuture tr self writepipetransportwaiterwaiter self loop rununtilcompletewaiter self protocol connectionmade assertcalledwithtr self loop assertreader5 tr readready self assertequalnone waiter result def testcanwriteeofself tr self writepipetransport self asserttruetr canwriteeof mock patch os write def testwriteself mwrite tr self writepipetransport mwrite returnvalue 4 tr writeb data mwrite assertcalledwith5 b data self assertfalseself loop writers self assertequalbytearray tr buffer mock patch os write def testwritenodataself mwrite tr self writepipetransport tr writeb self assertfalsemwrite called self assertfalseself loop writers self assertequalbytearrayb tr buffer mock patch os write def testwritepartialself mwrite tr self writepipetransport mwrite returnvalue 2 tr writeb data self loop assertwriter5 tr writeready self assertequalbytearrayb ta tr buffer mock patch os write def testwritebufferself mwrite tr self writepipetransport self loop addwriter5 tr writeready tr buffer bytearrayb previous tr writeb data self assertfalsemwrite called self loop assertwriter5 tr writeready self assertequalbytearrayb previousdata tr buffer mock patch os write def testwriteagainself mwrite tr self writepipetransport mwrite sideeffect blockingioerror tr writeb data mwrite assertcalledwith5 bytearrayb data self loop assertwriter5 tr writeready self assertequalbytearrayb data tr buffer mock patch asyncio unixevents logger mock patch os write def testwriteerrself mwrite mlog tr self writepipetransport err oserror mwrite sideeffect err tr fatalerror mock mock tr writeb data mwrite assertcalledwith5 b data self assertfalseself loop writers self assertequalbytearray tr buffer tr fatalerror assertcalledwith err fatal write error on pipe transport self assertequal1 tr connlost tr writeb data self assertequal2 tr connlost tr writeb data tr writeb data tr writeb data tr writeb data this is a bit overspecified mlog warning assertcalledwith pipe closed by peer or os writepipe data raised exception tr close mock patch os write def testwritecloseself mwrite tr self writepipetransport tr readready pipe was closed by peer tr writeb data self assertequaltr connlost 1 tr writeb data self assertequaltr connlost 2 def testreadreadyself tr self writepipetransport tr readready self assertfalseself loop readers self assertfalseself loop writers self asserttruetr isclosing testutils runbrieflyself loop self protocol connectionlost assertcalledwithnone mock patch os write def testwritereadyself mwrite tr self writepipetransport self loop addwriter5 tr writeready tr buffer bytearrayb data mwrite returnvalue 4 tr writeready self assertfalseself loop writers self assertequalbytearray tr buffer mock patch os write def testwritereadypartialself mwrite tr self writepipetransport self loop addwriter5 tr writeready tr buffer bytearrayb data mwrite returnvalue 3 tr writeready self loop assertwriter5 tr writeready self assertequalbytearrayb a tr buffer mock patch os write def testwritereadyagainself mwrite tr self writepipetransport self loop addwriter5 tr writeready tr buffer bytearrayb data mwrite sideeffect blockingioerror tr writeready mwrite assertcalledwith5 bytearrayb data self loop assertwriter5 tr writeready self assertequalbytearrayb data tr buffer mock patch os write def testwritereadyemptyself mwrite tr self writepipetransport self loop addwriter5 tr writeready tr buffer bytearrayb data mwrite returnvalue 0 tr writeready mwrite assertcalledwith5 bytearrayb data self loop assertwriter5 tr writeready self assertequalbytearrayb data tr buffer mock patch asyncio log logger error mock patch os write def testwritereadyerrself mwrite mlogexc tr self writepipetransport self loop addwriter5 tr writeready tr buffer bytearrayb data mwrite sideeffect err oserror tr writeready self assertfalseself loop writers self assertfalseself loop readers self assertequalbytearray tr buffer self asserttruetr isclosing mlogexc assertnotcalled self assertequal1 tr connlost testutils runbrieflyself loop self protocol connectionlost assertcalledwitherr mock patch os write def testwritereadyclosingself mwrite tr self writepipetransport self loop addwriter5 tr writeready tr closing true tr buffer bytearrayb data mwrite returnvalue 4 tr writeready self assertfalseself loop writers self assertfalseself loop readers self assertequalbytearray tr buffer self protocol connectionlost assertcalledwithnone self pipe close assertcalledwith mock patch os write def testabortself mwrite tr self writepipetransport self loop addwriter5 tr writeready self loop addreader5 tr readready tr buffer b da b ta tr abort self assertfalsemwrite called self assertfalseself loop readers self assertfalseself loop writers self assertequal tr buffer self asserttruetr isclosing testutils runbrieflyself loop self protocol connectionlost assertcalledwithnone def testcallconnectionlostself tr self writepipetransport self assertisnotnonetr protocol self assertisnotnonetr loop err none tr callconnectionlosterr self protocol connectionlost assertcalledwitherr self pipe close assertcalledwith self assertisnonetr protocol self assertisnonetr loop def testcallconnectionlostwitherrself tr self writepipetransport self assertisnotnonetr protocol self assertisnotnonetr loop err oserror tr callconnectionlosterr self protocol connectionlost assertcalledwitherr self pipe close assertcalledwith self assertisnonetr protocol self assertisnonetr loop def testcloseself tr self writepipetransport tr writeeof mock mock tr close tr writeeof assertcalledwith closing the transport twice must not fail tr close def testcloseclosingself tr self writepipetransport tr writeeof mock mock tr closing true tr close self assertfalsetr writeeof called def testwriteeofself tr self writepipetransport tr writeeof self asserttruetr isclosing self assertfalseself loop readers testutils runbrieflyself loop self protocol connectionlost assertcalledwithnone def testwriteeofpendingself tr self writepipetransport tr buffer b data tr writeeof self asserttruetr isclosing self assertfalseself protocol connectionlost called class abstractchildwatchertestsunittest testcase def testwarnsonsubclassingself with self assertwarnsdeprecationwarning class mywatcherasyncio abstractchildwatcher pass def testnotimplementedself f mock mock watcher asyncio abstractchildwatcher self assertraises notimplementederror watcher addchildhandler f f self assertraises notimplementederror watcher removechildhandler f self assertraises notimplementederror watcher attachloop f self assertraises notimplementederror watcher close self assertraises notimplementederror watcher isactive self assertraises notimplementederror watcher enter self assertraises notimplementederror watcher exit f f f class basechildwatchertestsunittest testcase def testnotimplementedself f mock mock watcher unixevents basechildwatcher self assertraises notimplementederror watcher dowaitpid f class childwatchertestsmixin ignorewarnings mock patch objectlog logger warning def setupself super setup self loop self newtestloop self running false self zombies with mock patch object self loop addsignalhandler as self maddsignalhandler self watcher self createwatcher self watcher attachloopself loop def waitpidself pid flags if isinstanceself watcher asyncio safechildwatcher or pid 1 self assertgreaterpid 0 try if pid 0 return self zombies popitem else return pid self zombies poppid except keyerror pass if self running return 0 0 else raise childprocesserror def addzombieself pid status self zombiespid status def waitstatustoexitcodeself status if status 32768 return status 32768 elif 32700 status 32768 return status 32768 else return status def testcreatewatcherself self maddsignalhandler assertcalledoncewith signal sigchld self watcher sigchld def waitpidmocksfunc def wrappedfuncself def patchtarget wrapper return mock patchtarget wrapswrapper newcallablemock mock with patch asyncio unixevents waitstatustoexitcode self waitstatustoexitcode patch os waitpid self waitpid as mwaitpid funcself mwaitpid return wrappedfunc waitpidmocks def testsigchldself mwaitpid register a child callback mock mock with self watcher self running true self watcher addchildhandler42 callback 9 10 14 self assertfalsecallback called child is running self watcher sigchld self assertfalsecallback called child terminates returncode 12 self running false self addzombie42 exitcode12 self watcher sigchld callback assertcalledoncewith42 12 9 10 14 callback resetmock ensure that the child is effectively reaped self addzombie42 exitcode13 with self ignorewarnings self watcher sigchld self assertfalsecallback called sigchld called again self zombies clear self watcher sigchld self assertfalsecallback called waitpidmocks def testsigchldtwochildrenself mwaitpid callback1 mock mock callback2 mock mock register child 1 with self watcher self running true self watcher addchildhandler43 callback1 7 8 self assertfalsecallback1 called self assertfalsecallback2 called register child 2 with self watcher self watcher addchildhandler44 callback2 147 18 self assertfalsecallback1 called self assertfalsecallback2 called children are running self watcher sigchld self assertfalsecallback1 called self assertfalsecallback2 called child 1 terminates signal 3 self addzombie43 signal3 self watcher sigchld callback1 assertcalledoncewith43 3 7 8 self assertfalsecallback2 called callback1 resetmock child 2 still running self watcher sigchld self assertfalsecallback1 called self assertfalsecallback2 called child 2 terminates code 108 self addzombie44 exitcode108 self running false self watcher sigchld callback2 assertcalledoncewith44 108 147 18 self assertfalsecallback1 called callback2 resetmock ensure that the children are effectively reaped self addzombie43 exitcode14 self addzombie44 exitcode15 with self ignorewarnings self watcher sigchld self assertfalsecallback1 called self assertfalsecallback2 called sigchld called again self zombies clear self watcher sigchld self assertfalsecallback1 called self assertfalsecallback2 called waitpidmocks def testsigchldtwochildrenterminatingtogetherself mwaitpid callback1 mock mock callback2 mock mock register child 1 with self watcher self running true self watcher addchildhandler45 callback1 17 8 self assertfalsecallback1 called self assertfalsecallback2 called register child 2 with self watcher self watcher addchildhandler46 callback2 1147 18 self assertfalsecallback1 called self assertfalsecallback2 called children are running self watcher sigchld self assertfalsecallback1 called self assertfalsecallback2 called child 1 terminates code 78 child 2 terminates signal 5 self addzombie45 exitcode78 self addzombie46 signal5 self running false self watcher sigchld callback1 assertcalledoncewith45 78 17 8 callback2 assertcalledoncewith46 5 1147 18 callback1 resetmock callback2 resetmock ensure that the children are effectively reaped self addzombie45 exitcode14 self addzombie46 exitcode15 with self ignorewarnings self watcher sigchld self assertfalsecallback1 called self assertfalsecallback2 called waitpidmocks def testsigchldraceconditionself mwaitpid register a child callback mock mock with self watcher child terminates before being registered self addzombie50 exitcode4 self watcher sigchld self watcher addchildhandler50 callback 1 12 callback assertcalledoncewith50 4 1 12 callback resetmock ensure that the child is effectively reaped self addzombie50 signal1 with self ignorewarnings self watcher sigchld self assertfalsecallback called waitpidmocks def testsigchldreplacehandlerself mwaitpid callback1 mock mock callback2 mock mock register a child with self watcher self running true self watcher addchildhandler51 callback1 19 self assertfalsecallback1 called self assertfalsecallback2 called register the same child again with self watcher self watcher addchildhandler51 callback2 21 self assertfalsecallback1 called self assertfalsecallback2 called child terminates signal 8 self running false self addzombie51 signal8 self watcher sigchld callback2 assertcalledoncewith51 8 21 self assertfalsecallback1 called callback2 resetmock ensure that the child is effectively reaped self addzombie51 exitcode13 with self ignorewarnings self watcher sigchld self assertfalsecallback1 called self assertfalsecallback2 called waitpidmocks def testsigchldremovehandlerself mwaitpid callback mock mock register a child with self watcher self running true self watcher addchildhandler52 callback 1984 self assertfalsecallback called unregister the child self watcher removechildhandler52 self assertfalsecallback called child terminates code 99 self running false self addzombie52 exitcode99 with self ignorewarnings self watcher sigchld self assertfalsecallback called waitpidmocks def testsigchldunknownstatusself mwaitpid callback mock mock register a child with self watcher self running true self watcher addchildhandler53 callback 19 self assertfalsecallback called terminate with unknown status self zombies53 1178 self running false self watcher sigchld callback assertcalledoncewith53 1178 19 callback resetmock ensure that the child is effectively reaped self addzombie53 exitcode101 with self ignorewarnings self watcher sigchld self assertfalsecallback called waitpidmocks def testremovechildhandlerself mwaitpid callback1 mock mock callback2 mock mock callback3 mock mock register children with self watcher self running true self watcher addchildhandler54 callback1 1 self watcher addchildhandler55 callback2 2 self watcher addchildhandler56 callback3 3 remove child handler 1 self asserttrueself watcher removechildhandler54 remove child handler 2 multiple times self asserttrueself watcher removechildhandler55 self assertfalseself watcher removechildhandler55 self assertfalseself watcher removechildhandler55 all children terminate self addzombie54 exitcode0 self addzombie55 exitcode1 self addzombie56 exitcode2 self running false with self ignorewarnings self watcher sigchld self assertfalsecallback1 called self assertfalsecallback2 called callback3 assertcalledoncewith56 2 3 waitpidmocks def testsigchldunhandledexceptionself mwaitpid callback mock mock register a child with self watcher self running true self watcher addchildhandler57 callback raise an exception mwaitpid sideeffect valueerror with mock patch objectlog logger error as merror self assertequalself watcher sigchld none self asserttruemerror called waitpidmocks def testsigchldchildreapedelsewhereself mwaitpid register a child callback mock mock with self watcher self running true self watcher addchildhandler58 callback self assertfalsecallback called child terminates self running false self addzombie58 exitcode4 waitpid is called elsewhere os waitpid58 os wnohang mwaitpid resetmock sigchld with self ignorewarnings self watcher sigchld if isinstanceself watcher asyncio fastchildwatcher here the fastchildwatcher enters a deadlock there is no way to prevent it self assertfalsecallback called else callback assertcalledoncewith58 255 waitpidmocks def testsigchldunknownpidduringregistrationself mwaitpid register two children callback1 mock mock callback2 mock mock with self ignorewarnings self watcher self running true child 1 terminates self addzombie591 exitcode7 an unknown child terminates self addzombie593 exitcode17 self watcher sigchld self watcher addchildhandler591 callback1 self watcher addchildhandler592 callback2 callback1 assertcalledoncewith591 7 self assertfalsecallback2 called waitpidmocks def testsetloopself mwaitpid register a child callback mock mock with self watcher self running true self watcher addchildhandler60 callback attach a new loop oldloop self loop self loop self newtestloop patch mock patch object with patcholdloop removesignalhandler as moldremove patchself loop addsignalhandler as mnewadd self watcher attachloopself loop moldremove assertcalledoncewith signal sigchld mnewadd assertcalledoncewith signal sigchld self watcher sigchld child terminates self running false self addzombie60 exitcode9 self watcher sigchld callback assertcalledoncewith60 9 waitpidmocks def testsetloopraceconditionself mwaitpid register 3 children callback1 mock mock callback2 mock mock callback3 mock mock with self watcher self running true self watcher addchildhandler61 callback1 self watcher addchildhandler62 callback2 self watcher addchildhandler622 callback3 detach the loop oldloop self loop self loop none with mock patch object oldloop removesignalhandler as mremovesignalhandler with self assertwarnsregex runtimewarning a loop is being detached self watcher attachloopnone mremovesignalhandler assertcalledoncewith signal sigchld child 1 2 terminate self addzombie61 exitcode11 self addzombie62 signal5 sigchld was not caught self assertfalsecallback1 called self assertfalsecallback2 called self assertfalsecallback3 called attach a new loop self loop self newtestloop with mock patch object self loop addsignalhandler as maddsignalhandler self watcher attachloopself loop maddsignalhandler assertcalledoncewith signal sigchld self watcher sigchld callback1 assertcalledoncewith61 11 race condition callback2 assertcalledoncewith62 5 race condition self assertfalsecallback3 called callback1 resetmock callback2 resetmock child 3 terminates self running false self addzombie622 exitcode19 self watcher sigchld self assertfalsecallback1 called self assertfalsecallback2 called callback3 assertcalledoncewith622 19 waitpidmocks def testcloseself mwaitpid register two children callback1 mock mock with self watcher self running true child 1 terminates self addzombie63 exitcode9 other child terminates self addzombie65 exitcode18 self watcher sigchld self watcher addchildhandler63 callback1 self watcher addchildhandler64 callback1 self assertequallenself watcher callbacks 1 if isinstanceself watcher asyncio fastchildwatcher self assertequallenself watcher zombies 1 with mock patch object self loop removesignalhandler as mremovesignalhandler self watcher close mremovesignalhandler assertcalledoncewith signal sigchld self assertfalseself watcher callbacks if isinstanceself watcher asyncio fastchildwatcher self assertfalseself watcher zombies class safechildwatchertests childwatchertestsmixin testutils testcase def createwatcherself with warnings catchwarnings warnings simplefilterignore deprecationwarning return asyncio safechildwatcher class fastchildwatchertests childwatchertestsmixin testutils testcase def createwatcherself with warnings catchwarnings warnings simplefilterignore deprecationwarning return asyncio fastchildwatcher class policytestsunittest testcase def createpolicyself return asyncio defaulteventlooppolicy mock patch asyncio unixevents canusepidfd def testgetdefaultchildwatcherself mcanusepidfd mcanusepidfd returnvalue false policy self createpolicy self assertisnonepolicy watcher with self assertwarnsdeprecationwarning watcher policy getchildwatcher self assertisinstancewatcher asyncio threadedchildwatcher self assertispolicy watcher watcher with self assertwarnsdeprecationwarning self assertiswatcher policy getchildwatcher mcanusepidfd returnvalue true policy self createpolicy self assertisnonepolicy watcher with self assertwarnsdeprecationwarning watcher policy getchildwatcher self assertisinstancewatcher asyncio pidfdchildwatcher self assertispolicy watcher watcher with self assertwarnsdeprecationwarning self assertiswatcher policy getchildwatcher def testgetchildwatcheraftersetself policy self createpolicy with warnings catchwarnings warnings simplefilterignore deprecationwarning watcher asyncio fastchildwatcher policy setchildwatcherwatcher self assertispolicy watcher watcher with self assertwarnsdeprecationwarning self assertiswatcher policy getchildwatcher def testgetchildwatcherthreadself def f policy seteventlooppolicy neweventloop self assertisinstancepolicy geteventloop asyncio abstracteventloop with warnings catchwarnings warnings simplefilterignore deprecationwarning watcher policy getchildwatcher self assertisinstancewatcher asyncio safechildwatcher self assertisnonewatcher loop policy geteventloop close policy self createpolicy with warnings catchwarnings warnings simplefilterignore deprecationwarning policy setchildwatcherasyncio safechildwatcher th threading threadtargetf th start th join def testchildwatcherreplacemainloopexistingself policy self createpolicy loop policy neweventloop policy seteventlooploop explicitly setup safechildwatcher default threadedchildwatcher has no loop property with warnings catchwarnings warnings simplefilterignore deprecationwarning watcher asyncio safechildwatcher policy setchildwatcherwatcher watcher attachlooploop self assertiswatcher loop loop newloop policy neweventloop policy seteventloopnewloop self assertiswatcher loop newloop policy seteventloopnone self assertiswatcher loop none loop close newloop close class testfunctionalunittest testcase def setupself self loop asyncio neweventloop asyncio seteventloopself loop def teardownself self loop close asyncio seteventloopnone def testaddreaderinvalidargumentself def assertraises return self assertraisesregexvalueerror r invalid file object cb lambda none with assertraises self loop addreaderobject cb with assertraises self loop addwriterobject cb with assertraises self loop removereaderobject with assertraises self loop removewriterobject def testaddreaderorwritertransportfdself def assertraises return self assertraisesregex runtimeerror r file descriptor is used by transport async def runner tr pr await self loop createconnection lambda asyncio protocol sockrsock try cb lambda none with assertraises self loop addreaderrsock cb with assertraises self loop addreaderrsock fileno cb with assertraises self loop removereaderrsock with assertraises self loop removereaderrsock fileno with assertraises self loop addwriterrsock cb with assertraises self loop addwriterrsock fileno cb with assertraises self loop removewriterrsock with assertraises self loop removewriterrsock fileno finally tr close rsock wsock socket socketpair try self loop rununtilcompleterunner finally rsock close wsock close unittest skipunlesshasattros fork requires os fork class testforkunittest isolatedasynciotestcase async def testforknotshareeventloopself the forked process should not share the event loop with the parent loop asyncio getrunningloop r w os pipe self addcleanupos close r self addcleanupos close w pid os fork if pid 0 child try with self assertwarnsdeprecationwarning loop asyncio geteventlooppolicy geteventloop os writew b loop stridloop encode except runtimeerror os writew b no loop except baseexception as e os writew b error asciie encode finally os exit0 else parent result os readr 100 self assertequalresult 5 b loop result self assertnotequalintresult5 idloop waitprocesspid exitcode0 hashlibhelper requireshashdigest md5 def testforksignalhandlingself self addcleanupmultiprocessingcleanuptests sending signal to the forked process should not affect the parent process ctx multiprocessing getcontext fork manager ctx manager self addcleanupmanager shutdown childstarted manager event childhandled manager event parenthandled manager event def childmain def onsigtermargs childhandled set sys exit signal signalsignal sigterm onsigterm childstarted set while true time sleep1 async def main loop asyncio getrunningloop loop addsignalhandlersignal sigterm lambda args parenthandled set process ctx processtargetchildmain process start childstarted wait os killprocess pid signal sigterm process jointimeoutsupport shorttimeout async def func await asyncio sleep0 1 return 42 test parent s loop is still functional self assertequalawait asyncio createtaskfunc 42 asyncio runmain childhandled waittimeoutsupport shorttimeout self assertfalseparenthandled isset self asserttruechildhandled isset hashlibhelper requireshashdigest md5 def testforkasynciorunself self addcleanupmultiprocessingcleanuptests ctx multiprocessing getcontext fork manager ctx manager self addcleanupmanager shutdown result manager value i 0 async def childmain await asyncio sleep0 1 result value 42 process ctx processtargetlambda asyncio runchildmain process start process join self assertequalresult value 42 hashlibhelper requireshashdigest md5 def testforkasynciosubprocessself self addcleanupmultiprocessingcleanuptests ctx multiprocessing getcontext fork manager ctx manager self addcleanupmanager shutdown result manager value i 1 async def childmain proc await asyncio createsubprocessexecsys executable c pass result value await proc wait process ctx processtargetlambda asyncio runchildmain process start process join self assertequalresult value 0 if name main unittest main don t call transport close because the event loop and the selector are mocked callback must not be a coroutine function if pathlib path wasn t supported the exception would be different create the file ensure that the socket is closed on any bind error 160 kib the second call should do nothing the pipe is not paused resuming should do nothing this is a bit overspecified pipe was closed by peer closing the transport twice must not fail register a child child is running child terminates returncode 12 ensure that the child is effectively reaped sigchld called again register child 1 register child 2 children are running child 1 terminates signal 3 child 2 still running child 2 terminates code 108 ensure that the children are effectively reaped sigchld called again register child 1 register child 2 children are running child 1 terminates code 78 child 2 terminates signal 5 ensure that the children are effectively reaped register a child child terminates before being registered ensure that the child is effectively reaped register a child register the same child again child terminates signal 8 ensure that the child is effectively reaped register a child unregister the child child terminates code 99 register a child terminate with unknown status ensure that the child is effectively reaped register children remove child handler 1 remove child handler 2 multiple times all children terminate register a child raise an exception register a child child terminates waitpid is called elsewhere sigchld here the fastchildwatcher enters a deadlock there is no way to prevent it register two children child 1 terminates an unknown child terminates register a child attach a new loop child terminates register 3 children detach the loop child 1 2 terminate sigchld was not caught attach a new loop race condition race condition child 3 terminates register two children child 1 terminates other child terminates explicitly setup safechildwatcher default threadedchildwatcher has no _loop property the forked process should not share the event loop with the parent child parent sending signal to the forked process should not affect the parent process test parent s loop is still functional
import contextlib
import errno
import io
import multiprocessing
from multiprocessing.util import _cleanup_tests as multiprocessing_cleanup_tests
import os
import pathlib
import signal
import socket
import stat
import sys
import threading
import time
import unittest
from unittest import mock
import warnings

from test import support
from test.support import os_helper
from test.support import socket_helper
from test.support import wait_process
from test.support import hashlib_helper

if sys.platform == 'win32':
    raise unittest.SkipTest('UNIX only')


import asyncio
from asyncio import log
from asyncio import unix_events
from test.test_asyncio import utils as test_utils


def tearDownModule():
    asyncio.set_event_loop_policy(None)


MOCK_ANY = mock.ANY


def EXITCODE(exitcode):
    return 32768 + exitcode


def SIGNAL(signum):
    if not 1 <= signum <= 68:
        raise AssertionError(f'invalid signum {signum}')
    return 32768 - signum


def close_pipe_transport(transport):
    
    
    if transport._pipe is None:
        return
    transport._pipe.close()
    transport._pipe = None


@unittest.skipUnless(signal, 'Signals are not supported')
class SelectorEventLoopSignalTests(test_utils.TestCase):

    def setUp(self):
        super().setUp()
        self.loop = asyncio.SelectorEventLoop()
        self.set_event_loop(self.loop)

    def test_check_signal(self):
        self.assertRaises(
            TypeError, self.loop._check_signal, '1')
        self.assertRaises(
            ValueError, self.loop._check_signal, signal.NSIG + 1)

    def test_handle_signal_no_handler(self):
        self.loop._handle_signal(signal.NSIG + 1)

    def test_handle_signal_cancelled_handler(self):
        h = asyncio.Handle(mock.Mock(), (),
                           loop=mock.Mock())
        h.cancel()
        self.loop._signal_handlers[signal.NSIG + 1] = h
        self.loop.remove_signal_handler = mock.Mock()
        self.loop._handle_signal(signal.NSIG + 1)
        self.loop.remove_signal_handler.assert_called_with(signal.NSIG + 1)

    @mock.patch('asyncio.unix_events.signal')
    def test_add_signal_handler_setup_error(self, m_signal):
        m_signal.NSIG = signal.NSIG
        m_signal.valid_signals = signal.valid_signals
        m_signal.set_wakeup_fd.side_effect = ValueError

        self.assertRaises(
            RuntimeError,
            self.loop.add_signal_handler,
            signal.SIGINT, lambda: True)

    @mock.patch('asyncio.unix_events.signal')
    def test_add_signal_handler_coroutine_error(self, m_signal):
        m_signal.NSIG = signal.NSIG

        async def simple_coroutine():
            pass

        
        coro_func = simple_coroutine
        coro_obj = coro_func()
        self.addCleanup(coro_obj.close)
        for func in (coro_func, coro_obj):
            self.assertRaisesRegex(
                TypeError, 'coroutines cannot be used with add_signal_handler',
                self.loop.add_signal_handler,
                signal.SIGINT, func)

    @mock.patch('asyncio.unix_events.signal')
    def test_add_signal_handler(self, m_signal):
        m_signal.NSIG = signal.NSIG
        m_signal.valid_signals = signal.valid_signals

        cb = lambda: True
        self.loop.add_signal_handler(signal.SIGHUP, cb)
        h = self.loop._signal_handlers.get(signal.SIGHUP)
        self.assertIsInstance(h, asyncio.Handle)
        self.assertEqual(h._callback, cb)

    @mock.patch('asyncio.unix_events.signal')
    def test_add_signal_handler_install_error(self, m_signal):
        m_signal.NSIG = signal.NSIG
        m_signal.valid_signals = signal.valid_signals

        def set_wakeup_fd(fd):
            if fd == -1:
                raise ValueError()
        m_signal.set_wakeup_fd = set_wakeup_fd

        class Err(OSError):
            errno = errno.EFAULT
        m_signal.signal.side_effect = Err

        self.assertRaises(
            Err,
            self.loop.add_signal_handler,
            signal.SIGINT, lambda: True)

    @mock.patch('asyncio.unix_events.signal')
    @mock.patch('asyncio.base_events.logger')
    def test_add_signal_handler_install_error2(self, m_logging, m_signal):
        m_signal.NSIG = signal.NSIG
        m_signal.valid_signals = signal.valid_signals

        class Err(OSError):
            errno = errno.EINVAL
        m_signal.signal.side_effect = Err

        self.loop._signal_handlers[signal.SIGHUP] = lambda: True
        self.assertRaises(
            RuntimeError,
            self.loop.add_signal_handler,
            signal.SIGINT, lambda: True)
        self.assertFalse(m_logging.info.called)
        self.assertEqual(1, m_signal.set_wakeup_fd.call_count)

    @mock.patch('asyncio.unix_events.signal')
    @mock.patch('asyncio.base_events.logger')
    def test_add_signal_handler_install_error3(self, m_logging, m_signal):
        class Err(OSError):
            errno = errno.EINVAL
        m_signal.signal.side_effect = Err
        m_signal.NSIG = signal.NSIG
        m_signal.valid_signals = signal.valid_signals

        self.assertRaises(
            RuntimeError,
            self.loop.add_signal_handler,
            signal.SIGINT, lambda: True)
        self.assertFalse(m_logging.info.called)
        self.assertEqual(2, m_signal.set_wakeup_fd.call_count)

    @mock.patch('asyncio.unix_events.signal')
    def test_remove_signal_handler(self, m_signal):
        m_signal.NSIG = signal.NSIG
        m_signal.valid_signals = signal.valid_signals

        self.loop.add_signal_handler(signal.SIGHUP, lambda: True)

        self.assertTrue(
            self.loop.remove_signal_handler(signal.SIGHUP))
        self.assertTrue(m_signal.set_wakeup_fd.called)
        self.assertTrue(m_signal.signal.called)
        self.assertEqual(
            (signal.SIGHUP, m_signal.SIG_DFL), m_signal.signal.call_args[0])

    @mock.patch('asyncio.unix_events.signal')
    def test_remove_signal_handler_2(self, m_signal):
        m_signal.NSIG = signal.NSIG
        m_signal.SIGINT = signal.SIGINT
        m_signal.valid_signals = signal.valid_signals

        self.loop.add_signal_handler(signal.SIGINT, lambda: True)
        self.loop._signal_handlers[signal.SIGHUP] = object()
        m_signal.set_wakeup_fd.reset_mock()

        self.assertTrue(
            self.loop.remove_signal_handler(signal.SIGINT))
        self.assertFalse(m_signal.set_wakeup_fd.called)
        self.assertTrue(m_signal.signal.called)
        self.assertEqual(
            (signal.SIGINT, m_signal.default_int_handler),
            m_signal.signal.call_args[0])

    @mock.patch('asyncio.unix_events.signal')
    @mock.patch('asyncio.base_events.logger')
    def test_remove_signal_handler_cleanup_error(self, m_logging, m_signal):
        m_signal.NSIG = signal.NSIG
        m_signal.valid_signals = signal.valid_signals
        self.loop.add_signal_handler(signal.SIGHUP, lambda: True)

        m_signal.set_wakeup_fd.side_effect = ValueError

        self.loop.remove_signal_handler(signal.SIGHUP)
        self.assertTrue(m_logging.info)

    @mock.patch('asyncio.unix_events.signal')
    def test_remove_signal_handler_error(self, m_signal):
        m_signal.NSIG = signal.NSIG
        m_signal.valid_signals = signal.valid_signals
        self.loop.add_signal_handler(signal.SIGHUP, lambda: True)

        m_signal.signal.side_effect = OSError

        self.assertRaises(
            OSError, self.loop.remove_signal_handler, signal.SIGHUP)

    @mock.patch('asyncio.unix_events.signal')
    def test_remove_signal_handler_error2(self, m_signal):
        m_signal.NSIG = signal.NSIG
        m_signal.valid_signals = signal.valid_signals
        self.loop.add_signal_handler(signal.SIGHUP, lambda: True)

        class Err(OSError):
            errno = errno.EINVAL
        m_signal.signal.side_effect = Err

        self.assertRaises(
            RuntimeError, self.loop.remove_signal_handler, signal.SIGHUP)

    @mock.patch('asyncio.unix_events.signal')
    def test_close(self, m_signal):
        m_signal.NSIG = signal.NSIG
        m_signal.valid_signals = signal.valid_signals

        self.loop.add_signal_handler(signal.SIGHUP, lambda: True)
        self.loop.add_signal_handler(signal.SIGCHLD, lambda: True)

        self.assertEqual(len(self.loop._signal_handlers), 2)

        m_signal.set_wakeup_fd.reset_mock()

        self.loop.close()

        self.assertEqual(len(self.loop._signal_handlers), 0)
        m_signal.set_wakeup_fd.assert_called_once_with(-1)

    @mock.patch('asyncio.unix_events.sys')
    @mock.patch('asyncio.unix_events.signal')
    def test_close_on_finalizing(self, m_signal, m_sys):
        m_signal.NSIG = signal.NSIG
        m_signal.valid_signals = signal.valid_signals
        self.loop.add_signal_handler(signal.SIGHUP, lambda: True)

        self.assertEqual(len(self.loop._signal_handlers), 1)
        m_sys.is_finalizing.return_value = True
        m_signal.signal.reset_mock()

        with self.assertWarnsRegex(ResourceWarning,
                                   "skipping signal handlers removal"):
            self.loop.close()

        self.assertEqual(len(self.loop._signal_handlers), 0)
        self.assertFalse(m_signal.signal.called)


@unittest.skipUnless(hasattr(socket, 'AF_UNIX'),
                     'UNIX Sockets are not supported')
class SelectorEventLoopUnixSocketTests(test_utils.TestCase):

    def setUp(self):
        super().setUp()
        self.loop = asyncio.SelectorEventLoop()
        self.set_event_loop(self.loop)

    @socket_helper.skip_unless_bind_unix_socket
    def test_create_unix_server_existing_path_sock(self):
        with test_utils.unix_socket_path() as path:
            sock = socket.socket(socket.AF_UNIX)
            sock.bind(path)
            sock.listen(1)
            sock.close()

            coro = self.loop.create_unix_server(lambda: None, path)
            srv = self.loop.run_until_complete(coro)
            srv.close()
            self.loop.run_until_complete(srv.wait_closed())

    @socket_helper.skip_unless_bind_unix_socket
    def test_create_unix_server_pathlib(self):
        with test_utils.unix_socket_path() as path:
            path = pathlib.Path(path)
            srv_coro = self.loop.create_unix_server(lambda: None, path)
            srv = self.loop.run_until_complete(srv_coro)
            srv.close()
            self.loop.run_until_complete(srv.wait_closed())

    def test_create_unix_connection_pathlib(self):
        with test_utils.unix_socket_path() as path:
            path = pathlib.Path(path)
            coro = self.loop.create_unix_connection(lambda: None, path)
            with self.assertRaises(FileNotFoundError):
                
                
                self.loop.run_until_complete(coro)

    def test_create_unix_server_existing_path_nonsock(self):
        path = test_utils.gen_unix_socket_path()
        self.addCleanup(os_helper.unlink, path)
        
        open(path, "wb").close()

        coro = self.loop.create_unix_server(lambda: None, path)
        with self.assertRaisesRegex(OSError,
                                    'Address.*is already in use'):
            self.loop.run_until_complete(coro)

    def test_create_unix_server_ssl_bool(self):
        coro = self.loop.create_unix_server(lambda: None, path='spam',
                                            ssl=True)
        with self.assertRaisesRegex(TypeError,
                                    'ssl argument must be an SSLContext'):
            self.loop.run_until_complete(coro)

    def test_create_unix_server_nopath_nosock(self):
        coro = self.loop.create_unix_server(lambda: None, path=None)
        with self.assertRaisesRegex(ValueError,
                                    'path was not specified, and no sock'):
            self.loop.run_until_complete(coro)

    def test_create_unix_server_path_inetsock(self):
        sock = socket.socket()
        with sock:
            coro = self.loop.create_unix_server(lambda: None, path=None,
                                                sock=sock)
            with self.assertRaisesRegex(ValueError,
                                        'A UNIX Domain Stream.*was expected'):
                self.loop.run_until_complete(coro)

    def test_create_unix_server_path_dgram(self):
        sock = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)
        with sock:
            coro = self.loop.create_unix_server(lambda: None, path=None,
                                                sock=sock)
            with self.assertRaisesRegex(ValueError,
                                        'A UNIX Domain Stream.*was expected'):
                self.loop.run_until_complete(coro)

    @unittest.skipUnless(hasattr(socket, 'SOCK_NONBLOCK'),
                         'no socket.SOCK_NONBLOCK (linux only)')
    @socket_helper.skip_unless_bind_unix_socket
    def test_create_unix_server_path_stream_bittype(self):
        fn = test_utils.gen_unix_socket_path()
        self.addCleanup(os_helper.unlink, fn)

        sock = socket.socket(socket.AF_UNIX,
                             socket.SOCK_STREAM | socket.SOCK_NONBLOCK)
        with sock:
            sock.bind(fn)
            coro = self.loop.create_unix_server(lambda: None, path=None,
                                                sock=sock)
            srv = self.loop.run_until_complete(coro)
            srv.close()
            self.loop.run_until_complete(srv.wait_closed())

    def test_create_unix_server_ssl_timeout_with_plain_sock(self):
        coro = self.loop.create_unix_server(lambda: None, path='spam',
                                            ssl_handshake_timeout=1)
        with self.assertRaisesRegex(
                ValueError,
                'ssl_handshake_timeout is only meaningful with ssl'):
            self.loop.run_until_complete(coro)

    def test_create_unix_connection_path_inetsock(self):
        sock = socket.socket()
        with sock:
            coro = self.loop.create_unix_connection(lambda: None,
                                                    sock=sock)
            with self.assertRaisesRegex(ValueError,
                                        'A UNIX Domain Stream.*was expected'):
                self.loop.run_until_complete(coro)

    @mock.patch('asyncio.unix_events.socket')
    def test_create_unix_server_bind_error(self, m_socket):
        
        sock = mock.Mock()
        m_socket.socket.return_value = sock

        sock.bind.side_effect = OSError
        coro = self.loop.create_unix_server(lambda: None, path="/test")
        with self.assertRaises(OSError):
            self.loop.run_until_complete(coro)
        self.assertTrue(sock.close.called)

        sock.bind.side_effect = MemoryError
        coro = self.loop.create_unix_server(lambda: None, path="/test")
        with self.assertRaises(MemoryError):
            self.loop.run_until_complete(coro)
        self.assertTrue(sock.close.called)

    def test_create_unix_connection_path_sock(self):
        coro = self.loop.create_unix_connection(
            lambda: None, os.devnull, sock=object())
        with self.assertRaisesRegex(ValueError, 'path and sock can not be'):
            self.loop.run_until_complete(coro)

    def test_create_unix_connection_nopath_nosock(self):
        coro = self.loop.create_unix_connection(
            lambda: None, None)
        with self.assertRaisesRegex(ValueError,
                                    'no path and sock were specified'):
            self.loop.run_until_complete(coro)

    def test_create_unix_connection_nossl_serverhost(self):
        coro = self.loop.create_unix_connection(
            lambda: None, os.devnull, server_hostname='spam')
        with self.assertRaisesRegex(ValueError,
                                    'server_hostname is only meaningful'):
            self.loop.run_until_complete(coro)

    def test_create_unix_connection_ssl_noserverhost(self):
        coro = self.loop.create_unix_connection(
            lambda: None, os.devnull, ssl=True)

        with self.assertRaisesRegex(
            ValueError, 'you have to pass server_hostname when using ssl'):

            self.loop.run_until_complete(coro)

    def test_create_unix_connection_ssl_timeout_with_plain_sock(self):
        coro = self.loop.create_unix_connection(lambda: None, path='spam',
                                            ssl_handshake_timeout=1)
        with self.assertRaisesRegex(
                ValueError,
                'ssl_handshake_timeout is only meaningful with ssl'):
            self.loop.run_until_complete(coro)


@unittest.skipUnless(hasattr(os, 'sendfile'),
                     'sendfile is not supported')
class SelectorEventLoopUnixSockSendfileTests(test_utils.TestCase):
    DATA = b"12345abcde" * 16 * 1024  

    class MyProto(asyncio.Protocol):

        def __init__(self, loop):
            self.started = False
            self.closed = False
            self.data = bytearray()
            self.fut = loop.create_future()
            self.transport = None
            self._ready = loop.create_future()

        def connection_made(self, transport):
            self.started = True
            self.transport = transport
            self._ready.set_result(None)

        def data_received(self, data):
            self.data.extend(data)

        def connection_lost(self, exc):
            self.closed = True
            self.fut.set_result(None)

        async def wait_closed(self):
            await self.fut

    @classmethod
    def setUpClass(cls):
        with open(os_helper.TESTFN, 'wb') as fp:
            fp.write(cls.DATA)
        super().setUpClass()

    @classmethod
    def tearDownClass(cls):
        os_helper.unlink(os_helper.TESTFN)
        super().tearDownClass()

    def setUp(self):
        self.loop = asyncio.new_event_loop()
        self.set_event_loop(self.loop)
        self.file = open(os_helper.TESTFN, 'rb')
        self.addCleanup(self.file.close)
        super().setUp()

    def make_socket(self, cleanup=True):
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.setblocking(False)
        sock.setsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF, 1024)
        sock.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 1024)
        if cleanup:
            self.addCleanup(sock.close)
        return sock

    def run_loop(self, coro):
        return self.loop.run_until_complete(coro)

    def prepare(self):
        sock = self.make_socket()
        proto = self.MyProto(self.loop)
        port = socket_helper.find_unused_port()
        srv_sock = self.make_socket(cleanup=False)
        srv_sock.bind((socket_helper.HOST, port))
        server = self.run_loop(self.loop.create_server(
            lambda: proto, sock=srv_sock))
        self.run_loop(self.loop.sock_connect(sock, (socket_helper.HOST, port)))
        self.run_loop(proto._ready)

        def cleanup():
            proto.transport.close()
            self.run_loop(proto.wait_closed())

            server.close()
            self.run_loop(server.wait_closed())

        self.addCleanup(cleanup)

        return sock, proto

    def test_sock_sendfile_not_available(self):
        sock, proto = self.prepare()
        with mock.patch('asyncio.unix_events.os', spec=[]):
            with self.assertRaisesRegex(asyncio.SendfileNotAvailableError,
                                        "os[.]sendfile[(][)] is not available"):
                self.run_loop(self.loop._sock_sendfile_native(sock, self.file,
                                                              0, None))
        self.assertEqual(self.file.tell(), 0)

    def test_sock_sendfile_not_a_file(self):
        sock, proto = self.prepare()
        f = object()
        with self.assertRaisesRegex(asyncio.SendfileNotAvailableError,
                                    "not a regular file"):
            self.run_loop(self.loop._sock_sendfile_native(sock, f,
                                                          0, None))
        self.assertEqual(self.file.tell(), 0)

    def test_sock_sendfile_iobuffer(self):
        sock, proto = self.prepare()
        f = io.BytesIO()
        with self.assertRaisesRegex(asyncio.SendfileNotAvailableError,
                                    "not a regular file"):
            self.run_loop(self.loop._sock_sendfile_native(sock, f,
                                                          0, None))
        self.assertEqual(self.file.tell(), 0)

    def test_sock_sendfile_not_regular_file(self):
        sock, proto = self.prepare()
        f = mock.Mock()
        f.fileno.return_value = -1
        with self.assertRaisesRegex(asyncio.SendfileNotAvailableError,
                                    "not a regular file"):
            self.run_loop(self.loop._sock_sendfile_native(sock, f,
                                                          0, None))
        self.assertEqual(self.file.tell(), 0)

    def test_sock_sendfile_cancel1(self):
        sock, proto = self.prepare()

        fut = self.loop.create_future()
        fileno = self.file.fileno()
        self.loop._sock_sendfile_native_impl(fut, None, sock, fileno,
                                             0, None, len(self.DATA), 0)
        fut.cancel()
        with contextlib.suppress(asyncio.CancelledError):
            self.run_loop(fut)
        with self.assertRaises(KeyError):
            self.loop._selector.get_key(sock)

    def test_sock_sendfile_cancel2(self):
        sock, proto = self.prepare()

        fut = self.loop.create_future()
        fileno = self.file.fileno()
        self.loop._sock_sendfile_native_impl(fut, None, sock, fileno,
                                             0, None, len(self.DATA), 0)
        fut.cancel()
        self.loop._sock_sendfile_native_impl(fut, sock.fileno(), sock, fileno,
                                             0, None, len(self.DATA), 0)
        with self.assertRaises(KeyError):
            self.loop._selector.get_key(sock)

    def test_sock_sendfile_blocking_error(self):
        sock, proto = self.prepare()

        fileno = self.file.fileno()
        fut = mock.Mock()
        fut.cancelled.return_value = False
        with mock.patch('os.sendfile', side_effect=BlockingIOError()):
            self.loop._sock_sendfile_native_impl(fut, None, sock, fileno,
                                                 0, None, len(self.DATA), 0)
        key = self.loop._selector.get_key(sock)
        self.assertIsNotNone(key)
        fut.add_done_callback.assert_called_once_with(mock.ANY)

    def test_sock_sendfile_os_error_first_call(self):
        sock, proto = self.prepare()

        fileno = self.file.fileno()
        fut = self.loop.create_future()
        with mock.patch('os.sendfile', side_effect=OSError()):
            self.loop._sock_sendfile_native_impl(fut, None, sock, fileno,
                                                 0, None, len(self.DATA), 0)
        with self.assertRaises(KeyError):
            self.loop._selector.get_key(sock)
        exc = fut.exception()
        self.assertIsInstance(exc, asyncio.SendfileNotAvailableError)
        self.assertEqual(0, self.file.tell())

    def test_sock_sendfile_os_error_next_call(self):
        sock, proto = self.prepare()

        fileno = self.file.fileno()
        fut = self.loop.create_future()
        err = OSError()
        with mock.patch('os.sendfile', side_effect=err):
            self.loop._sock_sendfile_native_impl(fut, sock.fileno(),
                                                 sock, fileno,
                                                 1000, None, len(self.DATA),
                                                 1000)
        with self.assertRaises(KeyError):
            self.loop._selector.get_key(sock)
        exc = fut.exception()
        self.assertIs(exc, err)
        self.assertEqual(1000, self.file.tell())

    def test_sock_sendfile_exception(self):
        sock, proto = self.prepare()

        fileno = self.file.fileno()
        fut = self.loop.create_future()
        err = asyncio.SendfileNotAvailableError()
        with mock.patch('os.sendfile', side_effect=err):
            self.loop._sock_sendfile_native_impl(fut, sock.fileno(),
                                                 sock, fileno,
                                                 1000, None, len(self.DATA),
                                                 1000)
        with self.assertRaises(KeyError):
            self.loop._selector.get_key(sock)
        exc = fut.exception()
        self.assertIs(exc, err)
        self.assertEqual(1000, self.file.tell())


class UnixReadPipeTransportTests(test_utils.TestCase):

    def setUp(self):
        super().setUp()
        self.loop = self.new_test_loop()
        self.protocol = test_utils.make_test_protocol(asyncio.Protocol)
        self.pipe = mock.Mock(spec_set=io.RawIOBase)
        self.pipe.fileno.return_value = 5

        blocking_patcher = mock.patch('os.set_blocking')
        blocking_patcher.start()
        self.addCleanup(blocking_patcher.stop)

        fstat_patcher = mock.patch('os.fstat')
        m_fstat = fstat_patcher.start()
        st = mock.Mock()
        st.st_mode = stat.S_IFIFO
        m_fstat.return_value = st
        self.addCleanup(fstat_patcher.stop)

    def read_pipe_transport(self, waiter=None):
        transport = unix_events._UnixReadPipeTransport(self.loop, self.pipe,
                                                       self.protocol,
                                                       waiter=waiter)
        self.addCleanup(close_pipe_transport, transport)
        return transport

    def test_ctor(self):
        waiter = self.loop.create_future()
        tr = self.read_pipe_transport(waiter=waiter)
        self.loop.run_until_complete(waiter)

        self.protocol.connection_made.assert_called_with(tr)
        self.loop.assert_reader(5, tr._read_ready)
        self.assertIsNone(waiter.result())

    @mock.patch('os.read')
    def test__read_ready(self, m_read):
        tr = self.read_pipe_transport()
        m_read.return_value = b'data'
        tr._read_ready()

        m_read.assert_called_with(5, tr.max_size)
        self.protocol.data_received.assert_called_with(b'data')

    @mock.patch('os.read')
    def test__read_ready_eof(self, m_read):
        tr = self.read_pipe_transport()
        m_read.return_value = b''
        tr._read_ready()

        m_read.assert_called_with(5, tr.max_size)
        self.assertFalse(self.loop.readers)
        test_utils.run_briefly(self.loop)
        self.protocol.eof_received.assert_called_with()
        self.protocol.connection_lost.assert_called_with(None)

    @mock.patch('os.read')
    def test__read_ready_blocked(self, m_read):
        tr = self.read_pipe_transport()
        m_read.side_effect = BlockingIOError
        tr._read_ready()

        m_read.assert_called_with(5, tr.max_size)
        test_utils.run_briefly(self.loop)
        self.assertFalse(self.protocol.data_received.called)

    @mock.patch('asyncio.log.logger.error')
    @mock.patch('os.read')
    def test__read_ready_error(self, m_read, m_logexc):
        tr = self.read_pipe_transport()
        err = OSError()
        m_read.side_effect = err
        tr._close = mock.Mock()
        tr._read_ready()

        m_read.assert_called_with(5, tr.max_size)
        tr._close.assert_called_with(err)
        m_logexc.assert_called_with(
            test_utils.MockPattern(
                'Fatal read error on pipe transport'
                '\nprotocol:.*\ntransport:.*'),
            exc_info=(OSError, MOCK_ANY, MOCK_ANY))

    @mock.patch('os.read')
    def test_pause_reading(self, m_read):
        tr = self.read_pipe_transport()
        m = mock.Mock()
        self.loop.add_reader(5, m)
        tr.pause_reading()
        self.assertFalse(self.loop.readers)

    @mock.patch('os.read')
    def test_resume_reading(self, m_read):
        tr = self.read_pipe_transport()
        tr.pause_reading()
        tr.resume_reading()
        self.loop.assert_reader(5, tr._read_ready)

    @mock.patch('os.read')
    def test_close(self, m_read):
        tr = self.read_pipe_transport()
        tr._close = mock.Mock()
        tr.close()
        tr._close.assert_called_with(None)

    @mock.patch('os.read')
    def test_close_already_closing(self, m_read):
        tr = self.read_pipe_transport()
        tr._closing = True
        tr._close = mock.Mock()
        tr.close()
        self.assertFalse(tr._close.called)

    @mock.patch('os.read')
    def test__close(self, m_read):
        tr = self.read_pipe_transport()
        err = object()
        tr._close(err)
        self.assertTrue(tr.is_closing())
        self.assertFalse(self.loop.readers)
        test_utils.run_briefly(self.loop)
        self.protocol.connection_lost.assert_called_with(err)

    def test__call_connection_lost(self):
        tr = self.read_pipe_transport()
        self.assertIsNotNone(tr._protocol)
        self.assertIsNotNone(tr._loop)

        err = None
        tr._call_connection_lost(err)
        self.protocol.connection_lost.assert_called_with(err)
        self.pipe.close.assert_called_with()

        self.assertIsNone(tr._protocol)
        self.assertIsNone(tr._loop)

    def test__call_connection_lost_with_err(self):
        tr = self.read_pipe_transport()
        self.assertIsNotNone(tr._protocol)
        self.assertIsNotNone(tr._loop)

        err = OSError()
        tr._call_connection_lost(err)
        self.protocol.connection_lost.assert_called_with(err)
        self.pipe.close.assert_called_with()

        self.assertIsNone(tr._protocol)
        self.assertIsNone(tr._loop)

    def test_pause_reading_on_closed_pipe(self):
        tr = self.read_pipe_transport()
        tr.close()
        test_utils.run_briefly(self.loop)
        self.assertIsNone(tr._loop)
        tr.pause_reading()

    def test_pause_reading_on_paused_pipe(self):
        tr = self.read_pipe_transport()
        tr.pause_reading()
        
        tr.pause_reading()

    def test_resume_reading_on_closed_pipe(self):
        tr = self.read_pipe_transport()
        tr.close()
        test_utils.run_briefly(self.loop)
        self.assertIsNone(tr._loop)
        tr.resume_reading()

    def test_resume_reading_on_paused_pipe(self):
        tr = self.read_pipe_transport()
        
        
        tr.resume_reading()


class UnixWritePipeTransportTests(test_utils.TestCase):

    def setUp(self):
        super().setUp()
        self.loop = self.new_test_loop()
        self.protocol = test_utils.make_test_protocol(asyncio.BaseProtocol)
        self.pipe = mock.Mock(spec_set=io.RawIOBase)
        self.pipe.fileno.return_value = 5

        blocking_patcher = mock.patch('os.set_blocking')
        blocking_patcher.start()
        self.addCleanup(blocking_patcher.stop)

        fstat_patcher = mock.patch('os.fstat')
        m_fstat = fstat_patcher.start()
        st = mock.Mock()
        st.st_mode = stat.S_IFSOCK
        m_fstat.return_value = st
        self.addCleanup(fstat_patcher.stop)

    def write_pipe_transport(self, waiter=None):
        transport = unix_events._UnixWritePipeTransport(self.loop, self.pipe,
                                                        self.protocol,
                                                        waiter=waiter)
        self.addCleanup(close_pipe_transport, transport)
        return transport

    def test_ctor(self):
        waiter = self.loop.create_future()
        tr = self.write_pipe_transport(waiter=waiter)
        self.loop.run_until_complete(waiter)

        self.protocol.connection_made.assert_called_with(tr)
        self.loop.assert_reader(5, tr._read_ready)
        self.assertEqual(None, waiter.result())

    def test_can_write_eof(self):
        tr = self.write_pipe_transport()
        self.assertTrue(tr.can_write_eof())

    @mock.patch('os.write')
    def test_write(self, m_write):
        tr = self.write_pipe_transport()
        m_write.return_value = 4
        tr.write(b'data')
        m_write.assert_called_with(5, b'data')
        self.assertFalse(self.loop.writers)
        self.assertEqual(bytearray(), tr._buffer)

    @mock.patch('os.write')
    def test_write_no_data(self, m_write):
        tr = self.write_pipe_transport()
        tr.write(b'')
        self.assertFalse(m_write.called)
        self.assertFalse(self.loop.writers)
        self.assertEqual(bytearray(b''), tr._buffer)

    @mock.patch('os.write')
    def test_write_partial(self, m_write):
        tr = self.write_pipe_transport()
        m_write.return_value = 2
        tr.write(b'data')
        self.loop.assert_writer(5, tr._write_ready)
        self.assertEqual(bytearray(b'ta'), tr._buffer)

    @mock.patch('os.write')
    def test_write_buffer(self, m_write):
        tr = self.write_pipe_transport()
        self.loop.add_writer(5, tr._write_ready)
        tr._buffer = bytearray(b'previous')
        tr.write(b'data')
        self.assertFalse(m_write.called)
        self.loop.assert_writer(5, tr._write_ready)
        self.assertEqual(bytearray(b'previousdata'), tr._buffer)

    @mock.patch('os.write')
    def test_write_again(self, m_write):
        tr = self.write_pipe_transport()
        m_write.side_effect = BlockingIOError()
        tr.write(b'data')
        m_write.assert_called_with(5, bytearray(b'data'))
        self.loop.assert_writer(5, tr._write_ready)
        self.assertEqual(bytearray(b'data'), tr._buffer)

    @mock.patch('asyncio.unix_events.logger')
    @mock.patch('os.write')
    def test_write_err(self, m_write, m_log):
        tr = self.write_pipe_transport()
        err = OSError()
        m_write.side_effect = err
        tr._fatal_error = mock.Mock()
        tr.write(b'data')
        m_write.assert_called_with(5, b'data')
        self.assertFalse(self.loop.writers)
        self.assertEqual(bytearray(), tr._buffer)
        tr._fatal_error.assert_called_with(
                            err,
                            'Fatal write error on pipe transport')
        self.assertEqual(1, tr._conn_lost)

        tr.write(b'data')
        self.assertEqual(2, tr._conn_lost)
        tr.write(b'data')
        tr.write(b'data')
        tr.write(b'data')
        tr.write(b'data')
        
        m_log.warning.assert_called_with(
            'pipe closed by peer or os.write(pipe, data) raised exception.')
        tr.close()

    @mock.patch('os.write')
    def test_write_close(self, m_write):
        tr = self.write_pipe_transport()
        tr._read_ready()  

        tr.write(b'data')
        self.assertEqual(tr._conn_lost, 1)
        tr.write(b'data')
        self.assertEqual(tr._conn_lost, 2)

    def test__read_ready(self):
        tr = self.write_pipe_transport()
        tr._read_ready()
        self.assertFalse(self.loop.readers)
        self.assertFalse(self.loop.writers)
        self.assertTrue(tr.is_closing())
        test_utils.run_briefly(self.loop)
        self.protocol.connection_lost.assert_called_with(None)

    @mock.patch('os.write')
    def test__write_ready(self, m_write):
        tr = self.write_pipe_transport()
        self.loop.add_writer(5, tr._write_ready)
        tr._buffer = bytearray(b'data')
        m_write.return_value = 4
        tr._write_ready()
        self.assertFalse(self.loop.writers)
        self.assertEqual(bytearray(), tr._buffer)

    @mock.patch('os.write')
    def test__write_ready_partial(self, m_write):
        tr = self.write_pipe_transport()
        self.loop.add_writer(5, tr._write_ready)
        tr._buffer = bytearray(b'data')
        m_write.return_value = 3
        tr._write_ready()
        self.loop.assert_writer(5, tr._write_ready)
        self.assertEqual(bytearray(b'a'), tr._buffer)

    @mock.patch('os.write')
    def test__write_ready_again(self, m_write):
        tr = self.write_pipe_transport()
        self.loop.add_writer(5, tr._write_ready)
        tr._buffer = bytearray(b'data')
        m_write.side_effect = BlockingIOError()
        tr._write_ready()
        m_write.assert_called_with(5, bytearray(b'data'))
        self.loop.assert_writer(5, tr._write_ready)
        self.assertEqual(bytearray(b'data'), tr._buffer)

    @mock.patch('os.write')
    def test__write_ready_empty(self, m_write):
        tr = self.write_pipe_transport()
        self.loop.add_writer(5, tr._write_ready)
        tr._buffer = bytearray(b'data')
        m_write.return_value = 0
        tr._write_ready()
        m_write.assert_called_with(5, bytearray(b'data'))
        self.loop.assert_writer(5, tr._write_ready)
        self.assertEqual(bytearray(b'data'), tr._buffer)

    @mock.patch('asyncio.log.logger.error')
    @mock.patch('os.write')
    def test__write_ready_err(self, m_write, m_logexc):
        tr = self.write_pipe_transport()
        self.loop.add_writer(5, tr._write_ready)
        tr._buffer = bytearray(b'data')
        m_write.side_effect = err = OSError()
        tr._write_ready()
        self.assertFalse(self.loop.writers)
        self.assertFalse(self.loop.readers)
        self.assertEqual(bytearray(), tr._buffer)
        self.assertTrue(tr.is_closing())
        m_logexc.assert_not_called()
        self.assertEqual(1, tr._conn_lost)
        test_utils.run_briefly(self.loop)
        self.protocol.connection_lost.assert_called_with(err)

    @mock.patch('os.write')
    def test__write_ready_closing(self, m_write):
        tr = self.write_pipe_transport()
        self.loop.add_writer(5, tr._write_ready)
        tr._closing = True
        tr._buffer = bytearray(b'data')
        m_write.return_value = 4
        tr._write_ready()
        self.assertFalse(self.loop.writers)
        self.assertFalse(self.loop.readers)
        self.assertEqual(bytearray(), tr._buffer)
        self.protocol.connection_lost.assert_called_with(None)
        self.pipe.close.assert_called_with()

    @mock.patch('os.write')
    def test_abort(self, m_write):
        tr = self.write_pipe_transport()
        self.loop.add_writer(5, tr._write_ready)
        self.loop.add_reader(5, tr._read_ready)
        tr._buffer = [b'da', b'ta']
        tr.abort()
        self.assertFalse(m_write.called)
        self.assertFalse(self.loop.readers)
        self.assertFalse(self.loop.writers)
        self.assertEqual([], tr._buffer)
        self.assertTrue(tr.is_closing())
        test_utils.run_briefly(self.loop)
        self.protocol.connection_lost.assert_called_with(None)

    def test__call_connection_lost(self):
        tr = self.write_pipe_transport()
        self.assertIsNotNone(tr._protocol)
        self.assertIsNotNone(tr._loop)

        err = None
        tr._call_connection_lost(err)
        self.protocol.connection_lost.assert_called_with(err)
        self.pipe.close.assert_called_with()

        self.assertIsNone(tr._protocol)
        self.assertIsNone(tr._loop)

    def test__call_connection_lost_with_err(self):
        tr = self.write_pipe_transport()
        self.assertIsNotNone(tr._protocol)
        self.assertIsNotNone(tr._loop)

        err = OSError()
        tr._call_connection_lost(err)
        self.protocol.connection_lost.assert_called_with(err)
        self.pipe.close.assert_called_with()

        self.assertIsNone(tr._protocol)
        self.assertIsNone(tr._loop)

    def test_close(self):
        tr = self.write_pipe_transport()
        tr.write_eof = mock.Mock()
        tr.close()
        tr.write_eof.assert_called_with()

        
        tr.close()

    def test_close_closing(self):
        tr = self.write_pipe_transport()
        tr.write_eof = mock.Mock()
        tr._closing = True
        tr.close()
        self.assertFalse(tr.write_eof.called)

    def test_write_eof(self):
        tr = self.write_pipe_transport()
        tr.write_eof()
        self.assertTrue(tr.is_closing())
        self.assertFalse(self.loop.readers)
        test_utils.run_briefly(self.loop)
        self.protocol.connection_lost.assert_called_with(None)

    def test_write_eof_pending(self):
        tr = self.write_pipe_transport()
        tr._buffer = [b'data']
        tr.write_eof()
        self.assertTrue(tr.is_closing())
        self.assertFalse(self.protocol.connection_lost.called)


class AbstractChildWatcherTests(unittest.TestCase):

    def test_warns_on_subclassing(self):
        with self.assertWarns(DeprecationWarning):
            class MyWatcher(asyncio.AbstractChildWatcher):
                pass

    def test_not_implemented(self):
        f = mock.Mock()
        watcher = asyncio.AbstractChildWatcher()
        self.assertRaises(
            NotImplementedError, watcher.add_child_handler, f, f)
        self.assertRaises(
            NotImplementedError, watcher.remove_child_handler, f)
        self.assertRaises(
            NotImplementedError, watcher.attach_loop, f)
        self.assertRaises(
            NotImplementedError, watcher.close)
        self.assertRaises(
            NotImplementedError, watcher.is_active)
        self.assertRaises(
            NotImplementedError, watcher.__enter__)
        self.assertRaises(
            NotImplementedError, watcher.__exit__, f, f, f)


class BaseChildWatcherTests(unittest.TestCase):

    def test_not_implemented(self):
        f = mock.Mock()
        watcher = unix_events.BaseChildWatcher()
        self.assertRaises(
            NotImplementedError, watcher._do_waitpid, f)


class ChildWatcherTestsMixin:

    ignore_warnings = mock.patch.object(log.logger, "warning")

    def setUp(self):
        super().setUp()
        self.loop = self.new_test_loop()
        self.running = False
        self.zombies = {}

        with mock.patch.object(
                self.loop, "add_signal_handler") as self.m_add_signal_handler:
            self.watcher = self.create_watcher()
            self.watcher.attach_loop(self.loop)

    def waitpid(self, pid, flags):
        if isinstance(self.watcher, asyncio.SafeChildWatcher) or pid != -1:
            self.assertGreater(pid, 0)
        try:
            if pid < 0:
                return self.zombies.popitem()
            else:
                return pid, self.zombies.pop(pid)
        except KeyError:
            pass
        if self.running:
            return 0, 0
        else:
            raise ChildProcessError()

    def add_zombie(self, pid, status):
        self.zombies[pid] = status

    def waitstatus_to_exitcode(self, status):
        if status > 32768:
            return status - 32768
        elif 32700 < status < 32768:
            return status - 32768
        else:
            return status

    def test_create_watcher(self):
        self.m_add_signal_handler.assert_called_once_with(
            signal.SIGCHLD, self.watcher._sig_chld)

    def waitpid_mocks(func):
        def wrapped_func(self):
            def patch(target, wrapper):
                return mock.patch(target, wraps=wrapper,
                                  new_callable=mock.Mock)

            with patch('asyncio.unix_events.waitstatus_to_exitcode', self.waitstatus_to_exitcode), \
                 patch('os.waitpid', self.waitpid) as m_waitpid:
                func(self, m_waitpid)
        return wrapped_func

    @waitpid_mocks
    def test_sigchld(self, m_waitpid):
        
        callback = mock.Mock()

        with self.watcher:
            self.running = True
            self.watcher.add_child_handler(42, callback, 9, 10, 14)

        self.assertFalse(callback.called)

        
        self.watcher._sig_chld()

        self.assertFalse(callback.called)

        
        self.running = False
        self.add_zombie(42, EXITCODE(12))
        self.watcher._sig_chld()

        callback.assert_called_once_with(42, 12, 9, 10, 14)

        callback.reset_mock()

        
        self.add_zombie(42, EXITCODE(13))
        with self.ignore_warnings:
            self.watcher._sig_chld()

        self.assertFalse(callback.called)

        
        self.zombies.clear()
        self.watcher._sig_chld()

        self.assertFalse(callback.called)

    @waitpid_mocks
    def test_sigchld_two_children(self, m_waitpid):
        callback1 = mock.Mock()
        callback2 = mock.Mock()

        
        with self.watcher:
            self.running = True
            self.watcher.add_child_handler(43, callback1, 7, 8)

        self.assertFalse(callback1.called)
        self.assertFalse(callback2.called)

        
        with self.watcher:
            self.watcher.add_child_handler(44, callback2, 147, 18)

        self.assertFalse(callback1.called)
        self.assertFalse(callback2.called)

        
        self.watcher._sig_chld()

        self.assertFalse(callback1.called)
        self.assertFalse(callback2.called)

        
        self.add_zombie(43, SIGNAL(3))
        self.watcher._sig_chld()

        callback1.assert_called_once_with(43, -3, 7, 8)
        self.assertFalse(callback2.called)

        callback1.reset_mock()

        
        self.watcher._sig_chld()

        self.assertFalse(callback1.called)
        self.assertFalse(callback2.called)

        
        self.add_zombie(44, EXITCODE(108))
        self.running = False
        self.watcher._sig_chld()

        callback2.assert_called_once_with(44, 108, 147, 18)
        self.assertFalse(callback1.called)

        callback2.reset_mock()

        
        self.add_zombie(43, EXITCODE(14))
        self.add_zombie(44, EXITCODE(15))
        with self.ignore_warnings:
            self.watcher._sig_chld()

        self.assertFalse(callback1.called)
        self.assertFalse(callback2.called)

        
        self.zombies.clear()
        self.watcher._sig_chld()

        self.assertFalse(callback1.called)
        self.assertFalse(callback2.called)

    @waitpid_mocks
    def test_sigchld_two_children_terminating_together(self, m_waitpid):
        callback1 = mock.Mock()
        callback2 = mock.Mock()

        
        with self.watcher:
            self.running = True
            self.watcher.add_child_handler(45, callback1, 17, 8)

        self.assertFalse(callback1.called)
        self.assertFalse(callback2.called)

        
        with self.watcher:
            self.watcher.add_child_handler(46, callback2, 1147, 18)

        self.assertFalse(callback1.called)
        self.assertFalse(callback2.called)

        
        self.watcher._sig_chld()

        self.assertFalse(callback1.called)
        self.assertFalse(callback2.called)

        
        
        self.add_zombie(45, EXITCODE(78))
        self.add_zombie(46, SIGNAL(5))
        self.running = False
        self.watcher._sig_chld()

        callback1.assert_called_once_with(45, 78, 17, 8)
        callback2.assert_called_once_with(46, -5, 1147, 18)

        callback1.reset_mock()
        callback2.reset_mock()

        
        self.add_zombie(45, EXITCODE(14))
        self.add_zombie(46, EXITCODE(15))
        with self.ignore_warnings:
            self.watcher._sig_chld()

        self.assertFalse(callback1.called)
        self.assertFalse(callback2.called)

    @waitpid_mocks
    def test_sigchld_race_condition(self, m_waitpid):
        
        callback = mock.Mock()

        with self.watcher:
            
            self.add_zombie(50, EXITCODE(4))
            self.watcher._sig_chld()

            self.watcher.add_child_handler(50, callback, 1, 12)

        callback.assert_called_once_with(50, 4, 1, 12)
        callback.reset_mock()

        
        self.add_zombie(50, SIGNAL(1))
        with self.ignore_warnings:
            self.watcher._sig_chld()

        self.assertFalse(callback.called)

    @waitpid_mocks
    def test_sigchld_replace_handler(self, m_waitpid):
        callback1 = mock.Mock()
        callback2 = mock.Mock()

        
        with self.watcher:
            self.running = True
            self.watcher.add_child_handler(51, callback1, 19)

        self.assertFalse(callback1.called)
        self.assertFalse(callback2.called)

        
        with self.watcher:
            self.watcher.add_child_handler(51, callback2, 21)

        self.assertFalse(callback1.called)
        self.assertFalse(callback2.called)

        
        self.running = False
        self.add_zombie(51, SIGNAL(8))
        self.watcher._sig_chld()

        callback2.assert_called_once_with(51, -8, 21)
        self.assertFalse(callback1.called)

        callback2.reset_mock()

        
        self.add_zombie(51, EXITCODE(13))
        with self.ignore_warnings:
            self.watcher._sig_chld()

        self.assertFalse(callback1.called)
        self.assertFalse(callback2.called)

    @waitpid_mocks
    def test_sigchld_remove_handler(self, m_waitpid):
        callback = mock.Mock()

        
        with self.watcher:
            self.running = True
            self.watcher.add_child_handler(52, callback, 1984)

        self.assertFalse(callback.called)

        
        self.watcher.remove_child_handler(52)

        self.assertFalse(callback.called)

        
        self.running = False
        self.add_zombie(52, EXITCODE(99))
        with self.ignore_warnings:
            self.watcher._sig_chld()

        self.assertFalse(callback.called)

    @waitpid_mocks
    def test_sigchld_unknown_status(self, m_waitpid):
        callback = mock.Mock()

        
        with self.watcher:
            self.running = True
            self.watcher.add_child_handler(53, callback, -19)

        self.assertFalse(callback.called)

        
        self.zombies[53] = 1178
        self.running = False
        self.watcher._sig_chld()

        callback.assert_called_once_with(53, 1178, -19)

        callback.reset_mock()

        
        self.add_zombie(53, EXITCODE(101))
        with self.ignore_warnings:
            self.watcher._sig_chld()

        self.assertFalse(callback.called)

    @waitpid_mocks
    def test_remove_child_handler(self, m_waitpid):
        callback1 = mock.Mock()
        callback2 = mock.Mock()
        callback3 = mock.Mock()

        
        with self.watcher:
            self.running = True
            self.watcher.add_child_handler(54, callback1, 1)
            self.watcher.add_child_handler(55, callback2, 2)
            self.watcher.add_child_handler(56, callback3, 3)

        
        self.assertTrue(self.watcher.remove_child_handler(54))

        
        self.assertTrue(self.watcher.remove_child_handler(55))
        self.assertFalse(self.watcher.remove_child_handler(55))
        self.assertFalse(self.watcher.remove_child_handler(55))

        
        self.add_zombie(54, EXITCODE(0))
        self.add_zombie(55, EXITCODE(1))
        self.add_zombie(56, EXITCODE(2))
        self.running = False
        with self.ignore_warnings:
            self.watcher._sig_chld()

        self.assertFalse(callback1.called)
        self.assertFalse(callback2.called)
        callback3.assert_called_once_with(56, 2, 3)

    @waitpid_mocks
    def test_sigchld_unhandled_exception(self, m_waitpid):
        callback = mock.Mock()

        
        with self.watcher:
            self.running = True
            self.watcher.add_child_handler(57, callback)

        
        m_waitpid.side_effect = ValueError

        with mock.patch.object(log.logger,
                               'error') as m_error:

            self.assertEqual(self.watcher._sig_chld(), None)
            self.assertTrue(m_error.called)

    @waitpid_mocks
    def test_sigchld_child_reaped_elsewhere(self, m_waitpid):
        
        callback = mock.Mock()

        with self.watcher:
            self.running = True
            self.watcher.add_child_handler(58, callback)

        self.assertFalse(callback.called)

        
        self.running = False
        self.add_zombie(58, EXITCODE(4))

        
        os.waitpid(58, os.WNOHANG)

        m_waitpid.reset_mock()

        
        with self.ignore_warnings:
            self.watcher._sig_chld()

        if isinstance(self.watcher, asyncio.FastChildWatcher):
            
            
            self.assertFalse(callback.called)
        else:
            callback.assert_called_once_with(58, 255)

    @waitpid_mocks
    def test_sigchld_unknown_pid_during_registration(self, m_waitpid):
        
        callback1 = mock.Mock()
        callback2 = mock.Mock()

        with self.ignore_warnings, self.watcher:
            self.running = True
            
            self.add_zombie(591, EXITCODE(7))
            
            self.add_zombie(593, EXITCODE(17))

            self.watcher._sig_chld()

            self.watcher.add_child_handler(591, callback1)
            self.watcher.add_child_handler(592, callback2)

        callback1.assert_called_once_with(591, 7)
        self.assertFalse(callback2.called)

    @waitpid_mocks
    def test_set_loop(self, m_waitpid):
        
        callback = mock.Mock()

        with self.watcher:
            self.running = True
            self.watcher.add_child_handler(60, callback)

        
        old_loop = self.loop
        self.loop = self.new_test_loop()
        patch = mock.patch.object

        with patch(old_loop, "remove_signal_handler") as m_old_remove, \
             patch(self.loop, "add_signal_handler") as m_new_add:

            self.watcher.attach_loop(self.loop)

            m_old_remove.assert_called_once_with(
                signal.SIGCHLD)
            m_new_add.assert_called_once_with(
                signal.SIGCHLD, self.watcher._sig_chld)

        
        self.running = False
        self.add_zombie(60, EXITCODE(9))
        self.watcher._sig_chld()

        callback.assert_called_once_with(60, 9)

    @waitpid_mocks
    def test_set_loop_race_condition(self, m_waitpid):
        
        callback1 = mock.Mock()
        callback2 = mock.Mock()
        callback3 = mock.Mock()

        with self.watcher:
            self.running = True
            self.watcher.add_child_handler(61, callback1)
            self.watcher.add_child_handler(62, callback2)
            self.watcher.add_child_handler(622, callback3)

        
        old_loop = self.loop
        self.loop = None

        with mock.patch.object(
                old_loop, "remove_signal_handler") as m_remove_signal_handler:

            with self.assertWarnsRegex(
                    RuntimeWarning, 'A loop is being detached'):
                self.watcher.attach_loop(None)

            m_remove_signal_handler.assert_called_once_with(
                signal.SIGCHLD)

        
        self.add_zombie(61, EXITCODE(11))
        self.add_zombie(62, SIGNAL(5))

        
        self.assertFalse(callback1.called)
        self.assertFalse(callback2.called)
        self.assertFalse(callback3.called)

        
        self.loop = self.new_test_loop()

        with mock.patch.object(
                self.loop, "add_signal_handler") as m_add_signal_handler:

            self.watcher.attach_loop(self.loop)

            m_add_signal_handler.assert_called_once_with(
                signal.SIGCHLD, self.watcher._sig_chld)
            callback1.assert_called_once_with(61, 11)  
            callback2.assert_called_once_with(62, -5)  
            self.assertFalse(callback3.called)

        callback1.reset_mock()
        callback2.reset_mock()

        
        self.running = False
        self.add_zombie(622, EXITCODE(19))
        self.watcher._sig_chld()

        self.assertFalse(callback1.called)
        self.assertFalse(callback2.called)
        callback3.assert_called_once_with(622, 19)

    @waitpid_mocks
    def test_close(self, m_waitpid):
        
        callback1 = mock.Mock()

        with self.watcher:
            self.running = True
            
            self.add_zombie(63, EXITCODE(9))
            
            self.add_zombie(65, EXITCODE(18))
            self.watcher._sig_chld()

            self.watcher.add_child_handler(63, callback1)
            self.watcher.add_child_handler(64, callback1)

            self.assertEqual(len(self.watcher._callbacks), 1)
            if isinstance(self.watcher, asyncio.FastChildWatcher):
                self.assertEqual(len(self.watcher._zombies), 1)

            with mock.patch.object(
                    self.loop,
                    "remove_signal_handler") as m_remove_signal_handler:

                self.watcher.close()

                m_remove_signal_handler.assert_called_once_with(
                    signal.SIGCHLD)
                self.assertFalse(self.watcher._callbacks)
                if isinstance(self.watcher, asyncio.FastChildWatcher):
                    self.assertFalse(self.watcher._zombies)


class SafeChildWatcherTests (ChildWatcherTestsMixin, test_utils.TestCase):
    def create_watcher(self):
        with warnings.catch_warnings():
            warnings.simplefilter("ignore", DeprecationWarning)
            return asyncio.SafeChildWatcher()


class FastChildWatcherTests (ChildWatcherTestsMixin, test_utils.TestCase):
    def create_watcher(self):
        with warnings.catch_warnings():
            warnings.simplefilter("ignore", DeprecationWarning)
            return asyncio.FastChildWatcher()


class PolicyTests(unittest.TestCase):

    def create_policy(self):
        return asyncio.DefaultEventLoopPolicy()

    @mock.patch('asyncio.unix_events.can_use_pidfd')
    def test_get_default_child_watcher(self, m_can_use_pidfd):
        m_can_use_pidfd.return_value = False
        policy = self.create_policy()
        self.assertIsNone(policy._watcher)
        with self.assertWarns(DeprecationWarning):
            watcher = policy.get_child_watcher()
        self.assertIsInstance(watcher, asyncio.ThreadedChildWatcher)

        self.assertIs(policy._watcher, watcher)
        with self.assertWarns(DeprecationWarning):
            self.assertIs(watcher, policy.get_child_watcher())

        m_can_use_pidfd.return_value = True
        policy = self.create_policy()
        self.assertIsNone(policy._watcher)
        with self.assertWarns(DeprecationWarning):
            watcher = policy.get_child_watcher()
        self.assertIsInstance(watcher, asyncio.PidfdChildWatcher)

        self.assertIs(policy._watcher, watcher)
        with self.assertWarns(DeprecationWarning):
            self.assertIs(watcher, policy.get_child_watcher())

    def test_get_child_watcher_after_set(self):
        policy = self.create_policy()
        with warnings.catch_warnings():
            warnings.simplefilter("ignore", DeprecationWarning)
            watcher = asyncio.FastChildWatcher()
            policy.set_child_watcher(watcher)

        self.assertIs(policy._watcher, watcher)
        with self.assertWarns(DeprecationWarning):
            self.assertIs(watcher, policy.get_child_watcher())

    def test_get_child_watcher_thread(self):

        def f():
            policy.set_event_loop(policy.new_event_loop())

            self.assertIsInstance(policy.get_event_loop(),
                                  asyncio.AbstractEventLoop)
            with warnings.catch_warnings():
                warnings.simplefilter("ignore", DeprecationWarning)
                watcher = policy.get_child_watcher()

            self.assertIsInstance(watcher, asyncio.SafeChildWatcher)
            self.assertIsNone(watcher._loop)

            policy.get_event_loop().close()

        policy = self.create_policy()
        with warnings.catch_warnings():
            warnings.simplefilter("ignore", DeprecationWarning)
            policy.set_child_watcher(asyncio.SafeChildWatcher())

        th = threading.Thread(target=f)
        th.start()
        th.join()

    def test_child_watcher_replace_mainloop_existing(self):
        policy = self.create_policy()
        loop = policy.new_event_loop()
        policy.set_event_loop(loop)

        
        
        with warnings.catch_warnings():
            warnings.simplefilter("ignore", DeprecationWarning)
            watcher = asyncio.SafeChildWatcher()
            policy.set_child_watcher(watcher)
        watcher.attach_loop(loop)

        self.assertIs(watcher._loop, loop)

        new_loop = policy.new_event_loop()
        policy.set_event_loop(new_loop)

        self.assertIs(watcher._loop, new_loop)

        policy.set_event_loop(None)

        self.assertIs(watcher._loop, None)

        loop.close()
        new_loop.close()


class TestFunctional(unittest.TestCase):

    def setUp(self):
        self.loop = asyncio.new_event_loop()
        asyncio.set_event_loop(self.loop)

    def tearDown(self):
        self.loop.close()
        asyncio.set_event_loop(None)

    def test_add_reader_invalid_argument(self):
        def assert_raises():
            return self.assertRaisesRegex(ValueError, r'Invalid file object')

        cb = lambda: None

        with assert_raises():
            self.loop.add_reader(object(), cb)
        with assert_raises():
            self.loop.add_writer(object(), cb)

        with assert_raises():
            self.loop.remove_reader(object())
        with assert_raises():
            self.loop.remove_writer(object())

    def test_add_reader_or_writer_transport_fd(self):
        def assert_raises():
            return self.assertRaisesRegex(
                RuntimeError,
                r'File descriptor .* is used by transport')

        async def runner():
            tr, pr = await self.loop.create_connection(
                lambda: asyncio.Protocol(), sock=rsock)

            try:
                cb = lambda: None

                with assert_raises():
                    self.loop.add_reader(rsock, cb)
                with assert_raises():
                    self.loop.add_reader(rsock.fileno(), cb)

                with assert_raises():
                    self.loop.remove_reader(rsock)
                with assert_raises():
                    self.loop.remove_reader(rsock.fileno())

                with assert_raises():
                    self.loop.add_writer(rsock, cb)
                with assert_raises():
                    self.loop.add_writer(rsock.fileno(), cb)

                with assert_raises():
                    self.loop.remove_writer(rsock)
                with assert_raises():
                    self.loop.remove_writer(rsock.fileno())

            finally:
                tr.close()

        rsock, wsock = socket.socketpair()
        try:
            self.loop.run_until_complete(runner())
        finally:
            rsock.close()
            wsock.close()


@unittest.skipUnless(hasattr(os, 'fork'), 'requires os.fork()')
class TestFork(unittest.IsolatedAsyncioTestCase):

    async def test_fork_not_share_event_loop(self):
        
        loop = asyncio.get_running_loop()
        r, w = os.pipe()
        self.addCleanup(os.close, r)
        self.addCleanup(os.close, w)
        pid = os.fork()
        if pid == 0:
            
            try:
                with self.assertWarns(DeprecationWarning):
                    loop = asyncio.get_event_loop_policy().get_event_loop()
                os.write(w, b'LOOP:' + str(id(loop)).encode())
            except RuntimeError:
                os.write(w, b'NO LOOP')
            except BaseException as e:
                os.write(w, b'ERROR:' + ascii(e).encode())
            finally:
                os._exit(0)
        else:
            
            result = os.read(r, 100)
            self.assertEqual(result[:5], b'LOOP:', result)
            self.assertNotEqual(int(result[5:]), id(loop))
            wait_process(pid, exitcode=0)

    @hashlib_helper.requires_hashdigest('md5')
    def test_fork_signal_handling(self):
        self.addCleanup(multiprocessing_cleanup_tests)

        
        
        ctx = multiprocessing.get_context('fork')
        manager = ctx.Manager()
        self.addCleanup(manager.shutdown)
        child_started = manager.Event()
        child_handled = manager.Event()
        parent_handled = manager.Event()

        def child_main():
            def on_sigterm(*args):
                child_handled.set()
                sys.exit()

            signal.signal(signal.SIGTERM, on_sigterm)
            child_started.set()
            while True:
                time.sleep(1)

        async def main():
            loop = asyncio.get_running_loop()
            loop.add_signal_handler(signal.SIGTERM, lambda *args: parent_handled.set())

            process = ctx.Process(target=child_main)
            process.start()
            child_started.wait()
            os.kill(process.pid, signal.SIGTERM)
            process.join(timeout=support.SHORT_TIMEOUT)

            async def func():
                await asyncio.sleep(0.1)
                return 42

            
            self.assertEqual(await asyncio.create_task(func()), 42)

        asyncio.run(main())

        child_handled.wait(timeout=support.SHORT_TIMEOUT)
        self.assertFalse(parent_handled.is_set())
        self.assertTrue(child_handled.is_set())

    @hashlib_helper.requires_hashdigest('md5')
    def test_fork_asyncio_run(self):
        self.addCleanup(multiprocessing_cleanup_tests)

        ctx = multiprocessing.get_context('fork')
        manager = ctx.Manager()
        self.addCleanup(manager.shutdown)
        result = manager.Value('i', 0)

        async def child_main():
            await asyncio.sleep(0.1)
            result.value = 42

        process = ctx.Process(target=lambda: asyncio.run(child_main()))
        process.start()
        process.join()

        self.assertEqual(result.value, 42)

    @hashlib_helper.requires_hashdigest('md5')
    def test_fork_asyncio_subprocess(self):
        self.addCleanup(multiprocessing_cleanup_tests)

        ctx = multiprocessing.get_context('fork')
        manager = ctx.Manager()
        self.addCleanup(manager.shutdown)
        result = manager.Value('i', 1)

        async def child_main():
            proc = await asyncio.create_subprocess_exec(sys.executable, '-c', 'pass')
            result.value = await proc.wait()

        process = ctx.Process(target=lambda: asyncio.run(child_main()))
        process.start()
        process.join()

        self.assertEqual(result.value, 0)

if __name__ == '__main__':
    unittest.main()