o
    Fhe2                  
   @   sh  d dl Z d dlZd dlZd dlZd dlmZmZmZmZm	Z	m
Z
mZmZmZ d dlZddlmZmZmZmZmZmZmZmZ ddlmZmZmZ ddlmZ zd dlZW n e y_   dZY nw G dd deZ!G d	d
 d
eZ"G dd deZ#dd Z$dd Z%ej&dd Z'ej&dd Z(ej)dd Z*ej)d6ddZ+dd Z,dd Z-dd Z.ej&d eg e j/f de0de0d!e	e j/ fd"d#Z1ej&d!e	e j/ fd$d%Z2ej&d&eg e3f d!eg e3f fd'd(Zej&d!eg e3f fd)d&Z4ej&d#e j/d!e	e" fd*d+Z5ej&d,d- Z6ej&d#e j/d!e	e# fd.d/Z7ej&d0d1 Z8ej&d#e j/d!e	e! fd2d3Z9ej&d4d5 Z:dS )7    N)	Any	AwaitableCallableDictIteratorOptionalProtocolUnionoverload   )BaseTestServerRawTestServer
TestClient
TestServerloop_contextsetup_test_loopteardown_test_loopunused_port)ApplicationBaseRequestRequest)_RequestHandlerc                   @   s|   e Zd Zedddedeeeef  dede	e
ef fddZedddedeeeef  dede	edf fd	dZdS )
AiohttpClientNserver_kwargs_AiohttpClient__paramr   kwargsreturnc                      d S N selfr   r   r   r    r    H/var/www/html/venv/lib/python3.10/site-packages/aiohttp/pytest_plugin.py__call__'      zAiohttpClient.__call__c                   r   r   r    r!   r    r    r#   r$   /   r%   )__name__
__module____qualname__r
   r   r   r   strr   r   r   r$   r   r   r    r    r    r#   r   &   s.    

r   c                
   @   4   e Zd Zdddedee dedee fddZ	dS )	AiohttpServerNportappr-   r   r   c                K      d S r   r    )r"   r.   r-   r   r    r    r#   r$   :      zAiohttpServer.__call__)
r&   r'   r(   r   r   intr   r   r   r$   r    r    r    r#   r+   9       r+   c                
   @   r*   )	AiohttpRawServerNr,   handlerr-   r   r   c                K   r/   r   r    )r"   r4   r-   r   r    r    r#   r$   @   r0   zAiohttpRawServer.__call__)
r&   r'   r(   r   r   r1   r   r   r   r$   r    r    r    r#   r3   ?   r2   r3   c                 C   s:   | j ddddd | j dddd	d | j d
ddd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_addoptionE   s$   
r?   c                    sd   | j  t rdn
t rdndS dd| jvr%|  jd7  _d fdd}|| _ dS )zjSet up pytest fixture.

    Allow fixtures to be coroutines. Run coroutine fixtures in an event loop.
    TFNrequestr@   c                     sz   |d }r	|d= d|j vrtd|d r3| i | fdd}||   S  | i |S )Nr@   loopz^Asynchronous fixtures must depend on the 'loop' fixture or be used in tests depending from it.c                      s&   z   W S  ty   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addfinalizerrC   rD   )argsr   r@   rI   funcis_async_genstrip_requestrF   r#   wrapperp   s   


z%pytest_fixture_setup.<locals>.wrapper)rP   inspectisasyncgenfunctioniscoroutinefunctionargnames)
fixturedefrS   r    rO   r#   pytest_fixture_setupZ   s   



 rY   c                 C      | j dS )z--fast config optionr5   config	getoptionrA   r    r    r#   fast      r^   c                 C   rZ   )z!--enable-loop-debug config optionr<   r[   rA   r    r    r#   
