U  W[4@sdZddlmZmZddlZddlZddlZddlZddlZddl m Z ddl m Z ddl mZddlmZmZmZmZddlmZdd lmZGd d d ejZGd d d ejZGdddeZdS)zD Test running processes with the APIs in L{twisted.internet.utils}. )divisionabsolute_importN)_PY3)platform)unittest)errorreactorutils interfaces)Deferred)SuppressedWarningsTestsc@seZdZdZeeddkr dZdZdZ e j Z ddZ ddZdd Zd d Zd d ZddZddZertde_ddZddZddZddZddZddZdd Zd!d"ZdS)#ProcessUtilsTestszt Test running a process using L{getProcessOutput}, L{getProcessValue}, and L{getProcessOutputAndValue}. Nz)reactor doesn't implement IReactorProcessc CsB|}t|d}|tj|tjW5QRXtj|S)zj Write the given list of lines to a text file and return the absolute path to it. Zwt)mktempopenwriteoslinesepjoinpathabspath)selfZ sourceLinesZscript scriptFiler:/usr/lib/python3/dist-packages/twisted/test/test_iutils.pymakeSourceFile!s "z ProcessUtilsTests.makeSourceFilec Cs>|ddddddddd d g }t|jd |g}||jd S) 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% if hasattr(sys.stdout, 'buffer'):z # Python 3z s = bytes([s])z" sys.stdout.buffer.write(s)z else:z # Python 2z sys.stdout.write(s)z sys.stdout.flush()-us hello world rr getProcessOutputexe 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)errrrrcbFailedMsz?ProcessUtilsTests.test_outputWithErrorIgnored..cbFailed)rr rrr&IOErrorr )rrr#r)rr(rtest_outputWithErrorIgnored@s   z-ProcessUtilsTests.test_outputWithErrorIgnoredcCs8|dddddg}tj|jd|gdd}||jd S) 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)Zerrortoosfoofoorr"rrrtest_outputWithErrorCollectedSs 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)rr getProcessValuerr r!r"rrr test_valuefs zProcessUtilsTests.test_valuec sFddddddddd d g }fd d }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!if hasattr(sys.stdout, 'buffer'):z # Python 3z. sys.stdout.buffer.write(b'hello world!\n')z0 sys.stderr.buffer.write(b'goodbye world!\n')zelse:z # Python 2z' sys.stdout.write(b'hello world!\n')z) sys.stderr.write(b'goodbye world!\n')z sys.exit(1)csJ|\}}}|dtr(|dn|dtj|ddS)Ns hello world! sgoodbye world! sgoodbye world!r/)r!rrr)Z out_err_codeoutr'coder(rrgotOutputAndValues  z@ProcessUtilsTests.test_outputAndValue..gotOutputAndValuer)rr getProcessOutputAndValuerr rrr4r#rr(rtest_outputAndValueqs z%ProcessUtilsTests.test_outputAndValuecsJddddddg}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!signalSIGKILL)Z out_err_sigr2r'Zsigr(rrr4s   z>ProcessUtilsTests.test_outputSignal..gotOutputAndValuer)rr r5rr&tupler r6rr(rtest_outputSignals   z#ProcessUtilsTests.test_outputSignalz"Windows doesn't have real signals.cCsVtj|}t||ddg}||jd|g|d}|||t |S)Nzimport os, syssys.stdout.write(os.getcwd())r)r) rrrrmakedirsrrr encodesysgetfilesystemencodingrZutilFunccheckdirrr#rrr _pathTests zProcessUtilsTests._pathTestcCs|tj|jS)z L{getProcessOutput} runs the given command with the working directory given by the C{path} parameter. )rDr rr!r(rrrtest_getProcessOutputPathsz+ProcessUtilsTests.test_getProcessOutputPathcsfdd}tj|S)z~ L{getProcessValue} runs the given command with the working directory given by the C{path} parameter. cs|ddSNrr!resultZignoredr(rrrBsz9ProcessUtilsTests.test_getProcessValuePath..check)rDr r0rrBrr(rtest_getProcessValuePaths z*ProcessUtilsTests.test_getProcessValuePathcsfdd}tj|S)z L{getProcessOutputAndValue} runs the given command with the working directory given by the C{path} parameter. cs&|\}}}|||ddSrFrGZout_err_statusrCr2r'Zstatusr(rrrBs  zBProcessUtilsTests.test_getProcessOutputAndValuePath..check)rDr r5rJrr(r!test_getProcessOutputAndValuePaths z3ProcessUtilsTests.test_getProcessOutputAndValuePathc Cstj|}t||dd|fdg}|tjtt||tj |t t dj t |d||j dd|g}|||t|S)Nzimport os, sys, statzos.chmod(%r, stat.S_IXUSR)r<.rz-Sr)rrrrr=rZ addCleanupchdirgetcwdchmodstatS_IMODEst_moderr r>r?r@rArrr_defaultPathTests$    z"ProcessUtilsTests._defaultPathTestcCs|tj|jS)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. )rUr rr!r(rrr test_getProcessOutputDefaultPathsz2ProcessUtilsTests.test_getProcessOutputDefaultPathcsfdd}tj|S)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. cs|ddSrFrGrHr(rrrB sz@ProcessUtilsTests.test_getProcessValueDefaultPath..check)rUr r0rJrr(rtest_getProcessValueDefaultPaths z1ProcessUtilsTests.test_getProcessValueDefaultPathcsfdd}tj|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. cs&|\}}}|||ddSrFrGrLr(rrrBs  zIProcessUtilsTests.test_getProcessOutputAndValueDefaultPath..check)rUr r5rJrr(r(test_getProcessOutputAndValueDefaultPaths  z:ProcessUtilsTests.test_getProcessOutputAndValueDefaultPath)__name__ __module__ __qualname____doc__r ZIReactorProcessrskipoutputvaluer? executablerrr$r+r.r1r7r;rZ isWindowsrDrErKrMrUrVrWrXrrrrr s.  !      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)rakwrIrr showwarning*sz@SuppressWarningsTests.test_suppressWarnings..showwarningrfcSst|dSr%)warningswarn)msgrrrf.sz6SuppressWarningsTests.test_suppressWarnings..f)ignorezThis is messagemessagezSanity check messager/zUnignored messageN)Zpatchrgr ZsuppressWarningsdictr!len)rrfrjgrrertest_suppressWarnings$s z+SuppressWarningsTests.test_suppressWarningsN)rYrZr[r\rrrrrrra srac@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. rkz.*foo.*rkz.*bar.*csSr%rrrerrUzGDeferredSuppressedWarningsTests.test_deferredCallback.. ignore foo ignore foo 2cSsg|] }|dqSrlr.0wrrr ZszIDeferredSuppressedWarningsTests.test_deferredCallback..N)r runWithWarningsSuppressedrgrhcallbackr! flushWarnings)rfiltersrrertest_deferredCallbackLs   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. rtrucsSr%rrrerrrvfrwzFDeferredSuppressedWarningsTests.test_deferredErrback..rxcSs |tSr%)ZtrapZeroDivisionError)rjrrrrvirwrzcSsg|] }|dqSrlrr{rrrr~lszHDeferredSuppressedWarningsTests.test_deferredErrback..N) r rrgrhZerrbackrZ addErrbackr!r)rrr#rrertest_deferredErrback]s   z4DeferredSuppressedWarningsTests.test_deferredErrbackN) rYrZr[r\ staticmethodr rrrrrrrrsCs rs)r\Z __future__rrrgrrRr?r8Ztwisted.python.compatrZtwisted.python.runtimerZ twisted.trialrZtwisted.internetrrr r Ztwisted.internet.deferr Ztwisted.python.test.test_utilr ZTestCaser ZSynchronousTestCaserarsrrrrs(     #