o 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@sHeZdZUdZdZded<ded<dd 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 transportreturnNonecCst|_t|_i|_dSN)rDeferred onConnection onCompletiondataselfr#]C:\Users\psant\Downloads\PAF backend\school\venv\Lib\site-packages\twisted/test/test_stdio.py__init__=s   z&StandardIOTestProcessProtocol.__init__cCs|jddSr)rcallbackr!r#r#r$connectionMadeBz,StandardIOTestProcessProtocol.connectionMadecCsB|j|d||j|<|jdur|jd}|_||dSdS)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)rr&)r"reasonr#r#r$ processEndedOr(z*StandardIOTestProcessProtocol.processEnded)rr) __name__ __module__ __qualname____doc__r__annotations__r%r'r.r0r#r#r#r$r's    rc@seZdZereddurdZd5ddZd6ddZddZ d7d8ddZ d9ddZ d9d d!Z d9d"d#Z d$d%Zd&d'Zd(d)Zd*d+Zd,d-Zd.d/Zd0d1Zeed2d3d4ZdS):StandardInputOutputTestsZ win32processNzIOn windows, spawnProcess is not available in the absence of win32process.protor sibling str | bytesargsstrkwrrr cOsPt|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.envN) isinstancer,decodesys executabler __class__r2listr Z spawnProcess properEnv)r"r7r8r:r<Zprocargsr#r#r$ _spawnProcessZs z&StandardInputOutputTests._spawnProcessr-defer.Deferred[None]r&Callable[[Failure], object]cs$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&rIrKr#)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_loseconncsbt}|D] }td|qWdn1swYdj|tjdS)NzChild logged: ) openrmsgrstripZfailIfInr trapr ProcessDone)r/fline errorLogFilepr"r#r$r0s zBStandardInputOutputTests.test_loseConnection..processEndedN)mktemprrOrrrErLr"r-r0r#rUr$test_loseConnections z,StandardInputOutputTests.test_loseConnectionFexampleoutboolcs\|}tt_d fdd }j|dd}|j|}||||S) NignoredrrrFcs&j}r j|Sj|Sr)rrZ closeStdoutZ closeStdin)r^r-r\rWr#r$cbBytess   zGStandardInputOutputTests.exampleOutputsAndZeroExitCode..cbBytescSs|tjdSr)rQrrRr/r#r#r$r0r(zLStandardInputOutputTests.exampleOutputsAndZeroExitCode..processEnded)r^rrrF)rXrrrr addCallbackrLrE)r"r[r\rVr`r0r-r#r_r$exampleOutputsAndZeroExitCodes  z6StandardInputOutputTests.exampleOutputsAndZeroExitCodecC |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_halfcloseNrcr!r#r#r$test_readConnectionLost z0StandardInputOutputTests.test_readConnectionLostcCrd) 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_buggyNrer!r#r#r$test_buggyReadConnectionLostrgz5StandardInputOutputTests.test_buggyReadConnectionLostcCs|jdddS)rhZ stdio_test_halfclose_buggy_writeT)r\Nrer!r#r#r$test_buggyWriteConnectionLostsz6StandardInputOutputTests.test_buggyWriteConnectionLostc s^tz jdtddWnty }ztt|d}~wwfdd}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. rMz Received z) from child, did not find expected bytes.N) assertTruer endswithUNIQUE_LAST_WRITE_STRINGrQrrRrarWr"r#r$r0s  zEStandardInputOutputTests.test_lastWriteReceived..processEnded)rrErm ValueErrorrr;rLr)r"er0r#rnr$test_lastWriteReceiveds    z/StandardInputOutputTests.test_lastWriteReceivedc2tj}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)NrM)r splitlinesrkrQrrR)r/hostZpeerrnr#r$r0s  z?StandardInputOutputTests.test_hostAndPeer..processEndedNrrrErLrYr#rnr$test_hostAndPeers   z)StandardInputOutputTests.test_hostAndPeercrr)z Verify that the C{write} method of the transport of a protocol connected to L{StandardIO} sends bytes to standard out. sstdio_test_writec"jdd|tjdSNrMsok! assertEqualr rQrrRrarnr#r$r0z9StandardInputOutputTests.test_write..processEndedNrurYr#rnr$ test_write   z#StandardInputOutputTests.test_writecrr)z Verify that the C{writeSequence} method of the transport of a protocol connected to L{StandardIO} sends bytes to standard out. sstdio_test_writeseqcrwrxryrarnr#r$r0 r{zAStandardInputOutputTests.test_writeSequence..processEndedNrurYr#rnr$test_writeSequencer}z+StandardInputOutputTests.test_writeSequencecCsV|}t|d}tdD] }|d|fqWd|S1s$wY|S)Nwbi%d )rXrNrangewrite)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. dcs<rdfdtdddSdS)Nrg{Gz?)appendpoprr callLater)Zign)r'proctoWritewrittenr#r$r'8s z>StandardInputOutputTests.test_producer..connectionMadesstdio_test_producercs>jdddtf|tjdS)NrMr)z*Connection lost with %d writes left to go.)rzr joinZ assertFalselenrQrrRra)rWr"rrr#r$r0Bs z.processEndedN)rrrCrrErrbrLrYr#)r'rWrr"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_consumercsPtd}jd|Wdn1swY|tjdS)NrbrM)rNrzr readrQrrR)r/rSrrWr"r#r$r0Ws z.processEndedN)rrrrErLrYr#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 s@t\}}tj |tj |||d<tj|fi|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. r)stdoutstdincs@D]}|krdSd|ftddS)N%dr)ZloseConnectionrr r)value) connectioncounthowManyspinr#r$rszAStandardInputOutputTests.test_normalFileStandardOut..spinrcs<tddddtDdS)NrMr)css|]}d|fVqdS)rNr#).0rr#r#r$ szVStandardInputOutputTests.test_normalFileStandardOut..cbLost..)rznextZ getContentrrra)rrpathr"r#r$cbLostszCStandardInputOutputTests.test_normalFileStandardOut..cbLostN)rrrrZFilePathrXrNnormalZ addCleanupclosedictfilenor isWindowsospiper Z StandardIO itertoolsrr rrb)r"Z onConnLostr7rkwargsrwrr#)rrrrr"rr$test_normalFileStandardOut^s&     z3StandardInputOutputTests.test_normalFileStandardOut) r7r r8r9r:r;r<rrr )r-rFr&rGrrF)F)r[r;r\r]rrF)rrF)r1r2r3rrrskiprErLrZrcrfrirjrqrvr|r~rrrrrr#r#r#r$r6Ss0       '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.unittestrrrmrenvironrDpathseprrrr6r#r#r#r$s*        ,