← Back
Editing: 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 ( #
Save File
Cancel