o b@5@sdZddlZddlZddlZddlZddlZddlmZddlm Z m Z m Z m Z ddl mZddlmZddlmZddlmZmZGd d d eZGd d d eZGd ddeZdS)zD Test running processes with the APIs in L{twisted.internet.utils}. N)skipIf)error interfacesreactorutils)Deferred)platform)SuppressedWarningsTests)SynchronousTestCaseTestCasec@seZdZdZeeddurdZdZdZ e j Z ddZ ddZdd Zd d Zd d ZddZeedddZddZddZddZddZddZddZdd Zd!d"Zd#d$ZdS)%ProcessUtilsTestszt Test running a process using L{getProcessOutput}, L{getProcessValue}, and L{getProcessOutputAndValue}. Nz)reactor doesn't implement IReactorProcesscCsV|}t|d}|tj|tjWdn1s wYtj|S)zj Write the given list of lines to a text file and return the absolute path to it. wtN)mktempopenwriteoslinesepjoinpathabspath)self sourceLinesscript scriptFiler:/usr/lib/python3/dist-packages/twisted/test/test_iutils.pymakeSourceFile$s   z ProcessUtilsTests.makeSourceFilecCs.|gd}t|jd|g}||jdS)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 assertEqualrrdrrr test_output.s  zProcessUtilsTests.test_outputcsFddg}tjd|g}|t}fdd}|||S)z The L{Deferred} returned by L{getProcessOutput} is fired with an L{IOError} L{Failure} if the child process writes to stderr. rz!sys.stderr.write("hello world\n")rcs|jtjSN) assertFailure processEndedr ProcessDone)errrrrcbFailedLz?ProcessUtilsTests.test_outputWithErrorIgnored..cbFailed)rrr r!r(IOErrorr")rrr%r-rr,rtest_outputWithErrorIgnored?s   z-ProcessUtilsTests.test_outputWithErrorIgnoredcCs2|gd}tj|jd|gdd}||jdS)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. )rzsys.stdout.write("foo")sys.stdout.flush()zsys.stderr.write("foo")sys.stderr.flush()rT)errortoosfoofoorr$rrrtest_outputWithErrorCollectedRs  z/ProcessUtilsTests.test_outputWithErrorCollectedcCs,|dg}t|jd|g}||jdS)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$rrr test_valuegs zProcessUtilsTests.test_valuecs6gd}fdd}tjd|g}||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. )rz*sys.stdout.buffer.write(b'hello world!\n')z,sys.stderr.buffer.write(b'goodbye world!\n')z sys.exit(1)cs2|\}}}|d|d|ddS)Ns hello world! sgoodbye world! r5r# out_err_codeoutr+coder,rrgotOutputAndValues   z@ProcessUtilsTests.test_outputAndValue..gotOutputAndValuerrrgetProcessOutputAndValuer!r"rrr=r%rr,rtest_outputAndValueqs   z%ProcessUtilsTests.test_outputAndValuez"Windows doesn't have real signals.csBgd}fdd}tjd|g}|t}||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')r1r2z$os.kill(os.getpid(), signal.SIGKILL)cs4|\}}}|d|d|tjdS)Ns stdout bytes s stderr bytes )r#signalSIGKILL) out_err_sigr;r+sigr,rrr=s   z>ProcessUtilsTests.test_outputSignal..gotOutputAndValuer)rrr?r!r(tupler"r@rr,rtest_outputSignals    z#ProcessUtilsTests.test_outputSignalcCsVtj|}t||ddg}||jd|g|d}|||t |S)Nimport os, syszsys.stdout.write(os.getcwd())r)r) rrrrmakedirsrr!r"encodesysgetfilesystemencoding)rutilFunccheckdirrr%rrr _pathTests zProcessUtilsTests._pathTestcC|tj|jS)z L{getProcessOutput} runs the given command with the working directory given by the C{path} parameter. )rPrr r#r,rrrtest_getProcessOutputPathsz+ProcessUtilsTests.test_getProcessOutputPathcfdd}tj|S)z~ L{getProcessValue} runs the given command with the working directory given by the C{path} parameter. c|ddSNrr8resultignoredr,rrrNr.z9ProcessUtilsTests.test_getProcessValuePath..check)rPrr6rrNrr,rtest_getProcessValuePaths z*ProcessUtilsTests.test_getProcessValuePathcrS)z L{getProcessOutputAndValue} runs the given command with the working directory given by the C{path} parameter. c&|\}}}|||ddSrUr8out_err_statusrOr;r+statusr,rrrN  zBProcessUtilsTests.test_getProcessOutputAndValuePath..check)rPrr?rYrr,r!test_getProcessOutputAndValuePaths z3ProcessUtilsTests.test_getProcessOutputAndValuePathcCstj|}t||gd}|tjtt|t t dj }t |t j t jB|tj ||||jd|g}|||t|S)N)rHzcdir = os.getcwd()zsys.stdout.write(cdir).r)rrrrrIr addCleanupchdirgetcwdstatS_IMODEst_modechmodS_IXUSRS_IRUSRr!r"rJrKrL)rrMrNrOr originalModer%rrr_defaultPathTests  z"ProcessUtilsTests._defaultPathTestcCrQ)a If no value is supplied for the C{path} parameter, L{getProcessOutput} runs the given command in the same working directory as the parent process and succeeds even if the current working directory is not accessible. )rlrr r#r,rrr test_getProcessOutputDefaultPathsz2ProcessUtilsTests.test_getProcessOutputDefaultPathcrS)a If no value is supplied for the C{path} parameter, L{getProcessValue} runs the given command in the same working directory as the parent process and succeeds even if the current working directory is not accessible. crTrUr8rVr,rrrNr.z@ProcessUtilsTests.test_getProcessValueDefaultPath..check)rlrr6rYrr,rtest_getProcessValueDefaultPaths z1ProcessUtilsTests.test_getProcessValueDefaultPathcrS)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. cr[rUr8r\r,rrrNr_zIProcessUtilsTests.test_getProcessOutputAndValueDefaultPath..check)rlrr?rYrr,r(test_getProcessOutputAndValueDefaultPaths z:ProcessUtilsTests.test_getProcessOutputAndValueDefaultPathcsDddg}dtjjd|gd}fdd}|||S)z Standard input can be made available to the child process by passing bytes for the `stdinBytes` parameter. rz"sys.stdout.write(sys.stdin.read())sThese are the bytes to see.r) stdinBytescs&|\}}}|d|dSrU)assertInr#r9rrprrr=(s  zPProcessUtilsTests.test_get_processOutputAndValueStdin..gotOutputAndValuer>)rrr%r=rrrr#test_get_processOutputAndValueStdins z5ProcessUtilsTests.test_get_processOutputAndValueStdin) __name__ __module__ __qualname____doc__rIReactorProcessrskipoutputvaluerK executabler!rr&r0r4r7rArr isWindowsrGrPrRrZr`rlrmrnrorsrrrrr s0        r c@seZdZdZddZdS)SuppressWarningsTestsz. Tests for L{utils.suppressWarnings}. csgfdd}|td|dd}t|dtddf}|d |td |d|td |d |td d S)zs L{utils.suppressWarnings} decorates a function so that the given warnings are suppressed. cs||fdSr')append)rakwrWrr showwarning@sz@SuppressWarningsTests.test_suppressWarnings..showwarningrcSst|dSr')warningswarn)msgrrrfEsz6SuppressWarningsTests.test_suppressWarnings..f)ignorezThis is messagemessagezSanity check messager5zUnignored messageN)patchrrsuppressWarningsdictr#len)rrrgrrrtest_suppressWarnings9s z+SuppressWarningsTests.test_suppressWarningsN)rtrurvrwrrrrrr~4s r~c@s*eZdZdZeejZddZddZdS)DeferredSuppressedWarningsTestsz` Tests for L{utils.runWithWarningsSuppressed}, the version that supports Deferreds. cshdifdifg}t||fddtddtd|dgdd |Dd S) 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'rrrrrlzGDeferredSuppressedWarningsTests.test_deferredCallback.. ignore foo ignore foo 2cSg|]}|dqSrr.0wrrr pzIDeferredSuppressedWarningsTests.test_deferredCallback..N)rrunWithWarningsSuppressedrrcallbackr# flushWarnings)rfiltersrrrtest_deferredCallbackds    z5DeferredSuppressedWarningsTests.test_deferredCallbackcsxdifdifg}t||fdd}tdt|ddtd|dgdd |Dd S) 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. rrcrr'rrrrrrzrzFDeferredSuppressedWarningsTests.test_deferredErrback..rcSs |tSr')trapZeroDivisionError)rrrrr}s rcSrrrrrrrrrzHDeferredSuppressedWarningsTests.test_deferredErrback..N) rrrrerrbackr addErrbackr#r)rrr%rrrtest_deferredErrbackrs    z4DeferredSuppressedWarningsTests.test_deferredErrbackN) rtrurvrw staticmethodrrrrrrrrrZs   r)rwrrBrerKrunittestrtwisted.internetrrrrtwisted.internet.deferrtwisted.python.runtimertwisted.python.test.test_utilr twisted.trial.unittestr r r r~rrrrrs"    &