U
'7`+ @ s d dl Z d dlZd dlZd dlmZ d dlZd dlmZmZ d dl m
Z
ddl m Z m
Z
mZmZmZmZmZmZ z d dlZW n ek
r dZY nX z d dlZW n ek
r dZY nX dd Zd d
Zejd d Zejd
d Z ej dd Z ej d.ddZ dd Z dd Z!dd Z"ejdd Z#ejd d Z$ejd d Zejd d! Z%ejd"d# Z&ejd$d% Z'ejd&d' Z(ejd(d) Z)ejd*d+ Z*ejd,d- Z+dS )/ N)Callable)PY_37isasyncgenfunction) Application )BaseTestServer
RawTestServer
TestClient
TestServer loop_contextsetup_test_loopteardown_test_loop unused_portc 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)actiondefaulthelpz--aiohttp-loopZstorepyloopz:run tests with specific loop: pyloop, uvloop, tokio or all--aiohttp-enable-loop-debugz enable event loop debug mode)Z addoption)parser r H/opt/alt/python38/lib64/python3.8/site-packages/aiohttp/pytest_plugin.pypytest_addoption! s$ r c sb | j t rdnt r$dndS dd| jkrH| jd7 _d fdd}|| _ dS )zS
Allow fixtures to be coroutines. Run coroutine fixtures in an event loop.
TFNrequestr c sv |d }r|d= d|j kr$td|d rb| | fdd}|| S | |S d S )Nr loopz^Asynchronous fixtures must depend on the 'loop' fixture or be used in tests depending from it.c s* z W S tk
r$ Y nX d S N)run_until_complete __anext__StopAsyncIterationr _loopgenr r finalizer` s z8pytest_fixture_setup.<locals>.wrapper.<locals>.finalizer) fixturenames ExceptionZgetfixturevalueZ addfinalizerr r )argskwargsr r$ funcZ is_async_genZ
strip_requestr! r wrapperK s
z%pytest_fixture_setup.<locals>.wrapper)r* r asyncioiscoroutinefunctionargnames)Z
fixturedefr+ r r) r pytest_fixture_setup6 s
r/ c C s | j dS )z--fast config optionr config getoptionr r r r fastn s r3 c C s | j dS )z!--enable-loop-debug config optionr r0 r r r r
loop_debugt s r4 c c s` t jddJ} dV dd | D }|rRtdt|t|dkrBdnd d
|W 5 Q R X dS ) z
Context manager which checks for RuntimeWarnings, 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|dqS )z#{w.filename}:{w.lineno}:{w.message})w)categoryRuntimeWarningformat).0r6 r r r
<listcomp> s
z,_runtime_warning_context.<locals>.<listcomp>z{} Runtime Warning{},
{}r s
)warningscatch_warnings RuntimeErrorr9 lenjoin) _warningsZrwr r r _runtime_warning_contextz s rE Fc c s( | r | V nt } | V t| |d dS )z{
setups and tears down a loop unless one is passed in via the loop
argument when it's passed straight through.
r3 N)r r
)r r3 r r r _passthrough_loop_context s
rG c C s( | |r$t|r$t| ||S dS )z/
Fix pytest collecting for coroutines.
N)Zfuncnamefilterr, r- listZ
_genfunctions)Z collectornameobjr r r pytest_pycollect_makeitem s rK c
s j d}t jr jdp0 jdd}t F t||d.} fdd j j
D }| j f | W 5 Q R X W 5 Q R X dS dS ) zL
Run coroutines in an event loop instead of a normal function call.
r
proactor_loopr NrF c s i | ]}| j | qS r )funcargs)r: arg
pyfuncitemr r
<dictcomp> s z&pytest_pyfunc_call.<locals>.<dictcomp>T)
r1 r2 r, r- ZfunctionrM getrE rG Z _fixtureinfor. r rJ )rP r3 Z
existing_loopr" Ztestargsr rO r pytest_pyfunc_call s
&rS c C s d| j krd S | jjj}dtji}td k r4tj|d< td k rFtj|d< |dkrRd}i }| dD ]N}|
d }| d }||kr|r`t d
|t
| f nq`|| ||< q`| jdt
| t
| d d S ) N loop_factoryr uvlooptokioallzpyloop,uvloop?,tokio?,?z ?z&Unknown loop '%s', available loops: %s)Zids)r% r1 ZoptionZ aiohttp_loopr, ZDefaultEventLoopPolicyrU ZEventLoopPolicyrV splitendswithstrip
ValueErrorrH keysZ parametrizevalues)ZmetafuncZloopsZavail_factoriesZ factoriesrI Zrequiredr r r pytest_generate_tests s8
r` c c sH | }t | t|d$}|r*|d t | |V W 5 Q R X dS )z%Return an instance of the event loop.rF TN)r, set_event_loop_policyr Z set_debugset_event_loop)rT r3 r4 policyr" r r r r s
r c c sR t st } tj| _nt } t| t| j}t | |V W 5 Q R X d S r )
r r, Zget_event_loop_policyZProactorEventLoopZ
_loop_factoryZ WindowsProactorEventLoopPolicyra r Znew_event_looprb )rc r" r r r rL s
rL c C s t jdtdd | S )Nz3Deprecated, use aiohttp_unused_port fixture instead
stacklevelr? warnDeprecationWarning)aiohttp_unused_portr r r r s r c C s t S )z1Return a port that is unused on the current host.) _unused_portr r r r rj s rj c # s: g dd fdd
}|V fdd} | dS )z^Factory to create a TestServer instance, given an app.
aiohttp_server(app, **kwargs)
Nportc s4 t | |d}|jf d i|I d H | |S Nrl r )r
start_serverappend)Zapprm r( serverr serversr r go s
zaiohttp_server.<locals>.goc s r I d H q d S r popcloser rs r r finalize s z aiohttp_server.<locals>.finalizer r rt ry r rr r aiohttp_server s
r| c C s t jdtdd | S )Nz.Deprecated, use aiohttp_server fixture insteadrd re rg )r| r r r test_server s r} c # s: g dd fdd
}|V fdd} | dS )zpFactory to create a RawTestServer instance, given a web handler.
aiohttp_raw_server(handler, **kwargs)
Nrl c s4 t | |d}|jf d i|I d H | |S rn )r ro rp )Zhandlerrm r( rq rr r r rt 2 s
z aiohttp_raw_server.<locals>.goc s r I d H q d S r ru r rx r r ry : s z$aiohttp_raw_server.<locals>.finalizerz r{ r rr r aiohttp_raw_server* s
r~ c C s t jdtdd | S )Nz2Deprecated, use aiohttp_raw_server fixture insteadrd re rg )r~ r r r raw_test_serverA s r c # s: g dd fdd
}|V fdd} | dS )zFactory to create a TestClient instance.
aiohttp_client(app, **kwargs)
aiohttp_client(server, **kwargs)
aiohttp_client(raw_server, **kwargs)
N)
server_kwargsc s t | tr.t | ttfs.| f||} i }n |r:tdt | trv|pJi }t| fdi|}t|fdi|}n0t | trt| fdi|}ntdt| | I d H
| |S )Nzargs should be emptyr zUnknown argument type: %r)
isinstancer r r AssertionErrorr
r r] typero rp )Z__paramr r' r( rq Zclientclientsr r r rt U s"
zaiohttp_client.<locals>.goc s r I d H q d S r ru r )r r r ry n s z aiohttp_client.<locals>.finalizerz r{ r r r aiohttp_clientK s
r c C s t jdtdd | S )Nz.Deprecated, use aiohttp_client fixture insteadrd re rg )r r r r test_clientu s r )F),r,
contextlibr? Zcollections.abcr ZpytestZaiohttp.helpersr r Z aiohttp.webr Z
test_utilsr r r r
r r r
r rk rU ImportErrorrV r r/ Zfixturer3 r4 contextmanagerrE rG rK rS r` r rL rj r| r} r~ r r r r r r r <module> s` (
8
"
)