|
Server : LiteSpeed System : Linux server51.dnsbootclub.com 4.18.0-553.62.1.lve.el8.x86_64 #1 SMP Mon Jul 21 17:50:35 UTC 2025 x86_64 User : nandedex ( 1060) PHP Version : 8.1.33 Disable Function : NONE Directory : /opt/cppython/lib/python3.8/test/__pycache__/ |
U
>��g�: � @ s& d dl Z d dl mZ d dlmZ d dlZd dlZd dlZd dlZd dlZd dl Z d dl
Z
d dl Z d dl Z d dl
Z
d dlZd dlZd dlZd dlZd dlZd dlZd dlZd dlmZ z d dlZW n ek
r� dZY nX ejr�e �d��ejdkZ e r�dZ ndZ d Z e e e!fZ"ej#d
d fa$d d
� Z%G dd� de j&�Z'G dd� de(�Z)G dd� dej*�Z+G dd� de'�Z,G dd� de'�Z-e �.e d�G dd� de'��Z/e �0e d�G d d � d e'��Z1G d d � d e j&�Z2e �0e3ed �d!�G d"d#� d#e,��Z4e �0e d$�G d%d&� d&e'��Z5G d'd(� d(e'�Z6e7d)k�r"e �8� dS )*� N)�mock)�support)�FakePathztest is not helpful for PGO�win32z@import msvcrt; msvcrt.setmode(sys.stdout.fileno(), os.O_BINARY);� )Znonexisting_i_hope�-c�passc C s@ t �d�} | d krd S t�| tj�r<t�| g�jdkr<| fad S )N�truer ) �shutil�which�os�access�X_OK�
subprocess�run�
returncode�ZERO_RETURN_CMD)Z
shell_true� r �3/opt/cppython/lib/python3.8/test/test_subprocess.py� setUpModule3 s
�r c @ s& e Zd Zdd� Zdd� Zddd�ZdS ) � BaseTestCasec C s t �� d S �N)r �
reap_children��selfr r r �setUp> s zBaseTestCase.setUpc C sB t s.tjD ] }|�� q
t�� | �tjd� | �� t�� d S )Nz subprocess._active not empty) � mswindowsr �_active�wait�_cleanup� assertFalseZ
doCleanupsr r )r �instr r r �tearDownC s
�zBaseTestCase.tearDownNc C s$ t �|�}|�� }| �|||� d S r )r �strip_python_stderr�strip� assertEqual)r �stderr�expected�msg�actualr r r �assertStderrEqualO s
z BaseTestCase.assertStderrEqual)N)�__name__�
__module__� __qualname__r r" r* r r r r r = s r c @ s e Zd ZdS )�PopenTestExceptionN)r+ r, r- r r r r r. Y s r. c @ s e Zd ZdZdd� ZdS )�PopenExecuteChildRaiseszePopen subclass for testing cleanup of subprocess.PIPE filehandles when
_execute_child fails.
c O s t d��d S )NzForced Exception for Test)r. �r �args�kwargsr r r �_execute_childa s z&PopenExecuteChildRaises._execute_childN)r+ r, r- �__doc__r3 r r r r r/ ] s r/ c @ s2 e Zd Zdd� Zdd� Zdd� Zdd� Zd d
� Zd d � Zd
d� Z dd� Z
dd� Z dd� Z dd� Z
dd� Zdd� Zdd � Zd d � Zd d � Zd!d"� Zd#d$� Zd%d&� Zd'd(� Zd)d*� Zd+d,� Zd-d.� Zd/d0� Zd1d2� Zd3d4� Z d5d6� Z e � e d7�d8d9� �Z!e � e d7�d:d;� �Z"e � e d7�d<d=� �Z#d>d?� Z$d@dA� Z%dBdC� Z&dDdE� Z'dFdG� Z(dHdI� Z)e � e dJ�dKdL� �Z*e � e dJ�dMdN� �Z+dOdP� Z,e � e-j.e-j/kdQ�dRdS� �Z0e � e-j.e-j/kdQ�e � e1�2� dT�dUdV� ��Z3dWdX� Z4dYdZ� Z5d[d\� Z6d]d^� Z7d_d`� Z8dadb� Z9dcdd� Z:dedf� Z;dgdh� Z<didj� Z=dkdl� Z>dmdn� Z?dodp� Z@dqdr� ZAdsdt� ZBdudv� ZCdwdx� ZDe � e-jEdykdz�e � e1�Fd{�d|kd}�d~d� ��ZGd�d�� ZHd�d�� ZId�d�� ZJd�d�� ZKd�d�� ZLd�d�� ZMd�d�� ZNd�d�� ZOd�d�� ZPd�d�� ZQd�d�� ZRd�d�� ZSd�d�� ZTd�d�� ZUd�d�� ZVd�d�� ZWd�d�� ZXd�d�� ZYd�d�� ZZd�d�� Z[d�d�� Z\d�d�� Z]d�d�� Z^d�d�� Z_d�d�� Z`d�d�� Zad�d�� Zbd�d�� Zcd�d�� Zdd�d�� Zed�d�� Zfd�d�� Zgd�d�� Zhd�dÄ Zid�dń Zjd�dDŽ Zkd�dɄ Zle �meneodʃdˡe �menepd̃d͡e �menepddϡd�dф ���Zqe �mepjr�sd�ep�t� �dӡd�dՄ �Zue � e d֡d�d �Zve � e d֡d�dڄ �Zwd�S )��ProcessTestCasec C sz t jtt jt jt jd�}z4| � |jt
j � | � |jt
j � | � |jt
j � W 5 |j�� |j�� |j�� |�� X d S )N��stdin�stdoutr& ) r �Popenr �PIPEr7 �closer8 r& r �assertIsInstance�io�BufferedIOBase�r �pr r r �test_io_buffered_by_defaultg s �
z+ProcessTestCase.test_io_buffered_by_defaultc C s| t jtt jt jt jdd�}z4| � |jt
j � | � |jt
j � | � |jt
j � W 5 |j�� |j�� |j�� |�� X d S )Nr )r7 r8 r& �bufsize) r r9 r r: r7 r; r8 r& r r< r= � RawIOBaser? r r r �test_io_unbuffered_worksu s �
z(ProcessTestCase.test_io_unbuffered_worksc C s" t �tjddg�}| �|d� d S �Nr �import sys; sys.exit(47)�/ �r �call�sys�
executabler% �r �rcr r r �
test_call_seq� s
�z ProcessTestCase.test_call_seqc C s" | j tjtjtjddgdd� d S )Nr �while True: pass皙�����?��timeout)� assertRaisesr �TimeoutExpiredrI rJ rK r r r r �test_call_timeout� s
�z!ProcessTestCase.test_call_timeoutc C s t �t�}| �|d� d S �Nr )r �
check_callr r% rL r r r �test_check_call_zero� s
z$ProcessTestCase.test_check_call_zeroc C s> | � tj��}t�tjddg� W 5 Q R X | �|jjd� d S rE ) rS r �CalledProcessErrorrW rJ rK r% � exceptionr �r �cr r r �test_check_call_nonzero� s
�z'ProcessTestCase.test_check_call_nonzeroc C s" t �tjddg�}| �d|� d S )Nr �
print('BDFL')� BDFL�r � check_outputrJ rK �assertIn�r �outputr r r �test_check_output� s
�z!ProcessTestCase.test_check_outputc C s> | � tj��}t�tjddg� W 5 Q R X | �|jjd� d S )Nr zimport sys; sys.exit(5)� ) rS r rY ra rJ rK r% rZ r r[ r r r �test_check_output_nonzero� s
�z)ProcessTestCase.test_check_output_nonzeroc C s( t jtjddgt jd�}| �d|� d S )Nr �$import sys; sys.stderr.write('BDFL')�r& r_ )r ra rJ rK �STDOUTrb rc r r r �test_check_output_stderr� s
�z(ProcessTestCase.test_check_output_stderrc C sN t �� }| �|j� |�d� |�d� tjtj ddg|d�}| �
d|� d S )N� pearr r �6import sys; sys.stdout.write(sys.stdin.read().upper())�r7 � PEAR) �tempfile�
TemporaryFile�
addCleanupr; �write�seekr ra rJ rK rb )r �tfrd r r r �test_check_output_stdin_arg� s
��z+ProcessTestCase.test_check_output_stdin_argc C s& t jtjddgdd�}| �d|� d S )Nr rm rl ��inputro r` rc r r r �test_check_output_input_arg� s ��z+ProcessTestCase.test_check_output_input_argc C s& t jtjddgdd�}| �d|� dS )z<input=None has a legacy meaning of input='' on check_output.r �3import sys; print('XX' if sys.stdin.read() else '')Nrw s XX�r ra rJ rK � assertNotInrc r r r � test_check_output_input_none� s ��z,ProcessTestCase.test_check_output_input_nonec C s( t jtjddgd dd�}| �d|� d S )Nr rz T)rx �text�XXr{ rc r r r �!test_check_output_input_none_text� s � �z1ProcessTestCase.test_check_output_input_none_textc C s( t jtjddgd dd�}| �d|� d S )Nr rz T)rx �universal_newlinesr r{ rc r r r �/test_check_output_input_none_universal_newlines� s � �z?ProcessTestCase.test_check_output_input_none_universal_newlinesc C sP | � t��(}tjtjddgtjd�}| �d� W 5 Q R X | �d|j j
d � d S )Nr �print('will not be run')�r8 z-Expected ValueError when stdout arg supplied.r8 r ) rS �
ValueErrorr ra rJ rK r8 �failrb rZ r1 �r r\ rd r r r � test_check_output_stdout_arg� s
�z,ProcessTestCase.test_check_output_stdout_argc C s� t �� }| �|j� |�d� |�d� | �t��(}tj t
j ddg|dd�}| � d� W 5 Q R X | �
d|jjd � | �
d |jjd � d S )
Nrl r r r� � hare�r7 rx �7Expected ValueError when stdin and input args supplied.r7 rx )rp rq rr r; rs rt rS r� r ra rJ rK r� rb rZ r1 �r ru r\ rd r r r �&test_check_output_stdin_with_input_arg� s
�z6ProcessTestCase.test_check_output_stdin_with_input_argc C sL | � tj��&}tjtjddgdd�}| �d� W 5 Q R X | �|jj d� d S )Nr �Mimport sys, time
sys.stdout.write('BDFL')
sys.stdout.flush()
time.sleep(3600)� rQ zExpected TimeoutExpired.r_ )
rS r rT ra rJ rK r� r% rZ rd r� r r r �test_check_output_timeout� s �� z)ProcessTestCase.test_check_output_timeoutc C s8 t j�� }d|d< tjtjddg|d�}| �|d� d S )N�banana�FRUITr z5import sys, os;sys.exit(os.getenv("FRUIT")=="banana")��env� )r �environ�copyr rI rJ rK r% )r �newenvrM r r r �test_call_kwargs s
��z ProcessTestCase.test_call_kwargsc C sh t �� �F}| jttjdd� tjjjj}dg|d }| jttjf|�� W 5 Q R X | � |�
� d� d S )Nr� )Zinvalid_arg_namer r ) r Zcaptured_stderrrS � TypeErrorr r9 �__init__�__code__� co_argcountr% �getvalue)r �sZargcountZ
too_many_argsr r r �test_invalid_args s
z!ProcessTestCase.test_invalid_argsc C sR t jtjddgt jt jd�}| �|jj� | �|jj� |� � | �
|j d � d S )Nr �print("banana")�r8 r& ) r r9 rJ rK r: rr r8 r; r& r r% r7 r? r r r �test_stdin_none s �z ProcessTestCase.test_stdin_nonec C sl d}t jtjd|gt jt jd�}| �|jj� | �|jj� |� � \}}| �
|j d|� | �
|� � d� d S )Nz�import sys; from subprocess import Popen, PIPE;p = Popen([sys.executable, "-c", "print('test_stdout_none')"], stdin=PIPE, stderr=PIPE);p.wait(); assert p.stdout is None;r r� r s test_stdout_none�
r r9 rJ rK r: rr r8 r; r& � communicater% r �rstrip�r �coder@ �out�errr r r �test_stdout_none s � z ProcessTestCase.test_stdout_nonec C sR t jtjddgt jt jd�}| �|jj� | �|jj� |� � | �
|j d � d S )Nr r� �r7 r8 ) r r9 rJ rK r: rr r8 r; r7 r r% r& r? r r r �test_stderr_none6 s �z ProcessTestCase.test_stderr_nonec K s2 |dg }t j|f|�}|�� | �d|j� d S �NrF rG )r r9 r r% r )r �pre_argsr2 r1 r@ r r r �_assert_python? s
z ProcessTestCase._assert_pythonc C s0 t j�t j�tj�d�}| j|dgtjd� d S �N� doesnotexistr �rK )r �path�join�dirnamerJ rK r� �r r� r r r �test_executableG s �z ProcessTestCase.test_executablec C s6 t j�t j�tj�d�}| j|dgt �tj�d� d S r� )r r� r� r� rJ rK r� �fsencoder� r r r �test_bytes_executableR s �
�z%ProcessTestCase.test_bytes_executablec C s4 t j�t j�tj�d�}| j|dgttj�d� d S r� )r r� r� r� rJ rK r� r r� r r r �test_pathlike_executableX s �
�z(ProcessTestCase.test_pathlike_executablec C s0 t jdg}| �|� | jt| j|td d� d S )Nr r r� )rJ rK r� rS �NONEXISTING_ERRORS�NONEXISTING_CMD)r r� r r r � test_executable_takes_precedence^ s
�z0ProcessTestCase.test_executable_takes_precedencez"executable argument replaces shellc C s | j g tjdd� d S �NT)rK �shell)r� rJ rK r r r r � test_executable_replaces_shellh s z.ProcessTestCase.test_executable_replaces_shellc C s | j g t�tj�dd� d S r� )r� r r� rJ rK r r r r �$test_bytes_executable_replaces_shelln s �z4ProcessTestCase.test_bytes_executable_replaces_shellc C s | j g ttj�dd� d S r� )r� r rJ rK r r r r �'test_pathlike_executable_replaces_shells s �z7ProcessTestCase.test_pathlike_executable_replaces_shellc
C s* t �|�� t�� W 5 Q R � S Q R X d S r )r Z
change_cwdr �getcwd)r �cwdr r r �_normalize_cwdy s z ProcessTestCase._normalize_cwdc C s t j�tj�}t j�|�S r )r r� �realpathrJ rK �split)r Z python_pathr r r �_split_python_path� s z"ProcessTestCase._split_python_pathc K sl t j|ddgfdt ji|��}| �|jj� |�� | �d|j� t j
j }| �||�||j� � �
� �� d S )Nr zcimport os, sys; buf = sys.stdout.buffer; buf.write(os.getcwd().encode()); buf.flush(); sys.exit(47)r8 rG )r r9 r: rr r8 r; r r% r r r� �normcase�read�decode)r Z expected_cwdZ
python_argr2 r@ r� r r r � _assert_cwd� s ���
�zProcessTestCase._assert_cwdc C s( t �� }| �|�}| j|tj|d� d S �N�r� )rp �
gettempdirr� r� rJ rK �r �temp_dirr r r �test_cwd� s
zProcessTestCase.test_cwdc C s. t �� }| �|�}| j|tjt�|�d� d S r� )rp r� r� r� rJ rK r r� r� r r r �test_cwd_with_bytes� s
z#ProcessTestCase.test_cwd_with_bytesc C s, t �� }| �|�}| j|tjt|�d� d S r� )rp r� r� r� rJ rK r r� r r r �test_cwd_with_pathlike� s
z&ProcessTestCase.test_cwd_with_pathlikez"pending resolution of issue #15533c C sv | � � \}}tj�tj|�}t�� �H}| �tt j
|g� | jtt j
|g|d� | � |�}| j |||d� W 5 Q R X d S r� �
r� r r� r� �curdirr Ztemp_cwdrS �FileNotFoundErrorr r9 r� r� )r �
python_dir� python_base�
rel_python� wrong_dirr r r �test_cwd_with_relative_arg� s
�
�
z*ProcessTestCase.test_cwd_with_relative_argc C s� | � � \}}tj�tj|�}d}t�� �P}| jtt j
|g|d� | jtt j
|g||d� | � |�}| j ||||d� W 5 Q R X d S )N�somethingyoudonthaver� �rK r� r� )r r� r� r� Z doesntexistr� r r r �!test_cwd_with_relative_executable� s"
�
�
�z1ProcessTestCase.test_cwd_with_relative_executablec C sr | � � \}}tj�||�}tj�tj|�}t�� �6}| jtt j
|g|d� | � |�}| j |||d� W 5 Q R X d S r� )
r� r r� r� r� r r� rS r� r r9 r� r� )r r� r� Z
abs_pythonr� r� r r r �test_cwd_with_absolute_arg� s
�
z*ProcessTestCase.test_cwd_with_absolute_argzTest is not venv-compatiblec C s. | � � \}}| �|�}| j|dtj|d� d S )Nr� r� )r� r� r� rJ rK )r r� r� r r r �test_executable_with_cwd� s
�z(ProcessTestCase.test_executable_with_cwdz#need an installed Python. See #7774c C s | j t�� dtjd� d S )Nr� r� )r� r r� rJ rK r r r r �test_executable_without_cwd� s �z+ProcessTestCase.test_executable_without_cwdc C sH t jtjddgt jd�}|j�d� |j�� |�� | � |j
d� d S )Nr �0import sys; sys.exit(sys.stdin.read() == "pear")rn rl r� ) r r9 rJ rK r: r7 rs r; r r% r r? r r r �test_stdin_pipe� s
��
z ProcessTestCase.test_stdin_pipec C sf t �� }| �|j� |�� }t�|d� t�|dd� tj t
j ddg|d�}|� � | �
|jd� d S �Nrl r r r� rn r� )rp rq rr r; �filenor rs �lseekr r9 rJ rK r r% r �r ru �dr@ r r r �test_stdin_filedes� s
��z"ProcessTestCase.test_stdin_filedesc C sX t �� }| �|j� |�d� |�d� tjtj ddg|d�}|�
� | � |j d� d S r� )
rp rq rr r; rs rt r r9 rJ rK r r% r �r ru r@ r r r �test_stdin_fileobj s
��z"ProcessTestCase.test_stdin_fileobjc C s> t jtjddgt jd�}|� | �|j�� d� W 5 Q R X d S )Nr �&import sys; sys.stdout.write("orange")r� � orange�r r9 rJ rK r: r% r8 r� r? r r r �test_stdout_pipe s
��z ProcessTestCase.test_stdout_pipec C s` t �� }| �|j� |�� }tjtjddg|d�}|� � t
� |dd� | � t
�
|d�d� d S )Nr r� r� r � r� )rp rq rr r; r� r r9 rJ rK r r r� r% r� r� r r r �test_stdout_filedes s
��z#ProcessTestCase.test_stdout_filedesc C sP t �� }| �|j� tjtjddg|d�}|�� |� d� | �
|� � d� d S )Nr r� r� r r� ) rp rq rr r; r r9 rJ rK r rt r% r� r� r r r �test_stdout_fileobj$ s
��
z#ProcessTestCase.test_stdout_fileobjc C s> t jtjddgt jd�}|� | �|j�� d� W 5 Q R X d S )Nr �*import sys; sys.stderr.write("strawberry")ri �
strawberry)r r9 rJ rK r: r* r& r� r? r r r �test_stderr_pipe/ s
��z ProcessTestCase.test_stderr_pipec C s` t �� }| �|j� |�� }tjtjddg|d�}|� � t
� |dd� | � t
�
|d�d� d S )Nr r� ri r r� r� )rp rq rr r; r� r r9 rJ rK r r r� r* r� r� r r r �test_stderr_filedes7 s
��z#ProcessTestCase.test_stderr_filedesc C sP t �� }| �|j� tjtjddg|d�}|�� |� d� | �
|� � d� d S )Nr r� ri r r� � rp rq rr r; r r9 rJ rK r rt r* r� r� r r r �test_stderr_fileobjC s
��
z#ProcessTestCase.test_stderr_fileobjc C sR t jtjddgt jt jd�}|�� \}}| �|d� | �|d� | �|jd� d S )Nr z�import sys, subprocess;rc = subprocess.call([sys.executable, "-c", "import sys;" "sys.stderr.write('42')"], stderr=subprocess.STDOUT);sys.exit(rc)r� s 42� r ) r r9 rJ rK r: r� r* r% r �r r@ r8 r& r r r �,test_stderr_redirect_with_no_stdout_redirectN s
�� z<ProcessTestCase.test_stderr_redirect_with_no_stdout_redirectc C sB t jtjddgt jt jd�}|� | �|j�� d� W 5 Q R X d S )Nr �Rimport sys;sys.stdout.write("apple");sys.stdout.flush();sys.stderr.write("orange")r� � appleorange) r r9 rJ rK r: rj r* r8 r� r? r r r �test_stdout_stderr_pipec s
��z'ProcessTestCase.test_stdout_stderr_pipec C sR t �� }| �|j� tjtjddg||d�}|�� |� d� | �
|� � d� d S )Nr r� r� r r� r� r� r r r �test_stdout_stderr_fileo s
��
z'ProcessTestCase.test_stdout_stderr_filec C sl d}t jtjd|gt jt jd�}| �|jj� | �|jj� |� � \}}| �
|j d|� | �
|� � d� d S )Nz�import sys, subprocess; rc = subprocess.call([sys.executable, "-c", "import os, sys; sys.exit(os.write(sys.stdout.fileno(), b'test with stdout=1'))"], stdout=1); assert rc == 18r r� r s test with stdout=1r� r� r r r � test_stdout_filedes_of_stdout~ s � z-ProcessTestCase.test_stdout_filedes_of_stdoutc C s2 t jtjddgt jd�}|�� | �|jd � d S )Nr z'for i in range(10240):print("x" * 1024)r� )r r9 rJ rK �DEVNULLr r% r8 r? r r r �test_stdout_devnull� s
��z#ProcessTestCase.test_stdout_devnullc C s2 t jtjddgt jd�}|�� | �|jd � d S )Nr z=import sys
for i in range(10240):sys.stderr.write("x" * 1024)ri )r r9 rJ rK r r r% r& r? r r r �test_stderr_devnull� s
��z#ProcessTestCase.test_stderr_devnullc C s2 t jtjddgt jd�}|�� | �|jd � d S )Nr z import sys;sys.stdin.read(1)rn )r r9 rJ rK r r r% r7 r? r r r �test_stdin_devnull� s
��z"ProcessTestCase.test_stdin_devnullc C sT t j�� }d|d< tjtjddgtj|d�� }|�� \}}| � |d� W 5 Q R X d S )N�oranger� r �2import sys,os;sys.stdout.write(os.getenv("FRUIT"))�r8 r� r� )
r r� r� r r9 rJ rK r: r� r% �r r� r@ r8 r& r r r �test_env� s
�� zProcessTestCase.test_envr z#cannot test an empty env on WindowsZPy_ENABLE_SHAREDr� zEThe Python shared library cannot be loaded with an empty environment.c st dd� � t jtjddgt ji d��H}|�� \}}t|�� �}| �|t � � fdd�|D �}| �
|g � W 5 Q R X dS ) z+Verify that env={} is as empty as possible.c S s* d| kp(d| kp(| dkp(| � d�p(| dkS )z:Determine if an environment variable is under our control.Z VERSIONERZ__CFZ
LD_PRELOADZSANDBOX�LC_CTYPE)�
startswith)�nr r r �is_env_var_to_ignore� s ���z<ProcessTestCase.test_empty_env.<locals>.is_env_var_to_ignorer z)import os; print(list(os.environ.keys()))r c s g | ]}� |�s|�qS r r )�.0�k�r r r �
<listcomp>� s �z2ProcessTestCase.test_empty_env.<locals>.<listcomp>N) r r9 rJ rK r: r� �evalr$ r< �listr% )r r@ r8 r& Zchild_env_namesr r r �test_empty_env� s
� � z ProcessTestCase.test_empty_envc C s\ t jd }| �t�� t�|ddg� W 5 Q R X | �t�� t�t jddg� W 5 Q R X d S )N� r r zpass# )rJ rK rS r� r r9 )r �cmdr r r �test_invalid_cmd� s
z ProcessTestCase.test_invalid_cmdc C s� t j�� }d|d< | �t�� tjt|d� W 5 Q R X t j�� }d|d< | �t�� tjt|d� W 5 Q R X t j�� }d|d< | �t�� tjt|d� W 5 Q R X t j�� }d|d< tjtj d d
gtj
|d �� }|� � \}}| � |d � W 5 Q R X d S )
NZcabbagezFRUIT VEGETABLEr� zorange VEGETABLE=cabbager� Zlemonz FRUIT=ORANGEz orange=lemonr z3import sys, os;sys.stdout.write(os.getenv("FRUIT"))r s orange=lemon)
r r� r� rS r� r r9 r rJ rK r: r� r% r
r r r �test_invalid_env� s.
�� z ProcessTestCase.test_invalid_envc C s4 t jtjddgt jd�}|�d� | �|jd� d S )Nr z/import sys;sys.exit(sys.stdin.read() == "pear")rn rl r� �r r9 rJ rK r: r� r% r r? r r r �test_communicate_stdin� s
��
z&ProcessTestCase.test_communicate_stdinc C s@ t jtjddgt jd�}|�� \}}| �|d� | �|d � d S )Nr z)import sys; sys.stdout.write("pineapple")r� � pineapple)r r9 rJ rK r: r� r% r� r r r �test_communicate_stdout s
�� z'ProcessTestCase.test_communicate_stdoutc C s@ t jtjddgt jd�}|�� \}}| �|d � | �|d� d S )Nr z)import sys; sys.stderr.write("pineapple")ri r )r r9 rJ rK r: r� r% r* r� r r r �test_communicate_stderr
s
�� z'ProcessTestCase.test_communicate_stderrc C st t jtjddgt jt jt jd�}| �|jj� | �|jj� | �|j j� |�
d�\}}| � |d� | � |d� d S )Nr zNimport sys,os;sys.stderr.write("pineapple");sys.stdout.write(sys.stdin.read())r6 � bananar )
r r9 rJ rK r: rr r8 r; r& r7 r� r% r* r� r r r �test_communicate s
�� z ProcessTestCase.test_communicatec C sd t jtjddgdt jt jt jd�}| jt j|jddd� |�� \}}| �|d� | � |�
� d� d S ) Nr z~import sys,os,time;sys.stderr.write("pineapple\n");time.sleep(1);sys.stderr.write("pear\n");sys.stdout.write(sys.stdin.read())T)r� r7 r8 r& r� g333333�?rQ s pineapple
pear
) r r9 rJ rK r: rS rT r� r% r* �encoder� r r r �test_communicate_timeout! s
��
� z(ProcessTestCase.test_communicate_timeoutc C sL t jtjddgt jd�}| jt j|jdd� |�� \}}| �t |�d� d S )Nr z�import sys,os,time;sys.stdout.write("a" * (64 * 1024));time.sleep(0.2);sys.stdout.write("a" * (64 * 1024));time.sleep(0.2);sys.stdout.write("a" * (64 * 1024));time.sleep(0.2);sys.stdout.write("a" * (64 * 1024));r� g�������?rQ i )
r r9 rJ rK r: rS rT r� r% �len)r r@ r8 �_r r r �%test_communicate_timeout_large_output4 s
� �
z5ProcessTestCase.test_communicate_timeout_large_outputc C s� dD ]�}dD ]�}dD ]�}i }|r*t j|d< |r8t j|d< |rFt j|d< |sLqt jtf|�}|�� |jd k rz| �|jj� |jd k r�| �|jj� |j d k r| �|j j� qq qd S )N)FTr7 r8 r& )
r r: r9 r r� r7 �
assertTrue�closedr8 r& )r Z
stdin_pipeZ stdout_pipeZ stderr_pipe�optionsr@ r r r � test_communicate_pipe_fd_leakE s(
z-ProcessTestCase.test_communicate_pipe_fd_leakc C s: t �tjddg�}|�� \}}| �|d � | �|d � d S )Nr rF )r r9 rJ rK r� r% r� r r r �test_communicate_returns[ s
� z(ProcessTestCase.test_communicate_returnsc C s� t �� \}}t �|� t �|� tjtjddtj gtj tj tj d�}| �
|j j� | �
|j j� | �
|j
j� dtj }|�|�\}}| �||� d S )Nr zpimport sys,os;sys.stdout.write(sys.stdin.read(47));sys.stderr.write("x" * %d);sys.stdout.write(sys.stdin.read())r6 � a)r �piper; r r9 rJ rK r �
PIPE_MAX_SIZEr: rr r8 r& r7 r� r% )r �x�yr@ Zstring_to_writer8 r& r r r �test_communicate_pipe_bufc s$
���
z)ProcessTestCase.test_communicate_pipe_bufc C s� t jtjddgt jt jt jd�}| �|jj� | �|jj� | �|j j� |j �
d� |� d�\}}| � |d� | �
|d� d S )Nr z0import sys,os;sys.stdout.write(sys.stdin.read())r6 r s splits bananasplitr� )r r9 rJ rK r: rr r8 r; r& r7 rs r� r% r* r� r r r � test_writes_before_communicatez s
�� z.ProcessTestCase.test_writes_before_communicatec
C s� t jddt d g}dD ]�}tj|fdtjdtj|di�}|�� |j�d� |j�� | � |j
� � d� |j�d � |j� � | �
|j
j � | � |j
� � d
� | � |j
�d �d � | � |j
�� d � W 5 Q R X qd S )
Nr �import sys,os;a@ buf = sys.stdout.buffer;buf.write(sys.stdin.readline().encode());buf.flush();buf.write(b"line2\n");buf.flush();buf.write(sys.stdin.read().encode());buf.flush();buf.write(b"line4\n");buf.flush();buf.write(b"line5\r\n");buf.flush();buf.write(b"line6\r");buf.flush();buf.write(b"\nline7");buf.flush();buf.write(b"\nline8");)r� r~ r7 r8 Tzline1
zline3
zline2
� z line4
line5
line6
line7
line8)rJ rK � SETBINARYr r9 r: r7 rs �flushr% r8 �readliner; rr r� )r r1 Z extra_kwargr@ r r r � test_universal_newlines_and_text� s: �� �
�� �z0ProcessTestCase.test_universal_newlines_and_textc C s^ t jtjddt d gt jt jdd�}| �|jj� | �|j j� |�
� \}}| � |d� d S )Nr r3 z�buf = sys.stdout.buffer;buf.write(b"line2\n");buf.flush();buf.write(b"line4\n");buf.flush();buf.write(b"line5\r\n");buf.flush();buf.write(b"line6\r");buf.flush();buf.write(b"\nline7");buf.flush();buf.write(b"\nline8");r� )r& r8 r� z#line2
line4
line5
line6
line7
line8) r r9 rJ rK r5 r: rr r8 r; r& r� r% r� r r r �#test_universal_newlines_communicate� s
��� �z3ProcessTestCase.test_universal_newlines_communicatec C sH t jtjddt t�d� gt jdd�}|�d�\}}| � |j
d� d S )Nr r3 a
s = sys.stdin.readline()
assert s == "line1\n", repr(s)
s = sys.stdin.read()
assert s == "line3\n", repr(s)
r� )r7 r� � line1
line3
r ) r r9 rJ rK r5 �textwrap�dedentr: r� r% r r� r r r �)test_universal_newlines_communicate_stdin� s
�� z9ProcessTestCase.test_universal_newlines_communicate_stdinc C s0 t jtt jt jdd�}|�� | �|jd� d S )NT)r7 r8 r� r )r r9 r r: r� r% r r? r r r �.test_universal_newlines_communicate_input_none� s �z>ProcessTestCase.test_universal_newlines_communicate_input_nonec C s� t jtjddt t�d� gt jt jt jdd�}| �|j j
� | �|j j
� |� d�\}}| �
|jd� | �
d|� | �|�d �� d S )
Nr r3 a�
s = sys.stdin.buffer.readline()
sys.stdout.buffer.write(s)
sys.stdout.buffer.write(b"line2\r")
sys.stderr.buffer.write(b"eline2\n")
s = sys.stdin.buffer.read()
sys.stdout.buffer.write(s)
sys.stdout.buffer.write(b"line4\n")
sys.stdout.buffer.write(b"line5\r\n")
sys.stderr.buffer.write(b"eline6\r")
sys.stderr.buffer.write(b"eline7\r\nz")
T)r7 r& r8 r� r: r z line1
line2
line3
line4
line5
zeline2
eline6
eline7
)r r9 rJ rK r5 r; r< r: rr r8 r; r& r� r% r r' r
r� r r r �7test_universal_newlines_communicate_stdin_stdout_stderr� s
�
� zGProcessTestCase.test_universal_newlines_communicate_stdin_stdout_stderrc C sT dD ]J}d| }t jd|g}tj|tjtj|d�}|jdd�\}}| �|d� qd S )N)zutf-16z utf-32-bez@import sys; sys.stdout.buffer.write('1\r\n2\r3\n4'.encode('%s'))r )r7 r8 �encodingr rw z1
2
3
4)rJ rK r r9 r: r� r% )r r@ r� r1 �popenr8 r& r r r �-test_universal_newlines_communicate_encodings s � �z=ProcessTestCase.test_universal_newlines_communicate_encodingsc C s\ dD ]R\}}d}t jd|g}tj|tjtjd|d�}|jdd�\}}| �|d�|�� qd S ) N))�ignorer )�replaceu ��)�surrogateescapeu ��)�backslashreplacez\x80\x80z2import sys; sys.stdout.buffer.write(b'[\x80\x80]')r �utf-8)r7 r8 r@ �errorsr rw z[{}])rJ rK r r9 r: r� r% �format)r rH r' r� r1 rA r8 r& r r r �test_communicate_errors s �z'ProcessTestCase.test_communicate_errorsc
C s0 t s
d}nd}g }t�� }z�t|�D ]n}z.tj� |t
j �}|� t�
|tjtjB �� W q$ tk
r� } z |jtjkrv� W Y � q�W 5 d }~X Y q$X q$| �d| � td�D ]}t�|�� � q�td�D ]>}tjtjddgtjtjtjd�}|�d �d
} | �| d � q�W 5 |D ]}t�|� �qt�|� X d S ) Ni i z4failed to reach the file descriptor limit (tried %d)�
� r z-import sys;sys.stdout.write(sys.stdin.read())r6 s limer ) r rp �mkdtempr r; r
�rmtree�ranger� r� r ZTESTFN�append�open�O_WRONLY�O_CREAT�OSError�errnoZEMFILE�skipTest�popr r9 rJ rK r: r� r% )
r Z max_handles�handlesZtmpdir�h�iZtmpfile�er@ �datar r r �test_no_leaking, s@ �
��z ProcessTestCase.test_no_leakingc C s� | � t�dddg�d� | � t�dddg�d� | � t�dddg�d � | � t�d
d d g�d
� | � t�dddg�d� | � t�dddg�d� | � t�dddg�d� | � t�ddg�d� d S )Nza b cr� r[ z "a b c" d ezab"c�\z ab\"c \ dz \z
ab\"c " \\" dza\\\bzde fgrY za\\\b "de fg" hza\"br\ z
a\\\"b c dza\\b cz "a\\b c" d eza\\b\ cz
"a\\b\ c" d e�abr zab "")r% r � list2cmdliner r r r �test_list2cmdlineS s0 ��������z!ProcessTestCase.test_list2cmdlinec C sb t jtjddgt jd�}| �|jj� | �|� � � t
� |j� � d� |�
� | �|� � d� d S )Nr zimport os; os.read(0, 1)rn � Ar )r r9 rJ rK r: rr r7 r; � assertIsNone�pollr rs r� r r% r? r r r � test_polle s
��zProcessTestCase.test_pollc C s. t �t�}| �|�� d� | �|�� d� d S rV �r r9 r r% r r? r r r � test_waitp s
zProcessTestCase.test_waitc C s` t �tjddg�}| �t j��}|jdd� W 5 Q R X | �dt|j �� | �
|jdd�d� d S )Nr z import time; time.sleep(0.3)�-C ��6?rQ z0.0001r� r ) r r9 rJ rK rS rT r rb �strrZ r% )r r@ r\ r r r �test_wait_timeoutv s �z!ProcessTestCase.test_wait_timeoutc C s& | � t�� t�td� W 5 Q R X d S )Nr )rS r� r r9 r r r r r �test_invalid_bufsize� s z$ProcessTestCase.test_invalid_bufsizec C s> t �td �}| �|�� d� t jtd d�}| �|�� d� d S )Nr )rB rf r? r r r �test_bufsize_is_none� s z$ProcessTestCase.test_bufsize_is_nonec
C s� t jtjddgt jt jt jd|d��j}|j�|� t� |j�
� � |j � � }t
�� �* z|j� � W n tk
rx Y nX W 5 Q R X d |_W 5 Q R X | �|jd� | �||� d S )Nr zDimport sys;sys.stdout.write(sys.stdin.readline());sys.stdout.flush()r� )r7 r8 r& rB r� r )r r9 rJ rK r: r r7 rs r r; r� r8 r7 r �SuppressCrashReportrT r% r )r �liner' r� r@ Z read_liner r r �_test_bufsize_equal_one� s& �
z'ProcessTestCase._test_bufsize_equal_onec C s d}| j ||dd� d S )Nzline
T�r� )ro �r rn r r r � test_bufsize_equal_one_text_mode� s z0ProcessTestCase.test_bufsize_equal_one_text_modec C s: dt j�� }| �td�� | j|ddd� W 5 Q R X d S )Ns linezline bufferingr� Frp )r �linesepr" ZassertWarnsRegex�RuntimeWarningro rq r r r �"test_bufsize_equal_one_binary_mode� s z2ProcessTestCase.test_bufsize_equal_one_binary_modec
C s< t d�D ].}| �t�� tjttjtjd� W 5 Q R X qd S )Nr� r� )rO rS r� r r9 r� r: )r rZ r r r �test_leaking_fds_on_error� s �z)ProcessTestCase.test_leaking_fds_on_errorc C s� zdd l }|j W n" ttfk
r4 | �d� Y nX t�dt�d��}tj d|g}t
j |t
j dd�}|� |�
� d }W 5 Q R X | �|d � | �|jd� d S )
Nr z need msvcrt.CrtSetReportModezL
import msvcrt
import subprocess
cmd = a
for report_type in [msvcrt.CRT_WARN,
msvcrt.CRT_ERROR,
msvcrt.CRT_ASSERT]:
msvcrt.CrtSetReportMode(report_type, msvcrt.CRTDBG_MODE_FILE)
msvcrt.CrtSetReportFile(report_type, msvcrt.CRTDBG_FILE_STDERR)
try:
subprocess.Popen(cmd,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE)
except OSError:
pass
r T)r& r� r� r )�msvcrtZCrtSetReportMode�AttributeError� ImportErrorrV r; r< r� rJ rK r r9 r: r� r% r )r rw r� r �procr& r r r �test_nonexisting_with_pipes� s
� z+ProcessTestCase.test_nonexisting_with_pipesc s� g � � fdd�}t j|d�}|�� z2| �t�� t j
t t j t j t j d� W 5 Q R X W 5 |�� d }� D ]:}zt�|� W qh tk
r� } z|}W 5 d }~X Y qhX qh|d k r�|�X d S )Nc s* t d�D ] } � �t�� � t�d� qd S )N� g����MbP?)rO �extendr r- �time�sleep)rZ ��fdsr r �open_fds� s z<ProcessTestCase.test_double_close_on_error.<locals>.open_fds)�targetr6 )
� threading�Thread�startr� r r; rT rS �EnvironmentErrorr r9 r� r: )r r� �t�exc�fdr[ r r� r �test_double_close_on_error� s( �z*ProcessTestCase.test_double_close_on_errorc s� t �tjddg�� | �� jd� g �� �fdd�}t�d|�}|�� t rPd}nd}� j
d d
� | j� j|d d � � �
� | j� j|d
d � |� � | �dd|fd|fg�� dS )z?Issue21291: Popen.wait() needs to be threadsafe for returncode.r zimport time; time.sleep(12)Nc sN �� d� �� f� � �� � �� �� d� jf� � �� �� d� jf� d S )N�thread-start-poll-result�thread-after-kill-and-wait�thread-after-second-wait)rP rd �killr r r �rz �resultsr r �kill_proc_timer_thread
s zDProcessTestCase.test_threadsafe_wait.<locals>.kill_proc_timer_thread皙�����?r� i����r| rQ z*unexpected result in wait from main thread�r( z&unexpected result in second main wait.)r� Nr� r� ) r r9 rJ rK r% r r� �Timerr� r r r� )r r� r� Zexpected_errorcoder r� r �test_threadsafe_wait s4
�
�
���z$ProcessTestCase.test_threadsafe_waitc C s6 d� d�}t�tjd|g�}| �|�d�t|�� d S )N�;)zimport subprocess, syszJretcode = subprocess.call([sys.executable, '-c', 'print("Hello World!")'])zassert retcode == 0r s Hello World!)r� r ra rJ rK r' r
�ascii)r r� rd r r r �test_issue87807 s
z ProcessTestCase.test_issue8780c C s� t �� \}}t �� \}}t �� \}}ztjdg|||d� W nP tk
r� t�|� t�|� t�|� t�|� t�|� t�|� Y nX | �tj �
|�� | �tj �
|�� | �tj �
|�� d S )N�*r6 ) rp �mkstempr r9 rT r r; �remover r� �exists)r ZifhandleZifnameZofhandleZofnameZefhandleZefnamer r r � test_handles_closed_on_exceptionB s" �
z0ProcessTestCase.test_handles_closed_on_exceptionc C sT t jtt jt jt jd�}| �|jj� | �|jj� | �|jj� |� dd � d S )Nr6 � x� )
r r9 r r: rr r8 r; r& r7 r� r? r r r �test_communicate_epipeV s �z&ProcessTestCase.test_communicate_epipec C s8 t jtt jd�}| �|jj� |�� |�dd � d S )Nrn r� r� ) r r9 r r: rr r7 r; r r� r? r r r �!test_communicate_epipe_only_stdina s �z1ProcessTestCase.test_communicate_epipe_only_stdin�SIGUSR1zRequires signal.SIGUSR1r� zRequires os.kill�getppidzRequires os.getppidc
C sn dd� }t � t j|�}| �t j t j|� tjddg}dD ]0}|tji}tj|f|��}|�� W 5 Q R X q8d S )Nc S s d S r r )�signum�framer r r �handlerq s z7ProcessTestCase.test_communicate_eintr.<locals>.handlerr z7import os, signal;os.kill(os.getppid(), signal.SIGUSR1)r� ) �signalr� rr rJ rK r r: r9 r� )r r� � old_handlerr1 �stream�kw�processr r r �test_communicate_eintri s �
z&ProcessTestCase.test_communicate_eintr� /proc/%d/fdzLinux specificc C s\ dt �� }t �|�}| �t�� tttjtjtjd� W 5 Q R X t �|�}| � ||� dS )z;Test for the fork() failure fd leak reported in issue16327.r� r6 N)
r �getpid�listdirrS r. r/ r r r: r% )r Z fd_directoryZfds_before_popenZfds_after_exceptionr r r �!test_failed_child_execute_fd_leak� s
�
z1ProcessTestCase.test_failed_child_execute_fd_leakz+behavior currently not supported on Windowsc C s< | � t��}t�ddddg� W 5 Q R X | �|jjd� d S )Nz/opt/nonexistent_binary�withZsomer1 )rS r� r rI r% rZ �filenamer[ r r r �%test_file_not_found_includes_filename� s z5ProcessTestCase.test_file_not_found_includes_filenamec C s< | � t��}tjddgdd� W 5 Q R X | �|jjd� d S )N�exit�0z/some/nonexistent/directoryr� )rS r� r r9 r% rZ r� r[ r r r � test_file_not_found_with_bad_cwd� s z0ProcessTestCase.test_file_not_found_with_bad_cwdN)xr+ r, r- rA rD rN rU rX r] re rg rk rv ry r} r� r� r� r� r� r� r� r� r� r� r� r� r� r� r� �unittest�skipIfr r� r� r� r� r� r� r� r� r� r� r� r� rJ � base_prefix�prefixr� � sysconfigZis_python_buildr� r� r� r� r� r� r� r� r� r� r� r r r r r r r �platformZget_config_varr r r r r r r! r# r& r* r+ r1 r2 r8 r9 r= r> r? rB rJ r] ra re rg rj rk rl ro rr ru rv r{ r� r� r� r� r� r� �
skipUnless�hasattrr� r r� r� �isdirr� r� r� r� r r r r r5 e s
�
�
�
��
' '
) 4 � � ��
r5 c @ s� e Zd Zdd� Zdd� Zdd� Zdd� Zd d
� Zd d � Zd
d� Z dd� Z
dd� Z dd� Z dd� Z
dd� Zdd� Zdd � Zd d � Zd d � Zd!d"� Zd#d$� Ze�ed%�d&d'� �Zd(S ))�RunFuncTestCasec K s t jd|g}tj|f|�S )z4Run Python code in a subprocess using subprocess.runr )rJ rK r r )r r� r2 �argvr r r �
run_python� s zRunFuncTestCase.run_pythonc C s< | � d�}| �|jd� | �tj�� |�� W 5 Q R X d S r� )r� r% r rS r rY �check_returncode�r �cpr r r �test_returncode� s
z RunFuncTestCase.test_returncodec C s: | � tj��}| jddd� W 5 Q R X | �|jjd� d S )NrF T��checkrG )rS r rY r� r% rZ r r[ r r r �
test_check� s zRunFuncTestCase.test_checkc C s t jtdd�}| �|jd� d S )NTr� r )r r r r% r r� r r r �test_check_zero� s z RunFuncTestCase.test_check_zeroc C s* | � tj�� | jddd� W 5 Q R X d S )NrO rh rQ )rS r rT r� r r r r � test_timeout� s z RunFuncTestCase.test_timeoutc C s" | j dtjd�}| �d|j� d S )Nr^ r� r_ �r� r r: rb r8 r� r r r �test_capture_stdout� s z#RunFuncTestCase.test_capture_stdoutc C s" | j dtjd�}| �d|j� d S )Nrh ri r_ )r� r r: rb r& r� r r r �test_capture_stderr� s �z#RunFuncTestCase.test_capture_stderrc C sL t �� }| �|j� |�d� |�d� | jd|tjd�}| � d|j
� d S )Nrl r rm r� ro ) rp rq rr r; rs rt r� r r: rb r8 )r ru r� r r r rv � s
�z+RunFuncTestCase.test_check_output_stdin_argc C s$ | j ddtjd�}| �d|j� d S )Nrm rl )rx r8 ro r� r� r r r ry � s �z+RunFuncTestCase.test_check_output_input_argc C s~ t �� }| �|j� |�d� |�d� | jtdd��}| jd|dd�}W 5 Q R X | � d|j
j d � | � d |j
j d � d S )
Nrl r r� r� r� r� r� r7 rx ) rp rq rr r; rs rt rS r� r� rb rZ r1 r� r r r r� � s
� �z6RunFuncTestCase.test_check_output_stdin_with_input_argc C sN | � tj��}| jddtjd�}W 5 Q R X | �|jjd� | �|jjd� d S )Nr� r� )rR r8 r_ ) rS r rT r� r: r% rZ rd r8 )r r\ r� r r r r� � s �z)RunFuncTestCase.test_check_output_timeoutc C s2 t j�� }d|d< | jd|d�}| �|jd� d S )Nr� r� zCimport sys, os;sys.exit(33 if os.getenv("FRUIT")=="banana" else 31)r� �! )r r� r� r� r% r )r r� r� r r r �test_run_kwargs� s
�z RunFuncTestCase.test_run_kwargsc C s� t rdnd}t�|�}|d kr.| �|� d�� t|�}tj|tjd�}| �|j d� | �
t �� tj|tjdd� W 5 Q R X d S )Nztree.comZlsz required for this testr� r T)r8 r� ) r r
r rV r r r r r% r rS r� )r �progr� �resr r r �test_run_with_pathlike_path� s
z+RunFuncTestCase.test_run_with_pathlike_pathc C s2 t �tj�}|ddg}t�|�}| �|jd� d S )Nr s import sys; sys.exit(57)�9 )r r� rJ rK r r r% r �r r� r1 r� r r r �&test_run_with_bytes_path_and_arguments
s
z6RunFuncTestCase.test_run_with_bytes_path_and_argumentsc C s0 t tj�}|ddg}t�|�}| �|jd� d S )Nr zimport sys; sys.exit(57)r� )r rJ rK r r r% r r� r r r �)test_run_with_pathlike_path_and_arguments s
z9RunFuncTestCase.test_run_with_pathlike_path_and_argumentsc C s. | j ddd�}| �d|j� | �d|j� d S )Nz>import sys;sys.stdout.write('BDFL'); sys.stderr.write('FLUFL')T)�capture_outputr_ s FLUFL)r� rb r8 r& r� r r r �test_capture_output s
�z#RunFuncTestCase.test_capture_outputc C sj t �� }| �|j� | jtdd��}| jdd|d�}W 5 Q R X | �d|jj d � | �d|jj d � d S ) NzAExpected ValueError when stdout and capture_output args supplied.r� r� T)r� r8 r8 r r� �
rp rq rr r; rS r� r� rb rZ r1 r� r r r �#test_stdout_with_capture_output_arg# s � �z3RunFuncTestCase.test_stdout_with_capture_output_argc C sj t �� }| �|j� | jtdd��}| jdd|d�}W 5 Q R X | �d|jj d � | �d|jj d � d S ) NzAExpected ValueError when stderr and capture_output args supplied.r� r� T)r� r& r& r r� r� r� r r r �#test_stderr_with_capture_output_arg/ s � �z3RunFuncTestCase.test_stderr_with_capture_output_argz)requires posix like 'sleep' shell commandc
C s~ t �� }ztjddddd� W n4 tjk
rR } zt �� }t�� }W 5 d}~X Y n X | �d� | j|| dd|� d �d
� dS ) zJOutput capturing after a timeout mustn't hang forever on open filehandles.zsleep 3TrP )r� rR r� NzTimeoutExpired not raised.g �?z/TimeoutExpired was delayed! Bad traceback:
```
z```r� ) r~ � monotonicr r rT � traceback�
format_excr� Z
assertLess)r Z before_secsr� Z
after_secsZstacksr r r �.test_run_with_shell_timeout_and_capture_output? s
�
�z>RunFuncTestCase.test_run_with_shell_timeout_and_capture_outputN)r+ r, r- r� r� r� r� r� r� r� rv ry r� r� r� r� r� r� r� r� r� r� r� r r� r r r r r� � s(
r� zPOSIX specific testsc sP e Zd Z� fdd�Zdd� Zdd� Zdd� Zd d
� ZG d d � d ej �Z
e � d
�dd� �Z
e � d
�dd� �Ze�ej�d� d�dd� �Zdd� Zdd� Zdd� Zd d � Zd d � Zd d!� Zd"d#� ZG d$d%� d%ej �Z e�ej�d&� d'�d(d)� �Z d*d+� Z e�e j d,kd-�d.d/� �Z!d0d1� Z"d2d3� Z#d4d5� Z$d6d7� Z%d8d9� Z&d:d;� Z'd<d=� Z(e�e j �)d>�d?�d@dA� �Z*dBdC� Z+dDdE� Z,dFdG� Z-dHdI� Z.dJdK� Z/dLdM� Z0dNdO� Z1dPdQ� Z2dRdS� Z3dTdU� Z4dVdW� Z5dXdY� Z6dZd[� Z7d\d]� Z8d^d_� Z9d`da� Z:dbdc� Z;ddde� Z<dfdg� Z=dhdi� Z>djdk� Z?dldm� Z@dndo� ZAdpdq� ZBdrds� ZCdtdu� ZDdvdw� ZEdxdy� ZFe�e j �)dz��one�Gd{�jHe�Gd|�jHkd}�d~d� �ZIeJ�Kd�d��d�d�� �ZLd�d�� ZMd�d�� ZNd�d�� ZOd�d�� ZPd�d�� ZQd�d�� ZRd�d�� ZSd�d�� ZTd�d�� ZUd�d�� ZVeJjWd�d�� �ZXeJjWd�d�� �ZYd�d�� ZZd�d�� Z[d�d�� Z\d�d�� Z]e�^e_d�k �o2e`e_d��d��d�d�� �Zad�d�� Zb� ZcS )��POSIXProcessTestCasec s t � �� d| _d S )Nz/_this/pa.th/does/not/exist)�superr �_nonexistent_dirr �� __class__r r r S s
zPOSIXProcessTestCase.setUpc
C sL zt �| j� W n& tk
r6 } z|}W 5 d }~X Y nX | �d| j � |S )Nz,chdir to nonexistent directory %s succeeded.)r �chdirr� rT r� )r r[ �desired_exceptionr r r �_get_chdir_exceptionW s �z)POSIXProcessTestCase._get_chdir_exceptionc
C s� | � � }z tjtjddg| jd�}W nR tk
rv } z4| �|j|j� | �|j |j � | �|j
|j
� W 5 d}~X Y nX | � d| � dS )z;Test error in the child raised in the parent for a bad cwd.r r r� N�Expected OSError: %s� r� r r9 rJ rK r� rT r% rU �strerrorr� r� �r r� r@ r[ r r r �test_exception_cwdd s �
"z'POSIXProcessTestCase.test_exception_cwdc
C s� | � � }z tjtjddg| jd�}W nR tk
rv } z4| �|j|j� | �|j |j � | �|j
|j
� W 5 d}~X Y nX | � d| � dS )zBTest error in the child raised in the parent for a bad executable.r r r� Nr� r� r� r r r � test_exception_bad_executables s �
"z2POSIXProcessTestCase.test_exception_bad_executablec
C s� | � � }zt�| jddg�}W nR tk
rp } z4| �|j|j� | �|j|j� | �|j|j� W 5 d}~X Y nX | � d| � dS )z?Test error in the child raised in the parent for a bad args[0].r r Nr� )
r� r r9 r� rT r% rU r� r� r� r� r r r �test_exception_bad_args_0� s "z.POSIXProcessTestCase.test_exception_bad_args_0c @ s e Zd Zdd� ZdS )z&POSIXProcessTestCase.PopenNoDestructorc C s d S r r r r r r �__del__� s z.POSIXProcessTestCase.PopenNoDestructor.__del__N)r+ r, r- r� r r r r �PopenNoDestructor� s r� z%subprocess._posixsubprocess.fork_execc
C sN dd� }||_ tjdtd��( | �t�� | �dg� W 5 Q R X W 5 Q R X dS )z>Test error passing done through errpipe_write in the good casec W s0 | d }d� tj��� }t�|d| d � dS )N�
z{:x}s OSError:� :r )rI rU ZEISDIRr" r rs )r1 �
errpipe_writeZerr_coder r r � proper_error� s zHPOSIXProcessTestCase.test_exception_errpipe_normal.<locals>.proper_error�subprocess.os.waitpid�� side_effect�non_existent_commandN)r� r �patch�ChildProcessErrorrS �IsADirectoryErrorr� )r � fork_execr� r r r � test_exception_errpipe_normal� s � z2POSIXProcessTestCase.test_exception_errpipe_normalc
sn d� � fdd�}||_ tjdtd��* | �tj��}| �dg� W 5 Q R X W 5 Q R X | �t � �t
|j �� dS )zZTest error passing done through errpipe_write where its not
in the expected formats � ޭc s | d }t �|� � dS )Nr� r )r rs )r1 r� �Z
error_datar r � bad_error� s zGPOSIXProcessTestCase.test_exception_errpipe_bad_data.<locals>.bad_errorr� r� r� N) r� r r� r� rS r �SubprocessErrorr� rb �reprri rZ )r r� r r[ r r r � test_exception_errpipe_bad_data� s � z4POSIXProcessTestCase.test_exception_errpipe_bad_data�/proc/self/statuszneed /proc/self/statusc C s� t jddgdd�}|�� D ]}|�d�r|} q<q| �d� t jddgdd�}|�� D ]}|�d�rV|} qnqV| j||dd � d S )
N�catr F)�restore_signalss SigIgnz&SigIgn not found in /proc/self/status.Tz=restore_signals=True should've unblocked SIGPIPE and friends.r� )r ra �
splitlinesr
rV �assertNotEqual)r Zdefault_proc_statusrn Zdefault_sig_ign_maskZrestored_proc_statusZrestored_sig_ign_maskr r r �test_restore_signals� s( �
�
�z)POSIXProcessTestCase.test_restore_signalsc
C sn zt jtjddgdd�}W n0 tk
rJ } z|jtjkr:� W 5 d }~X Y n X t�d�}t |�}| �
||� d S )Nr z import os; print(os.getsid(0))T)�start_new_sessionr ) r ra rJ rK rT rU �EPERMr �getsid�intr )r rd r[ Z
parent_sidZ child_sidr r r �test_start_new_session� s
�
z+POSIXProcessTestCase.test_start_new_sessionc C sD t �� � t�tjddg�}|�� W 5 Q R X | �|j t j
� d S )Nr zimport os; os.abort()) r rm r r9 rJ rK r r% r r� �SIGABRTr? r r r �test_run_abort� s
�z#POSIXProcessTestCase.test_run_abortc C sN t �ttj� d�}t|�}| �d|�� � | �d|� | �ttj�|� d S )N�fake cmdr� ZSIG)r rY r r� r ri rb �lower�r r� Z error_stringr r r �"test_CalledProcessError_str_signal� s
z7POSIXProcessTestCase.test_CalledProcessError_str_signalc C s$ t �dd�}t|�}| �d|� d S )Ni�Ki�r zunknown signal 9876543.�r rY ri rb r r r r �*test_CalledProcessError_str_unknown_signal� s z?POSIXProcessTestCase.test_CalledProcessError_str_unknown_signalc C s$ t �dd�}t|�}| �d|� d S )N� r znon-zero exit status 2.r r r r r �$test_CalledProcessError_str_non_zero s z9POSIXProcessTestCase.test_CalledProcessError_str_non_zeroc C sD t jtjddgt jdd� d�}|� | �|j�� d� W 5 Q R X d S )Nr r c S s t �dd�S )Nr� �apple)r �putenvr r r r �<lambda> r� z3POSIXProcessTestCase.test_preexec.<locals>.<lambda>)r8 �
preexec_fn� appler� r? r r r � test_preexec s
��z!POSIXProcessTestCase.test_preexecc
C s� dd� }zt jtjddg|d�}W nd t jk
rT } z| �t jd� W 5 d }~X Y n> tk
r� } z| �d|j d � W 5 d }~X Y n X | �
d � d S )
Nc S s t d��d S )Nz'What if two swallows carried a coconut?�r� r r r r �raise_it s z=POSIXProcessTestCase.test_preexec_exception.<locals>.raise_itr r �r z)Expected a ValueError from the preexec_fnZcoconutr zEException raised by preexec_fn did not make it to the parent process.) r r9 rJ rK r r' �_posixsubprocessr� rb r1 r� )r r! r@ r[ r r r �test_preexec_exception s �
�$z+POSIXProcessTestCase.test_preexec_exceptionc @ s e Zd ZdZdd� Zdd� ZdS )z+POSIXProcessTestCase._TestExecuteChildPopenz3Used to test behavior at the end of _execute_child.c O s || _ tjj| f|�|� d S r )� _testcaser r9 r� )r Ztestcaser1 r2 r r r r� $ s z4POSIXProcessTestCase._TestExecuteChildPopen.__init__c O s� zt j
j | f|�|� W 5 dd� t d�D �}z8|D ].}| jj|| j�� | j�� | j�� fdd� q2W 5 |D ]}t�|� qjX X d S )Nc S s g | ]}t �d t j��qS )� /dev/zero)r rQ �O_RDONLY�r r% r r r r / s �zNPOSIXProcessTestCase._TestExecuteChildPopen._execute_child.<locals>.<listcomp>� z!At least one fd was closed early.r� ) rO r r; r% r| r7 r� r8 r& r r9 r3 )r r1 r2 Z devzero_fdsr� r r r r3 ( s � �� z:POSIXProcessTestCase._TestExecuteChildPopen._execute_childN)r+ r, r- r4 r� r3 r r r r �_TestExecuteChildPopen" s r* r& z/dev/zero required.c C s@ dd� }| � tj��" | j| ttjtjtj|d� W 5 Q R X dS )z6Issue16140: Don't double close pipes on preexec error.c S s t �d��d S )Nz-force the _execute_child() errpipe_data path.)r r r r r r r! ? s �zWPOSIXProcessTestCase.test_preexec_errpipe_does_not_double_close_pipes.<locals>.raise_it)r7 r8 r& r N)rS r r r* r r: )r r! r r r �0test_preexec_errpipe_does_not_double_close_pipes; s �zEPOSIXProcessTestCase.test_preexec_errpipe_does_not_double_close_pipesc C s dd� }t �� }t j}t j}z�t �� | �t �� � tjtjddgdd� d� | �t �� d� t �� | � t �� � tjtjddgd d� d� | � t �� d
� |t _| j
t tj tjddgd d� d� t `| j
t
tj tjddgd d� d� W 5 |t _|t _|�st �� X d S )
Nc S s t d��d S )Nzthis shouldn't escape)� RuntimeErrorr r r r �raise_runtime_errorL s zPPOSIXProcessTestCase.test_preexec_gc_module_failure.<locals>.raise_runtime_errorr r c S s d S r r r r r r r U r� zEPOSIXProcessTestCase.test_preexec_gc_module_failure.<locals>.<lambda>r" z#Popen enabled gc when it shouldn't.c S s d S r r r r r r r \ r� zPopen left gc disabled.c S s d S r r r r r r r b r� c S s d S r r r r r r r g r� )�gc� isenabled�disabler r rI rJ rK �enabler' rS r, r9 rx )r r- ZenabledZorig_gc_disableZorig_gc_isenabledr r r � test_preexec_gc_module_failureI sB �
��
�
�
z3POSIXProcessTestCase.test_preexec_gc_module_failure�darwinz!setrlimit() seems to fail on OS Xc
C s� zddl m}m}m} W n, tk
rD } z| �|� W 5 d }~X Y nX ||�}|\}}||d|f� | �|||� z tjt j
ddgdd� d� W n t k
r� Y n X | �d� d S ) Nr )� getrlimit� setrlimit� RLIMIT_NPROCr r c S s d S r r r r r r r } r� z@POSIXProcessTestCase.test_preexec_fork_failure.<locals>.<lambda>r" z.RLIMIT_NPROC had no effect; probably superuser) �resourcer4 r5 r6 ry rV rr r rI rJ rK �BlockingIOError)r r4 r5 r6 r� Zlimitsr% Zhardr r r �test_preexec_fork_failuren s �
z.POSIXProcessTestCase.test_preexec_fork_failurec C s� t �� \}}t|ddd��&}|�dtj � |�dtj � W 5 Q R X t� |d� t
� |�}|� � t�
|� | �|jd� d S �N�wrE )rH z#!%s
z(exec '%s' -c 'import sys; sys.exit(47)'
i� rG )rp r� rQ rs r �
unix_shellrJ rK r �chmodr r9 r r� r% r )r r� �fname�fobjr@ r r r �test_args_string� s �
z%POSIXProcessTestCase.test_args_stringc C s< | j ttjtjddgdd� | j ttjtjddgdd� d S )Nr rF rG �� startupinfo��
creationflags�rS r� r rI rJ rK r r r r r� � s
��
��z&POSIXProcessTestCase.test_invalid_argsc C sT t j�� }d|d< tjdgdtj|d�}|� | �|j�� � d�d� W 5 Q R X d S �Nr r� z echo $FRUITr� �r� r8 r� s
r �
r r� r� r r9 r: r% r8 r� r$ �r r� r@ r r r �test_shell_sequence� s
�z(POSIXProcessTestCase.test_shell_sequencec C sR t j�� }d|d< tjddtj|d�}|� | �|j�� � d�d� W 5 Q R X d S rF rH rI r r r �test_shell_string� s
�z&POSIXProcessTestCase.test_shell_stringc C sv t �� \}}t|ddd��&}|�dtj � |�dtj � W 5 Q R X t� |d� t
� |�}t� |� | �
|d� d S r: )rp r� rQ rs r r< rJ rK r r= r rI r� r% )r r� r> r? rM r r r �test_call_string� s �
z%POSIXProcessTestCase.test_call_stringc
C s� g }dD ]2}dD ](}t j�||�}t j�|�r|�|� qq|sJ| �d� d}t j�|�rpt j�|�sp|�|� |D ]D}tjd|dtj d�}|�" | �
|j � � �
� t|d�� W 5 Q R X qtd S ) N)z/binz /usr/bin/z/usr/local/bin)ZbashZkshz"bash or ksh required for this testz/bin/shzecho $0T)rK r� r8 r� )r r� r� �isfilerP rV �islinkr r9 r: r% r8 r� r$ �bytes)r Zshellsr� �name�shr@ r r r �test_specific_shell� s"
�z(POSIXProcessTestCase.test_specific_shellc
G sf t � t jt j�}z&tjtjddgdtjtjtjd�}W 5 t � t j|� X |j� d� t
||�|� |S )Nr z�if 1:
import sys, time
sys.stdout.write('x\n')
sys.stdout.flush()
time.sleep(30)
T�� close_fdsr7 r8 r& r� ) r� �SIGINT�default_int_handlerr r9 rJ rK r: r8 r� �getattr)r �methodr1 r� r@ r r r �
_kill_process� s �
z"POSIXProcessTestCase._kill_process)ZnetbsdZopenbsdz"Due to known OS bug (issue #16762)c G sR t jtjddgdt jt jt jd�}|j�d� t�d� t ||�|� |�
� d S )Nr z�if 1:
import sys, time
sys.stdout.write('x\n')
sys.stdout.flush()
TrS r� ) r r9 rJ rK r: r8 r� r~ r rW r� )r rX r1 r@ r r r �_kill_dead_process� s �
z'POSIXProcessTestCase._kill_dead_processc C s: | � dtj�}|�� \}}| �d|� | �|�� d� d S )N� send_signals KeyboardInterruptr )rY r� rU r� rb r r �r r@ r% r& r r r �test_send_signal s z%POSIXProcessTestCase.test_send_signalc C s: | � d�}|�� \}}| �|d� | �|�� tj � d S )Nr� r� )rY r� r* r% r r� �SIGKILLr\ r r r � test_kill s
z POSIXProcessTestCase.test_killc C s: | � d�}|�� \}}| �|d� | �|�� tj � d S )N� terminater� )rY r� r* r% r r� �SIGTERMr\ r r r �test_terminate s
z#POSIXProcessTestCase.test_terminatec C s | � dtj� d S �Nr[ )rZ r� rU r r r r �test_send_signal_dead s z*POSIXProcessTestCase.test_send_signal_deadc C s | � d� d S �Nr� �rZ r r r r �test_kill_dead s z#POSIXProcessTestCase.test_kill_deadc C s | � d� d S �Nr` rf r r r r �test_terminate_dead s z(POSIXProcessTestCase.test_terminate_deadc C s6 g }|D ](}t �|�}t �|�}|�|||f� q|S r )r �get_inheritable�duprP )r Zsave_fdsr� r� � inheritable�savedr r r � _save_fds s
z POSIXProcessTestCase._save_fdsc C s. |D ]$\}}}t j|||d� t �|� qd S )N)rl )r �dup2r; )r r� r� rm rl r r r � _restore_fds( s z!POSIXProcessTestCase._restore_fdsc C s� d}| � |�}|D ]\}}}|dkr|} q.qzX|D ]}t�|� q4tjtjddg|tjtjd�� � \}}t
� |�}| � ||fd� W 5 | �|� X d S )Nr r r� r6 )r r� )
rn rp r r; r r9 rJ rK r: r� r r# r% ) r r� r7 � saved_fdsr� rm rl r� r� r r r �check_close_std_fds- s&
��
z(POSIXProcessTestCase.check_close_std_fdsc C s | � dg� d S rV �rr r r r r �test_close_fd_0F s z$POSIXProcessTestCase.test_close_fd_0c C s | � dg� d S �Nr� rs r r r r �test_close_fd_1I s z$POSIXProcessTestCase.test_close_fd_1c C s | � dg� d S )Nr rs r r r r �test_close_fd_2L s z$POSIXProcessTestCase.test_close_fd_2c C s | � ddg� d S )Nr r� rs r r r r �test_close_fds_0_1O s z'POSIXProcessTestCase.test_close_fds_0_1c C s | � ddg� d S )Nr r rs r r r r �test_close_fds_0_2R s z'POSIXProcessTestCase.test_close_fds_0_2c C s | � ddg� d S )Nr� r rs r r r r �test_close_fds_1_2U s z'POSIXProcessTestCase.test_close_fds_1_2c C s | � dddg� d S �Nr r� r rs r r r r �test_close_fds_0_1_2X s z)POSIXProcessTestCase.test_close_fds_0_1_2c
C sv t �d�}t �d�}z.t �d� t �d� t�tjddg��� W 5 t �|d� t �|d� t �|� t �|� X dS )z=Issue #15798: Popen should work when stdio fds are available.r r� r z*print('AssertionError:0:CLOEXEC failure.')N) r rk ro r; r r9 rJ rK r )r Z new_stdinZ
new_stdoutr r r �test_small_errpipe_write_fd] s
�
z0POSIXProcessTestCase.test_small_errpipe_write_fdc
C s@ dd� t d�D �}�zdd� |D �}|D ]\}}t�|� q(t�|d d� t�|d dd� | �t d��}zPt|�D ]\}}t� ||� qvt
j t j
dd g|d |d
|d d �}|�� W 5 | �|� X |D ]}t�|dd� q�t�|d
d �}t�t�|d d ��} | �|d
� | �| d� W 5 |D ]}t�|� �q(X d S )Nc S s g | ] }t �� �qS r �rp r� �r rZ r r r r t s z?POSIXProcessTestCase.test_remapping_std_fds.<locals>.<listcomp>r� c S s g | ] \}}|�qS r r �r r� r> r r r r v s r� � STDINr r �Zimport sys; got = sys.stdin.read();sys.stdout.write("got %s"%got); sys.stderr.write("err")r r6 r� � got STDIN� err)rO r r; �unlinkrs r� rn rp � enumeratero r r9 rJ rK r r� r r# r% )
r �temps�temp_fdsr� r> rq �temp_fdr@ r� r� r r r �test_remapping_std_fdsr s8
�� z+POSIXProcessTestCase.test_remapping_std_fdsc
C s" dd� t d�D �}dd� |D �}z�|D ]\}}t�|� q&| �t d��}z�t|�D ]\}} t�| |� qRt�|d� t� |dd� t
j t j
ddg|||d �}
|
�� |D ]}t� |dd� q�t�|d
�} t�t�|d
��} W 5 | �|� X | �| d � | �| d � W 5 |D ]}t�|� �q
X d S )
Nc S s g | ] }t �� �qS r r~ r r r r r � s z7POSIXProcessTestCase.check_swap_fds.<locals>.<listcomp>r� c S s g | ] \}}|�qS r r r� r r r r � s r� r r r� r6 r� r� r� )rO r r; r� rn rp r� ro rs r� r r9 rJ rK r r� r r# r% )
r Zstdin_noZ stdout_noZ stderr_nor� r� r� r> rq r� r@ r� r� r r r �check_swap_fds� s8
�� z#POSIXProcessTestCase.check_swap_fdsc C sX | � ddd� | � ddd� | � ddd� | � ddd� | � ddd� | � ddd� d S r{ )r� r r r r �
test_swap_fds� s z"POSIXProcessTestCase.test_swap_fdsc C sl | � td��}�zJ|D ](}t�� �}t�|�� |� W 5 Q R X qttd��t|� � � }t�
|� dddg}i }t ||�D ]\}} |||| < qzt �
d�}
ttd��t|� � � } tjtjd|
t| �gf|�} | �| d� t ||�D ]r\}} t�|dtj� t�|d�}
ttt|
�d ���}t �
d
|� d |� d |� d
|� d| � d� �}| �| g||� q�W 5 | �|� X d S )Nr� r7 r8 r& z�
import os, sys
skipped_fd = int(sys.argv[1])
for fd in range(3):
if fd != skipped_fd:
os.write(fd, str(fd).encode('ascii'))
r r r� r� z"
When testing z to z4 redirection,
parent descriptor z5 got redirected
to descriptor(s) z instead of descriptor z.
)rn rO rp rp rq r ro r� �setrW r; �zipr; r< r rI rJ rK ri r% r� �SEEK_SETr� r �mapr r� )r �from_fds�to_fdsrq Zfrom_fd�fZ fd_to_closeZ arg_namesr2 Zto_fdr� Z
skipped_fdrM Z
read_bytesZread_fdsr( r r r �#_check_swap_std_fds_with_one_closed� sD
� �����
z8POSIXProcessTestCase._check_swap_std_fds_with_one_closedc C s< t �td�d�D ]&}t �td�d�D ]}| �||� q$qd S )Nr� r )� itertools� combinationsrO � permutationsr� )r r� r� r r r �!test_swap_std_fds_with_one_closed s z6POSIXProcessTestCase.test_swap_std_fds_with_one_closedc
C s� dd� }zt jt|d� W n| tk
rX } z | �t j� | �t|�d� W 5 d }~X Y nJ t jk
r� } z | � t j� | �t|�d� W 5 d }~X Y n X | �
d� d S )Nc S s t d��d S )N�
surrogate:�r r r r r �prepare s zCPOSIXProcessTestCase.test_surrogates_error_message.<locals>.preparer" r� z!Exception occurred in preexec_fn.z1Expected ValueError or subprocess.SubprocessError) r rI r r� rc r# r% ri r ZassertIsNotNoner� )r r� r� r r r � test_surrogates_error_message s �
"z2POSIXProcessTestCase.test_surrogates_error_messagec C s� dD ]�\}}|� dd�}dt|� }tj�� }|||< d|d< |}tjtjd|g|d�}|� d �}| �
|� d�t |�� |� dd�}d
t|� }tj�� }|||< tjtjd|g|d�}|� d �}| �
|� d�t |�� qd S ) N))�testu abc�)u test�Z42r� rE z&import os; print(ascii(os.getenv(%s)))�C�LC_ALLr r� s
z'import os; print(ascii(os.getenvb(%s))))
r" r r r� r� r ra rJ rK r� r% r� r� )r �key�valueZ
encoded_value�scriptr� Z
decoded_valuer8 r r r �test_undecodable_env s.
�
�
z)POSIXProcessTestCase.test_undecodable_envc C s� t �td �}ttdd � �}t j�td �\}}t �|�}t�|g| �}| �|d� d|d� |��
d�f }tj|dd�}| �|d� t j � � }||d< tj|g| |d �}| �|d� t j
� � }t �|�|d
< tj|g| |d �}| �|d� d S ) Nr r� s '%s' %s� rG T�r� �PATHr� s PATH)r r� r r r� r� r rI r% r� r" r� r� �environb) r Z abs_programr1 r� Zprogram�exitcoder r� Zenvbr r r �test_bytes_program8 s"
z'POSIXProcessTestCase.test_bytes_programc C s� t jddd�}t jddd�}tjtj|gtjtjtjdd�}| �|jd� tjtj|gtjdd�}|�� \}}t t
t |� d ���}t |j
�� |j�� |j�� g�}| �||@ d
|||@ f � d S ) Nzinput_reader.py�subprocessdata�Zsubdir� fd_status.pyF)r7 r8 r& rT r� �r8 rT � ,z5Expected no fds from %r to be open in child, found %r)r �findfiler r9 rJ rK r: rr r� r� r� r r� r7 r� r8 r& r ) r Zsleeper� fd_status�p1�p2rd �errorZ
result_fdsZ unwanted_fdsr r r �test_pipe_cloexecS s. � � �
��z&POSIXProcessTestCase.test_pipe_cloexecc
s t jddd�}t jddd�}d}|d d }tjtj|gtjtjdd �� tjtj||g� jtjdd ��| �� j � | ��j � � fd
d �}�fd d
�}| �|� | �|� � j
� |� � j
� � t
�
�jgg g d�\}}} | �|d� | ��j�� |� � j� � �j� � d S )Nzqcat.pyr� r� zqgrep.pys zxcvbn� �
F)r7 r8 rT c s& z � � � W n tk
r Y nX d S r �r` �ProcessLookupErrorr )r� r r �kill_p1{ s zBPOSIXProcessTestCase.test_pipe_cloexec_real_tools.<locals>.kill_p1c s& z � � � W n tk
r Y nX d S r r� r )r� r r �kill_p2� s zBPOSIXProcessTestCase.test_pipe_cloexec_real_tools.<locals>.kill_p2rK zThe child hung)r r� r r9 rJ rK r: r8 rr r r7 rs r; �selectr' r% r� )
r ZqcatZqgrepZsubdatar\ r� r� Z readfilesZignored1Zignored2r )r� r� r � test_pipe_cloexec_real_toolsj s6 � �
z1POSIXProcessTestCase.test_pipe_cloexec_real_toolsc
s� t jddd�}t�� }| �tj|d � | �tj|d � t|�� td�D ],}t�tj tj
�}| �tj|� � � |� qJ� D ]}t� |d� q|t
jtj|gt
jdd �}|�� \}}ttt|�d
���}| �|� @ � d � t
jtj|gt
jdd �}|�� \}}ttt|�d
���}| �|� @ d � | �d|d
� t� fdd�td�D ��} t
jtj|gt
jd| d�}|�� \}}ttt|�d
���}| �|| � @ d� | �d|d
� d S )Nr� r� r� r r� � TFr� r� zSome fds were closedzSome fds were left openzSubprocess failedc 3 s | ]}� � � V qd S r )rW r( �r� r r � <genexpr>� s z6POSIXProcessTestCase.test_close_fds.<locals>.<genexpr>r) �r8 rT �pass_fdsz'Some fds not in pass_fds were left open)r r� r r- rr r; r� rO rQ �devnullr' �add�set_inheritabler r9 rJ rK r: r� r� r r� r% r rb )
r r� r� r% r� r@ rd �ignored�
remaining_fds� fds_to_keepr r� r �test_close_fds� sT � � �
� � �z#POSIXProcessTestCase.test_close_fdsZfreebsdz/devz/dev/fdz/Requires fdescfs mounted on /dev/fd on FreeBSD.c C s� t jddd�}tjtjdt�d| �gtjd�}|� � \}}|�
� }| j t |�dd| d � t
tt|d
�� �d ���}t
tt|d �� �d ���}| j||@ d
d � dS )z;Confirm that issue21618 is fixed (may fail under valgrind).r� r� r� r aK
import os, resource, subprocess, sys, textwrap
open_fds = set()
# Add a bunch more fds to pass down.
for _ in range(40):
fd = os.open(os.devnull, os.O_RDONLY)
open_fds.add(fd)
# Leave a two pairs of low ones available for use by the
# internal child error pipe and the stdout pipe.
# We also leave 10 more open as some Python buildbots run into
# "too many open files" errors during the test if we do not.
for fd in sorted(open_fds)[:14]:
os.close(fd)
open_fds.remove(fd)
for fd in open_fds:
#self.addCleanup(os.close, fd)
os.set_inheritable(fd, True)
max_fd_open = max(open_fds)
# Communicate the open_fds to the parent unittest.TestCase process.
print(','.join(map(str, sorted(open_fds))))
sys.stdout.flush()
rlim_cur, rlim_max = resource.getrlimit(resource.RLIMIT_NOFILE)
try:
# 29 is lower than the highest fds we are leaving open.
resource.setrlimit(resource.RLIMIT_NOFILE, (29, rlim_max))
# Launch a new Python interpreter with our low fd rlim_cur that
# inherits open fds above that limit. It then uses subprocess
# with close_fds=True to get a report of open fds in the child.
# An explicit list of fds to check is passed to fd_status.py as
# letting fd_status rely on its default logic would miss the
# fds above rlim_cur as it normally only checks up to that limit.
subprocess.Popen(
[sys.executable, '-c',
textwrap.dedent("""
import subprocess, sys
subprocess.Popen([sys.executable, %r] +
[str(x) for x in range({max_fd})],
close_fds=True).wait()
""".format(max_fd=max_fd_open+1))],
close_fds=False).wait()
finally:
resource.setrlimit(resource.RLIMIT_NOFILE, (rlim_cur, rlim_max))
r� r z(expected exactly two lines of output:
%rr� r r� r� zSome fds were left open.N)r r� r r9 rJ rK r; r< r: r� r r% r$ r� r� r r$ r� r )r r� r@ rd Z
unused_stderrZ output_linesZ
opened_fdsr� r r r �%test_close_fds_when_max_fd_is_lowered� s$
/��0�2 �
�z:POSIXProcessTestCase.test_close_fds_when_max_fd_is_loweredrK rf c
C s* t jddd�}t� }td�D ]Z}t�� }| �tj|d � | �tj|d � t�|d d� t�|d d� |� |� q |D ]�}t
j t j
|gt
jd|fd�}|�� \}}ttt|�d ���} ||h }
| �|| d
� | �| |
@ d � | �t�� } | �t
jtd |fd
�� W 5 Q R X | �dt| j�� q|d S )Nr� r� r� rf r r� Tr� r� zfd to be passed not passedzfd to be closed passedF)rT r� zoverriding close_fds)r r� r� rO r r- rr r; r� �updater r9 rJ rK r: r� r� r r� rb r Z assertWarnsrt rI r ri �warning) r r� r� r/ r� r� r@ rd r� r� Z to_be_closed�contextr r r �
test_pass_fds
s: �
� �z"POSIXProcessTestCase.test_pass_fdsc
C s� t jddd�}t�� \}}| �tj|� | �tj|� t�|d� t�|d� ||f}tj|g}|t t
t |��7 }t j
|t jd|d�}|�� \}}tt
t|�d���} | �| t|�d| � | �t�|�d� | �t�|�d� d S ) Nr� r� r� TFr� r� z output=%a)r r� r r- rr r; r� rJ rK r r� ri r r9 r: r� r� r r� r% rj )
r r� rl Znon_inheritabler� r1 r@ rd r� r� r r r �test_pass_fds_inheritable8
s&
� z.POSIXProcessTestCase.test_pass_fds_inheritablec C s� t jddd�}g }td�D ],}t�tjtj�}| �tj|� |� |� qt�
� \}}| �tj|� | �tj|� |� d|� t j
tj|g|d |d |d d|d�� t�|d �}W 5 Q R X d
d � |�d �D �}| �|dddht|�B d
|��� dS )z7Regression test for https://bugs.python.org/issue32270.r� r� r� r r� r T)r7 r8 r& rT r� r� c S s h | ] }t |��qS r )r )r �numr r r � <setcomp>k
s z@POSIXProcessTestCase.test_pass_fds_redirected.<locals>.<setcomp>r� zoutput=N)r r� rO r rQ r� �O_RDWRrr r; rP r- �insertr r9 rJ rK r� r� r% � frozenset) r r� r� r% r� Zstdout_rZstdout_wrd r� r r r �test_pass_fds_redirectedV
s( �z-POSIXProcessTestCase.test_pass_fds_redirectedc C s6 t �tjd�� }tjt||d�}|�� W 5 Q R X d S )N�r+)r8 r7 �r= rQ r r� r r9 r r �r Zinoutr@ r r r �%test_stdout_stdin_are_single_inout_fdp
s �z:POSIXProcessTestCase.test_stdout_stdin_are_single_inout_fdc C s6 t �tjd�� }tjt||d�}|�� W 5 Q R X d S )Nr� r� r� r� r r r �&test_stdout_stderr_are_single_inout_fdv
s �z;POSIXProcessTestCase.test_stdout_stderr_are_single_inout_fdc C s6 t �tjd�� }tjt||d�}|�� W 5 Q R X d S )Nr� )r& r7 r� r� r r r �%test_stderr_stdin_are_single_inout_fd|
s �z:POSIXProcessTestCase.test_stderr_stdin_are_single_inout_fdc C sR t jddd�}tjtj|gtjtjd�}|�� \}}| �d|j d|�
d� � d S )Nzsigchild_ignore.pyr� r� r� r z6sigchild_ignore.py exited non-zero with this error:
%srG ) r r� r r9 rJ rK r: r� r% r r� )r Zsigchild_ignorer@ r8 r& r r r � test_wait_when_sigchild_ignored�
s � � �z4POSIXProcessTestCase.test_wait_when_sigchild_ignoredc C sz t �d�}tjtjddgtjdd�}|j}| �|j � z4| � |� d�d� | �
||�|gg g d�d � W 5 |�
� X d S ) Nr� r z$import sys;sys.stdout.write("apple")r )r8 rB r� s applg )r �
import_moduler r9 rJ rK r: r8 rr r; r r% r� rb r� )r r� r@ r� r r r �test_select_unbuffered�
s
�� "z+POSIXProcessTestCase.test_select_unbufferedc C s� t jtjddgt jt jd�}| �|jj� | �|jj� t |�}|j
}t � dt
f��
d }W 5 Q R X trv| �t j� n| �|dd� t jD �� d S )Nr z import sys, time;time.sleep(0.2)r� r c S s g | ] }t |��qS r ��id�r �or r r r �
s zEPOSIXProcessTestCase.test_zombie_fast_process_del.<locals>.<listcomp>)r r9 rJ rK r: rr r8 r; r& r� �pidr �check_warnings�ResourceWarningr rc r rb )r r@ �identr� r r r � test_zombie_fast_process_del�
s
��z1POSIXProcessTestCase.test_zombie_fast_process_delc
C s t jtjddgt jt jd�}| �|jj� | �|jj� t |�}|j
}t � dt
f��
d }W 5 Q R X t�|tj� tr�| �t j� n| �|dd� t jD �� t�d� | �t��& t jtt jt jd��}W 5 Q R X W 5 Q R X | �ttj|d� t�r| �t j� n| � |d d� t jD �� d S )
Nr zimport time;time.sleep(3)r� r c S s g | ] }t |��qS r r� r� r r r r �
s zJPOSIXProcessTestCase.test_leak_fast_process_del_killed.<locals>.<listcomp>r� r c S s g | ] }t |��qS r r� r� r r r r �
s ) r r9 rJ rK r: rr r8 r; r& r� r� r r� r� r r� r� r^ r rc r rb r~ r rS rT r� �waitpidr| )r r@ r� r� rz r r r �!test_leak_fast_process_del_killed�
s8
��
�z6POSIXProcessTestCase.test_leak_fast_process_del_killedc sx t jddd�}t�d�� | �tj� � tjtj |gtj
d� fdd�d�}|� � \}}t t
t|�d ���}| �� |� d S )
Nr� r� r� r� Tc s t �d� �S ru )r ro r �r� r r r �
r� zCPOSIXProcessTestCase.test_close_fds_after_preexec.<locals>.<lambda>)r8 rT r r� )r r� r rk rr r; r r9 rJ rK r: r� r� r� r r� r| )r r� r@ rd r� r� r r� r � test_close_fds_after_preexec�
s
� z1POSIXProcessTestCase.test_close_fds_after_preexecc C s� dd l }t�� }z�dd� }t�� ddgd dgfdgdd dgfdgdgddgfdgdgd dffD ]L\}}}}| �t��0 |�||dd ||d
d
d
d
d d d
ddd|� W 5 Q R X q`W 5 |s�t�� X d S )Nr c S s d S r r r r r r r �
r� z5POSIXProcessTestCase.test_fork_exec.<locals>.<lambda>�{ s exe� envs argTr ���r� r r� r� )r# r. r/ r0 r1 rS r� r� )r r# �
gc_enabled�funcr1 Zexe_listr� �env_listr r r �test_fork_exec�
s@ � �z#POSIXProcessTestCase.test_fork_execc C s� dd l }G dd� d�}t�� }z�t�� ddddd|d �|d
�ffD ]f}| jtd �|�d ��6}|�d
gd
gd|d dgddddd d
ddddd � W 5 Q R X | � dt
|j �� qDW 5 |s�t�� X d S )Nr c @ s e Zd ZdZdd� Zdd� ZdS )zJPOSIXProcessTestCase.test_fork_exec_sorted_fd_sanity_check.<locals>.BadIntTc S s
|| _ d S r )r� )r r� r r r r� s zSPOSIXProcessTestCase.test_fork_exec_sorted_fd_sanity_check.<locals>.BadInt.__init__c S s | j rd| _ | jS t�d S )NF)�firstr� r� r r r r �__int__ s zRPOSIXProcessTestCase.test_fork_exec_sorted_fd_sanity_check.<locals>.BadInt.__int__N)r+ r, r- r� r� r� r r r r �BadInt s r� )r� r r� r� rf )ri r� )� � �* l )rf r� )r4 � r� r) r� r zfds_to_keep={}r� s falseTr� r� r� r� r� ) r# r. r/ r0 r1 rS r� rI r� rb ri rZ )r r# r� r� r� r\ r r r �%test_fork_exec_sorted_fd_sanity_check sN
�� �z:POSIXProcessTestCase.test_fork_exec_sorted_fd_sanity_checkc
C sR t �t�}|�: tj�|d�� }t|j_|� � |j�
� W 5 Q R X W 5 Q R X d S )Nr7 ) r r9 r r r� �object�BrokenPipeErrorr; r� r� Zassert_called_with�r rz �mock_proc_stdinr r r �,test_communicate_BrokenPipeError_stdin_close7 s
zAPOSIXProcessTestCase.test_communicate_BrokenPipeError_stdin_closec
C s` t �t�}|�H tj�|d��.}t|j_|� d� |j�
d� |j �
� W 5 Q R X W 5 Q R X d S )Nr7 � stuff) r r9 r r r� r� r� rs r� r� �assert_called_once_withr; r� r r r �,test_communicate_BrokenPipeError_stdin_write@ s
zAPOSIXProcessTestCase.test_communicate_BrokenPipeError_stdin_writec C s� t jtjdgt jt jd�}|�` tj�|d��F}tt j
d��.}t |j _
|�� |j_|�d� |j �� W 5 Q R X W 5 Q R X W 5 Q R X d S )N�-hr� r7 �wbr� )r r9 rJ rK r: r r� r� rQ r r� r� r6 r� r� � return_valuer� r� )r rz r� Zdev_nullr r r �,test_communicate_BrokenPipeError_stdin_flushJ s �
�
zAPOSIXProcessTestCase.test_communicate_BrokenPipeError_stdin_flushc
C sf t jtjdgt jt jd�}|�> tj�|d��$}t|j _
|j dd� |j � � W 5 Q R X W 5 Q R X d S )Nr r� r7 i� rQ )
r r9 rJ rK r: r r� r� r� r; r� r� r� r� r r r �9test_communicate_BrokenPipeError_stdin_close_with_timeoutY s � zNPOSIXProcessTestCase.test_communicate_BrokenPipeError_stdin_close_with_timeoutN�
W_STOPCODEzneed _testcapi.W_STOPCODEc C sp t }t�|�}|j}t�|d�\}}| �|d� t�d�}t j
d||fd�� |� � }W 5 Q R X | �|d� dS )zATest wait() behavior when waitpid returns WIFSTOPPED; issue29335.r r� r� )r ���N) r r r9 r� r r� r% � _testcapir r r� r )r r1 rz r� �statusr r r r � test_stoppede s
z!POSIXProcessTestCase.test_stoppedc C sH t jtjddgt jd�}z|jdd� W d S t jk
r@ Y qX qd S )Nr z+import os, time; os.close(1), time.sleep(2)r� rP rQ )r r9 rJ rK r: r� rT �r rz r r r �1test_communicate_repeated_call_after_stdout_closex s
�� zFPOSIXProcessTestCase.test_communicate_repeated_call_after_stdout_close)dr+ r, r- r r� r� r� r� r r9 r� r r� r� r r� r� r r� r� r
r r r r r r r$ r* r+ r2 rJ r� r9 r@ r� rJ rK rL rR rY r
rZ r] r_ rb rd rg ri rn rp rr rt rv rw rx ry rz r| r} r� r� r� r� r� r� r� r� r� r� r� �stat�st_devr� r Zrequires_mac_verr� r� r� r� r� r� r� r� r� r� r� � cpython_onlyr� r� r� r r r r� r r� r
r �
__classcell__r r r� r r� P s�
�
% �
�
-0, )0�
P
! *
'
�
r� zWindows specific testsc @ s� e Zd Zdd� Zdd� Zdd� Zdd� Zd d
� Zej d d � �Z
d
d� Z dd� Z dd� Z
dd� Zdd� Zdd� Zdd� Zdd � Zd d � Zd d � Zd!d"� Zd#d$� Zd%d&� Zd'd(� Zd)d*� Zd+d,� Zd-S ).�Win32ProcessTestCasec C s. d}d}t �� }||_||_t jt|d� d S )Nr� r� rA )r � STARTUPINFO�dwFlags� wShowWindowrI r )r �STARTF_USESHOWWINDOW� SW_MAXIMIZErB r r r �test_startupinfo� s �z%Win32ProcessTestCase.test_startupinfoc C s( d}d}t j||d�}t jt|d� d S )Nr� r� )r r rA )r r rI r )r ZSTARTF_USERSHOWWINDOWr rB r r r �test_startupinfo_keywords� s ��z.Win32ProcessTestCase.test_startupinfo_keywordsc C s� t �� }t j|_t j|_td�D ]�}t}tt j
d��@}t j ||t j |d�}|� |�
� W 5 Q R X | �|jd� W 5 Q R X | �|jt j� | �|j� | �|j� | �|j� | �|jt j� | �|jdg i� q d S )Nr r; )r8 r& rB r � handle_list)r r r r ZSW_HIDEr rO r rQ r r� r9 rj r� r% r rc Z hStdInputZ
hStdOutputZ hStdError�lpAttributeList)r rB r% r Znullrz r r r �test_startupinfo_copy� s, �� z*Win32ProcessTestCase.test_startupinfo_copyc C s( d}t j�d� tjt jd |d� d S )N� z' a DOS box should flash briefly ...
z# -c "import time; time.sleep(0.25)"rC )rJ r& rs r rI rK )r ZCREATE_NEW_CONSOLEr r r �test_creationflags� s ��z'Win32ProcessTestCase.test_creationflagsc C s$ | j ttjtjddgdd� d� d S )Nr rF c S s dS ru r r r r r r � r� z8Win32ProcessTestCase.test_invalid_args.<locals>.<lambda>r" rE r r r r r� � s
��z&Win32ProcessTestCase.test_invalid_argsc C s: G dd� dt �}| �t�� tjt|� d� W 5 Q R X d S )Nc @ s e Zd ZdZdS )z4Win32ProcessTestCase.test_issue31471.<locals>.BadEnvN)r+ r, r- �keysr r r r �BadEnv� s r r� )�dictrS r� r r9 r )r r r r r �test_issue31471� s z$Win32ProcessTestCase.test_issue31471c C s& t jtjddgdd�}| �|d� d S )Nr rF T)rT rG rH rL r r r r� � s
��z#Win32ProcessTestCase.test_close_fdsc
C s� dd l }t�� }| �tj|d � | �tj|d � g }|D ] }t�|d� |�|�|�� q<tj t
j dd� |d �gtj
dd�}|�� \}}| �|jd� t|�� � tj t
j dd� |d �gtj
tj
dd�}|�� \}}| �|jd� | �d |� g }t�� } d
|i| _tj t
j dd� |d �gtj
tj
| dd �}|�� \}}| �|jd� | �d |� t�d tf��d t�� } d
|d d � i| _tj t
j dd� |d �gtj
tj
| dd �}|�� \}}| �|jd� W 5 Q R X d S )
Nr r� Tr z2import msvcrt; print(msvcrt.open_osfhandle({}, 0))Fr� )r8 r& rT s OSErrorr )r8 r& rB rT z.*overriding close_fds)rw r r- rr r; r� rP Z
get_osfhandler r9 rJ rK rI r: r� r% r r r$ rb r r r r� rt )
r rw r� rX r� r@ r8 r& r rB r r r �test_close_fds_with_stdio� sl
� �
� �
� �
� � z.Win32ProcessTestCase.test_close_fds_with_stdioc C s t �� }i |_t jt|d� d S )NrA �r r r rI r �r rB r r r �test_empty_attribute_list s
�z.Win32ProcessTestCase.test_empty_attribute_listc C s$ t �� }dg i|_t jt|d� d S )Nr rA r# r$ r r r �test_empty_handle_list s
�z+Win32ProcessTestCase.test_empty_handle_listc C sN t j�� }d|d< tjdgdtj|d�}|� | �d|j�� � W 5 Q R X d S �N�physalisr� r� r� rG s physalis� r r� r� r r9 r: rb r8 r� rI r r r rJ s
�z(Win32ProcessTestCase.test_shell_sequencec C sL t j�� }d|d< tjddtj|d�}|� | �d|j�� � W 5 Q R X d S r' r) rI r r r rK ' s
�z&Win32ProcessTestCase.test_shell_stringc
C sZ dD ]P}t j�� }d|d< tjddtj||d�}|� | �d|j�� |� W 5 Q R X qd S )N)�ansiZoemr( r� r� r� )r� r8 r� r@ r) )r �encr� r@ r r r �test_shell_encodings1 s
�z)Win32ProcessTestCase.test_shell_encodingsc C s t �tjd �}| �|d� d S )Nz -c "import sys; sys.exit(47)"rG rH rL r r r rL = s �z%Win32ProcessTestCase.test_call_stringc G sz t jtjddgt jt jt jd�}|�@ |j�d� t||�|� |�� \}}| � |d� |�
� }W 5 Q R X | � |d� d S )Nr z�if 1:
import sys, time
sys.stdout.write('x\n')
sys.stdout.flush()
time.sleep(30)
r6 r� r� r ) r r9 rJ rK r: r8 r� rW r� r* r r )r rX r1 r@ r% r& r r r r rY C s � z"Win32ProcessTestCase._kill_processc G s� t jtjddgt jt jt jd�}|�J |j�d� t�d� t ||�|� |�
� \}}| � |d� |� � }W 5 Q R X | �
|d� d S )Nr z�if 1:
import sys, time
sys.stdout.write('x\n')
sys.stdout.flush()
sys.exit(42)
r6 r� r� r� )r r9 rJ rK r: r8 r� r~ r rW r� r* r r% )r rX r1 r@ r% r& rM r r r rZ X s �
z'Win32ProcessTestCase._kill_dead_processc C s | � dtj� d S rc )rY r� ra r r r r r] o s z%Win32ProcessTestCase.test_send_signalc C s | � d� d S re �rY r r r r r_ r s z Win32ProcessTestCase.test_killc C s | � d� d S rh r- r r r r rb u s z#Win32ProcessTestCase.test_terminatec C s | � dtj� d S rc )rZ r� ra r r r r rd x s z*Win32ProcessTestCase.test_send_signal_deadc C s | � d� d S re rf r r r r rg { s z#Win32ProcessTestCase.test_kill_deadc C s | � d� d S rh rf r r r r ri ~ s z(Win32ProcessTestCase.test_terminate_deadN)r+ r, r- r r r r r� r r r! r� r" r% r& rJ rK r, rL rY rZ r] r_ rb rd rg ri r r r r r � s.
1
r c @ s^ e Zd ZG dd� dej�Zej�ejd�dd� �Z dd� Z
dd � Z d
d � Z d d
� Z
dd� ZdS )� MiscTestsc s$ e Zd ZdZg Z� fdd�Z� ZS )zMiscTests.RecordingPopenz<A Popen that saves a reference to each instance for testing.c s t � j||� | j�| � d S r )r� r� �instances_createdrP r0 r� r r r� � s z!MiscTests.RecordingPopen.__init__)r+ r, r- r4 r/ r� r r r r� r �RecordingPopen� s r0 � _communicatec K s8 t |_z�tj �
t j d���}t |_tj �
t d| j��: | �
t ��$ |tjddgfdt ji|�� W 5 Q R X W 5 Q R X |jdd� D ]"}| �|tjdd�d |j� �� q�g }|jD ] }|tjd
d�kr�|�|� q�| j|jd |jd � | jt|�d|jd � W 5 Q R X W 5 | jj�� }|�� |�� | �g | jj� X dS )
a� Fake a SIGINT happening during Popen._communicate() and ._wait().
This avoids the need to actually try and get test environments to send
and receive signals reliably across platforms. The net effect of a ^C
happening during a blocking subprocess execution which we want to clean
up from is a KeyboardInterrupt coming out of communicate() or wait().
�_waitr9 r zLimport time
time.sleep(9)
import sys
sys.stderr.write('\n!runaway child!\n')r8 r� NrQ z.no open-ended wait() after the first allowed: g �?r r� )�KeyboardInterruptr� r0 r/ rW r� r r% r r� r� r r9 rS rJ rK r Zcall_args_listr rI rP ZassertLessEqualZ
call_countr$ )r ZpopenerZmock__communicater2 r� Z
mock__waitrI Z sigint_callsr r r � _test_keyboardinterrupt_no_kill� sF
� ���
�
� � z)MiscTests._test_keyboardinterrupt_no_killc C s | j tjdd� d S �NgT㥛� @rQ )r4 r rI r r r r �#test_call_keyboardinterrupt_no_kill� s z-MiscTests.test_call_keyboardinterrupt_no_killc C s | j tjdd� d S r5 )r4 r r r r r r �"test_run_keyboardinterrupt_no_kill� s z,MiscTests.test_run_keyboardinterrupt_no_killc C s dd� }| � |� d S )Nc _ s t j| |��
}t�W 5 Q R X d S r )r r9 r3 )r1 r2 Zunused_processr r r �popen_via_context_manager� s z[MiscTests.test_context_manager_keyboardinterrupt_no_kill.<locals>.popen_via_context_manager)r4 )r r8 r r r �.test_context_manager_keyboardinterrupt_no_kill� s z8MiscTests.test_context_manager_keyboardinterrupt_no_killc C s� | � t�d�d� | � t�d�d� d }z@t�� }tj� |d�}t�t
rLdnd| �\}}| � |d� W 5 |d k r|t�|� X d S )Nz
echo xyzzy�xyzzy)r r: Zfooztype zcat r ) r% r � getoutput�getstatusoutputr �rmdirrp rM r� r� r r )r �dirrP r rd r r r �test_getoutput� s ��zMiscTests.test_getoutputc C sn ddh}t tj�}t � }ddl}tj�� D ].\}}|�d�r>q*t||jf�rNq*|� |� q*| �
||| � dS )z*Ensure that __all__ is populated properly.r` �Handler Nr% ) r� r �__all__�types�__dict__�itemsr
�
isinstance�
ModuleTyper� r% )r Zintentionally_excludedZexportedZpossible_exportsrB rP r� r r r � test__all__� s
zMiscTests.test__all__N)r+ r, r- r r9 r0 r r� r� r4 r6 r7 r9 r? rG r r r r r. � s
+r. � PollSelectorz!Test needs selectors.PollSelectorc @ s e Zd Zdd� Zdd� ZdS )�ProcessTestCaseNoPollc C s t j| _tjt _t�| � d S r )r �_PopenSelector�
orig_selector� selectors�SelectSelectorr5 r r r r r r � s zProcessTestCaseNoPoll.setUpc C s | j t_t�| � d S r )rK r rJ r5 r" r r r r r" � s z ProcessTestCaseNoPoll.tearDownN)r+ r, r- r r" r r r r rI � s rI zWindows-specific testsc sP e Zd Z� fdd�Z� fdd�Zdd� Zdd� Zd d
� Zd d � Zd
d� Z � Z
S )�CommandsWithSpacesc s> t � �� t�dd�\}}|�� | _t�|d� t�|� d S )Nz.pyzte stsV import sys;sys.stdout.write('%d %s' % (len(sys.argv), [a.lower () for a in sys.argv]))) r� r rp r� r r> r rs r; )r r� r> r� r r r � s
zCommandsWithSpaces.setUpc s t �| j� t� �� d S r )r r� r> r� r" r r� r r r" � s zCommandsWithSpaces.tearDownc O sH t j|d< t j||�}|�$ | �|j�� �d�d| j � W 5 Q R X d S )Nr8 �mbcsz2 [%r, 'ab cd'])r r: r9 r% r8 r� r� r> )r r1 r2 r@ r r r � with_spaces
s
�z CommandsWithSpaces.with_spacesc C s | j dtj| jdf dd� d S )N�"%s" "%s" "%s"�ab cdr� r� �rP rJ rK r> r r r r � test_shell_string_with_spaces
s
��z0CommandsWithSpaces.test_shell_string_with_spacesc C s | j tj| jdgdd� d S )NrR r� r� rS r r r r � test_shell_sequence_with_spaces
s z2CommandsWithSpaces.test_shell_sequence_with_spacesc C s | � dtj| jdf � d S )NrQ rR rS r r r r � test_noshell_string_with_spaces
s �z2CommandsWithSpaces.test_noshell_string_with_spacesc C s | � tj| jdg� d S )NrR rS r r r r �!test_noshell_sequence_with_spaces
s z4CommandsWithSpaces.test_noshell_sequence_with_spaces) r+ r, r- r r" rP rT rU rV rW r r r r� r rN � s rN c @ s4 e Zd Zdd� Zdd� Zdd� Zdd� Zd d
� Zd S ) �ContextManagerTestsc C sl t jtjddgt jt jd��*}| �|j�� d� | �|j �� d� W 5 Q R X | �
|jj � | �
|j j � d S )Nr zAimport sys;sys.stdout.write('stdout');sys.stderr.write('stderr');r� s stdouts stderr) r r9 rJ rK r: r% r8 r� r* r& r' r( r r r r � test_pipe
s
�� z ContextManagerTests.test_pipec C s0 t �tjddg��}W 5 Q R X | �|jd� d S )Nr zimport sys; sys.exit(100)�d )r r9 rJ rK r% r r r r r r� -
s
�
z#ContextManagerTests.test_returncodec C s@ t jtjddgt jd�� }|�d� | �|jd� W 5 Q R X d S )Nr z2import sys;sys.exit(sys.stdin.read() == 'context')rn s contextr� r r r r r r 4
s
��
z*ContextManagerTests.test_communicate_stdinc
C s: | � t��& tjttjtjd��}W 5 Q R X W 5 Q R X d S )Nr� )rS r� r r9 r� r: r r r r r� <
s �z%ContextManagerTests.test_invalid_argsc C sr t jtt jtjd d�}|�� }|j�dtj � | � |j
� | � t |j
ddd� | �|j
d� | �|jj� dS )z9Broken pipe error should not prevent wait() (Issue 21619)r )r7 rB r� Nr )r r9 r r: r r. � __enter__r7 rs rc r rS rT �__exit__r% r' r( r r r r �test_broken_pipe_cleanupC
s � z,ContextManagerTests.test_broken_pipe_cleanupN)r+ r, r- rY r� r r� r] r r r r rX
s
rX �__main__)9r� r r� r r rJ r� r= r� r rU rp r~ r� rL r� r� r
r� r. r; Z test.supportr r ry ZPGOZSkipTestr� r r5 r� r� �NotADirectoryError�PermissionErrorr� rK r r ZTestCaser � Exceptionr. r9 r/ r5 r� r� r� r� r r. r� rI rN rX r+ �mainr r r r �<module> s�
F 0
=
}d �
+6