o >h8@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#w/var/www/vedio/testing/chatpythonscript.ninositsolution.com/env/lib/python3.10/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):StandardInputOutputTests 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.env) isinstancer,decodesys executabler __class__r2listr spawnProcess properEnv)r"r8r9r;r=procargsr#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: )fail)resultr!r#r$cb}sz4StandardInputOutputTests._requireFailure..cbcs|Srr#)err)r&r#r$ebsz4StandardInputOutputTests._requireFailure..eb) addCallbacks)r"r-r&rMrOr#)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: ) openrmsgrstripfailIfInr trapr ProcessDone)r/fline errorLogFilepr"r#r$r0s zBStandardInputOutputTests.test_loseConnection..processEnded)mktemprrTrrrHrQr"r-r0r#r[r$test_loseConnections z,StandardInputOutputTests.test_loseConnectionFexampleoutboolcs\|}tt_d fdd }j|dd}|j|}||||S) NignoredrrrIcs&j}r j|Sj|Sr)rr closeStdout closeStdin)rdr-rbr]r#r$cbBytess   zGStandardInputOutputTests.exampleOutputsAndZeroExitCode..cbBytescSs|tjdSr)rWrrXr/r#r#r$r0r(zLStandardInputOutputTests.exampleOutputsAndZeroExitCode..processEnded)rdrrrI)r^rrrr addCallbackrQrH)r"rarbr\rhr0r-r#rgr$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. stdio_test_halfcloserkr!r#r#r$test_readConnectionLost z0StandardInputOutputTests.test_readConnectionLostcCrl) 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. stdio_test_halfclose_buggyrnr!r#r#r$test_buggyReadConnectionLostrpz5StandardInputOutputTests.test_buggyReadConnectionLostcCs|jdddS)rq stdio_test_halfclose_buggy_writeT)rbrnr!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)usePTYNcs2jdtdjd|tjdS)z Asserts that the parent received the bytes written by the child immediately after the child starts. rRz Received z) from child, did not find expected bytes.N) assertTruer endswithUNIQUE_LAST_WRITE_STRINGrWrrXrir]r"r#r$r0s  zEStandardInputOutputTests.test_lastWriteReceived..processEnded)rrHry ValueErrorrr<rQr)r"er0r#rzr$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)NrR)r splitlinesrwrWrrX)r/hostpeerrzr#r$r0s  z?StandardInputOutputTests.test_hostAndPeer..processEndedrrrHrQr_r#rzr$test_hostAndPeers   z)StandardInputOutputTests.test_hostAndPeercr~)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|tjdSNrRsok! assertEqualr rWrrXrirzr#r$r0z9StandardInputOutputTests.test_write..processEndedrr_r#rzr$ test_write   z#StandardInputOutputTests.test_writecr~)z Verify that the C{writeSequence} method of the transport of a protocol connected to L{StandardIO} sends bytes to standard out. sstdio_test_writeseqcrrrrirzr#r$r0 rzAStandardInputOutputTests.test_writeSequence..processEndedrr_r#rzr$test_writeSequencerz+StandardInputOutputTests.test_writeSequencecCsV|}t|d}tdD] }|d|fqWd|S1s$wY|S)Nwbi%d )r^rSrangewrite)r"junkPathjunkFileir#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)ign)r'proctoWritewrittenr#r$r'8s z>StandardInputOutputTests.test_producer..connectionMadesstdio_test_producercs>jdddtf|tjdS)NrRr)z*Connection lost with %d writes left to go.)rr join assertFalselenrWrrXri)r]r"rrr#r$r0Bs z.processEnded)rrrDrrHrrjrQr_r#)r'r]rr"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)NrbrR)rSrr readrWrrX)r/rYrr]r"r#r$r0Ws z.processEnded)rrrrHrQr_r#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)loseConnectionrr r)value) connectioncounthowManyspinr#r$rszAStandardInputOutputTests.test_normalFileStandardOut..spinrcs<tddddtDdS)NrRr)css|]}d|fVqdS)rNr#).0rr#r#r$ szVStandardInputOutputTests.test_normalFileStandardOut..cbLost..)rnext getContentrrri)rrpathr"r#r$cbLostszCStandardInputOutputTests.test_normalFileStandardOut..cbLost)rrrrFilePathr^rSnormal addCleanupclosedictfilenor isWindowsospiper StandardIO itertoolsrr rrj)r" onConnLostr8rkwargsrwrr#)rrrrr"rr$test_normalFileStandardOut^s&     z3StandardInputOutputTests.test_normalFileStandardOut) r8r r9r:r;r<r=rrr )r-rIr&rJrrI)F)rar<rbrcrrI)rrI)r1r2r3rrrskiprHrQr`rkrorsrur}rrrrrrrrr#r#r#r$r6Ss0       'r6)-r4 __future__rrrrAtypingrrunittestrtwisted.internetrrrr r twisted.internet.interfacesr r twisted.internet.protocolr twisted.pythonrrtwisted.python.failurertwisted.python.reflectrtwisted.python.runtimertwisted.test.test_tcprtwisted.trial.unittestrrryrenvironrFpathseprrrr6r#r#r#r$s*        ,