o i@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)selfZ sourceLinesscript scriptFiler^C:\Users\psant\Downloads\PAF backend\school\venv\Lib\site-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 NrrgetProcessOutputexe 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) assertFailureZ processEndedrZ ProcessDone)errrrrcbFailedLz?ProcessUtilsTests.test_outputWithErrorIgnored..cbFailedN)rrrr 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)ZerrortoosfoofooNrr#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)rN)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! r1r"Z out_err_codeoutr(coder)rrgotOutputAndValues   z@ProcessUtilsTests.test_outputAndValue..gotOutputAndValuerNrrgetProcessOutputAndValuer r!rrr8r$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')r.r/z$os.kill(os.getpid(), signal.SIGKILL)cs4|\}}}|d|d|tjdS)Ns stdout bytes s stderr bytes )r"signalZSIGKILL)Z out_err_sigr6r(sigr)rrr8s   z>ProcessUtilsTests.test_outputSignal..gotOutputAndValuerN)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. N)rIrrr"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|ddSNrr4resultZignoredr)rrrGr+z9ProcessUtilsTests.test_getProcessValuePath..checkN)rIrr2rrGrr)rtest_getProcessValuePaths z*ProcessUtilsTests.test_getProcessValuePathcrL)z L{getProcessOutputAndValue} runs the given command with the working directory given by the C{path} parameter. c&|\}}}|||ddSrNr4Zout_err_statusrHr6r(statusr)rrrG  zBProcessUtilsTests.test_getProcessOutputAndValuePath..checkN)rIrr:rQrr)r!test_getProcessOutputAndValuePaths z3ProcessUtilsTests.test_getProcessOutputAndValuePathcCstj|}t||gd}|tjtt|t t dj }t |t j t jB|tj ||||jd|g}|||t|S)N)rAzcdir = os.getcwd()zsys.stdout.write(cdir).r)rrrrrBrZ addCleanupchdirgetcwdstatS_IMODEst_modechmodS_IXUSRS_IRUSRr r!rCrDrE)rrFrGrHrZ originalModer$rrr_defaultPathTests  z"ProcessUtilsTests._defaultPathTestcCrJ)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. N)rarrr"r)rrr test_getProcessOutputDefaultPathsz2ProcessUtilsTests.test_getProcessOutputDefaultPathcrL)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. crMrNr4rOr)rrrGr+z@ProcessUtilsTests.test_getProcessValueDefaultPath..checkN)rarr2rQrr)rtest_getProcessValueDefaultPaths z1ProcessUtilsTests.test_getProcessValueDefaultPathcrL)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. crSrNr4rTr)rrrGrVzIProcessUtilsTests.test_getProcessOutputAndValueDefaultPath..checkN)rarr:rQrr)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|dSrN)ZassertInr"r5rrerrr8(s  zPProcessUtilsTests.test_get_processOutputAndValueStdin..gotOutputAndValueNr9)rrr$r8rrfr#test_get_processOutputAndValueStdins z5ProcessUtilsTests.test_get_processOutputAndValueStdin) __name__ __module__ __qualname____doc__rZIReactorProcessrskipoutputvaluerD executabler rr%r-r0r3r<rrZ isWindowsr@rIrKrRrWrarbrcrdrgrrrrr 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)rakwrPrr showwarning@sz@SuppressWarningsTests.test_suppressWarnings..showwarningrucSst|dSr&)warningswarn)msgrrrfEsz6SuppressWarningsTests.test_suppressWarnings..f)ignorezThis is messagemessagezSanity check messager1zUnignored messageN)patchrvrZsuppressWarningsdictr"len)rrurygrrtrtest_suppressWarnings9s z+SuppressWarningsTests.test_suppressWarningsN)rhrirjrkrrrrrrp4s rpc@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. rzz.*foo.*rzz.*bar.*cSr&rrrtrrlzGDeferredSuppressedWarningsTests.test_deferredCallback.. ignore foo ignore foo 2cSg|]}|dqSr{r.0wrrr pzIDeferredSuppressedWarningsTests.test_deferredCallback..N)rrunWithWarningsSuppressedrvrwcallbackr" flushWarnings)rfiltersrrtrtest_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&rrrtrrrzrzFDeferredSuppressedWarningsTests.test_deferredErrback..rcSs |tSr&)ZtrapZeroDivisionError)ryrrrr}s rcSrr{rrrrrrrzHDeferredSuppressedWarningsTests.test_deferredErrback..N) rrrvrwZerrbackrZ addErrbackr"r)rrr$rrtrtest_deferredErrbackrs    z4DeferredSuppressedWarningsTests.test_deferredErrbackN) rhrirjrk staticmethodrrrrrrrrrZs   r)rkrr=r[rDrvZunittestrZtwisted.internetrrrrZtwisted.internet.deferrZtwisted.python.runtimerZtwisted.python.test.test_utilr Ztwisted.trial.unittestr r r rprrrrrs"    &