Operation   : Linux premium131.web-hosting.com 4.18.0-553.44.1.lve.el8.x86_64 #1 SMP Thu Mar 13 14:29:12 UTC 2025 x86_64
Software     : Apache
Server IP    : 162.0.232.56 | Your IP: 216.73.216.111
Domains      : 1034 Domain(s)
Permission   : [ 0755 ]

�
j f�2�@s�dZd
dl
Z
d
dlZd
dlZd
dlZd
dlZd
dlZd
dlZd
dlZd
dl	Z	d
dl
Z
d
dlZd
dlZd
dl
Z
d
dl
mZ
d
dlmZ
d
dlmZmZ
yd
dlZWnek
r



dZYn
XddlmZ
ddlmZ
dd	lmZ
dd
lmZ
ddlmZ
ddlmZ
dd
lmZ
ddlm Z 
e	j!dkr�
ddl"m#Z#
nd
dlm#Z#
dd�Z$dd�Z%ddd�
Z&dd�Z'Gdd�de�Z(Gdd�de�Z)Gdd�d�Z*Gd d!�d!e*e)�Z+d"d#d$d%�
Z,e-ed&�rGd'd(�d(ej.e�Z/Gd)d*�d*e/e�Z0Gd+d,�d,e0�Z1Gd-d.�d.e*e1�Z2d/d0�Z3ej4d1d2��
Z5ej4d"d#d3d4�
�
Z6nej4d5d6d7d
d"d#d8d9��
Z7d:d;�Z8Gd<d=�d=ej9�Z:Gd>d?�d?ej;�Z<d@dA�Z=GdBdC�dCe>�Z?dDdE�Z@GdFdG�dGe
jA�ZAej4dHdI��
ZBejCejDejEdJdK�ZFdLdM�ZGdS)NzUtilities shared by tests.�N)
�mock)
�
HTTPServer)�WSGIRequestHandler�
WSGIServer�
)
�base_events)
�compat)
�events)
�futures)
�	selectors)
�tasks)
�	coroutine)
�logger�win32)
�
socketpaircCs$tdkrdStj
tj�
SdS)
N)�sslZ
SSLContextZPROTOCOL_SSLv23�rr�7/opt/alt/python34/lib64/python3.4/asyncio/test_utils.py�dummy_ssl_context*s

rc

CsVtd
d��
}
|
�}|j
|�
}d|_z|j|�

Wd|j�
XdS)Nc
SsdS)
Nrrrrr�once2szrun_briefly.<locals>.onceF)r
Zcreate_taskZ_log_destroy_pending�run_until_complete�close)�loopr�gen�
trrr�run_briefly1s
	
	

r�cCsztj�|}xc|
�su|dk	rV|tj�}|d
k
rVt
j��
qVn|jtjdd|�

�

qWdS)Nr
g����MbP?r)�timer
�TimeoutErrorrrZsleep)rZpred�timeoutZdeadlinerrr�	run_until@s






r c

Cs|j|j
�

|j�
d
S)z�Legacy API to run once through the event loop.

    This is the recommended pattern for test code.  It will poll the
    selector once and run all callbacks scheduled in response to I/O
    events.
    N)Z	call_soon�stopZrun_forever)
rrrr�run_onceJs
r"c@s(eZ
dZd
d�Zdd�ZdS)�SilentWSGIRequestHandlerc


Cs
tj
�S)
N)�io�StringIO)
�selfrrr�
get_stderrWs
z#SilentWSGIRequestHandler.get_stderrc
GsdS)
Nr)r&�format�argsrrr�log_messageZs
z$SilentWSGIRequestHandler.log_messageN)�__name__�
__module__�__qualname__r'r*rrrrr#Usr#cs4eZ
dZd
Z�f
dd�Zdd�Z�S)�SilentWSGIServer�c
s/t�j
�\}
}|
j|j�

|
|fS)
N)�super�get_request�
settimeout�request_timeout)r&�request�client_addr)
�	__class__rrr1bs


