1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
|
#!/usr/bin/env python3
# --------------------( LICENSE )--------------------
# Copyright (c) 2014-2025 Beartype authors.
# See "LICENSE" for further details.
'''
:mod:`pytest` **global test configuration** (i.e., early-time configuration
guaranteed to be run by :mod:`pytest` *after* passed command-line arguments are
parsed).
:mod:`pytest` implicitly imports *all* functionality defined by this module
into *all* submodules of this subpackage.
See Also
----------
https://github.com/pytest-dev/pytest-asyncio/blob/master/pytest_asyncio/plugin.py
:mod:`pytest` plugin strongly inspiring this implementation. Despite its
popularity,, pytest-asyncio is mostly unmaintained, poorly commented and
documented, overly obfuscatory, has an extreme number of unresolved issues
and unmerged pull requests, and just generally exhibits code smells.
'''
# ....................{ TODO }....................
#FIXME: Consider refactoring the pytest_pyfunc_call() hook defined below into:
#* A pull request against pytest itself. Pytest absolutely requires support for
# asynchronous test functions. This is 2021, people.
#* A new competing "pytest-async" plugin. This is substantially easier but less
# ideal, as pytest *REALLY* both wants and needs this functionality.
# ....................{ IMPORTS }....................
from asyncio import (
get_event_loop_policy,
new_event_loop,
set_event_loop,
)
from functools import wraps
from inspect import iscoroutinefunction
from pytest import hookimpl
from warnings import (
catch_warnings,
simplefilter,
)
# ....................{ HOOKS ~ configure }....................
@hookimpl(hookwrapper=True, tryfirst=True)
def pytest_pyfunc_call(pyfuncitem: 'Function') -> None:
'''
Hook wrapper called immediately *before* calling the passed test function.
Specifically, this hook wrapper:
* If this test function is synchronous (i.e., declared with ``def``),
preserves this test function as is.
* If this test function is asynchronous (i.e., declared with ``async
def``), wraps this test function in a synchronous wrapper function
synchronously running this test function under an event loop uniquely
isolated to this test function. For safety, each asynchronous test
function is run under a new event loop.
This wrapper wraps all non-wrapper ``pytest_pyfunc_call()`` hooks and is
hopefully called *before* all wrapper ``pytest_pyfunc_call()`` hooks. See
also `the official pytest hook wrapper documentation <hook wrapper_>`__.
Parameters
----------
pyfuncitem : Function
:mod:`pytest` object encapsulating the test function to be run.
.. _hook wrapper:
https://docs.pytest.org/en/6.2.x/writing_plugins.html#hookwrapper-executing-around-other-hooks
'''
# Test function to be called by this hook.
test_func = pyfuncitem.obj
# If this test function is an asynchronous coroutine function (i.e.,
# callable declared with "async def" containing *NO* "yield"
# expressions)...
#
# Note that we intentionally prefer calling this well-tested tester of the
# well-tested "inspect" module rather than our comparable
# beartype._util.func.utilfunctest.is_func_coro() tester, which
# is hopefully but *NOT* necessarily known to be working here.
if iscoroutinefunction(test_func):
@wraps(test_func)
def test_func_synchronous(*args, **kwargs):
'''
Closure synchronously calling the current asynchronous test
coroutine function under a new event loop uniquely isolated to this
coroutine.
'''
# With a warning context manager...
with catch_warnings():
# Ignore *ALL* deprecating warnings emitted by the
# get_event_loop() function called below. For unknown reasons,
# CPython 3.11 devs thought that emitting a "There is no current
# event loop" warning (erroneously classified as a
# "deprecation") was a wonderful idea. "asyncio" is arduous
# enough to portably support as it is. Work with me here, guys!
simplefilter('ignore', DeprecationWarning)
# Current event loop for the current threading context if any
# *OR* create a new event loop otherwise. Note that the
# higher-level asyncio.get_event_loop() getter is intentionally
# *NOT* called here, as Python 3.10 broke backward compatibility
# by refactoring that getter to be an alias for the wildly
# different asyncio.get_running_loop() getter, which *MUST* be
# called only from within either an asynchronous callable or
# running event loop. In either case, asyncio.get_running_loop()
# and thus asyncio.get_event_loop() is useless in this context.
# Instead, we call the lower-level
# get_event_loop_policy().get_event_loop() getter -- which
# asyncio.get_event_loop() used to wrap. *facepalm*
#
# This getter should ideally return "None" rather than creating
# a new event loop without our permission if no loop has been
# set. This getter instead does the latter, implying that this
# closure will typically instantiate two event loops per
# asynchronous coroutine test function:
# * The first useless event loop implicitly created by this
# get_event_loop() call.
# * The second useful event loop explicitly created by the
# subsequent new_event_loop() call.
#
# Since there exists *NO* other means of querying the current
# event loop, we reluctantly bite the bullet and pay the piper.
event_loop_old = get_event_loop_policy().get_event_loop()
# Close this loop, regardless of whether the prior
# get_event_loop() call just implicitly created this loop,
# because the "asyncio" API offers *NO* means of differentiating
# these two common edge cases. *double facepalm*
event_loop_old.close()
# New event loop isolated to this coroutine.
#
# Note that this event loop has yet to be set as the current event
# loop for the current threading context. Explicit is better than
# implicit.
event_loop = new_event_loop()
# Set this as the current event loop for this threading context.
set_event_loop(event_loop)
# Coroutine object produced by this asynchronous coroutine test
# function. Technically, coroutine functions are *NOT* actually
# coroutines; they're just syntactic sugar implemented as standard
# synchronous functions dynamically creating and returning
# asynchronous coroutine objects on each call.
test_func_coroutine = test_func(*args, **kwargs)
# Synchronously run a new asynchronous task implicitly scheduled to
# run this coroutine, ignoring the value returned by this coroutine
# (if any) while reraising any exception raised by this coroutine
# up the call stack to pytest.
event_loop.run_until_complete(test_func_coroutine)
# Close this event loop.
event_loop.close()
# Replace this asynchronous coroutine test function with this
# synchronous closure wrapping this test function.
pyfuncitem.obj = test_func_synchronous
# Perform this test by calling this test function.
yield
|