loop_debug   r_   r`   c               	   c   sx    t jdd+} dV  dd | D }|r*tdt|t|dkr"dnd	d
|W d   dS 1 s5w   Y  dS )zContext 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 tkrd j|dqS )z#{w.filename}:{w.lineno}:{w.message})w)categoryRuntimeWarningformat).0rb   r    r    r#   
<listcomp>   s
    

z,_runtime_warning_context.<locals>.<listcomp>z{} Runtime Warning{},
{}r    s
)warningscatch_warningsRuntimeErrorre   lenjoin)	_warningsrwr    r    r#   _runtime_warning_context   s   	 "rr   Fc                 c   s,    | r| V  dS t  } | V  t| |d dS )zPassthrough 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   )rB   r^   r    r    r#   _passthrough_loop_context   s   
rt   c                 C   s,   |  |rt|rt| ||S dS dS )z%Fix pytest collecting for coroutines.N)funcnamefilterrT   rV   list_genfunctions)	collectornameobjr    r    r#   pytest_pycollect_makeitem   s   r{   c              	      s    j d}t jro jdp$ jdp$ jdp$ jdd}t = t||d} fdd	 j	j
D }| jdi | W d   n1 sPw   Y  W d   d
S W d   d
S 1 shw   Y  d
S dS )zBRun coroutines in an event loop instead of a normal function call.r5   proactor_loopselector_loopuvloop_looprB   Nrs   c                    s   i | ]}| j | qS r    )funcargs)rf   arg
pyfuncitemr    r#   
<dictcomp>   s    
z&pytest_pyfunc_call.<locals>.<dictcomp>Tr    )r\   r]   rT   rV   functionr   getrr   rt   _fixtureinforW   rC   rz   )r   r^   existing_looprG   testargsr    r   r#   pytest_pyfunc_call   s0   




r   c                 C   s   d| j vrd S | jjj}dtji}td urtj|d< |dkr d}i }|dD ]&}|d }|	d}||vrG|rFt
d	|t| f q'|| ||< q'| jdt| t| d
 d S )Nloop_factoryr;   uvloopallzpyloop,uvloop?,?z ?z&Unknown loop '%s', available loops: %s)ids)rJ   r\   optionaiohttp_loopasyncionew_event_loopr   splitendswithstrip
ValueErrorrv   keysparametrizevalues)metafuncloopsavail_factories	factoriesry   requiredr    r    r#   pytest_generate_tests   s0   





r   r   r   c                 c   sR    t | |d}|r|d t| |V  W d   dS 1 s"w   Y  dS )z%Return an instance of the event loop.rs   TN)r   	set_debugr   set_event_loop)r   r^   r`   rG   r    r    r#   rB     s   

"rB   c                  c   sF    t j} t| }t | |V  W d    d S 1 sw   Y  d S r   )r   ProactorEventLoopr   r   )factoryrG   r    r    r#   r|     s   

"r|   aiohttp_unused_portc                 C      t jdtdd | S )Nz3Deprecated, use aiohttp_unused_port fixture instead   
stacklevelrk   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                 #   sZ    g ddddt dtdtt dtdtf
 fd	d
}|V  dfdd} |  dS )z^Factory to create a TestServer instance, given an app.

    aiohttp_server(app, **kwargs)
    z	127.0.0.1Nhostr-   r.   r   r-   r   r   c                   s8   t | ||d}|jdd i|I d H  | |S )Nr   rB   r    )r   start_serverappend)r.   r   r-   r   serverrB   serversr    r#   go3  s
   
zaiohttp_server.<locals>.goc                      $    r    I d H   sd S d S r   popcloser    r   r    r#   finalizeA     z aiohttp_server.<locals>.finalizer   N)r   r)   r   r1   r   r   rC   rB   r   r   r    r   r#   aiohttp_server+  s$   r   c                 C   r   )Nz.Deprecated, use aiohttp_server fixture insteadr   r   r   )r   r    r    r#   test_serverH  r   r   c              	   #   sT    g dddt dtt dtdtf fdd}|V  dfd	d
} |  dS )zpFactory to create a RawTestServer instance, given a web handler.

    aiohttp_raw_server(handler, **kwargs)
    Nr,   r4   r-   r   r   c                   s6   t | |d}|jdd i|I d H  | |S )Nr,   rB   r    )r   r   r   )r4   r-   r   r   r   r    r#   r   Z  s
   
zaiohttp_raw_server.<locals>.goc                      r   r   r   r    r   r    r#   r   d  r   z$aiohttp_raw_server.<locals>.finalizer   )r   r   r1   r   r   rC   r   r    r   r#   aiohttp_raw_serverR  s   r   c                 C   r   )Nz2Deprecated, use aiohttp_raw_server fixture insteadr   r   r   )r   r    r    r#   raw_test_serverk  s   r   c                 #   s    g  t dddtdttttf  dtdtttf fdd}t dddtdttttf  dtdtt	df fd	d}dddt
ttf d
tdttttf  dtdtttf f
 fdd}|V  d fdd}|  dS )zFactory 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        zaiohttp_client.<locals>.goc                   r   r   r    r   r    r    r#   r     r   rN   c                   s   t | trt | ttfs| g|R i |} i }n|r!J dt | tr?|p)i }t| fdi|}t|fdi|}nt | trOt| fdi|}ntdt|  | I d H   	| |S )Nzargs should be emptyrB   zUnknown argument type: %r)

isinstancer   r   r   r   r   r   typer   r   )r   r   rN   r   r   clientclientsrB   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	   rC   r   r    r   r#   aiohttp_clientw  sN   




r   c                 C   r   )Nz.Deprecated, use aiohttp_client fixture insteadr   r   r   )r   r    r    r#   test_client  r   r   )F);r   
contextlibrT   rk   typingr   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+   r3   r?   rY   fixturer^   r`   contextmanagerrr   rt   r{   r   r   AbstractEventLoopboolrB   r|   r1   r   r   r   r   r   r   r   r    r    r    r#   <module>   st    ,(
9


 $	
	
=