Ϫf@5dZddlZddlZddlZddlZddlZddlmZddlm Z m Z m Z m Z ddl mZddlmZddlmZddlmZmZGd d eZGd d eZGd deZy)zD Test running processes with the APIs in L{twisted.internet.utils}. N)skipIf)error interfacesreactorutils)Deferred)platform)SuppressedWarningsTests)SynchronousTestCaseTestCaseceZdZdZej eddZdZdZ e jZ dZ dZdZdZdZd Zeej*d d Zd Zd ZdZdZdZdZdZdZdZy)ProcessUtilsTestszt Test running a process using L{getProcessOutput}, L{getProcessValue}, and L{getProcessOutputAndValue}. Nz)reactor doesn't implement IReactorProcessc |j}t|d5}|jtjj |tjzdddtj j|S#1swY(xYw)zj Write the given list of lines to a text file and return the absolute path to it. wtN)mktempopenwriteoslinesepjoinpathabspath)self sourceLinesscript scriptFiles :/usr/lib/python3/dist-packages/twisted/test/test_iutils.pymakeSourceFilez ProcessUtilsTests.makeSourceFile$sk  &$  H:   RZZ__[9BJJF G Hwwv&& H Hs ABB c|jgd}tj|jd|g}|j |j dS)z L{getProcessOutput} returns a L{Deferred} which fires with the complete output of the process it runs after that process exits. ) import syszfor s in b'hello world\n':z s = bytes([s])z sys.stdout.buffer.write(s)z sys.stdout.flush()-us hello world rrgetProcessOutputexe addCallback assertEqualrrds r test_outputzProcessUtilsTests.test_output.sN ((    " "488dJ-? @}}T--/?@@cjddg}tjjd|g}j |t }fd}|j ||S)z The L{Deferred} returned by L{getProcessOutput} is fired with an L{IOError} L{Failure} if the child process writes to stderr. r z!sys.stderr.write("hello world\n")r!cXj|jtjSN) assertFailure processEndedr ProcessDone)errrs rcbFailedz?ProcessUtilsTests.test_outputWithErrorIgnored..cbFailedLs"%%c&6&68I8IJ Jr*)rrr#r$r.IOErrorr%)rrr(r2s` rtest_outputWithErrorIgnoredz-ProcessUtilsTests.test_outputWithErrorIgnored?sf (( ? @   " "488dJ-? @   q' * K hr*c|jgd}tj|jd|gd}|j |j dS)z If a C{True} value is supplied for the C{errortoo} parameter to L{getProcessOutput}, the returned L{Deferred} fires with the child's stderr output as well as its stdout output. )r zsys.stdout.write("foo")sys.stdout.flush()zsys.stderr.write("foo")sys.stderr.flush()r!T)errortoosfoofoor"r's rtest_outputWithErrorCollectedz/ProcessUtilsTests.test_outputWithErrorCollectedRsO ((     " "488dJ-?$ O}}T--y99r*c|jdg}tj|jd|g}|j |j dS)z| The L{Deferred} returned by L{getProcessValue} is fired with the exit status of the child process. zraise SystemExit(1)r!)rrgetProcessValuer$r%r&r's r test_valuezProcessUtilsTests.test_valuegsJ ((*?)@A  ! !$((T:,> ?}}T--q11r*cjgd}fd}tjjd|g}|j |S)a The L{Deferred} returned by L{getProcessOutputAndValue} fires with a three-tuple, the elements of which give the data written to the child's stdout, the data written to the child's stderr, and the exit status of the child. )r z*sys.stdout.buffer.write(b'hello world!\n')z,sys.stderr.buffer.write(b'goodbye world!\n')z sys.exit(1)c~|\}}}j|dj|dj|dy)Ns hello world! sgoodbye world! r;r&) out_err_codeoutr1coders rgotOutputAndValuez@ProcessUtilsTests.test_outputAndValue..gotOutputAndValues@)NCd   S"3 4   S"5 6   T1 %r*r!rrgetProcessOutputAndValuer$r%rrrDr(s` rtest_outputAndValuez%ProcessUtilsTests.test_outputAndValueqsL((   &  * *488dJ5G H}}.//r*z"Windows doesn't have real signals.cjgd}fd}tjjd|g}j |t }|j |S)z If the child process exits because of a signal, the L{Deferred} returned by L{getProcessOutputAndValue} fires a L{Failure} of a tuple containing the child's stdout, stderr, and the signal which caused it to exit. )zimport sys, os, signalz"sys.stdout.write('stdout bytes\n')z"sys.stderr.write('stderr bytes\n')r6r7z$os.kill(os.getpid(), signal.SIGKILL)c|\}}}j|dj|dj|tjy)Ns stdout bytes s stderr bytes )r&signalSIGKILL) out_err_sigrBr1sigrs rrDz>ProcessUtilsTests.test_outputSignal..gotOutputAndValuesD'MCc   S"3 4   S"3 4   S&.. 1r*r!)rrrFr$r.tupler%rGs` rtest_outputSignalz#ProcessUtilsTests.test_outputSignals^((    2  * *488dJ5G H   q% (}}.//r*cDtjj|j}tj||j ddg}||j d|g|}|j||jtj|S)Nimport os, syszsys.stdout.write(os.getcwd())r!)r) rrrrmakedirsrr$r%encodesysgetfilesystemencoding)rutilFunccheckdirrr(s r _pathTestzProcessUtilsTests._pathTests}ggoodkkm, C(( > ?  TXXj1 < eSZZ(A(A(CDEr*cV|jtj|jS)z L{getProcessOutput} runs the given command with the working directory given by the C{path} parameter. )rZrr#r&rs rtest_getProcessOutputPathz+ProcessUtilsTests.test_getProcessOutputPaths! ~~e44d6F6FGGr*cNfd}jtj|S)z~ L{getProcessValue} runs the given command with the working directory given by the C{path} parameter. c*j|dyNrr@resultignoredrs rrXz9ProcessUtilsTests.test_getProcessValuePath..check   VQ 'r*)rZrr<rrXs` rtest_getProcessValuePathz*ProcessUtilsTests.test_getProcessValuePaths!  (~~e33U;;r*cNfd}jtj|S)z L{getProcessOutputAndValue} runs the given command with the working directory given by the C{path} parameter. cZ|\}}}j||j|dyr`r@out_err_statusrYrBr1statusrs rrXzBProcessUtilsTests.test_getProcessOutputAndValuePath..check/- Cf   S# &   VQ 'r*)rZrrFres` r!test_getProcessOutputAndValuePathz3ProcessUtilsTests.test_getProcessOutputAndValuePaths!  ( ~~e<.checkrdr*)rzrr<res` rtest_getProcessValueDefaultPathz1ProcessUtilsTests.test_getProcessValueDefaultPaths# ($$U%:%:EBBr*cNfd}jtj|S)a  If no value is supplied for the C{path} parameter, L{getProcessOutputAndValue} runs the given command in the same working directory as the parent process and succeeds even if the current working directory is not accessible. cZ|\}}}j||j|dyr`r@ris rrXzIProcessUtilsTests.test_getProcessOutputAndValueDefaultPath..checkrlr*)rzrrFres` r(test_getProcessOutputAndValueDefaultPathz:ProcessUtilsTests.test_getProcessOutputAndValueDefaultPaths# ( $$U%C%CUKKr*cjddg}dtjjd|g}fd}|j ||S)z Standard input can be made available to the child process by passing bytes for the `stdinBytes` parameter. r z"sys.stdout.write(sys.stdin.read())sThese are the bytes to see.r!) stdinBytescZ|\}}}j|jd|yr`)assertInr&)rArBr1rCrrs rrDzPProcessUtilsTests.test_get_processOutputAndValueStdin..gotOutputAndValue(s.)NCd MM*c *   Q %r*rE)rrr(rDrs` @r#test_get_processOutputAndValueStdinz5ProcessUtilsTests.test_get_processOutputAndValueStdinsb ((4  4  * * HH : !   & '(r*) __name__ __module__ __qualname____doc__rIReactorProcessrskipoutputvaluerU executabler$rr)r4r9r=rHrr isWindowsrPrZr]rfrmrzr|rrrr*rrrs "z!!'408: F E ..C'A"&:*202 H   "FG0H0>H < E<O C Lr*rceZdZdZdZy)SuppressWarningsTestsz. Tests for L{utils.suppressWarnings}. c^gfd}|jtd|d}tj|dt df}|d|j t d|d|j t d|d |j t d y ) zs L{utils.suppressWarnings} decorates a function so that the given warnings are suppressed. c,j||fyr-)append)rakwrbs r showwarningz@SuppressWarningsTests.test_suppressWarnings..showwarning@s MM1b' "r*rc.tj|yr-)warningswarn)msgs rfz6SuppressWarningsTests.test_suppressWarnings..fEs MM# r*)ignorezThis is message)messagezSanity check messager;zUnignored messageN)patchrrsuppressWarningsdictr&len)rrrgrbs @rtest_suppressWarningsz+SuppressWarningsTests.test_suppressWarnings9s  # 8]K8   " "1{DAR4S&T U ! Va(  Va(  Va(r*N)rrrrrrr*rrr4s )r*rc@eZdZdZeej ZdZdZy)DeferredSuppressedWarningsTestsz` Tests for L{utils.runWithWarningsSuppressed}, the version that supports Deferreds. c8difdifg}t|j|fdtjdj dtjd|j dg|j Dcgc]}|d c}ycc}w) z If the function called by L{utils.runWithWarningsSuppressed} returns a C{Deferred}, the warning filters aren't removed until the Deferred fires. rz.*foo.*rz.*bar.*cSr-rrbsrzGDeferredSuppressedWarningsTests.test_deferredCallback..lsr* ignore foo ignore foo 2rN)rrunWithWarningsSuppressedrrcallbackr& flushWarnings)rfilterswrbs @rtest_deferredCallbackz5DeferredSuppressedWarningsTests.test_deferredCallbackds *2.1F0KL &&w? l# n% .)$BTBTBV+WQAiL+WX+Ws B cldifdifg}t|j|fd}tjdj t |j dtjd|jdg|jDcgc]}|d c}ycc}w) z If the function called by L{utils.runWithWarningsSuppressed} returns a C{Deferred}, the warning filters aren't removed until the Deferred fires with an errback. rrcSr-rrsrrzFDeferredSuppressedWarningsTests.test_deferredErrback..zsFr*rc,|jtSr-)trapZeroDivisionError)rs rrzFDeferredSuppressedWarningsTests.test_deferredErrback..}sqvv&78r*rrN) rrrrerrbackr addErrbackr&r)rrr(rrbs @rtest_deferredErrbackz4DeferredSuppressedWarningsTests.test_deferredErrbackrs *2.1F0KL  * *7N C l#(*+ 89 n% .)$BTBTBV+WQAiL+WX+Ws B1 N) rrrr staticmethodrrrrrr*rrrZs&!-U-L-L M Y Yr*r)rrrKrsrUrunittestrtwisted.internetrrrrtwisted.internet.deferrtwisted.python.runtimer twisted.python.test.test_utilr twisted.trial.unittestr r rrrrr*rrs\  >>++A@ZZz#)/#)L%Y&=%Yr*