U  W[d3@s$dZddlmZmZddlZddlZddlZddlmZddl m Z m Z ddl m Z ddlmZddlmZmZmZdd lmZmZmZmZmZdd lmZd ZdZere d dkrd Zeej Z!ej"#ej$e!d<n"eZ!ej"#ej$%e&e!d<Gdddej'Z(Gdddej)Z*dS)z Tests for L{twisted.internet.stdio}. @var properEnv: A copy of L{os.environ} which has L{bytes} keys/values on POSIX platforms and native L{str} keys/values on Windows. )absolute_importdivisionN)unittest)filepathlog) requireModule)platform)range intToBytes bytesEnviron)errordeferprotocolstdioreactor)ConnectionLostNotifyingProtocolsxyz123abc Twisted is great!Z win32processzIOn windows, spawnProcess is not available in the absence of win32process. PYTHONPATHs PYTHONPATHc@s4eZdZdZdZddZddZddZd d ZdS) StandardIOTestProcessProtocola Test helper for collecting output from a child process and notifying something when it exits. @ivar onConnection: A L{defer.Deferred} which will be called back with L{None} when the connection to the child process is established. @ivar onCompletion: A L{defer.Deferred} which will be errbacked with the failure associated with the child process exiting when it exits. @ivar onDataReceived: A L{defer.Deferred} which will be called back with this instance whenever C{childDataReceived} is called, or L{None} to suppress these callbacks. @ivar data: A C{dict} mapping file descriptors to strings containing all bytes received from the child process on each file descriptor. NcCst|_t|_i|_dSN)r Deferred onConnection onCompletiondataselfr9/usr/lib/python3/dist-packages/twisted/test/test_stdio.py__init__Bs  z&StandardIOTestProcessProtocol.__init__cCs|jddSr)rcallbackrrrrconnectionMadeHsz,StandardIOTestProcessProtocol.connectionMadecCs>|j|d||j|<|jdk r:|jd}|_||dS)z Record all bytes received from the child process in the C{data} dictionary. Fire C{onDataReceived} if it is not L{None}. N)rgetonDataReceivedr)rnamebytesdrrrchildDataReceivedLs z/StandardIOTestProcessProtocol.childDataReceivedcCs|j|dSr)rr)rreasonrrr processEndedWsz*StandardIOTestProcessProtocol.processEnded) __name__ __module__ __qualname____doc__r"rrr&r(rrrrr.s  rc@s~eZdZeZddZddZddZddZd d Z d d Z d dZ ddZ ddZ ddZddZddZerzde_dS)StandardInputOutputTestscOs:tjdd|tjjgt|}tj|tj|fdti|S)a_ Launch a child Python process and communicate with it using the given ProcessProtocol. @param proto: A L{ProcessProtocol} instance which will be connected to the child process. @param sibling: The basename of a file containing the Python program to run in the child process. @param *args: strings which will be passed to the child process on the command line as C{argv[2:]}. @param **kw: additional arguments to pass to L{reactor.spawnProcess}. @return: The L{IProcessTransport} provider for the spawned process. s-ms twisted.test.env)sys executabler __class__r*listZ spawnProcess properEnv)rprotoZsiblingargskwrrr _spawnProcess`s z&StandardInputOutputTests._spawnProcesscs$fdd}fdd}|||S)Ncsd|fdS)Nz'Process terminated with non-Failure: %r)Zfail)resultrrrcb~sz4StandardInputOutputTests._requireFailure..cbcs|Srr)err)rrrebsz4StandardInputOutputTests._requireFailure..eb)Z addCallbacks)rr%rr9r;r)rrr_requireFailure}s  z(StandardInputOutputTests._requireFailurecsLtdtj}dfdd}||S)z Verify that a protocol connected to L{StandardIO} can disconnect itself using C{transport.loseConnection}. Child process logging to sstdio_test_loseconnc sPtd"}|D]}td|qW5QRXdj|tjdS)NrzChild logged: ) openrmsgrstripZfailIfInrtrapr ProcessDone)r'fline errorLogFileprrrr(s  zBStandardInputOutputTests.test_loseConnection..processEnded)mktemprrArrr7r<rr%r(rrGrtest_loseConnectionsz,StandardInputOutputTests.test_loseConnectioncsf|}td|tt_fdd}j|dd}|j|}| d||S)z When stdin is closed and the protocol connected to it implements L{IHalfCloseableProtocol}, the protocol's C{readConnectionLost} method is called. r=csj}j|Sr)rZ transportZ closeStdin)Zignoredr%rIrrcbBytess zAStandardInputOutputTests.test_readConnectionLost..cbBytescSs|tjdSr)rCr rDr'rrrr(szFStandardInputOutputTests.test_readConnectionLost..processEndedsstdio_test_halfclose) rJrrArr rr" addCallbackr<r7)rrHrNr(r%rrMrtest_readConnectionLosts   z0StandardInputOutputTests.test_readConnectionLostc sjtzjdtddWn0tk rL}ztt|W5d}~XYnXfdd}j|S)z Verify that a write made directly to stdout using L{os.write} after StandardIO has finished is reliably received by the process reading that stdout. sstdio_test_lastwriteT)ZusePTYNcs0jdtdjf|tjdS)z Asserts that the parent received the bytes written by the child immediately after the child starts. r?z4Received %r from child, did not find expected bytes.N) assertTruerendswithUNIQUE_LAST_WRITE_STRINGrCr rDrOrIrrrr(szEStandardInputOutputTests.test_lastWriteReceived..processEnded) rr7rT ValueErrorrZSkipTeststrr<r)rer(rrUrtest_lastWriteReceiveds    z/StandardInputOutputTests.test_lastWriteReceivedcs2tj}dfdd}||S)z Verify that the transport of a protocol connected to L{StandardIO} has C{getHost} and C{getPeer} methods. sstdio_test_hostpeercs6jd\}}|||tjdS)Nr?)r splitlinesrRrCr rD)r'ZhostZpeerrUrrr(s  z?StandardInputOutputTests.test_hostAndPeer..processEndedrrr7r<rKrrUrtest_hostAndPeers  z)StandardInputOutputTests.test_hostAndPeercs2tj}dfdd}||S)z Verify that the C{write} method of the transport of a protocol connected to L{StandardIO} sends bytes to standard out. sstdio_test_writecs"jdd|tjdSNr?sok! assertEqualrrCr rDrOrUrrr(sz9StandardInputOutputTests.test_write..processEndedr[rKrrUr test_writes  z#StandardInputOutputTests.test_writecs2tj}dfdd}||S)z Verify that the C{writeSequence} method of the transport of a protocol connected to L{StandardIO} sends bytes to standard out. sstdio_test_writeseqcs"jdd|tjdSr]r^rOrUrrr(szAStandardInputOutputTests.test_writeSequence..processEndedr[rKrrUrtest_writeSequences  z+StandardInputOutputTests.test_writeSequencec CsB|}t|d&}tdD]}|t|dqW5QRX|S)Nwbi )rJr@r writer )rjunkPathZjunkFileirrr _junkPath s   z"StandardInputOutputTests._junkPathcsdtj}gttdfdddjfdd}||S)z Verify that the transport of a protocol connected to L{StandardIO} is a working L{IProducer} provider. dcs:r6tddtdddS)Nrcg{Gz?)appendr poprdr callLater)Zign)rproctoWritewrittenrrr sz>StandardInputOutputTests.test_producer..connectionMadesstdio_test_producercs>jdddtf|tjdS)Nr?r z*Connection lost with %d writes left to go.)r_rjoinZ assertFalselenrCr rDrO)rIrrnrorrr(*s  z.processEnded)rrr2r r7rrPr<rKr)rrIrmrrnror test_producers   z&StandardInputOutputTests.test_producercs>tj}dfdd}||S)z Verify that the transport of a protocol connected to L{StandardIO} is a working L{IConsumer} provider. sstdio_test_consumerc s<td}jd|W5QRX|tjdS)Nrbr?)r@r_rreadrCr rD)r'rErerIrrrr(?s  z.processEnded)rrrgr7r<rKrrur test_consumer3s z&StandardInputOutputTests.test_consumercst}t|}td_}|j t | d}t st\}}tj |tj |||d<tj|f|dtfddtdfdd }|||S) aE If L{StandardIO} is created with a file descriptor which refers to a normal file (ie, a file from the filesystem), L{StandardIO.write} writes bytes to that file. In particular, it does not immediately consider the file closed or call its protocol's C{connectionLost} method. rb)stdoutstdincsBD],}|krdSt|q2qtddS)Nr)ZloseConnectionrdr rrl)value) connectioncounthowManyspinrrr~eszAStandardInputOutputTests.test_normalFileStandardOut..spinrc s8tddtttdS)Nr?r )r_nextZ getContentrpmapr r rO)r|r}pathrrrcbLostqs zCStandardInputOutputTests.test_normalFileStandardOut..cbLost)r rrrZFilePathrJr@normalZ addCleanupclosedictfilenor isWindowsospiperZ StandardIO itertoolsr|rrlrP)rZ onConnLostr4rkwargsr>wrr)r{r|r}rrr~rtest_normalFileStandardOutFs&    z3StandardInputOutputTests.test_normalFileStandardOutzpStandardIO does not accept stdout as an argument to Windows. Testing redirection to a file is therefore harder.N)r)r*r+skipWindowsNopywin32skipr7r<rLrQrYr\r`rargrrrvrrrrrrrr-\s '3r-)+r,Z __future__rrrr/rZ twisted.trialrZtwisted.pythonrrZtwisted.python.reflectrZtwisted.python.runtimerZtwisted.python.compatr r r Ztwisted.internetr r rrrZtwisted.test.test_tcprrTrrrenvironr3pathseprprencodegetfilesystemencodingZProcessProtocolrZTestCaser-rrrrs0      .