U n5{i8@sdZddlmZddlZddlZddlZddlmZmZddl m Z ddl m Z m Z mZmZmZddlmZmZddlmZdd lmZmZdd lmZdd lmZdd lmZdd lm Z ddl!m"Z"m#Z#dZ$e%ej&Z'ej()ej*e'd<GdddejZ+Gddde#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. ) annotationsN)AnyCallable)skipIf)defererrorprotocolreactorstdio)IProcessTransportIReactorProcess)ProcessProtocol)filepathlog)Failure) requireModule)platform)ConnectionLostNotifyingProtocol)SkipTestTestCasesxyz123abc Twisted is great! PYTHONPATHc@sLeZdZUdZdZded<ded<ddd d Zd d Zd dZddZ dS)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. Nzdefer.Deferred[None] | NoneonDataReceivedr transportNonereturncCst|_t|_i|_dSN)rDeferred onConnection onCompletiondataselfr$;/tmp/pip-unpacked-wheel-mnhg7r0q/twisted/test/test_stdio.py__init__=s  z&StandardIOTestProcessProtocol.__init__cCs|jddSr)rcallbackr"r$r$r%connectionMadeBsz,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)r!getrr')r#namebytesdr$r$r%childDataReceivedEs z/StandardIOTestProcessProtocol.childDataReceivedcCs|j|dSr)r r')r#reasonr$r$r% processEndedOsz*StandardIOTestProcessProtocol.processEnded) __name__ __module__ __qualname____doc__r__annotations__r&r(r.r0r$r$r$r%r's   rc@seZdZer eddkr dZdddddd d d Zd d d dddZddZ d0ddd dddZ d dddZ d dddZ d dddZ dd Zd!d"Zd#d$Zd%d&Zd'd(Zd)d*Zd+d,Zeed-d.d/ZdS)1StandardInputOutputTestsZ win32processNzIOn windows, spawnProcess is not available in the absence of win32process.r z str | bytesstrrr )protosiblingargskwrcOsPt|tr|}tjdd|tjjgt|}t tj |tj|fdt i|S)ak 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. z-mz twisted.test.env) isinstancer,decodesys executabler __class__r2listr Z spawnProcess properEnv)r#r8r9r:r;Zprocargsr$r$r% _spawnProcessZs$ z&StandardInputOutputTests._spawnProcessdefer.Deferred[None]zCallable[[Failure], object])r-r'rcs$fdd}fdd}|||S)Ncsd|dS)Nz%Process terminated with non-Failure: )Zfail)resultr"r$r%cb}sz4StandardInputOutputTests._requireFailure..cbcs|Srr$)err)r'r$r%ebsz4StandardInputOutputTests._requireFailure..eb)Z addCallbacks)r#r-r'rGrIr$)r'r#r%_requireFailurezs  z(StandardInputOutputTests._requireFailurecsLtdtj}dfdd}||S)z Verify that a protocol connected to L{StandardIO} can disconnect itself using C{transport.loseConnection}. zChild process logging to sstdio_test_loseconnc sNt"}|D]}td|qW5QRXdj|tjdS)NzChild logged: ) openrmsgrstripZfailIfInr!trapr ProcessDone)r/fline errorLogFilepr#r$r%r0s  zBStandardInputOutputTests.test_loseConnection..processEnded)mktemprrMrr rDrJr#r-r0r$rSr%test_loseConnections z,StandardInputOutputTests.test_loseConnectionFbool)exampleoutrcsb|}tt_dddfdd }j|dd}|j|}||||S)NrrE)ignoredrcs$j}rjn j|Sr)r rZ closeStdoutZ closeStdin)r\r-r[rUr$r%cbBytess   zGStandardInputOutputTests.exampleOutputsAndZeroExitCode..cbBytescSs|tjdSr)rOrrPr/r$r$r%r0szLStandardInputOutputTests.exampleOutputsAndZeroExitCode..processEnded)rVrrrr addCallbackrJrD)r#rZr[rTr^r0r-r$r]r%exampleOutputsAndZeroExitCodes  z6StandardInputOutputTests.exampleOutputsAndZeroExitCodercCs |dS)z When stdin is closed and the protocol connected to it implements L{IHalfCloseableProtocol}, the protocol's C{readConnectionLost} method is called. Zstdio_test_halfcloserar"r$r$r%test_readConnectionLostsz0StandardInputOutputTests.test_readConnectionLostcCs |dS) When stdin is closed and the protocol connnected to it implements L{IHalfCloseableProtocol} but its C{readConnectionLost} method raises an exception its regular C{connectionLost} method will be called. Zstdio_test_halfclose_buggyrbr"r$r$r%test_buggyReadConnectionLostsz5StandardInputOutputTests.test_buggyReadConnectionLostcCs|jdddS)rdZ stdio_test_halfclose_buggy_writeT)r[rbr"r$r$r%test_buggyWriteConnectionLostsz6StandardInputOutputTests.test_buggyWriteConnectionLostc shtzjdtddWn.tk rJ}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)ZusePTYNcs2jdtdjd|tjdS)z Asserts that the parent received the bytes written by the child immediately after the child starts. rKz Received z) from child, did not find expected bytes.N) assertTruer!endswithUNIQUE_LAST_WRITE_STRINGrOrrPr_rUr#r$r%r0s  zEStandardInputOutputTests.test_lastWriteReceived..processEnded)rrDri ValueErrorrr7rJr )r#er0r$rjr%test_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)NrK)r! splitlinesrgrOrrP)r/hostZpeerrjr$r%r0s  z?StandardInputOutputTests.test_hostAndPeer..processEndedrr rDrJrWr$rjr%test_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|tjdSNrKsok! assertEqualr!rOrrPr_rjr$r%r0sz9StandardInputOutputTests.test_write..processEndedrprWr$rjr% 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|tjdSrrrsr_rjr$r%r0 szAStandardInputOutputTests.test_writeSequence..processEndedrprWr$rjr%test_writeSequences  z+StandardInputOutputTests.test_writeSequencec Cs@|}t|d$}tdD]}|d|fqW5QRX|S)Nwbi%d )rVrLrangewrite)r#junkPathZjunkFileir$r$r% _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. dcs8r4dfdtdddS)Nrxg{Gz?)appendpoprzr callLater)Zign)r(proctoWritewrittenr$r%r(8sz>StandardInputOutputTests.test_producer..connectionMadesstdio_test_producercs>jdddtf|tjdS)NrKr)z*Connection lost with %d writes left to go.)rtr!joinZ assertFalselenrOrrPr_)rUr#rrr$r%r0Bs  z.processEnded)rr rBryrDrr`rJrWr$)r(rUrr#rrr% test_producer-s   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)NrbrK)rLrtr!readrOrrP)r/rQr{rUr#r$r%r0Ws  z.processEnded)rr r}rDrJrWr$rr% test_consumerKs z&StandardInputOutputTests.test_consumerzpStandardIO does not accept stdout as an argument to Windows. Testing redirection to a file is therefore harder.cst}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. rw)stdoutstdincsDD].}|krdSd|fq4qtddS)N%dr)ZloseConnectionrzr r)value) connectioncounthowManyspinr$r%rszAStandardInputOutputTests.test_normalFileStandardOut..spinrcs<tddddtDdS)NrKr)css|]}d|fVqdS)rNr$).0r|r$r$r% szVStandardInputOutputTests.test_normalFileStandardOut..cbLost..)rtnextZ getContentrryr_)rrpathr#r$r%cbLosts zCStandardInputOutputTests.test_normalFileStandardOut..cbLost)rrrrZFilePathrVrLnormalZ addCleanupclosedictfilenor isWindowsospiper Z StandardIO itertoolsrr rr`)r#Z onConnLostr8rkwargsrwrr$)rrrrr#rr%test_normalFileStandardOut^s&     z3StandardInputOutputTests.test_normalFileStandardOut)F)r1r2r3rrrskiprDrJrXrarcrerfrmrqrurvr}rrrrr$r$r$r%r6Ss.   'r6)-r4 __future__rrrr?typingrrZunittestrZtwisted.internetrrrr r Ztwisted.internet.interfacesr r Ztwisted.internet.protocolr Ztwisted.pythonrrZtwisted.python.failurerZtwisted.python.reflectrZtwisted.python.runtimerZtwisted.test.test_tcprZtwisted.trial.unittestrrrirenvironrCpathseprrrr6r$r$r$r%s(        ,