zSilentWSGIServer.get_requestc
CsdS)
Nr)r&r4�client_addressrrr�handle_errorgs
zSilentWSGIServer.handle_error)r+r,r-r3r1r8rr)
r6rr.^sr.c@seZ
dZd
d�ZdS)�SSLWSGIServerMixincCs�tj
jtj
jt�
d
d�}tj
j|�
s`tj
jtj
jtj�
dd�}ntj
j|d�}tj
j|d�}tj|
d|d|d	d
�
}y!|j|||�
|j	�
Wnt
k
r�


Yn
XdS)Nz..ZtestsZtestZtest_asynciozssl_key.pemzssl_cert.pem�keyfile�certfileZserver_sideT)�os�path�join�dirname�__file__�isdirrZwrap_socketZRequestHandlerClassr�OSError)r&r4r7�herer:r;Zssockrrr�finish_requestms$








	




z!SSLWSGIServerMixin.finish_requestN)r+r,r-rDrrrrr9ksr9c
@seZ
dZd
S)�
SSLWSGIServerN)r+r,r-rrrrrE�s
rE�use_sslFc
#s�d
d�}|
r|n|}||t���j
|�

�j�_tjd�f
dd��
}|j�
z	�V
Wd�j�
�j�
|j	�
XdS)NcSs#d
}dg
}|
||�
dg
S)Nz200 OK�Content-type�
text/plainsTest message)rGrHr)�environZstart_responseZstatusZheadersrrr�app�s

	


z_run_test_server.<locals>.app�targetcs�jd
d�
S)NZ
poll_intervalg�������?)
Z
serve_foreverr)
�httpdrr�<lambda>�sz"_run_test_server.<locals>.<lambda>)
r#Zset_appZserver_address�address�	threadingZThread�startZshutdownZserver_closer>)rNrF�
server_cls�server_ssl_clsrJZserver_classZ
server_threadr)
rLr�_run_test_server�s




	




	



rSZAF_UNIXc@seZ
dZd
d�ZdS)�UnixHTTPServerc

Cs&tj
j|�

d
|_d|_dS)Nz	127.0.0.1�P)�socketserver�UnixStreamServer�server_bindZserver_nameZserver_port)
r&rrrrX�s

	
zUnixHTTPServer.server_bindN)r+r,r-rXrrrrrT�srTcs4eZ
dZd
Zdd�Z�f
dd�Z�S)�UnixWSGIServerr/c

Cstj
|�

|j�
dS)
N)rTrXZ
setup_environ)
r&rrrrX�s


zUnixWSGIServer.server_bindc
s/t�j
�\}
}|
j|j�

|
dfS)N�	127.0.0.1�)rZr[)r0r1r2r3)r&r4r5)
r6rrr1�s

zUnixWSGIServer.get_request)r+r,r-r3rXr1rr)
r6rrY�srYc@seZ
dZd
d�ZdS)�SilentUnixWSGIServerc
CsdS)
Nr)r&r4r7rrrr8�s
z!SilentUnixWSGIServer.handle_errorN)r+r,r-r8rrrrr\�sr\c
@seZ
dZd
S)�UnixSSLWSGIServerN)r+r,r-rrrrr]�s
r]c
Cs tj
��}|jSWdQXdS)
N)�tempfileZNamedTemporaryFile�name)
�filerrr�gen_unix_socket_path�s

rac
cs@t�}z	|V
Wdyt
j|�

Wntk
r:


Yn
XXdS)
N)rar<�unlinkrB)
r=rrr�unix_socket_path�s	

	



rcc
cs:t��+}
t
d
|
d|dtdt�DdH
WdQXdS)NrNrFrQrR)rcrSr\r])rFr=rrr�run_test_unix_server�s


rd�hostz	127.0.0.1�portc	cs.td
||
fd|dt
dt�DdH
dS)NrNrFrQrR)rSr.rE)rerfrFrrr�run_test_server�s

rgc
Csqi}
xKt|�
D]=}|j
d
�
r=|jd
�
r=qntdd�
|
|<qWtd|f
|j|
��S)N�__�return_valueZTestProtocol)�dir�
startswith�endswith�MockCallback�type�	__bases__)�baseZdctr_rrr�make_test_protocol�s




rqc@sOeZ
dZd
d�Zddd�
Zdd�Zdd	�Zd
d�ZdS)�TestSelectorc

Cs
i|_dS)
N)
�keys)
r&rrr�__init__�s
zTestSelector.__init__NcCs)tj
|
d
||�}||j|
<|S)Nr
)rZSelectorKeyrs)r&�fileobjr	�data�keyrrr�register�s



zTestSelector.registercCs|jj
|
�
S)
N)rs�pop)r&rurrr�
unregister�s
zTestSelector.unregisterc
CsgS)
Nr)r&rrrr�select
s
zTestSelector.selectc


Cs|jS)
N)
rs)
r&rrr�get_map
s
zTestSelector.get_map)r+r,r-rtrxrzr{r|rrrrrr�s
rrcs�eZ
dZd
Zd�f
dd�
Zdd�Zdd�Z�f
d	d
�Zdd�Zd
d�Z	dd�Z
dd�Zdd�Zdd�Z
dd�Z�f
dd�Z�f
dd�Zdd�Zdd �Z�S)!�TestLoopa�
Loop for unittests.

    It manages self time directly.
    If something scheduled to be executed later then
    on next loop iteration after all ready handlers done
    generator passed to __init__ is calling.

    Generator should be like this:

        def gen():
            ...
            when = yield ...
            ... = yield time_advance

    Value returned by yield is absolute time of next scheduled handler.
    Value passed to yield is time advance to move loop's time forward.
    Ncs�t�j
�
|
dkr1d
d�}
d|_n	d|_|
�|_t|j�

d|_d|_g|_t�|_	i|_
i|_|j�
dS)Nc
ss	dV
dS)
Nrrrrrr
s
zTestLoop.__init__.<locals>.genFTr
g��&�.>)
r0rt�_check_on_close�_gen�next�_timeZ_clock_resolution�_timersrrZ	_selector�readers�writers�reset_counters)r&r)
r6rrrt
s


	


	
	
	
	
	
zTestLoop.__init__c


Cs|jS)
N)
r�)
r&rrrr/
s
z
TestLoop.timecCs|
r|j|
7_nd
S)zMove test time forward.N)
r�)r&�advancerrr�advance_time2
s
zTestLoop.advance_timec

sRt�j
�
|jrNy|jjd
�

Wntk
r>


YqNXtd�
�
ndS)Nr
zTime generator is not finished)r0rr~r�send�
StopIteration�AssertionError)
r&)
r6rrr7
s


	




zTestLoop.closecGs tj
|||�|j|
<dS)
N)r	�Handler�)r&�fd�callbackr)rrr�
add_readerA
s
zTestLoop.add_readercCs6|j|
d
7<|
|j
kr.|j
|
=dSdSdS)NrTF)�remove_reader_countr�)r&r�rrr�
remove_readerD
s





zTestLoop.remove_readercGs�|
|jks$t
d
j|
�
�
�
|j|
}|j|ks[t
dj|j|��
�
|j|ks�t
dj|j|��
�
dS)Nzfd {} is not registeredz{!r} != {!r})r�r�r(�	_callback�_args)r&r�r�r)�handlerrr�
assert_readerL
s
$





zTestLoop.assert_readercGs tj
|||�|j|
<dS)
N)r	r�r�)r&r�r�r)rrr�
add_writerT
s
zTestLoop.add_writercCs6|j|
d
7<|
|j
kr.|j
|
=dSdSdS)NrTF)�remove_writer_countr�)r&r�rrr�
remove_writerW
s





zTestLoop.remove_writercGs�|
|jks$t
d
j|
�
�
�
|j|
}|j|ks[t
dj|j|��
�
|j|ks�t
dj|j|��
�
dS)Nzfd {} is not registeredz{!r} != {!r})r�r�r(r�r�)r&r�r�r)r�rrr�
assert_writer_
s
$





zTestLoop.assert_writerc

Cs(tj
t�
|_tj
t�
|_dS)
N)�collections�defaultdict�intr�r�)
r&rrrr�g
s

zTestLoop.reset_countersc
sMt�j
�
x0|jD]%}
|jj|
�
}|j|�

qWg|_dS)
N)r0�	_run_oncer�rr�r�)r&�whenr�)
r6rrr�k
s






zTestLoop._run_oncecs&|jj
|
�

t�j|
||�S)
N)r��appendr0�call_at)r&r�r�r))
r6rrr�r
s

zTestLoop.call_atc
CsdS)
Nr)r&Z
event_listrrr�_process_eventsv
s
zTestLoop._process_eventsc


CsdS)
Nr)
r&rrr�_write_to_selfy
s
zTestLoop._write_to_self)r+r,r-�__doc__rtrr�rr�r�r�r�r�r�r�r�r�r�r�rr)
r6rr}
s 
r}c
Kstj
d
dg
|�
S)N�spec�__call__)rZMock)
�kwargsrrrrm}
s
rmc@s"eZ
dZd
Zdd�ZdS)�MockPatternz�A regex based str with a fuzzy __eq__.

    Use this helper with 'mock.assert_called_with', or anywhere
    where a regex comparison between strings is needed.

    For instance:
       mock_call.assert_called_with(MockPattern('spam.*ham'))
    cCs"tt
jt|�
|
t
j��
S)
N)�bool�re�search�str�
S)r&�otherrrr�__eq__�
s
zMockPattern.__eq__N)r+r,r-r�r�rrrrr��
s
r�c
Cs5tj
|�
}
|
dkr1td
|f
�
�
n|
S)Nzunable to get the source of %r)r	Z_get_function_source�
ValueError)�func�sourcerrr�get_function_source�
s



r�c@sUeZ
dZd
ddd�
Zddd�
Zdd	�ZejsQd
d�ZndS)�TestCase�cleanupTc
Cs<|
dk	st�
t
jd�

|r8|j|
j�

ndS)
N)r�r	�set_event_loopZ
addCleanupr)r&rr�rrrr��
s



zTestCase.set_event_loopNcCst|
�
}|j
|�

|S)
N)r}r�)r&rrrrr�
new_test_loop�
s



zTestCase.new_test_loopc

Cs'tj
d�

|jtj�d
�
dS)N)NNN)r	r�ZassertEqual�sys�exc_info)
r&rrr�tearDown�
s

zTestCase.tearDownc
OsGd
d�d�}|�S)Nc@s(eZ
dZd
d�Zdd�ZdS)z!TestCase.subTest.<locals>.EmptyCMc


SsdS)
Nr)
r&rrr�	__enter__�
s
z+TestCase.subTest.<locals>.EmptyCM.__enter__c

WsdS)
Nr)r&�excrrr�__exit__�
s
z*TestCase.subTest.<locals>.EmptyCM.__exit__N)r+r,r-r�r�rrrr�EmptyCM�
s
r�r)r&r)r�r�rrr�subTest�
s
zTestCase.subTest)	r+r,r-r�r�r�rZPY34r�rrrrr��
s

	r�c
cs;tj
}ztjtjd
�

dV
Wdtj|�

XdS)zrContext manager to disable asyncio logger.

    For example, it can be used to ignore warnings in debug mode.
    rN)r�levelZsetLevel�loggingZCRITICAL)
Z	old_levelrrr�disable_logger�
s
	


	r�cCs=tj
tj�
}||_|
|_||_d
|j_|S)z'Create a mock of a non-blocking socket.g)rZ	MagicMock�socket�protorn�familyZ
gettimeoutri)r�rnr�Zsockrrr�mock_nonblocking_socket�
s
	
	
	

r�cCstj
d
dd�

S)Nz'asyncio.sslproto._is_sslproto_availableriF)rZpatchrrrr�force_legacy_ssl_support�
s

r�)Hr�r��
contextlibr$r�r<r�r�rVr�r^rOrZunittestrZhttp.serverrZwsgiref.simple_serverrrr�ImportErrorr[rrr	r
rrZ
coroutinesr
�logr�platformZ
windows_utilsrrrr r"r#r.r9rErS�hasattrrWrTrYr\r]ra�contextmanagerrcrdrgrqZBaseSelectorrrZ
BaseEventLoopr}rmr�r�r�r�r�ZIPPROTO_TCPZSOCK_STREAMZAF_INETr�r�rrrr�<module>
s|


























	
	
	

v