|
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 mZ d dlZd dlZd dlZd dlZd dlZd dlmZ d dl Z d dl
Z
d dl Z d dl m
Z
d dlmZ e
�d�Zdd� Zd d
� ZdMd d �Zejd
d� �Zdd� Zi fdd�Zdd� ZG dd� dedd��Zejdd� �ZG dd� dedd ��Z G d d � d ed d ��Z d d!�d"d#�Z d$d%� Z d&d'� Z d(d)� Z!G d*d+� d+e j"�Z#G d,d-� d-e j"�Z$G d.d/� d/e j"�Z%G d0d1� d1e#�Z&G d2d3� d3e#�Z'G d4d5� d5e#�Z(G d6d7� d7e#�Z)G d8d9� d9e#�Z*G d:d;� d;e#�Z+G d<d=� d=e#�Z,G d>d?� d?e#�Z-G d@dA� dAe#�Z.G dBdC� dCe#�Z/G dDdE� dEe#�Z0G dFdG� dGedGdH��Z1e �2dI�G dJdK� dKe#��Z3e4dLk�r\e �5� dS )N� )�
namedtupleN)�dedent)�support)�
script_helperZ_xxsubinterpretersc s& t j�� fdd�tt� �d �D ��S )Nc 3 s | ]}t � |�V qd S �N)� combinations)�.0�r�Zsets� �;/opt/cppython/lib/python3.8/test/test__xxsubinterpreters.py� <genexpr> s �zpowerset.<locals>.<genexpr>� )� itertools�chain�
from_iterable�range�lenr
r r
r �powerset s �r c C s: t �� \}}| �dd�}td|� d|� d��}|t|�fS )N�
z
z-
import contextlib
with open(zU, 'w') as spipe:
with contextlib.redirect_stdout(spipe):
z
)�os�pipe�replacer �open)�scriptr �wZindented�wrappedr r r �_captured_script s ��
r c
C s> t |�\}}|�$ t�| ||� |�� W 5 Q R � S Q R X d S r )r � interpreters�
run_string�read)�interpZrequest�sharedr Zrpiper r r � _run_output( s r# c # s` t �� \�}� �fdd�}tj|d�}|�� d V t|d��}|�d� W 5 Q R X |�� d S )Nc s t �� td�� d��� d S )Nz7
# wait for "signal"
with open(z5) as rpipe:
rpipe.read()
)r r r r �r! r r r �run2 s
�z_running.<locals>.run��targetr �done)r r � threading�Thread�startr �write�join)r! r r% �tZspiper r$ r �_running/ s r/ c K s t | ||� d S r )� _run_interp)�id�sourcer" r r r �
run_interpM s r3 c C sF t |�}t�� }|| kr4t�� |kr(t�t||� nt�| ||� d S r )r r �get_main� get_current� RuntimeError�execr )r1 r2 r" Z_mainns�mainr r r r0 Q s r0 c s0 � ��fdd�}t j|d�}|�� |�� d S )Nc s t � ��� d S r )Z_runr �r1 r" r2 r r r% ^ s z run_interp_threaded.<locals>.runr& )r) r* r+ r- )r1 r2 r" r% r. r r9 r �run_interp_threaded] s r: c s* e Zd Zedd� �Zd� fdd� Z� ZS )� Interpreterc C s( t || �r|S t |t�r | |�S t�d S r )�
isinstance�str�NotImplementedError)�cls�rawr r r �from_rawg s
zInterpreter.from_rawNc s� t �� }||kr6|sd}n|dkr0td�|���|}nX|d k rp|sHd}n|dkrXtd��t|t j�s�t �|�}n |r||dkr�d}|}nt �� }t� �| ||�}|S )Nr8 z)name mismatch (expected "main", got "{}")r! z!name mismatch (unexpected "main")) r r4 �
ValueError�formatr< �
InterpreterID�create�super�__new__)r? �namer1 r8 �self�� __class__r r rG p s, � zInterpreter.__new__)NN)�__name__�
__module__� __qualname__� classmethodrA rG �
__classcell__r r rJ r r; e s
r; zname idc c s& z
d V W n t jk
r Y nX d S r )r �ChannelClosedErrorr r r r �expect_channel_closed� s
rR c s6 e Zd Zd
� fdd� Zdd� Zdd� Zdd � Z� ZS ) �
ChannelActionNc s& |sd}|sd}t � �| |||�}|S )N�bothr8 �rF rG )r? �action�endr! rI rJ r r rG � s zChannelAction.__new__c O sb | j dkr | jdkrJt| j��n*| j dkr@| jdkrJt| j��n
t| j ��| jdkr^t| j��d S )N�use)�same�opposite�send�recv��close� force-close)rT rY rZ r[ r\ )r8 rY �other�extra)rV rW rB r! )rI �args�kwargsr r r �__init__� s
zChannelAction.__init__c C s2 | j dkr|S | j dkr(|dkr$dS dS | j S d S )NrY rZ r[ r\ �rW �rI rW r r r � resolve_end� s
zChannelAction.resolve_endc C st | j dkr|S | j dkr(|d kr$t�|S | j dkrB|d kr>t�|S | j dkrp|jdkrZ|S |rl|jdkrl|S t�d S )NrY r` ra r8 )r! r6 rH )rI r! r` ra r r r �resolve_interp� s
z ChannelAction.resolve_interp)NN)rL rM rN rG rd rg rh rP r r rJ r rS � s rS zaction end interpc sB e Zd Zddd�� fdd� Zdd� Zdd � Zd
d �d d
�Z� ZS )� ChannelStater F��closedc s t � �| ||�}|S r rU )r? �pendingrk rI rJ r r rG � s zChannelState.__new__c C s t | �| jd | jd�S �Nr rj ��typerl rk �rI r r r �incr� s zChannelState.incrc C s t | �| jd | jd�S rm rn rp r r r �decr� s zChannelState.decrT��forcec C s2 | j r|r| jdkr| S t| �|r&dn| jdd�S )Nr Trj )rk rl ro )rI rt r r r r^ � s zChannelState.close)r )rL rM rN rG rq rr r^ rP r r rJ r ri � s ri zpending closedT��
hideclosedc C st |j r(|dkr"|dkr"|jr"d}q,d}nd}zt| |||�}W n( tjk
rf |sZ|sZ� |�� }Y n
X |rpd�|S )NrX r\ FT.)rk rl � _run_actionr rQ r^ )�cidrV rW �staterv Z
expectfail�resultr r r �
run_action� s r{ c C s� |dkr�|dkr$t �| d� |�� S |dkrz|jsfzt �| � W n t jk
rZ | Y S X td��q�t �| � |�� S nt|��nt|dkr�i }|dkr�d||< t j | f|� |�
� S |d kr�d
di}|dkr�d||< t j | f|� |j
dd �S t|��d S ) NrX r[ � spamr\ zexpected ChannelEmptyErrorr^ �r\ r[ Tr_ rt rs ) r � channel_sendrq rl � channel_recv�ChannelEmptyError� Exceptionrr rB �
channel_closer^ )rx rV rW ry rc r r r rw � s:
� rw c C s@ t �� D ]2} | dkrqzt �| � W q tk
r8 Y qX qd S )Nr )r �list_all�destroyr6 �r1 r r r �clean_up_interpreters s r� c C s8 t �� D ]*} zt �| � W q t jk
r0 Y qX qd S r )r �channel_list_all�channel_destroy�ChannelNotFoundError�rx r r r �clean_up_channels s
r� c @ s e Zd Zdd� ZdS )�TestBasec C s t � t� d S r �r� r� rp r r r �tearDown( s zTestBase.tearDownN)rL rM rN r� r r r r r� & s r� c @ s e Zd Zdd� Zdd� ZdS )�IsShareableTestsc
C sB d ddddg}|D ]*}| � |�� | �t�|�� W 5 Q R X qd S )Nr| �spam�
i����)�subTest�
assertTruer � is_shareable)rI Z
shareables�objr r r �test_default_shareables2 s � �z(IsShareableTests.test_default_shareablesc
C s~ G dd� d�}G dd� dt �}ddtdttt� t� d||d �|d
�g }|D ].}| �t|��� | �t� |�� W 5 Q R X qJd S ) Nc @ s e Zd Zdd� Zdd� ZdS )z3IsShareableTests.test_not_shareable.<locals>.Cheesec S s
|| _ d S r �rH )rI rH r r r rd C s z<IsShareableTests.test_not_shareable.<locals>.Cheese.__init__c S s | j S r r� rp r r r �__str__E s z;IsShareableTests.test_not_shareable.<locals>.Cheese.__str__N)rL rM rN rd r� r r r r �CheeseB s r� c @ s e Zd ZdS )z5IsShareableTests.test_not_shareable.<locals>.SubBytesN)rL rM rN r r r r �SubBytesH s r� TF.g Y@Z Wensleydaler| )
�bytes�NotImplementedro �objectr� r� �repr� assertFalser r� )rI r� r� Znot_shareablesr� r r r �test_not_shareableA s( ��z#IsShareableTests.test_not_shareableN)rL rM rN r� r� r r r r r� 0 s r� c s` e Zd Z� fdd�Z� fdd�Zdd� Zdd� Zd d
� Zd d � Zd
d� Z dd� Z
dd� Z � Z S )�ShareableTypeTestsc s t � �� t�� | _d S r )rF �setUpr �channel_createrx rp rJ r r r� d s
zShareableTypeTests.setUpc s t �| j� t� �� d S r )r r� rx rF r� rp rJ r r r� h s zShareableTypeTests.tearDownc
C s^ |D ]T}| � |��@ t�| j|� t�| j�}| �||� | �t|�t|�� W 5 Q R X qd S r )r� r r~ rx r � assertEqual�assertIsro )rI �valuesr� �gotr r r �_assert_valuesl s z!ShareableTypeTests._assert_valuesc
C sJ dD ]@}| � |��, t�| j|� t�| j�}| �||� W 5 Q R X qd S )Nr )r� r r~ rx r r� )rI r� r� r r r �test_singletonsw s
z"ShareableTypeTests.test_singletonsc C s | � dd| jg� d S )Nr| i' )r� rx rp r r r �
test_types� s
�z ShareableTypeTests.test_typesc C s | � dd� tdd�D �� d S )Nc s s | ]}|j d ddd�V qdS )� �littleT)�signedN)�to_bytes)r �ir r r r
� s �z0ShareableTypeTests.test_bytes.<locals>.<genexpr>���� )r� r rp r r r �
test_bytes� s
�z ShareableTypeTests.test_bytesc C s | � dddg� d S )Nz hello worldu 你好世界� )r� rp r r r � test_strs� s z ShareableTypeTests.test_strsc C s* | � t�tdd�tjtj d g�� d S )Nr� r� r )r� r r r �sys�maxsizerp r r r �test_int� s �zShareableTypeTests.test_intc C sd t jd t j d dd g}|D ]>}| �|��* | �t�� t�| j|� W 5 Q R X W 5 Q R X q d S )Nr r� �� )r� r� r� � assertRaises�
OverflowErrorr r~ rx )rI Zintsr� r r r �test_non_shareable_int� s
� z)ShareableTypeTests.test_non_shareable_int)
rL rM rN r� r� r� r� r� r� r� r� r� rP r r rJ r r� b s r� c @ s$ e Zd Zdd� Zdd� Zdd� ZdS )� ListAllTestsc C s" t �� }t �� }| �||g� d S r )r r4 r� r� )rI r8 �idsr r r � test_initial� s zListAllTests.test_initialc C s6 t �� }t �� }t �� }t �� }| �||||g� d S r )r r4 rE r� r� �rI r8 �first�secondr� r r r �test_after_creating� s
z ListAllTests.test_after_creatingc C s> t �� }t �� }t �� }t �|� t �� }| �|||g� d S r )r r4 rE r� r� r� r� r r r �test_after_destroying� s
z"ListAllTests.test_after_destroyingN)rL rM rN r� r� r� r r r r r� � s r� c @ s e Zd Zdd� Zdd� ZdS )�GetCurrentTestsc C s. t �� }t �� }| �||� | �|t j� d S r )r r4 r5 r� �assertIsInstancerD )rI r8 �curr r r � test_main� s zGetCurrentTests.test_mainc C sR t �� }t �� }t|td��}t|�� �}t �� \}}| �||� | � ||� d S )Nz�
import _xxsubinterpreters as _interpreters
cur = _interpreters.get_current()
print(cur)
assert isinstance(cur, _interpreters.InterpreterID)
)
r r4 rE r# r �int�stripr� r� ZassertNotEqual)rI r8 r! �outr� �_�expectedr r r �test_subinterpreter� s z#GetCurrentTests.test_subinterpreterN)rL rM rN r� r� r r r r r� � s r� c @ s e Zd Zdd� Zdd� ZdS )� GetMainTestsc C s0 t �� \}t �� }| �||� | �|t j� d S r )r r� r4 r� r� rD )rI r� r8 r r r �test_from_main� s
zGetMainTests.test_from_mainc C s< t �� \}t �� }t|td��}t|�� �}| �||� d S )Nz�
import _xxsubinterpreters as _interpreters
main = _interpreters.get_main()
print(main)
assert isinstance(main, _interpreters.InterpreterID)
)r r� rE r# r r� r� r� )rI r� r! r� r8 r r r �test_from_subinterpreter� s
z%GetMainTests.test_from_subinterpreterN)rL rM rN r� r� r r r r r� � s r� c @ s< e Zd Zdd� Zdd� Zdd� Zdd� Zd d
� Zd d � Zd
S )�IsRunningTestsc C s t �� }| �t �|�� d S r )r r4 r� �
is_running)rI r8 r r r r� � s zIsRunningTests.test_mainc C sP t �� }| �t �|�� t|�� | �t �|�� W 5 Q R X | �t �|�� d S r )r rE r� r� r/ r� �rI r! r r r r� � s
z"IsRunningTests.test_subinterpreterc C s2 t �� }t|td|� d���}| �|�� d� d S )Nz`
import _xxsubinterpreters as _interpreters
if _interpreters.is_running(zZ):
print(True)
else:
print(False)
�True)r rE r# r r� r� )rI r! r� r r r r� � s
� z'IsRunningTests.test_from_subinterpreterc C s6 t �� }t �|� | �t�� t �|� W 5 Q R X d S r )r rE r� r� r6 r� r� r r r �test_already_destroyed� s
z%IsRunningTests.test_already_destroyedc C s$ | � t�� t�d� W 5 Q R X d S �Ni@B )r� r6 r r� rp r r r �test_does_not_exist s z"IsRunningTests.test_does_not_existc C s$ | � t�� t�d� W 5 Q R X d S �Nr� )r� rB r r� rp r r r � test_bad_id s zIsRunningTests.test_bad_idN) rL rM rN r� r� r� r� r� r� r r r r r� � s r� c @ sD e Zd Zdd� Zdd� Zdd� Zdd� Zd d
� Zd d � Zd
d� Z dS )�InterpreterIDTestsc C s" t jddd�}| �t|�d� d S )Nr� Trs )r rD r� r� �rI r1 r r r �
test_with_int s z InterpreterIDTests.test_with_intc C s4 G dd� dt �}tj|� dd�}| �t|�d� d S )Nc @ s e Zd Zdd� ZdS )z.InterpreterIDTests.test_coerce_id.<locals>.Intc S s dS �Nr� r rp r r r � __index__ s z8InterpreterIDTests.test_coerce_id.<locals>.Int.__index__N�rL rM rN r� r r r r �Int s r� Trs r� )r= r rD r� r� )rI r� r1 r r r �test_coerce_id s z!InterpreterIDTests.test_coerce_idc C sf | � ttjt� � | � ttjd� | � ttjd� | � ttjd� | � ttjd� | � ttjd� d S �Ng $@�10s 10r� l )r� � TypeErrorr rD r� rB r� rp r r r r� s z InterpreterIDTests.test_bad_idc C s4 t �� }| �t�� t �t|�d � W 5 Q R X d S �Nr )r r� r� r6 rD r� r� r r r r� # s z&InterpreterIDTests.test_does_not_existc C s" t jddd�}| �t|�d� d S �Nr� Trs r� )r rD r� r= r� r r r �test_str( s zInterpreterIDTests.test_strc C s" t jddd�}| �t|�d� d S )Nr� Trs zInterpreterID(10))r rD r� r� r� r r r � test_repr, s z InterpreterIDTests.test_reprc C s( t �� }t �t|��}t �� }| �||k� | �||k� | �|t|�k� | �t|�|k� | �|tt|��k� | �tt|��|k� | �|tt|��d k� | �|tt|��k� | �|dd k� | �|td�k� | �|dk� | �||k� | �||k� | �||k� | �||k� d S �N皙�����?r� r� �infr� )r rE rD r� r� �floatr� r= �rI �id1�id2�id3r r r �
test_equality0 s$ z InterpreterIDTests.test_equalityN)
rL rM rN r� r� r� r� r� r� r� r r r r r� s r� c @ sN e Zd Zdd� Ze�d�dd� �Zdd� Zdd � Zd
d � Z d d
� Z
dd� Z dS )� CreateTestsc C s* t �� }| �|t j� | �|t �� � d S r )r rE r� rD �assertInr� r� r r r � test_in_mainI s zCreateTests.test_in_mainz*enable this test when working on pystate.cc C sD t � }td�D ] }t�� }t�|� |�|� q| �t|�d� d S )N�d )�setr r rE r� �addr� r )rI �seenr� r1 r r r �test_unique_idO s
zCreateTests.test_unique_idc sZ t �� �d � � �fdd�}t j|d�}�� |�� W 5 Q R X |�� | �� t�� � d S )Nc s t �� � ��� ��� d S r )r rE �acquire�releaser �r1 �lockr r �f\ s z%CreateTests.test_in_thread.<locals>.fr& )r) �Lockr* r+ r- r� r r� �rI r� r. r r� r �test_in_threadY s zCreateTests.test_in_threadc C sJ t �� \}t �� }t|td��}t|�� �}| �tt �� �|||h� d S )Nz�
import _xxsubinterpreters as _interpreters
id = _interpreters.create()
print(id)
assert isinstance(id, _interpreters.InterpreterID)
) r r� rE r# r r� r� r� r� )rI r8 r� r� r� r r r �test_in_subinterpreterh s
z"CreateTests.test_in_subinterpreterc s^ t �� \}t �� � d �� �fdd�}tj|d�}|�� |�� | �tt �� �|� �h� d S )Nc s t � td��} t| �� ��d S )Nz�
import _xxsubinterpreters as _interpreters
id = _interpreters.create()
print(id)
)r# r r� r� )r� �r� r� r r r� y s z6CreateTests.test_in_threaded_subinterpreter.<locals>.fr& ) r r� rE r) r* r+ r- r� r� )rI r8 r� r. r r r � test_in_threaded_subinterpreteru s
z+CreateTests.test_in_threaded_subinterpreterc C sj t t�� �}g }td�D ]}t�� }|�|� q|D ]}t�|� q4t�� }| �t t�� �||hB � d S �N� )r� r r� r rE �appendr� r� �rI �beforer� r� r1 r r r �test_after_destroy_all� s z"CreateTests.test_after_destroy_allc C s` t t�� �}t�� }t�� }t�� }t�|� t�|� t�� }| �t t�� �|||hB � d S r )r� r r� rE r� r� )rI r r� r� r� r1 r r r �test_after_destroy_some� s
z#CreateTests.test_after_destroy_someN) rL rM rN r� �unittest�skipr� r r r r
r r r r r r� G s
r� c @ s\ e Zd Zdd� Zdd� Zdd� Zdd� Zd d
� Zd d � Zd
d� Z dd� Z
dd� Z dd� Z dS )� DestroyTestsc C sf t �� }t �� }t �� }| �|t �� � t �|� | �|t �� � | �|t �� � | �|t �� � d S r )r rE r� r� r� � assertNotInr� r r r �test_one� s
zDestroyTests.test_onec C sv t t�� �}t � }td�D ]}t�� }|�|� q| �t t�� �||B � |D ]}t�|� qN| �t t�� �|� d S r )r� r r� r rE r� r� r� r r r r �test_all� s zDestroyTests.test_allc sX t �� \� ��t�� t �� � W 5 Q R X � �fdd�}tj|d�}|�� |�� d S )Nc s$ �� t�� t�� � W 5 Q R X d S r �r� r6 r r� r �r8 rI r r r� � s z!DestroyTests.test_main.<locals>.fr& ) r r� r� r6 r� r) r* r+ r- r r r r r� � s
zDestroyTests.test_mainc C s6 t �� }t �|� | �t�� t �|� W 5 Q R X d S r )r rE r� r� r6 r� r r r r� � s
z#DestroyTests.test_already_destroyedc C s$ | � t�� t�d� W 5 Q R X d S r� r rp r r r r� � s z DestroyTests.test_does_not_existc C s$ | � t�� t�d� W 5 Q R X d S r� )r� rB r r� rp r r r r� � s zDestroyTests.test_bad_idc C sJ t �� \}t �� }td|� d��}t �||� | �tt �� �||h� d S )Nzo
import _xxsubinterpreters as _interpreters
try:
_interpreters.destroy(zD)
except RuntimeError:
pass
�r r� rE r r r� r� )rI r8 r1 r r r r �test_from_current� s
�
z DestroyTests.test_from_currentc C sR t �� \}t �� }t �� }td|� d��}t �||� | �tt �� �||h� d S )NzZ
import _xxsubinterpreters as _interpreters
_interpreters.destroy(�)
r )rI r8 r� r� r r r r �test_from_sibling� s
�
z DestroyTests.test_from_siblingc s4 t �� � � fdd�}tj|d�}|�� |�� d S )Nc s t �� � d S r )r r� r r� r r r� � s z.DestroyTests.test_from_other_thread.<locals>.fr& )r rE r) r* r+ r- r r r� r �test_from_other_thread� s
z#DestroyTests.test_from_other_threadc
C s� t �� \}t �� }t|��^ | jt �|�d|� d�d� | jtd|� d�d�� t �|� W 5 Q R X | �t �|�� W 5 Q R X d S )NzInterp z& should be running before destruction.)�msgz%Should not be able to destroy interp z while it's still running.) r r� rE r/ r� r� r� r6 r� )rI r8 r! r r r �test_still_running� s
�
�z DestroyTests.test_still_runningN)
rL rM rN r r r� r� r� r� r r r r r r r r r � s
r c s e Zd Zed�ZdZ� fdd�Z� fdd�Zedd� �Z d d
� Z
d d � Z d
d� Z e
�eed�d�dd� �Zdd� Zdd� Zdd� Zdd� Zdd � Zd d � Zejd7d d!��Zd"d#� Zd$d%� Z d&d'� Z d(d)� Z d*d+� Z d,d-� Z d.d/� Z!d0d1� Z"d2d3� Z#e
�$d4�d5d6� �Z%� Z&S )8�RunStringTestszJ
with open('{}', 'w') as out:
out.write('{}')
r� c s t � �� t�� | _d | _d S r )rF r� r rE r1 �_fsrp rJ r r r� s
zRunStringTests.setUpc s" | j d k r| j �� t� �� d S r )r r^ rF r� rp rJ r r r� s
zRunStringTests.tearDownc C s | j d krt| �| _ | j S r )r Z FSFixturerp r r r �fs s
zRunStringTests.fsc C sB t d�\}}|� t�| j|� |�� }W 5 Q R X | �|d� d S )N�print("it worked!", end="")�
it worked!�r r r r1 r r� �rI r �filer� r r r � test_success s
zRunStringTests.test_successc s^ t d�\� }|�8 � �fdd�}tj|d�}|�� |�� |�� }W 5 Q R X ��|d� d S )Nr c s t ��j� � d S r )r r r1 r �r rI r r r� ) s z(RunStringTests.test_in_thread.<locals>.fr& r )r r) r* r+ r- r r� )rI r" r� r. r� r r$ r r & s z RunStringTests.test_in_threadc C sB t d�\}}|� t�| j|� |�� }W 5 Q R X | �|d� d S )Nz�
import threading
def f():
print('it worked!', end='')
t = threading.Thread(target=f)
t.start()
t.join()
r r r! r r r �test_create_thread3 s
z!RunStringTests.test_create_thread�forkztest needs os.fork()c C s| dd l }|�d��`}|�d� |�� d}td|j� d|� d��}t�| j|� |� d� |�
� }| � ||� W 5 Q R X d S )Nr zw+r� zspam spam spam spam spamz�
import os
try:
os.fork()
except RuntimeError:
with open('z3', 'w') as out:
out.write('z')
) �tempfile�NamedTemporaryFiler, �flushr rH r r r1 �seekr r� )rI r' r" r� r Zcontentr r r � test_forkC s
��
zRunStringTests.test_forkc
C s> t | j��* | �t�� t�| jd� W 5 Q R X W 5 Q R X d S )N�
print("spam"))r/ r1 r� r6 r r rp r r r �test_already_runningY s z#RunStringTests.test_already_runningc C s@ d}|t �� kr|d7 }q| �t�� t �|d� W 5 Q R X d S )Nr r r, )r r� r� r6 r r� r r r r� ^ s
z"RunStringTests.test_does_not_existc C s& | � t�� t�dd� W 5 Q R X d S )Nr� r, )r� rB r r rp r r r �
test_error_ide s z RunStringTests.test_error_idc C s& | � t�� t�dd� W 5 Q R X d S )Nr� r, )r� r� r r rp r r r r� i s zRunStringTests.test_bad_idc C s( | � t�� t�| jd� W 5 Q R X d S r� �r� r� r r r1 rp r r r �test_bad_scriptm s z RunStringTests.test_bad_scriptc C s( | � t�� t�| jd� W 5 Q R X d S )Ns
print("spam")r/ rp r r r �test_bytes_for_scriptq s z$RunStringTests.test_bytes_for_scriptNc c sf | � tj�� }d V W 5 Q R X |d krH| �t|j��d�d t|�� n| �t|j�d�||�� d S )N�:r z{}: {})r� r �RunFailedErrorr� r= � exception�splitrC )rI �exctyper Zcaughtr r r �assert_run_failedu s �
�z RunStringTests.assert_run_failedc C s( | � t�� t�| jd� W 5 Q R X d S )Nz print("spam")r7 � SyntaxErrorr r r1 rp r r r �test_invalid_syntax� s z"RunStringTests.test_invalid_syntaxc C s* | � td�� t�| jd� W 5 Q R X d S )Nr� zraise Exception("spam"))r7 r� r r r1 rp r r r � test_failure� s zRunStringTests.test_failurec C s* | � td�� t�| jd� W 5 Q R X d S )N�42zraise SystemExit(42))r7 �
SystemExitr r r1 rp r r r �test_SystemExit� s z RunStringTests.test_SystemExitc C sV | � t�� t�| jtd�� W 5 Q R X | � td�� t�| jtd�� W 5 Q R X d S )NzG
import sys
sys.exit()
r; zI
import sys
sys.exit(42)
)r7 r< r r r1 r rp r r r �
test_sys_exit� s z RunStringTests.test_sys_exitc C s� t �� \}}ddd d�}td|� d��}t�| j||� t|d��}t�|�}W 5 Q R X | � |d d� | � |d d
� | � |d d � | �
|d
� d S )N� hams -1)r� �eggs�cheddarz�
eggs = int(eggs)
spam = 42
result = spam + eggs
ns = dict(vars())
del ns['__builtins__']
import pickle
with open(�C, 'wb') as chan:
pickle.dump(ns, chan)
�rbr� �* r@ r� rz �) rA ) r r r r r r1 r �pickle�loadr� Z assertIsNone�rI r r r" r �chan�nsr r r �test_with_shared� s ��
z RunStringTests.test_with_sharedc C s� t �| jtd�� ddi}td�}t �| j||� t�� \}}td|� d��}t �| j|� t|d��}t�|�}W 5 Q R X | � |d d d � | � |d
d d� | � |d d� d S ) Nzj
spam = 'eggs'
ns1 = dict(vars())
del ns1['__builtins__']
r� r? zL
ns2 = dict(vars())
del ns2['__builtins__']
zr
ns = dict(vars())
del ns['__builtins__']
import pickle
with open(rB rC �ns1r@ �ns2)
r r r1 r r r r rF rG r� )rI r" r r r rI rJ r r r �test_shared_overwrites� s �
z%RunStringTests.test_shared_overwritesc C sh t �� \}}ddi}td|� d��}t�| j||� t|d��}t�|�}W 5 Q R X | � |d d� d S )NrL s not __main__z�
spam = 42
ns = dict(vars())
del ns['__builtins__']
import pickle
with open(rB rC )
r r r r r r1 r rF rG r� rH r r r �#test_shared_overwrites_default_vars� s �
z2RunStringTests.test_shared_overwrites_default_varsc C s� t �� \}}t�| jtd|� d��� t|d��}t�|�}W 5 Q R X t �� \}}t�| jtd|� d��� t|d��}t�|�}W 5 Q R X | � d|� | �
d|� | � d|� | � d|� d S )Nz�
spam = True
ns = dict(vars())
del ns['__builtins__']
import pickle
with open(zd, 'wb') as chan:
pickle.dump(ns, chan)
del ns, pickle, chan
rC z�
eggs = False
ns = dict(vars())
del ns['__builtins__']
import pickle
with open(rB r� r@ ) r r r r r1 r r rF rG r� r )rI r r rI rL rM r r r �test_main_reused� s �
� z RunStringTests.test_main_reusedc
C s| t �� \}}td|� d��}t�| j|� t|d��}t�|�}W 5 Q R X |� d� |� d� | �
|di d d d dd�� d S ) Nz�
spam = 42
ns = dict(vars())
ns['__builtins__'] = str(ns['__builtins__'])
import pickle
with open(rB rC � __builtins__�
__loader__�__main__rD )rL �__annotations__�__doc__� __package__�__spec__r� ) r r r r r r1 r rF rG �popr� )rI r r r rI rJ r r r � test_execution_namespace_is_main s" �
�z/RunStringTests.test_execution_namespace_is_mainzblocking foreverc
C sX t d�}t�� �2}t�|d|�}t�|��}|�� }W 5 Q R X W 5 Q R X | �|d� d S )Na�
from textwrap import dedent
import threading
import _xxsubinterpreters as _interpreters
id = _interpreters.create()
def f():
_interpreters.run_string(id, dedent('''
import time
# Give plenty of time for the main interpreter to finish.
time.sleep(1_000_000)
'''))
t = threading.Thread(target=f)
t.start()
r! r )r r Ztemp_dirr Z make_scriptZ spawn_python�waitr� )rI r �dirname�filename�proc�retcoder r r �test_still_running_at_exit$ s
z)RunStringTests.test_still_running_at_exit)N)'rL rM rN r ZSCRIPTZFILENAMEr� r� �propertyr r# r r% r Z
skipUnless�hasattrr r+ r- r� r. r� r0 r1 �
contextlib�contextmanagerr7 r9 r: r= r> rK rN rO rP rY r
r_ rP r r rJ r r s<
! r c @ sT e Zd Zdd� Zdd� Zdd� Zdd� Zd d
� Zd d � Zd
d� Z dd� Z
dd� Z dS )�ChannelIDTestsc C s0 t jddd�}| �t|�d� | �|jd� d S )Nr� Trs rT )r � _channel_idr� r� rW �rI rx r r r �test_default_kwargsB s z"ChannelIDTests.test_default_kwargsc C s� t jdddd�}| �|jd� t jddddd�}| �|jd� t jdddd�}| �|jd� t jddddd �}| �|jd� t jddddd�}| �|jd
� d S ) Nr� T�r[ rt r[ F�r[ r\ rt �r\ rt r\ )r\ r[ rt rT )r re r� rW rf r r r �test_with_kwargsH s z ChannelIDTests.test_with_kwargsc C s4 G dd� dt �}tj|� dd�}| �t|�d� d S )Nc @ s e Zd Zdd� ZdS )z*ChannelIDTests.test_coerce_id.<locals>.Intc S s dS r� r rp r r r r� Z s z4ChannelIDTests.test_coerce_id.<locals>.Int.__index__Nr� r r r r r� Y s r� Trs r� )r= r re r� r� )rI r� rx r r r r� X s z ChannelIDTests.test_coerce_idc C sf | � ttjt� � | � ttjd� | � ttjd� | � ttjd� | � ttjd� | � ttjd� d S r� )r� r� r re r� rB r� rp r r r r� ` s zChannelIDTests.test_bad_idc C s* | � t�� tjdddd� W 5 Q R X d S )Nr� F�r[ r\ )r� rB r re rp r r r �test_bad_kwargsh s z ChannelIDTests.test_bad_kwargsc C s6 t �� }| �t j�� t �t|�d � W 5 Q R X d S r� )r r� r� r� re r� rf r r r r� l s z"ChannelIDTests.test_does_not_existc C s" t jddd�}| �t|�d� d S r� )r re r� r= rf r r r r� q s zChannelIDTests.test_strc C s� t jddd�}| �t|�d� t jdddd�}| �t|�d� t jdddd�}| �t|�d� t jddddd �}| �t|�d� d S )
Nr� Trs z
ChannelID(10)rh zChannelID(10, send=True)rj zChannelID(10, recv=True)ri )r re r� r� rf r r r r� u s zChannelIDTests.test_reprc C s( t �� }t �t|��}t �� }| �||k� | �||k� | �|t|�k� | �t|�|k� | �|tt|��k� | �tt|��|k� | �|tt|��d k� | �|tt|��k� | �|dd k� | �|td�k� | �|dk� | �||k� | �||k� | �||k� | �||k� d S r� )r r� re r� r� r� r� r= )rI �cid1�cid2Zcid3r r r r� � s$ z ChannelIDTests.test_equalityN) rL rM rN rg rk r� r� rm r� r� r� r� r r r r rd @ s
rd c @ s� e Zd Zdd� Zdd� Zdd� Zdd� Zd d
� Zd d � Zd
d� Z dd� Z
dd� Z dd� Z dd� Z
dd� Ze�d�dd� �Zd d � Zd d � Zd d!� Zd"d#� Zd$d%� Zd&d'� Zd(d)� Zd*d+� Zd,d-� Zd.d/� Zd0d1� Z d2d3� Z d4d5� Z d6d7� Z d8S )9� ChannelTestsc C s t �� }| �|t j� d S r )r r� r� Z ChannelIDrf r r r �test_create_cid� s z ChannelTests.test_create_cidc C sr t �� }t �� }t �� }t �� }t �� }| �|t|�d � | �|t|�d � | �t|�t|� |||h� d S r� )r r� r� r� r� r� )rI r r� r� r� Zafterr r r �test_sequential_ids� s z ChannelTests.test_sequential_idsc C s\ t �� }t|td��}t|�� �}t �� }t|td��}t|�� �}| �|t|�d � d S )Nz�
import _xxsubinterpreters as _interpreters
cid = _interpreters.channel_create()
print(cid)
r )r rE r# r r� r� r� )rI r� r� rn r� ro r r r �test_ids_global� s z ChannelTests.test_ids_globalc C s> t �� }d}t �||� t �|�}| �||� | �||� d S �Nr| )r r� r~ r r� Z assertIsNot)rI rx Zorigr� r r r �test_send_recv_main� s
z ChannelTests.test_send_recv_mainc C s t �� }t|td��}d S )Na6
import _xxsubinterpreters as _interpreters
cid = _interpreters.channel_create()
orig = b'spam'
_interpreters.channel_send(cid, orig)
obj = _interpreters.channel_recv(cid)
assert obj is not orig
assert obj == orig
)r rE r# r )rI r� r� r r r � test_send_recv_same_interpreter� s z,ChannelTests.test_send_recv_same_interpreterc C s@ t �� }t �� }t|td|� d���}t �|�}| �|d� d S )N�_
import _xxsubinterpreters as _interpreters
_interpreters.channel_send(�, b'spam')
r| )r r� rE r# r r r� )rI rx r� r� r� r r r �%test_send_recv_different_interpreters� s �
z2ChannelTests.test_send_recv_different_interpretersc sV t �� � � fdd�}tj|d�}|�� t �� d� |�� t �� �}| �|d� d S )Nc sF zt �� �} W q6W q t jk
r2 t�d� Y q X q t �� | � d S )Nr� )r r r� �time�sleepr~ )r� r� r r r� � s
z8ChannelTests.test_send_recv_different_threads.<locals>.fr& r| ) r r� r) r* r+ r~ r- r r� �rI r� r. r� r r� r � test_send_recv_different_threads� s
z-ChannelTests.test_send_recv_different_threadsc sf t �� � t �� �d �� ��fdd�}tj|d�}|�� t �� d� |�� t �� �}| � |d� d S )Nc s t �td� � d� � d����d S )Nz�
import time
import _xxsubinterpreters as _interpreters
while True:
try:
obj = _interpreters.channel_recv(z�)
break
except _interpreters.ChannelEmptyError:
time.sleep(0.1)
assert(obj == b'spam')
_interpreters.channel_send(z, b'eggs')
)r# r r �rx r� r� r r r� � s
��zIChannelTests.test_send_recv_different_interpreters_and_threads.<locals>.fr& r| � eggs)
r r� rE r) r* r+ r~ r- r r� r| r r~ r �1test_send_recv_different_interpreters_and_threads� s
z>ChannelTests.test_send_recv_different_interpreters_and_threadsc C s( | � tj�� t�dd� W 5 Q R X d S )Nr� r| )r� r r� r~ rp r r r �test_send_not_found s z ChannelTests.test_send_not_foundc C s& | � tj�� t�d� W 5 Q R X d S r� )r� r r� r rp r r r �test_recv_not_found s z ChannelTests.test_recv_not_foundc C s. t �� }| �t j�� t �|� W 5 Q R X d S r )r r� r� r� r rf r r r �test_recv_empty s z ChannelTests.test_recv_emptyc C sR t �� }t �� }t|td�t|jd��}t �|�}| �|d� | �|� � d� d S )Nz�
import _xxsubinterpreters as _interpreters
print(cid.end)
_interpreters.channel_send(cid, b'spam')
r� r| r[ )
r r� rE r# r �dictr[ r r� r� �rI rx r! r� r� r r r � test_run_string_arg_unresolved s
�
z+ChannelTests.test_run_string_arg_unresolvedz(disabled until high-level channels existc C s` t �� }t j|dd�}t �� }t|td�t|jd��}t �|�}| � |d� | � |�
� d� d S )NT)Z_resolvez�
import _xxsubinterpreters as _interpreters
print(chan.id.end)
_interpreters.channel_send(chan.id, b'spam')
)rI r| r[ ) r r� re rE r# r r� r[ r r� r� r� r r r � test_run_string_arg_resolved. s
�
z)ChannelTests.test_run_string_arg_resolvedc C sr t �� }t �|d� t �|� t �|� | �t j�� t �|d� W 5 Q R X | �t j�� t �|� W 5 Q R X d S �Nr| r �r r� r~ r r� r� rQ rf r r r �test_close_single_userA s
z#ChannelTests.test_close_single_userc C s� t �� }t �� }t �� }t �|td|� d��� t �|td|� d��� t �|� | �t j�� }t �|td|� d��� W 5 Q R X | �dt |j
�� | �t j�� }t �|td|� d��� W 5 Q R X | �dt |j
�� d S )Nrw rx z_
import _xxsubinterpreters as _interpreters
_interpreters.channel_recv(r z,
_interpreters.channel_send(z, b'spam')
rQ ) r r� rE r r r� r� r3 r� r= r4 )rI rx r� r� �cmr r r �test_close_multiple_usersL s(
�
�
�
�z&ChannelTests.test_close_multiple_usersc C sN t �� }t �|d� t �|� t �|� | �t j�� t �|� W 5 Q R X d S rt r� rf r r r �test_close_multiple_timesd s
z&ChannelTests.test_close_multiple_timesc C s� ddddg}|D ]�\}}| � ||f��z t�� }t�|d� t�|� tj|||d� | �tj�� t�|d� W 5 Q R X | �tj�� t�|� W 5 Q R X W 5 Q R X qd S )N)FF�TF)FT)TTr| rl r )r� r r� r~ r r� r� rQ )rI Ztestsr[ r\ rx r r r �test_close_emptym s �
z ChannelTests.test_close_emptyc C s\ t �� }t �|d� t �|d� | �t j�� t �|� W 5 Q R X t �|� t �|d� d S )Nr| r? r �r r� r~ r� ZChannelNotEmptyErrorr� r rf r r r �%test_close_defaults_with_unused_items� s
z2ChannelTests.test_close_defaults_with_unused_itemsc C s� t �� }t �|d� t �|d� | �t j�� t j|dd� W 5 Q R X t �|� t �|d� t �|� t �|� t j|dd� d S )Nr| r? T�r\ r r� rf r r r �*test_close_recv_with_unused_items_unforced� s
z7ChannelTests.test_close_recv_with_unused_items_unforcedc C s� t �� }t �|d� t �|d� t j|dd� | �t j�� t �|d� W 5 Q R X t �|� t �|� | �t j�� t �|� W 5 Q R X d S )Nr| r? T�r[ r �r r� r~ r� r� rQ r rf r r r �*test_close_send_with_unused_items_unforced� s
z7ChannelTests.test_close_send_with_unused_items_unforcedc C s� t �� }t �|d� t �|d� | �t j�� t j|ddd� W 5 Q R X t �|� t �|d� t �|� t �|� t j|dd� d S )Nr| r? Tr} r r� r� rf r r r �*test_close_both_with_unused_items_unforced� s
z7ChannelTests.test_close_both_with_unused_items_unforcedc C sz t �� }t �|d� t �|d� t j|ddd� | �t j�� t �|d� W 5 Q R X | �t j�� t �|� W 5 Q R X d S )Nr| r? Trj r r� rf r r r �(test_close_recv_with_unused_items_forced� s z5ChannelTests.test_close_recv_with_unused_items_forcedc C sz t �� }t �|d� t �|d� t j|ddd� | �t j�� t �|d� W 5 Q R X | �t j�� t �|� W 5 Q R X d S )Nr| r? Trh r r� rf r r r �(test_close_send_with_unused_items_forced� s z5ChannelTests.test_close_send_with_unused_items_forcedc C s| t �� }t �|d� t �|d� t j|dddd� | �t j�� t �|d� W 5 Q R X | �t j�� t �|� W 5 Q R X d S )Nr| r? Tri r r� rf r r r �(test_close_both_with_unused_items_forced� s z5ChannelTests.test_close_both_with_unused_items_forcedc C s\ t �� }t �|� | �t j�� t �|d� W 5 Q R X | �t j�� t �|� W 5 Q R X d S rt )r r� r� r� rQ r~ r rf r r r �test_close_never_used� s
z"ChannelTests.test_close_never_usedc C s| t �� }t �|d� t �� }t �|td|� d��� | �t j�� t �|� W 5 Q R X | �t j�� t � |� W 5 Q R X d S )Nr| z`
import _xxsubinterpreters as _interpreters
_interpreters.channel_close(z, force=True)
)
r r� r~ rE r r r� rQ r r� �rI rx r! r r r �!test_close_by_unassociated_interp� s
� z.ChannelTests.test_close_by_unassociated_interpc C s� t �� }t �|d� t �|d� t �|d� t �|� t j|dd� | �t j�� t �|d� W 5 Q R X | �t j�� t �|� W 5 Q R X d S )Nr| Trs r r� rf r r r �-test_close_used_multiple_times_by_single_user� s
z:ChannelTests.test_close_used_multiple_times_by_single_userN) rL rM rN rq rr rs ru rv ry r} r� r� r� r� r� r r
r� r� r� r� r� r� r� r� r� r� r� r� r� r� r� r r r r rp � s8
rp c @ s\ e Zd Zdd� Zdd� Zdd� Zdd� Zd d
� Zd d � Zd
d� Z dd� Z
dd� Z dd� Z dS )�ChannelReleaseTestsc C sx t �� }t �|d� t �|� t j|ddd� | �t j�� t �|d� W 5 Q R X | �t j�� t �|� W 5 Q R X d S �Nr| Trl r �r r� r~ r �channel_releaser� rQ rf r r r �test_single_user! s
z$ChannelReleaseTests.test_single_userc C sx t �� }t �� }t �� }t �|td|� d��� t|td|� d|� d���}t �|td|� d��� | �|�� d� d S ) Nrw rx ze
import _xxsubinterpreters as _interpreters
obj = _interpreters.channel_recv(z,)
_interpreters.channel_release(z+)
print(repr(obj))
z+
_interpreters.channel_release(r zb'spam')r r� rE r r r# r� r� )rI rx r� r� r� r r r �test_multiple_users, s
� ��
� z'ChannelReleaseTests.test_multiple_usersc C sr t �� }t �|d� t �|� t �|� | �t j�� t �|d� W 5 Q R X | �t j�� t �|� W 5 Q R X d S r� r� rf r r r �test_no_kwargs@ s
z"ChannelReleaseTests.test_no_kwargsc C sZ t �� }t �|d� t �|� t j|ddd� | �t j�� t j|ddd� W 5 Q R X d S )Nr| Trl r� rf r r r �test_multiple_timesK s
z'ChannelReleaseTests.test_multiple_timesc C sV t �� }t �|d� t �|d� t j|ddd� | �t j�� t �|� W 5 Q R X d S )Nr| r? Trl )r r� r~ r� r� rQ r rf r r r �test_with_unused_itemsT s z*ChannelReleaseTests.test_with_unused_itemsc C s\ t �� }t �|� | �t j�� t �|d� W 5 Q R X | �t j�� t �|� W 5 Q R X d S rt )r r� r� r� rQ r~ r rf r r r �test_never_used] s
z#ChannelReleaseTests.test_never_usedc C s| t �� }t �|d� t �� }t �|td|� d��� t �|�}t �|� | �t j �� t �|d� W 5 Q R X | �
|d� d S )Nr| zb
import _xxsubinterpreters as _interpreters
_interpreters.channel_release(r r ) r r� r~ rE r r r r� r� rQ r� )rI rx r! r� r r r �test_by_unassociated_interpf s
�
z/ChannelReleaseTests.test_by_unassociated_interpc C sT t �� }t �� }t �|td|� d|� d��� | �t j�� t �|� W 5 Q R X d S )Nze
import _xxsubinterpreters as _interpreters
obj = _interpreters.channel_send(z5, b'spam')
_interpreters.channel_release(r )r r� rE r r r� rQ r r� r r r �test_close_if_unassociatedu s
�� z.ChannelReleaseTests.test_close_if_unassociatedc C sR t �� }t �|d � t �|� t �|d� t j|dd� t �|�}| �|d� d S )Nr| Tr� )r r� r~ r r� r� )rI rx r� r r r �test_partially� s
z"ChannelReleaseTests.test_partiallyc C s� t �� }t �|d� t �|d� t �|d� t �|� t j|ddd� | �t j�� t �|d� W 5 Q R X | �t j�� t �|� W 5 Q R X d S r� r� rf r r r �'test_used_multiple_times_by_single_user� s
z;ChannelReleaseTests.test_used_multiple_times_by_single_userN)
rL rM rN r� r� r� r� r� r� r� r� r� r� r r r r r� � s +
r� c sz e Zd ZdZ� fdd�Zedd� �Zedd� �Zdd � Zdd d �Z d
d� Z
dd� Z dd� Z dd� Z
dd� Zdd� Z� ZS )�ChannelCloseFixtureFc sx | j r i }n4t�|�}t�|�}t�|�}|j||j||j|i}|sHd}t� �| |||||�}t� |_t� |_ ||_
|S )NrY ) �QUICKr; rA rH rF rG r� �_preppedri �_state�_known)r? rW r! r` ra �creatorZknownrI rJ r r rG � s&
�zChannelCloseFixture.__new__c C s | j S r �r� rp r r r ry � s zChannelCloseFixture.statec C s@ z| j W S tk
r: | �| j�}| �|�| _ | j Y S X d S r )�_cid�AttributeError�_get_interpreterr� � _new_channel)rI r� r r r rx � s zChannelCloseFixture.cidc C s | � |�}| �|� |S r )r� �_prep_interpreterr� r r r �get_interpreter� s
z#ChannelCloseFixture.get_interpreterNc C s2 |d kr| j }|dkr&| jjdkr&dS t| jj�S )Nr\ r[ F)rW ry rk �boolrf r r r �expect_closed_error� s
z'ChannelCloseFixture.expect_closed_errorc C s | � |� d S r )r� r� r r r �prep_interpreter� s z$ChannelCloseFixture.prep_interpreterc C s
|| _ d S r r� )rI rV rz r r r �
record_action� s z!ChannelCloseFixture.record_actionc C s t � t� d S r r� rp r r r �clean_up� s z ChannelCloseFixture.clean_upc C s@ |j dkrt�� S t�� }t|jd|� d�� t�|�| _| jS )Nr8 a
import _xxsubinterpreters
cid = _xxsubinterpreters.channel_create()
# We purposefully send back an int to avoid tying the
# channel to the other interpreter.
_xxsubinterpreters.channel_send(zC, int(cid))
del _xxsubinterpreters
)rH r r� r3 r1 r r� )rI r� Zchr r r r� � s
�
z ChannelCloseFixture._new_channelc C sl |dkr| j S |dkr | jS |dkr*| jS |}z| j| }W n& tk
rb t|� }| j|< Y nX |S d S )N)rY r! r` ra )r! r` ra r� �KeyErrorr; )rI r! rH r r r r� � s z$ChannelCloseFixture._get_interpreterc C sF |j | jkrd S | j�|j � |jdkr,d S t|j d| j� d�� d S )Nr8 a
import _xxsubinterpreters as interpreters
import test.test__xxsubinterpreters as helpers
ChannelState = helpers.ChannelState
try:
cid
except NameError:
cid = interpreters._channel_id(r )r1 r� r� rH r3 rx r� r r r r� � s
�z%ChannelCloseFixture._prep_interpreter)N)rL rM rN r� rG r` ry rx r� r� r� r� r� r� r� r� rP r r rJ r r� � s
r� z end interp other extra creatorz%these tests take several hours to runc @ s� e Zd Zdd� Zdd� Zdd� Zdd� Zd d
� Zd d �d
d�Zdd� Z dd� Z
d!dd�Z dd� Z d"dd�Z
dd � Zd d � Zd d � ZdS )#�ExhaustiveChannelTestsc c s� g V | � dd�D ]"}|V | �� D ]}|| V q$q| � dd�D ]"}|V | �� D ]}|| V qTqB| �dd�D ]z}|V | � dd�D ]*}|| }|V | �� D ]}|| V q�q�| � dd�D ]*}|| }|V | �� D ]}|| V q�q�qr| �dd�D ]�}|V | � dd�D ].}|| }|V | �� D ]}|| V �q*�q| � dd�D ].}|| }|V | �� D ]}|| V �qf�qLq�d S )NrY r` ra )�_iter_close_action_sets� _iter_post_close_action_sets�_iter_use_action_sets)rI Z closeactionsZ postactionsZ
useactions�actionsr r r �iter_action_sets7 sB z'ExhaustiveChannelTests.iter_action_setsc c s ||f}t dd|�gV t dd|�gV t dd|�t dd|�gV t dd|�gV t dd|�gV t dd|�t dd|�gV |D ]<}|D ]2}|D ](}t dd|�t dd|�t dd|�gV q�q�q�|D ]P}|D ]F}|D ]<}|D ]2}t dd|�t dd|�t dd|�t dd|�gV q�q�q�q�d S )NrX r\ r[ �rS )rI �interp1�interp2�interpsZinterp3Zinterp4r r r r� d s>
�
�
�
�
�
�
�
�z,ExhaustiveChannelTests._iter_use_action_setsc
c s� d}||f}dD ]4}|r dnd}|D ] }|D ]}t |||�gV q,q$qdD ]<} dD ]2}
|D ](} |D ] } t | d| �t |
d| �gV qbqZqRqJd S )Nr} r� r_ r^ r] r\ r[ r� )
rI r� r� Zendsr� rt �opr! rW ZrecvopZsendopr\ r[ r r r r� � s
�
�z.ExhaustiveChannelTests._iter_close_action_setsc c s. dD ]$}t dd|�gV t dd|�gV qd S )N)rY ra r` rX r\ r[ r� r� r r r r� � s
�
�z3ExhaustiveChannelTests._iter_post_close_action_setsc C s |D ]}| � ||� qd S r )r{ )rI �fixr� rV r r r � run_actions� s z"ExhaustiveChannelTests.run_actionsTru c C s� |� |j�}|�|j|j|j�}|�|� |jdkrZt|j |j
||j |d�}|� ||� n�t
�� }t|jd|j � dt|j
�� dt|�� dt|j �� d|� d|� d|� d�� tt�t
�|�d �tt
�|��d
�}|� ||� d S ) Nr8 ru zB
result = helpers.run_action(
z,
z!,
hideclosed=zB,
)
interpreters.channel_send(zR, result.pending.to_bytes(1, 'little'))
interpreters.channel_send(z2, b'X' if result.closed else b'')
r� )rl rk )rg rW rh r! r` ra r� rH r{ rx rV ry r� r r� r3 r1 r� ri r� �
from_bytesr r� )rI r� rV rv rW r! rz r� r r r r{ � sB
��������
�z!ExhaustiveChannelTests.run_actionc c sF ddddg}|D ]0\}}}dD ] }dD ]}t |||||�V q&q qd S )N)r8 r! ra )r! r8 ra )r� r� ra )r� r� r8 )rY r` r� rl )r� )rI r r! r` ra r� rW r r r �
iter_fixtures� s �z$ExhaustiveChannelTests.iter_fixturesc C s` |rdnd}t ||jd�}|�� s4| j||dd� n(| �tj�� | j||dd� W 5 Q R X d S )Nr_ r^ rY Fru )rS rW r� r{ r� r rQ )rI r� rt r� r^ r r r �_close� s z ExhaustiveChannelTests._closeNc C s� |d ks|j dkr�| �tj�� t�|j� W 5 Q R X | �tj�� t�|jd� W 5 Q R X | �tj�� t�|j� W 5 Q R X | �tj�� tj|jdd� W 5 Q R X n0t|j d� t|j d� t|j d� t|j d� d S ) Nr8 r| Trs zz
with helpers.expect_channel_closed():
interpreters.channel_recv(cid)
z�
with helpers.expect_channel_closed():
interpreters.channel_send(cid, b'spam')
z{
with helpers.expect_channel_closed():
interpreters.channel_close(cid)
z�
with helpers.expect_channel_closed():
interpreters.channel_close(cid, force=True)
)
rH r� r rQ r rx r~ r� r3 r1 )rI r� r! r r r �_assert_closed_in_interp� s z/ExhaustiveChannelTests._assert_closed_in_interpc C s| | � |jj� t|jj�D ]}t�|j� q| �|� dD ]&}|� |�}|j
dkrTq:| �||� q:|� d�}| �||� d S )N)rY r` r8 Zfresh) r� ry rk r rl r r rx r� r� rH )rI r� r� r! r r r �_assert_closed s
z%ExhaustiveChannelTests._assert_closedFc c s� d}| � � D ]�}t� | �� D ]�}|d7 }|dkr: d S |rl|d d dkrTt� t||d�t|��� n2|d d dkr�tddd� td dd� tj�� |||fV q |r td
� q t� d S ) Nr r r� � z ({} actions)� r� re �.z---)r� �printr� rC r r� �stdoutr) )rI �verboser� r� r� r r r �_iter_close_tests s&
z(ExhaustiveChannelTests._iter_close_testsc C s dt _| �� D ]
\}}}qd S )NT)r� r� r� �rI r� r� r� r r r �_skim_close_tests- s z(ExhaustiveChannelTests._skim_close_testsc
C sp | � � D ]b\}}}| �d�|||���6 |�|j� | �||� | j|dd� | �|� W 5 Q R X |�� qd S )N� {} {} {}Frs � r� r� rC r� r! r� r� r� r� r� r r r �
test_close2 s z!ExhaustiveChannelTests.test_closec
C sp | � � D ]b\}}}| �d�|||���6 |�|j� | �||� | j|dd� | �|� W 5 Q R X |�� qd S )Nr� Trs r� r� r r r �test_force_close> s z'ExhaustiveChannelTests.test_force_close)N)F)rL rM rN r� r� r� r� r� r{ r� r� r� r� r� r� r� r� r r r r r� s +-1
r� rS )N)6� collectionsr rb r r rF r� �textwrapr r) rz r �testr Z test.supportr �
import_moduler r r r# rc r/ r3 r0 r: r; rR rS ri r{ rw r� r� ZTestCaser� r� r� r� r� r� r� r� r� r r rd rp r� r� r
r� rL r8 r r r r �<module> sr
&
3$
2?(;]b <Y _ &�q @