1
1
"""pytest-asyncio implementation."""
2
2
3
+ from __future__ import annotations
4
+
3
5
import asyncio
4
6
import contextlib
5
7
import enum
22
24
Any ,
23
25
Callable ,
24
26
Literal ,
25
- Optional ,
26
27
TypeVar ,
27
28
Union ,
28
29
overload ,
@@ -110,41 +111,41 @@ def pytest_addoption(parser: Parser, pluginmanager: PytestPluginManager) -> None
110
111
def fixture (
111
112
fixture_function : FixtureFunction ,
112
113
* ,
113
- scope : "Union[ _ScopeName, Callable[[str, Config], _ScopeName]]" = ...,
114
- loop_scope : Union [ _ScopeName , None ] = ...,
115
- params : Optional [ Iterable [object ]] = ...,
114
+ scope : _ScopeName | Callable [[str , Config ], _ScopeName ] = ...,
115
+ loop_scope : _ScopeName | None = ...,
116
+ params : Iterable [object ] | None = ...,
116
117
autouse : bool = ...,
117
- ids : Union [
118
- Iterable [Union [ str , float , int , bool , None ]],
119
- Callable [[Any ], Optional [ object ]],
120
- None ,
121
- ] = ...,
122
- name : Optional [ str ] = ...,
118
+ ids : (
119
+ Iterable [str | float | int | bool | None ]
120
+ | Callable [[Any ], object | None ]
121
+ | None
122
+ ) = ...,
123
+ name : str | None = ...,
123
124
) -> FixtureFunction : ...
124
125
125
126
126
127
@overload
127
128
def fixture (
128
129
fixture_function : None = ...,
129
130
* ,
130
- scope : "Union[ _ScopeName, Callable[[str, Config], _ScopeName]]" = ...,
131
- loop_scope : Union [ _ScopeName , None ] = ...,
132
- params : Optional [ Iterable [object ]] = ...,
131
+ scope : _ScopeName | Callable [[str , Config ], _ScopeName ] = ...,
132
+ loop_scope : _ScopeName | None = ...,
133
+ params : Iterable [object ] | None = ...,
133
134
autouse : bool = ...,
134
- ids : Union [
135
- Iterable [Union [ str , float , int , bool , None ]],
136
- Callable [[Any ], Optional [ object ]],
137
- None ,
138
- ] = ...,
139
- name : Optional [ str ] = None ,
135
+ ids : (
136
+ Iterable [str | float | int | bool | None ]
137
+ | Callable [[Any ], object | None ]
138
+ | None
139
+ ) = ...,
140
+ name : str | None = None ,
140
141
) -> FixtureFunctionMarker : ...
141
142
142
143
143
144
def fixture (
144
- fixture_function : Optional [ FixtureFunction ] = None ,
145
- loop_scope : Union [ _ScopeName , None ] = None ,
145
+ fixture_function : FixtureFunction | None = None ,
146
+ loop_scope : _ScopeName | None = None ,
146
147
** kwargs : Any ,
147
- ) -> Union [ FixtureFunction , FixtureFunctionMarker ] :
148
+ ) -> FixtureFunction | FixtureFunctionMarker :
148
149
if fixture_function is not None :
149
150
_make_asyncio_fixture_function (fixture_function , loop_scope )
150
151
return pytest .fixture (fixture_function , ** kwargs )
@@ -163,9 +164,7 @@ def _is_asyncio_fixture_function(obj: Any) -> bool:
163
164
return getattr (obj , "_force_asyncio_fixture" , False )
164
165
165
166
166
- def _make_asyncio_fixture_function (
167
- obj : Any , loop_scope : Union [_ScopeName , None ]
168
- ) -> None :
167
+ def _make_asyncio_fixture_function (obj : Any , loop_scope : _ScopeName | None ) -> None :
169
168
if hasattr (obj , "__func__" ):
170
169
# instance method, check the function object
171
170
obj = obj .__func__
@@ -288,7 +287,7 @@ def _add_kwargs(
288
287
return ret
289
288
290
289
291
- def _perhaps_rebind_fixture_func (func : _T , instance : Optional [ Any ] ) -> _T :
290
+ def _perhaps_rebind_fixture_func (func : _T , instance : Any | None ) -> _T :
292
291
if instance is not None :
293
292
# The fixture needs to be bound to the actual request.instance
294
293
# so it is bound to the same object as the test method.
@@ -388,9 +387,7 @@ class PytestAsyncioFunction(Function):
388
387
"""Base class for all test functions managed by pytest-asyncio."""
389
388
390
389
@classmethod
391
- def item_subclass_for (
392
- cls , item : Function , /
393
- ) -> Union [type ["PytestAsyncioFunction" ], None ]:
390
+ def item_subclass_for (cls , item : Function , / ) -> type [PytestAsyncioFunction ] | None :
394
391
"""
395
392
Returns a subclass of PytestAsyncioFunction if there is a specialized subclass
396
393
for the specified function item.
@@ -525,10 +522,8 @@ def runtest(self) -> None:
525
522
# see https://github.com/pytest-dev/pytest/issues/11307
526
523
@pytest .hookimpl (specname = "pytest_pycollect_makeitem" , tryfirst = True )
527
524
def pytest_pycollect_makeitem_preprocess_async_fixtures (
528
- collector : Union [pytest .Module , pytest .Class ], name : str , obj : object
529
- ) -> Union [
530
- pytest .Item , pytest .Collector , list [Union [pytest .Item , pytest .Collector ]], None
531
- ]:
525
+ collector : pytest .Module | pytest .Class , name : str , obj : object
526
+ ) -> pytest .Item | pytest .Collector | list [pytest .Item | pytest .Collector ] | None :
532
527
"""A pytest hook to collect asyncio coroutines."""
533
528
if not collector .funcnamefilter (name ):
534
529
return None
@@ -540,20 +535,17 @@ def pytest_pycollect_makeitem_preprocess_async_fixtures(
540
535
# see https://github.com/pytest-dev/pytest/issues/11307
541
536
@pytest .hookimpl (specname = "pytest_pycollect_makeitem" , hookwrapper = True )
542
537
def pytest_pycollect_makeitem_convert_async_functions_to_subclass (
543
- collector : Union [ pytest .Module , pytest .Class ] , name : str , obj : object
538
+ collector : pytest .Module | pytest .Class , name : str , obj : object
544
539
) -> Generator [None , pluggy .Result , None ]:
545
540
"""
546
541
Converts coroutines and async generators collected as pytest.Functions
547
542
to AsyncFunction items.
548
543
"""
549
544
hook_result = yield
550
545
try :
551
- node_or_list_of_nodes : Union [
552
- pytest .Item ,
553
- pytest .Collector ,
554
- list [Union [pytest .Item , pytest .Collector ]],
555
- None ,
556
- ] = hook_result .get_result ()
546
+ node_or_list_of_nodes : (
547
+ pytest .Item | pytest .Collector | list [pytest .Item | pytest .Collector ] | None
548
+ ) = hook_result .get_result ()
557
549
except BaseException as e :
558
550
hook_result .force_exception (e )
559
551
return
@@ -592,7 +584,7 @@ def pytest_pycollect_makeitem_convert_async_functions_to_subclass(
592
584
593
585
# A stack used to push package-scoped loops during collection of a package
594
586
# and pop those loops during collection of a Module
595
- __package_loop_stack : list [Union [ FixtureFunctionMarker , FixtureFunction ] ] = []
587
+ __package_loop_stack : list [FixtureFunctionMarker | FixtureFunction ] = []
596
588
597
589
598
590
@pytest .hookimpl
@@ -868,7 +860,7 @@ def _provide_clean_event_loop() -> None:
868
860
869
861
870
862
def _get_event_loop_no_warn (
871
- policy : Optional [ AbstractEventLoopPolicy ] = None ,
863
+ policy : AbstractEventLoopPolicy | None = None ,
872
864
) -> asyncio .AbstractEventLoop :
873
865
with warnings .catch_warnings ():
874
866
warnings .simplefilter ("ignore" , DeprecationWarning )
@@ -879,7 +871,7 @@ def _get_event_loop_no_warn(
879
871
880
872
881
873
@pytest .hookimpl (tryfirst = True , hookwrapper = True )
882
- def pytest_pyfunc_call (pyfuncitem : Function ) -> Optional [ object ] :
874
+ def pytest_pyfunc_call (pyfuncitem : Function ) -> object | None :
883
875
"""
884
876
Pytest hook called before a test case is run.
885
877
@@ -999,7 +991,7 @@ def _get_marked_loop_scope(asyncio_marker: Mark) -> _ScopeName:
999
991
return scope
1000
992
1001
993
1002
- def _retrieve_scope_root (item : Union [ Collector , Item ] , scope : str ) -> Collector :
994
+ def _retrieve_scope_root (item : Collector | Item , scope : str ) -> Collector :
1003
995
node_type_by_scope = {
1004
996
"class" : Class ,
1005
997
"module" : Module ,
0 commit comments