Z-& @ sJ d d l Z d d l Z d d l Z d d l Z d d l Z d d l m Z d d l m Z d d l m
Z
m Z m Z m
Z
m Z m Z m Z d d l m Z y d d l Z Wn e k
r d Z Yn Xy d d l Z Wn e k
r d Z Yn Xd d Z d d
Z e j d d Z e j d
d Z e j d d Z e j d d d Z d d Z d d Z d d Z e j d d Z! e j d d Z e j d d Z" e j d d! Z# e j d" d# Z$ e j d$ d% Z% e j d& d' Z& e j d( d) Z' e j d* d+ Z( d S), N)isasyncgenfunction) Application )BaseTestServer
RawTestServer
TestClient
TestServer loop_contextsetup_test_loopteardown_test_loop) unused_portc C sa | j d d d d d d d | j d d d d d
d d | j d d d d d d d
d S)Nz--aiohttp-fastaction
store_truedefaultFhelpz*run tests faster by disabling extra checksz--aiohttp-loopZstorepyloopz:run tests with specific loop: pyloop, uvloop, tokio or allz--aiohttp-enable-loop-debugz enable event loop debug mode)Z addoption)parser r /pytest_plugin.pypytest_addoption s r c s | j t r d n t j r6 d n d Sd d | j k rd | j d 7_ d f d d } | | _ d S)zS
Allow fixtures to be coroutines. Run coroutine fixtures in an event loop.
TFNrequestc s | d } r | d =d | j k r2 t d | j d r | | f d d } | j | j j S j | | Sd S)Nr loopz^Asynchronous fixtures must depend on the 'loop' fixture or be used in tests depending from it.c s0 y j j SWn t k
r+ Yn Xd S)N)run_until_complete __anext__StopAsyncIterationr )_loopgenr r finalizerQ s
z8pytest_fixture_setup.<locals>.wrapper.<locals>.finalizer) fixturenames ExceptionZgetfixturevalueZ addfinalizerr r )argskwargsr r )func is_async_gen
strip_request)r r r wrapper< s
z%pytest_fixture_setup.<locals>.wrapper)r )r" r asyncioiscoroutinefunctionargnames)Z
fixturedefr% r )r" r# r$ r pytest_fixture_setup' s r) c C s | j j d S)z--fast config optionz--aiohttp-fast)config getoption)r r r r fast_ s r, c C s | j j d S)z!--enable-loop-debug config optionz--aiohttp-enable-loop-debug)r* r+ )r r r r
loop_debuge s r- c
c s t j d d g } d Vd d | D } | ru t d j t | t | d k r] d n d d
j | Wd QRXd S) 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.
recordTNc S s1 g | ]' } | j t k r d j d | q S)z#{w.filename}:{w.lineno}:{w.message}w)categoryRuntimeWarningformat).0r/ r r r
<listcomp>u s z,_runtime_warning_context.<locals>.<listcomp>z{} Runtime Warning{},
{}r s
)warningscatch_warnings RuntimeErrorr2 lenjoin) _warningsrwr r r _runtime_warning_contextk s
r? Fc c s0 | r | Vn t } | Vt | d | d S)z{
setups and tears down a loop unless one is passed in via the loop
argument when it's passed straight through.
r, N)r
r )r r, r r r _passthrough_loop_context s
r@ c C s8 | j | r4 t j | r4 t | j | | Sd S)z/
Fix pytest collecting for coroutines.
N)Zfuncnamefilterr& r' listZ
_genfunctions)Z collectornameobjr r r pytest_pycollect_makeitem s rD c s j j d } t j j r j j d d } t V t | d | < } f d d j j
D } | j j | Wd QRXWd QRXd Sd S)zL
Run coroutines in an event loop instead of a normal function call.
z--aiohttp-fastr Nr, c s i | ] } j | | q Sr )funcargs)r3 arg)
pyfuncitemr r
<dictcomp> s z&pytest_pyfunc_call.<locals>.<dictcomp>T)
r* r+ r& r' functionrE getr? r@ Z _fixtureinfor( r rC )rG r, Z
existing_loopr Ztestargsr )rG r pytest_pyfunc_call s
$rK c C s. d | j k r d S| j j j } d t j i } t d k rJ t j | d <t d k rc t j | d <| d k ru d } i } x~ | j d D]m } | j d } | j
d } | | k r | r t d
| t | j
f n q | | | | <q W| j d t | j d t | j
d S) N loop_factoryr uvlooptokioallzpyloop,uvloop?,tokio?,?z ?z&Unknown loop '%s', available loops: %sZids)r r* ZoptionZ aiohttp_loopr& Znew_event_looprM rN splitendswithstrip
ValueErrorrA keysZ parametrizevalues)ZmetafuncZloopsZavail_factoriesZ factoriesrB Zrequiredr r r pytest_generate_tests s0
rX c c sE t | d | } | r( | j d | VWd QRXt j d d S)z%Return an instance of the event loop.r, TN)r Z set_debugr& Zset_event_loop)rL r, r- r r r r r s
r c C s t j d t | S)Nz3Deprecated, use aiohttp_unused_port fixture instead)r8 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 r[ s r[ c # sL g d d f d d } | V f d d } j | d S)z^Factory to create a TestServer instance, given an app.
aiohttp_server(app, **kwargs)
portNc s; t | d | } | j d | Id H j | | S)Nr] r )r start_serverappend)Zappr] r! server)r serversr r go s
zaiohttp_server.<locals>.goc s& x r! j j Id Hq Wd S)N)popcloser )ra r r finalize s z aiohttp_server.<locals>.finalize)r )r rb re r )r ra r aiohttp_server s
rf c C s t j d t | S)Nz.Deprecated, use aiohttp_server fixture instead)r8 rY rZ )rf r r r test_server s rg c # sL g d d f d d } | V f d d } j | d S)zpFactory to create a RawTestServer instance, given a web handler.
aiohttp_raw_server(handler, **kwargs)
r] Nc s; t | d | } | j d | Id H j | | S)Nr] r )r r^ r_ )handlerr] r! r` )r ra r r rb s
z aiohttp_raw_server.<locals>.goc s& x r! j j Id Hq Wd S)N)rc rd r )ra r r re
s z$aiohttp_raw_server.<locals>.finalize)r )r rb re r )r ra r aiohttp_raw_server s
ri c C s t j d t | S)Nz2Deprecated, use aiohttp_raw_server fixture instead)r8 rY rZ )ri r r r raw_test_server s rj c # sL g d d f d d } | V f d d } j | d S)zFactory to create a TestClient instance.
aiohttp_client(app, **kwargs)
aiohttp_client(server, **kwargs)
aiohttp_client(raw_server, **kwargs)
server_kwargsNc s t | t j rC t | t t f rC | | | } i } n | sV t d t | t r | pn i } t | d | } t | d | } n= t | t r t | d | } n t d t | | j
Id H j | | S)Nzargs should be emptyr zUnknown argument type: %r)
isinstance collectionsCallabler r AssertionErrorr r rU typer^ r_ )Z__paramrk r r! r` client)clientsr r r rb % s
zaiohttp_client.<locals>.goc s& x r! j j Id Hq Wd S)N)rc rd r )rr r r re = s z aiohttp_client.<locals>.finalize)r )r rb re r )rr r r aiohttp_client s
rs c C s t j d t | S)Nz.Deprecated, use aiohttp_client fixture instead)r8 rY rZ )rs r r r test_clientD s rt ))r& rm
contextlibr8 ZpytestZaiohttp.helpersr Z aiohttp.webr Z
test_utilsr r r r r r
r r r\ rM ImportErrorrN r r) Zfixturer, r- contextmanagerr? r@ rD rK rX r r[ rf rg ri rj rs rt r r r r <module> sF 4
8 !
)