Operation   : Linux premium88.web-hosting.com 4.18.0-553.54.1.lve.el8.x86_64 #1 SMP Wed Jun 4 13:01:13 UTC 2025 x86_64
Software     : Apache
Server IP    : 185.61.154.216 | Your IP: 216.73.216.44
Domains      : 1034 Domain(s)
Permission   : [ 0755 ]

3


 \�:�
@sdZd
dl
Z
d
dlZd
dlZd
dlZd
dlZd
dlZd
dlZd
dlZd
dl	Z	d
dl
Z
d
dlZd
dlZd
dl
Z
d
dlZd
dl
mZ
d
dlmZ
d
dlmZmZ
yd
dlZWnek
r�


dZYnXddlmZ
ddlmZ
dd	lmZ
dd
lmZ
ddlmZ
ddlmZ
dd
lmZ
ddl m!Z!
d
dl"m#Z#
e	j$dk�
rHddl%m&Z&
nd
dlm&Z&
dd�Z'e'd�
Z(e'd�
Z)dd�Z*dd�Z+dRdd�
Z,dd�Z-Gdd �d e�Z.Gd!d"�d"e�Z/Gd#d$�d$�Z0Gd%d&�d&e0e/�Z1d'd(�
d)d*�Z2e3ed+��rZGd,d-�d-ej4e�Z5Gd.d/�d/e5e�Z6Gd0d1�d1e6�Z7Gd2d3�d3e0e7�Z8d4d5�Z9ej:d6d7��
Z;ej:d'd(�
d8d9��
Z<ej:d:d
d'd;�d<d=��
Z=d>d?�Z>Gd@dA�dAej?�Z@GdBdC�dCejA�ZBdDdE�ZCGdFdG�dGeD�ZEdHdI�ZFGdJdK�dKe
jG�ZGej:dLdM��
ZHejIejJejKfdNdO�
ZLdPdQ�ZMdS)SzUtilities shared by tests.�N)
�mock)
�
HTTPServer)�WSGIRequestHandler�
WSGIServer�
)
�base_events)
�compat)
�events)
�futures)
�	selectors)
�tasks)
�	coroutine)
�logger)
�supportZwin32)
�
socketpairc
Cs`tt
d
�r*tjjt
j|�}
tjj|
�
r*|
Stjjtjjtj�
d|�}
tjj|
�
rT|
St	|�
�
dS)N�
TEST_HOME_DIR�test)
�hasattrr�os�path�joinr�isfile�dirname�__file__�FileNotFoundError)�filename�fullname�r�*/usr/lib64/python3.6/asyncio/test_utils.py�	data_file-s








rzssl_cert.pemzssl_key.pemcCstdkrdStj
tj�
SdS)
N)�ssl�
SSLContextZPROTOCOL_SSLv23rrrr�dummy_ssl_context<s

r"c

Cs@td
d��
}
|
�}|j
|�
}d|_z|j|�

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



r)�cCsTtj�|}xB|
�sN|dk	r8|tj�}|d
k
r8t
j��
|jtjd|d��

qWdS)Nr
g����MbP?)
r&)�timer
�TimeoutErrorr$rZsleep)r&Zpred�timeoutZdeadlinerrr�	run_untilRs






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)
r&rrr�run_once\s
r0c@seZ
dZd
d�Zdd�ZdS)�SilentWSGIRequestHandlerc


Cstj
�S)
N)�io�StringIO)
�selfrrr�
get_stderris
z#SilentWSGIRequestHandler.get_stderrc
GsdS)
Nr)r4�format�argsrrr�log_messagels
z$SilentWSGIRequestHandler.log_messageN)�__name__�
__module__�__qualname__r5r8rrrrr1gsr1cs(eZ
dZd
Z�f
dd�Zdd�Z�ZS)�SilentWSGIServer�c
s"t�j
�\}
}|
j|j�

|
|fS)
N)�super�get_request�
settimeout�request_timeout)r4�request�client_addr)
�	__class__rrr?ts


zSilentWSGIServer.get_requestc
CsdS)
Nr)r4rB�client_addressrrr�handle_errorys
zSilentWSGIServer.handle_error)r9r:r;rAr?rF�
__classcell__rr)
rDrr<psr<c@seZ
dZd
d�ZdS)�SSLWSGIServerMixincCs^t}t
}tj�}|j||�
|j|
d
d�}y|j|||�
|j�
Wntk
rX


YnXdS)NT)
Zserver_side)	�ONLYKEY�ONLYCERTr r!Zload_cert_chainZwrap_socketZRequestHandlerClassr%�OSError)r4rBrEZkeyfileZcertfile�contextZssockrrr�finish_requests






z!SSLWSGIServerMixin.finish_requestN)r9r:r;rMrrrrrH}srHc
@seZ
dZd
S)�
SSLWSGIServerN)r9r:r;rrrrrN�s
rNF)
�use_sslc
#svd
d�}|
r|n|}||t���j
|�

�j�_tj�f
dd�d�
}|j�
z
�V
Wd�j�
�j�
|j	�
XdS)NcSsd
}dg
}|
||�
dg
S)Nz200 OK�Content-type�
text/plainsTest message)rPrQr)�environZstart_responseZstatusZheadersrrr�app�s




z_run_test_server.<locals>.appcs�jd
d�
S)Ng�������?)
Z
poll_interval)
Z
serve_foreverr)
�httpdrr�<lambda>�sz"_run_test_server.<locals>.<lambda>)
�target)
r1Zset_appZserver_address�address�	threadingZThread�start�shutdownZserver_closer)rWrO�
server_cls�server_ssl_clsrSZserver_classZ
server_threadr)
rTr�_run_test_server�s












r]ZAF_UNIXc@seZ
dZd
d�ZdS)�UnixHTTPServerc

Cstj
j|�

d
|_d|_dS)Nz	127.0.0.1�P)�socketserver�UnixStreamServer�server_bindZserver_nameZserver_port)
r4rrrrb�s


zUnixHTTPServer.server_bindN)r9r:r;rbrrrrr^�sr^cs(eZ
dZd
Zdd�Z�f
dd�Z�ZS)�UnixWSGIServerr=c

Cstj
|�

|j�
dS)
N)r^rbZ
setup_environ)
r4rrrrb�s


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

|
dfS)N�	127.0.0.1�)rdre)r>r?r@rA)r4rBrC)
rDrrr?�s

zUnixWSGIServer.get_request)r9r:r;rArbr?rGrr)
rDrrc�srcc@seZ
dZd
d�ZdS)�SilentUnixWSGIServerc
CsdS)
Nr)r4rBrErrrrF�s
z!SilentUnixWSGIServer.handle_errorN)r9r:r;rFrrrrrf�srfc
@seZ
dZd
S)�UnixSSLWSGIServerN)r9r:r;rrrrrg�s
rgc
	Cstj
��}|jSQRXdS)
N)�tempfileZNamedTemporaryFile�name)
�filerrr�gen_unix_socket_path�s


rkc
cs<t�}z
|V
Wdyt
j|�

Wntk
r4


YnXXdS)
N)rkr�unlinkrK)
rrrr�unix_socket_path�s





rmc

cs,t��}
t
|
|ttd
�EdH
WdQRXdS)N)rWrOr[r\)rmr]rfrg)rOrrrr�run_test_unix_server�s


rnz	127.0.0.1)�host�portrOccst||
f|t
td
�EdH
dS)N)rWrOr[r\)r]r<rN)rorprOrrr�run_test_server�s


rqc
CsPi}
x4t|�
D](}|j
d
�
r(|jd
�
r(qtdd�
|
|<qWtd|f
|j|
��S)N�__)
�return_valueZTestProtocol)�dir�
startswith�endswith�MockCallback�type�	__bases__)�baseZdctrirrr�make_test_protocol�s




r{c@s6eZ
dZd
d�Zddd�
Zdd�Zdd	�Zd
d�ZdS)
�TestSelectorc

Cs
i|_dS)
N)
�keys)
r4rrr�__init__
s
zTestSelector.__init__NcCstj
|
d
||�}||j|
<|S)Nr
)rZSelectorKeyr})r4�fileobjr	�data�keyrrr�register
s



zTestSelector.registercCs|jj
|
�
S)
N)r}�pop)r4rrrr�
unregister
s
zTestSelector.unregisterc
CsgS)
Nr)r4r-rrr�select
s
zTestSelector.selectc


Cs|jS)
N)
r})
r4rrr�get_map
s
zTestSelector.get_map)
N)r9r:r;r~r�r�r�r�rrrrr|

s

r|cs�eZ
dZd
Zd-�f
dd�	Zdd�Zdd�Z�f
d	d
�Zdd�Zd
d�Z	dd�Z
dd�Zdd�Zdd�Z
dd�Zdd�Zdd�Zdd�Zdd �Zd!d"�Zd#d$�Z�f
d%d&�Z�f
d'd(�Zd)d*�Zd+d,�Z�ZS).�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.
    Ncsvt�j
�
|
dkr"d
d�}
d|_nd|_|
�|_t|j�

d|_d|_g|_t�|_	i|_
i|_|j�
t
j�|_dS)Nc
ss
dV
dS)
Nrrrrrr',
s
zTestLoop.__init__.<locals>.genFTr
g��&�.>)r>r~�_check_on_close�_gen�next�_timeZ_clock_resolution�_timersr|Z	_selector�readers�writers�reset_counters�weakref�WeakValueDictionary�_transports)r4r')
rDrrr~(
s










zTestLoop.__init__c


Cs|jS)
N)
r�)
r4rrrr+?
s
z
TestLoop.timecCs|
r|j|
7_d
S)zMove test time forward.N)
r�)r4�advancerrr�advance_timeB
s
zTestLoop.advance_timec

sBt�j
�
|jr>y|jjd
�

Wntk
r4


Yn
Xtd�
�
dS)Nr
zTime generator is not finished)r>r%r�r��send�
StopIteration�AssertionError)
r4)
rDrrr%G
s






zTestLoop.closecGstj
|||�|j|
<dS)
N)r	�Handler�)r4�fd�callbackr7rrr�_add_readerQ
s
zTestLoop._add_readercCs0|j|
d
7<|
|j
kr(|j
|
=dSdSdS)NrTF)�remove_reader_countr�)r4r�rrr�_remove_readerT
s





zTestLoop._remove_readercGsh|
|jkrt
d
|
�d��
�
|j|
}|j|krDt
d|j�d|���
�
|j|krdt
d|j�d|���
�
dS)Nzfd z is not registeredzunexpected callback: z != zunexpected callback args: )r�r��	_callback�_args)r4r�r�r7�handlerrr�
assert_reader\
s












zTestLoop.assert_readercCs|
|jkrt
d
|
�d��
�
dS)Nzfd z is registered)r�r�)r4r�rrr�assert_no_readerg
s


zTestLoop.assert_no_readercGstj
|||�|j|
<dS)
N)r	r�r�)r4r�r�r7rrr�_add_writerk
s
zTestLoop._add_writercCs0|j|
d
7<|
|j
kr(|j
|
=dSdSdS)NrTF)�remove_writer_countr�)r4r�rrr�_remove_writern
s





zTestLoop._remove_writercGs^|
|jkst
d
j|
�
�
�
|j|
}|j|ks>t
dj|j|��
�
|j|ksZt
dj|j|��
�
dS)Nzfd {} is not registeredz{!r} != {!r})r�r�r6r�r�)r4r�r�r7r�rrr�
assert_writerv
s






zTestLoop.assert_writercCs8y|j|
}Wnt
k
r"


YnXtd
j|
|��
�
dS)Nz.File descriptor {!r} is used by transport {!r})r��KeyError�RuntimeErrorr6)r4r�Z	transportrrr�_ensure_fd_no_transport~
s





z TestLoop._ensure_fd_no_transportcGs|j|
�

|j
|
|f|��S)
zAdd a reader callback.)r�r�)r4r�r�r7rrr�
add_reader�
s

zTestLoop.add_readercCs|j|
�

|j
|
�
S)
zRemove a reader callback.)r�r�)r4r�rrr�
remove_reader�
s

zTestLoop.remove_readercGs|j|
�

|j
|
|f|��S)
zAdd a writer callback..)r�r�)r4r�r�r7rrr�
add_writer�
s

zTestLoop.add_writercCs|j|
�

|j
|
�
S)
zRemove a writer callback.)r�r�)r4r�rrr�
remove_writer�
s

zTestLoop.remove_writerc

Cstj
t�
|_tj
t�
|_dS)
N)�collections�defaultdict�intr�r�)
r4rrrr��
s

zTestLoop.reset_countersc
s:t�j
�
x$|jD]}
|jj|
�
}|j|�

qWg|_dS)
N)r>�	_run_oncer�r�r�r�)r4�whenr�)
rDrrr��
s






zTestLoop._run_oncecs |jj
|
�

t�j|
|f|��S)
N)r��appendr>�call_at)r4r�r�r7)
rDrrr��
s

zTestLoop.call_atc
CsdS)
Nr)r4Z
event_listrrr�_process_events�
s
zTestLoop._process_eventsc


CsdS)
Nr)
r4rrr�_write_to_self�
s
zTestLoop._write_to_self)
N)r9r:r;�__doc__r~r+r�r%r�r�r�r�r�r�r�r�r�r�r�r�r�r�r�r�r�rGrr)
rDrr�
s,

r�c
Kstj
fd
dg
i
|��
S)N�spec�__call__)rZMock)
�kwargsrrrrw�
s
rwc@seZ
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'))
    cCstt
jt|�
|
t
j��
S)
N)�bool�re�search�str�
S)r4�otherrrr�__eq__�
s
zMockPattern.__eq__N)r9r:r;r�r�rrrrr��
s
r�c
Cs$tj
|�
}
|
dkr td
|f
�
�
|
S)Nzunable to get the source of %r)r	Z_get_function_source�
ValueError)�func�sourcerrr�get_function_source�
s




r�c@sVeZ
dZed
d��
Zdd�
dd�Zddd	�
Zd
d�Zdd
�Zdd�Z	e
jsRdd�ZdS)�TestCasec
Cs&|j}
|
dk	r|
j
d
d�

|j�
dS)NT)
�wait)Z_default_executorrZr%)r&Zexecutorrrr�
close_loop�
s


zTestCase.close_loopT)
�cleanupc
Cs,|
dk	st�
t
jd�

|r(|j|j|
�
dS)
N)r�r	�set_event_loopZ
addCleanupr�)r4r&r�rrrr��
s



zTestCase.set_event_loopNcCst|
�
}|j
|�

|S)
N)r�r�)r4r'r&rrr�
new_test_loop�
s



zTestCase.new_test_loopc

Cs|jt
_dS)
N)�_get_running_loopr	)
r4rrr�unpatch_get_running_loop�
s
z!TestCase.unpatch_get_running_loopc

Cs tj
|_
d
d�t_
tj�|_dS)Nc
SsdS)
NrrrrrrU�
sz TestCase.setUp.<locals>.<lambda>)r	r�rZthreading_setup�_thread_cleanup)
r4rrr�setUp�
s



zTestCase.setUpc

CsB|j�
t
jd�

|jtj�d
�
|j�
tj|j	�
tj
�
dS)N)NNN)r�r	r�ZassertEqual�sys�exc_infoZ
doCleanupsrZthreading_cleanupr�Z
reap_children)
r4rrr�tearDown�
s


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


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

WsdS)
Nr)r4�excrrr�__exit__�
s
z*TestCase.subTest.<locals>.EmptyCM.__exit__N)r9r:r;r�r�rrrr�EmptyCM�
s
r�r)r4r7r�r�rrr�subTest�
s
zTestCase.subTest)
N)
r9r:r;�staticmethodr�r�r�r�r�r�rZPY34r�rrrrr��
s


r�c
cs2tj
}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�protorx�familyZ
gettimeoutrs)r�rxr�Zsockrrr�mock_nonblocking_sockets




r�cCstj
d
dd�S)Nz'asyncio.sslproto._is_sslproto_availableF)
rs)rZpatchrrrr�force_legacy_ssl_supports

r�)
r*)Nr�r��
contextlibr2r�rr�r�r`r�rhrXr+Zunittestr�rZhttp.serverrZwsgiref.simple_serverrrr �ImportErrorrerrr	r
rrZ
coroutinesr
�logrrr�platformZ
windows_utilsrrrJrIr"r)r.r0r1r<rHrNr]rrar^rcrfrgrk�contextmanagerrmrnrqr{ZBaseSelectorr|Z
BaseEventLoopr�rwr�r�r�r�r�ZIPPROTO_TCPZSOCK_STREAMZAF_INETr�r�rrrr�<module>
s�





























	




4