File: //proc/self/root/lib/python3/dist-packages/twisted/test/__pycache__/test_iutils.cpython-38.pyc
U
W[�4 � @ s� d Z ddlmZmZ ddlZddlZddlZddlZddlZddl m
Z
ddl m Z ddl
mZ ddlmZmZmZmZ ddlmZ dd lmZ G d
d � d ej�ZG d d
� d
ej�ZG dd� de�Z dS )zD
Test running processes with the APIs in L{twisted.internet.utils}.
� )�division�absolute_importN)�_PY3)�platform)�unittest)�error�reactor�utils�
interfaces)�Deferred)�SuppressedWarningsTestsc @ s� e Zd ZdZe�ed�dkr dZdZdZ e
j Z dd� Z
dd� Zdd � Zd
d � Zd d
� Zdd� Zdd� Ze�� rtde_dd� Zdd� Zdd� Zdd� Zdd � Zd d � Zd d � Z d!d"� Z dS )#�ProcessUtilsTestszt
Test running a process using L{getProcessOutput}, L{getProcessValue}, and
L{getProcessOutputAndValue}.
Nz)reactor doesn't implement IReactorProcessc C sB | � � }t|d�� }|�tj�|�tj � W 5 Q R X tj�|�S )zj
Write the given list of lines to a text file and return the absolute
path to it.
Zwt)�mktemp�open�write�os�linesep�join�path�abspath)�selfZ sourceLinesZscript�
scriptFile� r �:/usr/lib/python3/dist-packages/twisted/test/test_iutils.py�makeSourceFile! s "z ProcessUtilsTests.makeSourceFilec C s> | � 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
�r r �getProcessOutput�exe� addCallback� assertEqual�r r �dr r r � test_output, s � z ProcessUtilsTests.test_outputc sF � � ddg�}t�� jd|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.
r z!sys.stderr.write("hello world\n")r c s � � | jtj�S �N)�
assertFailureZ processEndedr Z ProcessDone)�err�r r r �cbFailedM s z?ProcessUtilsTests.test_outputWithErrorIgnored.<locals>.cbFailed)r r r r r& �IOErrorr )r r r# r) r r( r �test_outputWithErrorIgnored@ s �
z-ProcessUtilsTests.test_outputWithErrorIgnoredc C s8 | � 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.
r zsys.stdout.write("foo")�sys.stdout.flush()zsys.stderr.write("foo")�sys.stderr.flush()r T)Zerrortoos foofoor r" r r r � test_outputWithErrorCollectedS s � z/ProcessUtilsTests.test_outputWithErrorCollectedc C s, | � dg�}t�| jd|g�}|�| jd�S )z|
The L{Deferred} returned by L{getProcessValue} is fired with the exit
status of the child process.
zraise SystemExit(1)r � )r r �getProcessValuer r r! r" r r r �
test_valuef s z ProcessUtilsTests.test_valuec sF � � ddddddddd d
g
�}� fd d �}t�� jd
|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.
r z!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)c sJ | \}}}� � |d� tr(� � |d� n� � |dtj � � � |d� d S )Ns
hello world!
s goodbye world!
s goodbye world!r/ )r! r r r )Z out_err_code�outr' �coder( r r �gotOutputAndValue� s
�z@ProcessUtilsTests.test_outputAndValue.<locals>.gotOutputAndValuer )r r �getProcessOutputAndValuer r �r r r4 r# r r( r �test_outputAndValueq s �
z%ProcessUtilsTests.test_outputAndValuec sJ � � ddddddg�}� fdd�}t�� jd |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)c s4 | \}}}� � |d� � � |d� � � |tj� d S )Ns
stdout bytes
s
stderr bytes
)r! �signal�SIGKILL)Z out_err_sigr2 r' Zsigr( r r r4 � s
z>ProcessUtilsTests.test_outputSignal.<locals>.gotOutputAndValuer )r r r5 r r& �tupler r6 r r( r �test_outputSignal� s
� z#ProcessUtilsTests.test_outputSignalz"Windows doesn't have real signals.c C sV t j�| �� �}t �|� | �ddg�}|| jd|g|d�}|�||�t �
� �� |S )Nzimport os, sys� sys.stdout.write(os.getcwd())r )r ) r r r r �makedirsr r r �encode�sys�getfilesystemencoding�r ZutilFunc�check�dirr r# r r r � _pathTest� s
�zProcessUtilsTests._pathTestc C s | � tj| j�S )z
L{getProcessOutput} runs the given command with the working directory
given by the C{path} parameter.
)rD r r r! r( r r r �test_getProcessOutputPath� s z+ProcessUtilsTests.test_getProcessOutputPathc s � fdd�}� � tj|�S )z~
L{getProcessValue} runs the given command with the working directory
given by the C{path} parameter.
c s � � | d� d S �Nr �r! ��resultZignoredr( r r rB � s z9ProcessUtilsTests.test_getProcessValuePath.<locals>.check)rD r r0 �r rB r r( r �test_getProcessValuePath� s z*ProcessUtilsTests.test_getProcessValuePathc s � fdd�}� � tj|�S )z�
L{getProcessOutputAndValue} runs the given command with the working
directory given by the C{path} parameter.
c s& | \}}}� � ||� � � |d� d S rF rG �Zout_err_statusrC r2 r' Zstatusr( r r rB � s
zBProcessUtilsTests.test_getProcessOutputAndValuePath.<locals>.check)rD r r5 rJ r r( r �!test_getProcessOutputAndValuePath� s z3ProcessUtilsTests.test_getProcessOutputAndValuePathc C s� t j�| �� �}t �|� | �dd|f dg�}| �t jt �� � t �|� | �t j |t
� t �
d�j �� t � |d� || j
dd|g�}|�||�t�� �� |S )Nzimport os, sys, statzos.chmod(%r, stat.S_IXUSR)r< �.r z-Sr )r r r r r= r Z
addCleanup�chdir�getcwd�chmod�stat�S_IMODE�st_moder r r> r? r@ rA r r r �_defaultPathTest� s$
�
� z"ProcessUtilsTests._defaultPathTestc C s | � tj| j�S )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.
)rU r r r! r( r r r � test_getProcessOutputDefaultPath� s z2ProcessUtilsTests.test_getProcessOutputDefaultPathc s � fdd�}� � 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.
c s � � | d� d S rF rG rH r( r r rB s z@ProcessUtilsTests.test_getProcessValueDefaultPath.<locals>.check)rU r r0 rJ r r( r � test_getProcessValueDefaultPath s z1ProcessUtilsTests.test_getProcessValueDefaultPathc s � fdd�}� � 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.
c s& | \}}}� � ||� � � |d� d S rF rG rL r( r r rB s
zIProcessUtilsTests.test_getProcessOutputAndValueDefaultPath.<locals>.check)rU r r5 rJ r r( r �(test_getProcessOutputAndValueDefaultPath s
�z:ProcessUtilsTests.test_getProcessOutputAndValueDefaultPath) �__name__�
__module__� __qualname__�__doc__r
ZIReactorProcessr �skip�output�valuer? �
executabler r r$ r+ r. r1 r7 r; r Z isWindowsrD rE rK rM rU rV rW rX r r r r r
s. !
r
c @ s e Zd ZdZdd� ZdS )�SuppressWarningsTestsz.
Tests for L{utils.suppressWarnings}.
c s� g � � fdd�}| � td|� dd� }t�|dtdd�f�}|d � | �t� �d
� |d� | �t� �d
� |d � | �t� �d � d
S )zs
L{utils.suppressWarnings} decorates a function so that the given
warnings are suppressed.
c s � � ||f� d S r% )�append)r �a�kw�rI r r � showwarning* s z@SuppressWarningsTests.test_suppressWarnings.<locals>.showwarningrf c S s t �| � d S r% )�warnings�warn)�msgr r r �f. s z6SuppressWarningsTests.test_suppressWarnings.<locals>.f)�ignorezThis is message��messagezSanity check messager/ zUnignored message� N)Zpatchrg r ZsuppressWarnings�dictr! �len)r rf rj �gr re r �test_suppressWarnings$ s z+SuppressWarningsTests.test_suppressWarningsN)rY rZ r[ r\ rr r r r r ra s ra c @ s* e Zd ZdZeej�Zdd� Zdd� ZdS )� DeferredSuppressedWarningsTestsz`
Tests for L{utils.runWithWarningsSuppressed}, the version that supports
Deferreds.
c sh di fdi fg}t � � | �|� fdd�� t�d� � �d� t�d� | �dgdd � | �� D �� d
S ) z�
If the function called by L{utils.runWithWarningsSuppressed} returns a
C{Deferred}, the warning filters aren't removed until the Deferred
fires.
�rk z.*foo.*�rk z.*bar.*c s � S r% r r re r r �<lambda>U � zGDeferredSuppressedWarningsTests.test_deferredCallback.<locals>.<lambda>�
ignore foo� � ignore foo 2c S s g | ] }|d �qS rl r ��.0�wr r r �
<listcomp>Z s zIDeferredSuppressedWarningsTests.test_deferredCallback.<locals>.<listcomp>N)r �runWithWarningsSuppressedrg rh �callbackr! �
flushWarnings)r �filtersr re r �test_deferredCallbackL s �
�z5DeferredSuppressedWarningsTests.test_deferredCallbackc sx di fdi fg}t � � | �|� fdd��}t�d� � �t� � |�dd� � t�d� | �dgdd � | �� D �� d
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.
rt ru c s � S r% r r re r r rv f rw zFDeferredSuppressedWarningsTests.test_deferredErrback.<locals>.<lambda>rx c S s
| � t�S r% )Ztrap�ZeroDivisionError)rj r r r rv i rw rz c S s g | ] }|d �qS rl r r{ r r r r~ l s zHDeferredSuppressedWarningsTests.test_deferredErrback.<locals>.<listcomp>N) r r rg rh Zerrbackr� Z
addErrbackr! r� )r r� r# r re r �test_deferredErrback] s �
�z4DeferredSuppressedWarningsTests.test_deferredErrbackN) rY rZ r[ r\ � staticmethodr r r� r� r r r r rs C s
rs ) r\ Z
__future__r r rg r rR r? r8 Ztwisted.python.compatr Ztwisted.python.runtimer Z
twisted.trialr Ztwisted.internetr r r r
Ztwisted.internet.deferr Z twisted.python.test.test_utilr ZTestCaser
ZSynchronousTestCasera rs r r r r �<module> s ( #