| Server IP : 209.38.156.173 / Your IP : 216.73.216.122 [ Web Server : Apache/2.4.52 (Ubuntu) System : Linux lakekumayuhotel 5.15.0-136-generic #147-Ubuntu SMP Sat Mar 15 15:53:30 UTC 2025 x86_64 User : root ( 0) PHP Version : 8.1.2-1ubuntu2.22 Disable Function : NONE Domains : 2 Domains MySQL : OFF | cURL : ON | WGET : ON | Perl : ON | Python : OFF | Sudo : ON | Pkexec : ON Directory : /lib/python3/dist-packages/twisted/test/__pycache__/ |
Upload File : |
o
�b@5 � @ s� d Z ddlZddlZddlZddlZddlZddlmZ ddlm Z m
Z
m Z m Z ddl
mZ ddlmZ ddlmZ ddlmZmZ G d d
� d
e�ZG d d � d e�ZG d
d� de�ZdS )zD
Test running processes with the APIs in L{twisted.internet.utils}.
� N)�skipIf)�error�
interfaces�reactor�utils)�Deferred)�platform)�SuppressedWarningsTests)�SynchronousTestCase�TestCasec @ s� e Zd ZdZe�ed�du rdZdZdZ e
j Z dd� Z
dd� Zdd � Zd
d � Zd d
� Zdd� Zee�� d�dd� �Zdd� Zdd� Zdd� Zdd� Zdd � Zd d � Z d d � Z d!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 sV | � � }t|d��}|�tj�|�tj � W d � n1 s w Y tj�|�S )zj
Write the given list of lines to a text file and return the absolute
path to it.
�wtN)�mktemp�open�write�os�linesep�join�path�abspath)�self� sourceLines�script�
scriptFile� r �:/usr/lib/python3/dist-packages/twisted/test/test_iutils.py�makeSourceFile$ s
� z ProcessUtilsTests.makeSourceFilec C s. | � g d��}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 s = bytes([s])z sys.stdout.buffer.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)�
assertFailure� processEndedr � ProcessDone)�err�r r r �cbFailedL � 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 s2 | � g d��}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)�errortoos foofoor r$ r r r � test_outputWithErrorCollectedR 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_valueg s z ProcessUtilsTests.test_valuec s6 � � g d��}� fdd�}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*sys.stdout.buffer.write(b'hello world!\n')z,sys.stderr.buffer.write(b'goodbye world!\n')z sys.exit(1)c s2 | \}}}� � |d� � � |d� � � |d� d S )Ns
hello world!
s goodbye world!
r5 �r# �� out_err_code�outr+ �coder, r r �gotOutputAndValue� s
z@ProcessUtilsTests.test_outputAndValue.<locals>.gotOutputAndValuer �r r �getProcessOutputAndValuer! r" �r r r= r% r r, r �test_outputAndValueq s �
z%ProcessUtilsTests.test_outputAndValuez"Windows doesn't have real signals.c sB � � g d��}� 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')r1 r2 z$os.kill(os.getpid(), signal.SIGKILL)c s4 | \}}}� � |d� � � |d� � � |tj� d S )Ns
stdout bytes
s
stderr bytes
)r# �signal�SIGKILL)� out_err_sigr; r+ �sigr, r r r= � s
z>ProcessUtilsTests.test_outputSignal.<locals>.gotOutputAndValuer )r r r? r! r( �tupler" r@ r r, r �test_outputSignal� s �
z#ProcessUtilsTests.test_outputSignalc C sV t j�| �� �}t �|� | �ddg�}|| jd|g|d�}|�||�t �
� �� |S )N�import os, sysz sys.stdout.write(os.getcwd())r )r ) r r r r �makedirsr r! r" �encode�sys�getfilesystemencoding)r �utilFunc�check�dirr r% r r r � _pathTest� s
�zProcessUtilsTests._pathTestc C � | � tj| j�S )z
L{getProcessOutput} runs the given command with the working directory
given by the C{path} parameter.
)rP r r r# r, r r r �test_getProcessOutputPath� s z+ProcessUtilsTests.test_getProcessOutputPathc � � fdd�}� � tj|�S )z~
L{getProcessValue} runs the given command with the working directory
given by the C{path} parameter.
c � � � | d� d S �Nr r8 ��result�ignoredr, r r rN � r. z9ProcessUtilsTests.test_getProcessValuePath.<locals>.check)rP r r6 �r rN r r, r �test_getProcessValuePath� s z*ProcessUtilsTests.test_getProcessValuePathc rS )z�
L{getProcessOutputAndValue} runs the given command with the working
directory given by the C{path} parameter.
c �& | \}}}� � ||� � � |d� d S rU r8 ��out_err_statusrO r; r+ �statusr, r r rN � �
zBProcessUtilsTests.test_getProcessOutputAndValuePath.<locals>.check)rP r r? rY r r, r �!test_getProcessOutputAndValuePath� s z3ProcessUtilsTests.test_getProcessOutputAndValuePathc C s� t j�| �� �}t �|� | �g d��}| �t jt �� � t �|� t �
t � d�j �}t � |t j
t jB � | �t j ||� || jd|g�}|�||�t�� �� |S )N)rH zcdir = os.getcwd()zsys.stdout.write(cdir)�.r )r r r r rI r �
addCleanup�chdir�getcwd�stat�S_IMODE�st_mode�chmod�S_IXUSR�S_IRUSRr! r" rJ rK rL )r rM rN rO r � originalModer% r r r �_defaultPathTest� s
�
z"ProcessUtilsTests._defaultPathTestc C rQ )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.
)rl r r r# r, r r r � test_getProcessOutputDefaultPath� s z2ProcessUtilsTests.test_getProcessOutputDefaultPathc rS )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 rT rU r8 rV r, r r rN r. z@ProcessUtilsTests.test_getProcessValueDefaultPath.<locals>.check)rl r r6 rY r r, r � test_getProcessValueDefaultPath� s z1ProcessUtilsTests.test_getProcessValueDefaultPathc rS )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 r[ rU r8 r\ r, r r rN r_ zIProcessUtilsTests.test_getProcessOutputAndValueDefaultPath.<locals>.check)rl r r? rY r r, r �(test_getProcessOutputAndValueDefaultPath s z:ProcessUtilsTests.test_getProcessOutputAndValueDefaultPathc sD � � ddg�}d�tj� jd|g�d�}� �fdd�}|�|� |S )z�
Standard input can be made available to the child process by passing
bytes for the `stdinBytes` parameter.
r z"sys.stdout.write(sys.stdin.read())s These are the bytes to see.r )�
stdinBytesc s& | \}}}� � �|� � �d|� d S rU )�assertInr# r9 �r rp r r r= ( s
zPProcessUtilsTests.test_get_processOutputAndValueStdin.<locals>.gotOutputAndValuer> )r r r% r= r rr r �#test_get_processOutputAndValueStdin s ���
z5ProcessUtilsTests.test_get_processOutputAndValueStdin) �__name__�
__module__� __qualname__�__doc__r �IReactorProcessr �skip�output�valuerK �
executabler! r r& r0 r4 r7 rA r r � isWindowsrG rP rR rZ r` rl rm rn ro rs r r r r r s0
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�rW r r � showwarning@ s z@SuppressWarningsTests.test_suppressWarnings.<locals>.showwarningr� c S s t �| � d S r' )�warnings�warn)�msgr r r �fE s z6SuppressWarningsTests.test_suppressWarnings.<locals>.f)�ignorezThis is message��messagezSanity check messager5 zUnignored message� N)�patchr� r �suppressWarnings�dictr# �len)r r� r� �gr r� r �test_suppressWarnings9 s z+SuppressWarningsTests.test_suppressWarningsN)rt ru rv rw r� r r r r r~ 4 s r~ 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.
�r� z.*foo.*�r� z.*bar.*c � � S r' r r r� r r �<lambda>l � zGDeferredSuppressedWarningsTests.test_deferredCallback.<locals>.<lambda>�
ignore foo� � ignore foo 2c S � g | ]}|d �qS r� r ��.0�wr r r �
<listcomp>p � zIDeferredSuppressedWarningsTests.test_deferredCallback.<locals>.<listcomp>N)r �runWithWarningsSuppressedr� r� �callbackr# �
flushWarnings)r �filtersr r� r �test_deferredCallbackd 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.
r� r� c r� r' r r r� r r r� z r� zFDeferredSuppressedWarningsTests.test_deferredErrback.<locals>.<lambda>r� c S s
| � t�S r' )�trap�ZeroDivisionError)r� r r r r� } s
r� c S r� r� r r� r r r r� r� zHDeferredSuppressedWarningsTests.test_deferredErrback.<locals>.<listcomp>N) r r� r� r� �errbackr� �
addErrbackr# r� )r r� r% r r� r �test_deferredErrbackr s
z4DeferredSuppressedWarningsTests.test_deferredErrbackN) rt ru rv rw � staticmethodr r� r� r� r r r r r� Z s
r� )rw r rB re rK r� �unittestr �twisted.internetr r r r �twisted.internet.deferr �twisted.python.runtimer � twisted.python.test.test_utilr �twisted.trial.unittestr
r r r~ r� r r r r �<module> s" &