File: //usr/local/lib/python3.10/dist-packages/aiohttp/__pycache__/pytest_plugin.cpython-310.pyc
o
���g�1 � @ s> d dl Z d dlZd dlZd dlZd dlmZmZmZmZm Z m
Z
m Z m Z m
Z
mZ d dlZddlmZmZmZmZmZmZmZmZ ddlmZm Z m Z ddl m Z zd dl Z W n e!ya dZ Y nw G dd� de �Z"G d d
� d
e �Z#G d d � d e �Z$d
d� Z%dd� Z&ej'dd� �Z(ej'dd� �Z)ej*dd� �Z+ej*d5dd��Z,dd� Z-d d � Z.d d � Z/ej'd d!� �Z0ej'd"d#� �Z1ej'd$eg e2f d%eg e2f fd&d'��Zej'd%eg e2f fd(d$��Z3ej'd!e j4d%e e# fd)d*��Z5ej'd+d,� �Z6ej'd!e j4d%e e$ fd-d.��Z7ej'd/d0� �Z8ej'd!e j4d%e e" fd1d2��Z9ej'd3d4� �Z:dS )6� N)
�Any� Awaitable�Callable�Dict�Iterator�Optional�Protocol�Type�Union�overload� )�BaseTestServer�
RawTestServer�
TestClient�
TestServer� loop_context�setup_test_loop�teardown_test_loop� unused_port)� Application� BaseRequest�Request)�_RequestHandlerc @ s| e Zd Zedd�dedeeeef dede e
ef fdd��Z edd�de deeeef dede e
df fd d��Z dS )
�
AiohttpClientN��
server_kwargs�_AiohttpClient__paramr �kwargs�returnc � � �d S �N� ��selfr r r r! r! �@/usr/local/lib/python3.10/dist-packages/aiohttp/pytest_plugin.py�__call__( � �zAiohttpClient.__call__c � r r r! r" r! r! r$ r% 0 r& )�__name__�
__module__� __qualname__r r r r �strr r r r% r
r r! r! r! r$ r ' s. ����
�����
�r c
@ �4 e Zd Zdd�dedee dedee fdd�Z dS ) �
AiohttpServerN��port�appr. r r c K � d S r r! )r# r/ r. r r! r! r$ r% ; � zAiohttpServer.__call__)
r' r( r) r r �intr r r r% r! r! r! r$ r, : � �����r, c
@ r+ ) �AiohttpRawServerNr- �handlerr. r r c K r0 r r! )r# r5 r. r r! r! r$ r% A r1 zAiohttpRawServer.__call__)
r' r( r) r r r2 r r r r% r! r! r! r$ r4 @ r3 r4 c C s: | j ddddd� | j dddd d� | j d
ddd d� d S ) N�--aiohttp-fast�
store_trueFz*run tests faster by disabling extra checks)�action�default�helpz--aiohttp-loop�store�pyloopz3run tests with specific loop: pyloop, uvloop or all�--aiohttp-enable-loop-debugz enable event loop debug mode)� addoption)�parserr! r! r$ �pytest_addoptionF s$ ��
�r@ c sd | j � t�� �r d�n
t�� �rd�ndS d�d| jvr%| jd7 _d�� ��fdd�}|| _ dS )zjSet up pytest fixture.
Allow fixtures to be coroutines. Run coroutine fixtures in an event loop.
TFN�request�rA c sz |d }�r |d= d|j vrtd��|�d�� �r3�| i |���� �fdd�}|�|� � ���� �S � ��| i |���S )NrA �loopz^Asynchronous fixtures must depend on the 'loop' fixture or be used in tests depending from it.c s& z� � ��� �W S ty Y d S w r )�run_until_complete� __anext__�StopAsyncIterationr! ��_loop�genr! r$ � finalizer� s
�z8pytest_fixture_setup.<locals>.wrapper.<locals>.finalizer)� fixturenames� Exception�getfixturevalue� addfinalizerrD rE )�argsr rA rJ ��func� is_async_gen�
strip_requestrG r$ �wrapperq s
�
z%pytest_fixture_setup.<locals>.wrapper)rQ �inspect�isasyncgenfunction�asyncio�iscoroutinefunction�argnames)�
fixturedefrT r! rP r$ �pytest_fixture_setup[ s
r[ c C � | j �d�S )z--fast config optionr6 ��config� getoptionrB r! r! r$ �fast� � r` c C r\ )z!--enable-loop-debug config optionr= r] rB r! r! r$ �
loop_debug� ra rb c c sx � t jdd��+} dV dd� | D �}|r*td�t|�t|�dkr"dnd d
�|����W d � dS 1 s5w Y dS ) z�Context manager which checks for RuntimeWarnings.
This exists specifically to
avoid "coroutine 'X' was never awaited" warnings being missed.
If RuntimeWarnings occur in the context a RuntimeError is raised.
T)�recordNc S s" g | ]
}|j tkrd j|d��qS )z#{w.filename}:{w.lineno}:{w.message})�w)�category�RuntimeWarning�format)�.0rd r! r! r$ �
<listcomp>� s
�z,_runtime_warning_context.<locals>.<listcomp>z{} Runtime Warning{},
{}r � �s�
)�warnings�catch_warnings� RuntimeErrorrg �len�join)� _warnings�rwr! r! r$ �_runtime_warning_context� s � � ���"�rt Fc c s, � | r| V dS t � } | V t| |d� dS )z�Passthrough loop context.
Sets up and tears down a loop unless one is passed in via the loop
argument when it's passed straight through.
�r` N)r r )rC r` r! r! r$ �_passthrough_loop_context� s �
rv c C s, | � |�rt�|�rt| �||��S dS dS )z%Fix pytest collecting for coroutines.N)�funcnamefilterrW rX �list�
_genfunctions)� collector�name�objr! r! r$ �pytest_pycollect_makeitem� s �r} c s� � j �d�}t�� j�rc� j�d�p� j�dd�}t� �= t||d�� }� fdd�� j j
D �}|� � j d i |��� W d � n1 sDw Y W d � dS W d � dS 1 s\w Y dS dS )
zBRun coroutines in an event loop instead of a normal function call.r6 �
proactor_looprC Nru c s i | ]}|� j | �qS r! )�funcargs)rh �arg��
pyfuncitemr! r$ �
<dictcomp>� s
��z&pytest_pyfunc_call.<locals>.<dictcomp>Tr! )
r^ r_ rW rX �functionr �getrt rv � _fixtureinforY rD r| )r� r` �
existing_looprH �testargsr! r� r$ �pytest_pyfunc_call� s, � �
� �
��
���r� c C s� d| j vrd S | jjj}dtji}td urtj|d< |dkr d}i }|�d�D ]&}|� d� }|�
d�}||vrG|rFt d |t |�
� �f ��q'|| ||< q'| jdt |�� �t |�
� �d
� d S ) N� loop_factoryr< �uvloop�allzpyloop,uvloop?�,�?z ?z&Unknown loop '%s', available loops: %s)�ids)rK r^ �option� aiohttp_looprW �DefaultEventLoopPolicyr� �EventLoopPolicy�split�endswith�strip�
ValueErrorrx �keys� parametrize�values)�metafunc�loops�avail_factories� factoriesr{ �requiredr! r! r$ �pytest_generate_tests� s0
��
�r� c c s` � | � }t �|� t|d��}|r|�d� t �|� |V W d � dS 1 s)w Y dS )z%Return an instance of the event loop.ru TN)rW �set_event_loop_policyr � set_debug�set_event_loop)r� r` rb �policyrH r! r! r$ rC s �
"�rC c c sT � t �� } t �| � t| j��}t �|� |V W d � d S 1 s#w Y d S r )rW � WindowsProactorEventLoopPolicyr� r �new_event_loopr� )r� rH r! r! r$ r~
s �
"�r~ �aiohttp_unused_portr c C � t jdtdd� | S )Nz3Deprecated, use aiohttp_unused_port fixture instead� ��
stacklevel�rm �warn�DeprecationWarning)r� r! r! r$ r � �r c C s t S )z1Return a port that is unused on the current host.)� _unused_portr! r! r! r$ r� ! s c # �T � g �dd�dt dtt dtdtf� �fdd�}|V d �fd d
� }� �|� � dS ) z^Factory to create a TestServer instance, given an app.
aiohttp_server(app, **kwargs)
Nr- r/ r. r r c � �6 �t | |d�}|jdd� i|��I d H ��|� |S �Nr- rC r! )r � start_server�append)r/ r. r �server�rC �serversr! r$ �go/ �
�
zaiohttp_server.<locals>.goc � �$ �� r� � � �� I d H � sd S d S r ��pop�closer! �r� r! r$ �finalize9 � � �z aiohttp_server.<locals>.finalize�r N)r r r2 r r rD �rC r� r� r! r� r$ �aiohttp_server' � ������r� c C r� )Nz.Deprecated, use aiohttp_server fixture insteadr� r� r� )r� r! r! r$ � test_server@ r� r� c # r� ) zpFactory to create a RawTestServer instance, given a web handler.
aiohttp_raw_server(handler, **kwargs)
Nr- r5 r. r r c � r� r� )r r� r� )r5 r. r r� r� r! r$ r� R r� z aiohttp_raw_server.<locals>.goc � r� r r� r! r� r! r$ r� \ r� z$aiohttp_raw_server.<locals>.finalizer� )r r r2 r r rD r� r! r� r$ �aiohttp_raw_serverJ r� r� c C r� )Nz2Deprecated, use aiohttp_raw_server fixture insteadr� r� r� )r� r! r! r$ �raw_test_serverc s �r� c
# s� � g � t dd�dtdttttf dtdtttf fdd��}t dd�dtdttttf dtdtt df fd d��}dd�dt
ttf d
tdttttf dtdtttf f
� �fd d�}|V d� fd d
� }�� |� � dS )z�Factory to create a TestClient instance.
aiohttp_client(app, **kwargs)
aiohttp_client(server, **kwargs)
aiohttp_client(raw_server, **kwargs)
Nr �__paramr r r c � r r r! �r� r r r! r! r$ r� y � �zaiohttp_client.<locals>.goc � r r r! r� r! r! r$ r� � r� rO c � s� �t | t�rt | ttf�s| �g|�R i |��} i }n|r!J d��t | t�r?|p)i }t| fd�i|��}t|fd�i|��}nt | t�rOt| fd�i|��}ntdt| � ��|�� I d H � � |� |S )Nzargs should be emptyrC zUnknown argument type: %r)
�
isinstancer r r
r r r� �typer� r� )r� r rO r r� �client��clientsrC r! r$ r� � s" � �
c � r� r r� r! )r� r! r$ r� � r� z aiohttp_client.<locals>.finalizer� ) r r r r r* r r r r
r r
rD r� r! r� r$ �aiohttp_cliento sN �����
�����
�
�
����
�r� c C r� )Nz.Deprecated, use aiohttp_client fixture insteadr� r� r� )r� r! r! r$ � test_client� r� r� )F);rW �
contextlibrU rm �typingr r r r r r r r r
r �pytest�
test_utilsr
r r r r r r r r� �webr r r � web_protocolr r� � ImportErrorr r, r4 r@ r[ �fixturer` rb �contextmanagerrt rv r} r� r� rC r~ r2 r� �AbstractEventLoopr� r� r� r� r� r� r! r! r! r$ �<module> sd 0
(
�9
$
=