plugin.py 29 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896
  1. """pytest-asyncio implementation."""
  2. from __future__ import annotations
  3. import asyncio
  4. import contextlib
  5. import contextvars
  6. import enum
  7. import functools
  8. import inspect
  9. import socket
  10. import sys
  11. import traceback
  12. import warnings
  13. from asyncio import AbstractEventLoop, AbstractEventLoopPolicy
  14. from collections.abc import (
  15. AsyncIterator,
  16. Awaitable,
  17. Callable,
  18. Generator,
  19. Iterable,
  20. Iterator,
  21. Sequence,
  22. )
  23. from types import AsyncGeneratorType, CoroutineType
  24. from typing import (
  25. Any,
  26. Literal,
  27. ParamSpec,
  28. TypeVar,
  29. overload,
  30. )
  31. import pluggy
  32. import pytest
  33. from _pytest.fixtures import resolve_fixture_function
  34. from _pytest.scope import Scope
  35. from pytest import (
  36. Config,
  37. FixtureDef,
  38. FixtureRequest,
  39. Function,
  40. Item,
  41. Mark,
  42. MonkeyPatch,
  43. Parser,
  44. PytestCollectionWarning,
  45. PytestDeprecationWarning,
  46. PytestPluginManager,
  47. )
  48. if sys.version_info >= (3, 11):
  49. from asyncio import Runner
  50. else:
  51. from backports.asyncio.runner import Runner
  52. if sys.version_info >= (3, 13):
  53. from typing import TypeIs
  54. else:
  55. from typing_extensions import TypeIs
  56. _ScopeName = Literal["session", "package", "module", "class", "function"]
  57. _R = TypeVar("_R", bound=Awaitable[Any] | AsyncIterator[Any])
  58. _P = ParamSpec("_P")
  59. FixtureFunction = Callable[_P, _R]
  60. class PytestAsyncioError(Exception):
  61. """Base class for exceptions raised by pytest-asyncio"""
  62. class Mode(str, enum.Enum):
  63. AUTO = "auto"
  64. STRICT = "strict"
  65. ASYNCIO_MODE_HELP = """\
  66. 'auto' - for automatically handling all async functions by the plugin
  67. 'strict' - for autoprocessing disabling (useful if different async frameworks \
  68. should be tested together, e.g. \
  69. both pytest-asyncio and pytest-trio are used in the same project)
  70. """
  71. def pytest_addoption(parser: Parser, pluginmanager: PytestPluginManager) -> None:
  72. group = parser.getgroup("asyncio")
  73. group.addoption(
  74. "--asyncio-mode",
  75. dest="asyncio_mode",
  76. default=None,
  77. metavar="MODE",
  78. help=ASYNCIO_MODE_HELP,
  79. )
  80. group.addoption(
  81. "--asyncio-debug",
  82. dest="asyncio_debug",
  83. action="store_true",
  84. default=None,
  85. help="enable asyncio debug mode for the default event loop",
  86. )
  87. parser.addini(
  88. "asyncio_mode",
  89. help="default value for --asyncio-mode",
  90. default="strict",
  91. )
  92. parser.addini(
  93. "asyncio_debug",
  94. help="enable asyncio debug mode for the default event loop",
  95. type="bool",
  96. default="false",
  97. )
  98. parser.addini(
  99. "asyncio_default_fixture_loop_scope",
  100. type="string",
  101. help="default scope of the asyncio event loop used to execute async fixtures",
  102. default=None,
  103. )
  104. parser.addini(
  105. "asyncio_default_test_loop_scope",
  106. type="string",
  107. help="default scope of the asyncio event loop used to execute tests",
  108. default="function",
  109. )
  110. @overload
  111. def fixture(
  112. fixture_function: FixtureFunction[_P, _R],
  113. *,
  114. scope: _ScopeName | Callable[[str, Config], _ScopeName] = ...,
  115. loop_scope: _ScopeName | None = ...,
  116. params: Iterable[object] | None = ...,
  117. autouse: bool = ...,
  118. ids: (
  119. Iterable[str | float | int | bool | None]
  120. | Callable[[Any], object | None]
  121. | None
  122. ) = ...,
  123. name: str | None = ...,
  124. ) -> FixtureFunction[_P, _R]: ...
  125. @overload
  126. def fixture(
  127. fixture_function: None = ...,
  128. *,
  129. scope: _ScopeName | Callable[[str, Config], _ScopeName] = ...,
  130. loop_scope: _ScopeName | None = ...,
  131. params: Iterable[object] | None = ...,
  132. autouse: bool = ...,
  133. ids: (
  134. Iterable[str | float | int | bool | None]
  135. | Callable[[Any], object | None]
  136. | None
  137. ) = ...,
  138. name: str | None = None,
  139. ) -> Callable[[FixtureFunction[_P, _R]], FixtureFunction[_P, _R]]: ...
  140. def fixture(
  141. fixture_function: FixtureFunction[_P, _R] | None = None,
  142. loop_scope: _ScopeName | None = None,
  143. **kwargs: Any,
  144. ) -> (
  145. FixtureFunction[_P, _R]
  146. | Callable[[FixtureFunction[_P, _R]], FixtureFunction[_P, _R]]
  147. ):
  148. if fixture_function is not None:
  149. _make_asyncio_fixture_function(fixture_function, loop_scope)
  150. return pytest.fixture(fixture_function, **kwargs)
  151. else:
  152. @functools.wraps(fixture)
  153. def inner(fixture_function: FixtureFunction[_P, _R]) -> FixtureFunction[_P, _R]:
  154. return fixture(fixture_function, loop_scope=loop_scope, **kwargs)
  155. return inner
  156. def _is_asyncio_fixture_function(obj: Any) -> bool:
  157. obj = getattr(obj, "__func__", obj) # instance method maybe?
  158. return getattr(obj, "_force_asyncio_fixture", False)
  159. def _make_asyncio_fixture_function(obj: Any, loop_scope: _ScopeName | None) -> None:
  160. if hasattr(obj, "__func__"):
  161. # instance method, check the function object
  162. obj = obj.__func__
  163. obj._force_asyncio_fixture = True
  164. obj._loop_scope = loop_scope
  165. def _is_coroutine_or_asyncgen(obj: Any) -> bool:
  166. return inspect.iscoroutinefunction(obj) or inspect.isasyncgenfunction(obj)
  167. def _get_asyncio_mode(config: Config) -> Mode:
  168. val = config.getoption("asyncio_mode")
  169. if val is None:
  170. val = config.getini("asyncio_mode")
  171. try:
  172. return Mode(val)
  173. except ValueError as e:
  174. modes = ", ".join(m.value for m in Mode)
  175. raise pytest.UsageError(
  176. f"{val!r} is not a valid asyncio_mode. Valid modes: {modes}."
  177. ) from e
  178. def _get_asyncio_debug(config: Config) -> bool:
  179. val = config.getoption("asyncio_debug")
  180. if val is None:
  181. val = config.getini("asyncio_debug")
  182. if isinstance(val, bool):
  183. return val
  184. else:
  185. return val == "true"
  186. _DEFAULT_FIXTURE_LOOP_SCOPE_UNSET = """\
  187. The configuration option "asyncio_default_fixture_loop_scope" is unset.
  188. The event loop scope for asynchronous fixtures will default to the fixture caching \
  189. scope. Future versions of pytest-asyncio will default the loop scope for asynchronous \
  190. fixtures to function scope. Set the default fixture loop scope explicitly in order to \
  191. avoid unexpected behavior in the future. Valid fixture loop scopes are: \
  192. "function", "class", "module", "package", "session"
  193. """
  194. def _validate_scope(scope: str | None, option_name: str) -> None:
  195. if scope is None:
  196. return
  197. valid_scopes = [s.value for s in Scope]
  198. if scope not in valid_scopes:
  199. raise pytest.UsageError(
  200. f"{scope!r} is not a valid {option_name}. "
  201. f"Valid scopes are: {', '.join(valid_scopes)}."
  202. )
  203. def pytest_configure(config: Config) -> None:
  204. default_fixture_loop_scope = config.getini("asyncio_default_fixture_loop_scope")
  205. _validate_scope(default_fixture_loop_scope, "asyncio_default_fixture_loop_scope")
  206. if not default_fixture_loop_scope:
  207. warnings.warn(PytestDeprecationWarning(_DEFAULT_FIXTURE_LOOP_SCOPE_UNSET))
  208. default_test_loop_scope = config.getini("asyncio_default_test_loop_scope")
  209. _validate_scope(default_test_loop_scope, "asyncio_default_test_loop_scope")
  210. config.addinivalue_line(
  211. "markers",
  212. "asyncio: "
  213. "mark the test as a coroutine, it will be "
  214. "run using an asyncio event loop",
  215. )
  216. @pytest.hookimpl(tryfirst=True)
  217. def pytest_report_header(config: Config) -> list[str]:
  218. """Add asyncio config to pytest header."""
  219. mode = _get_asyncio_mode(config)
  220. debug = _get_asyncio_debug(config)
  221. default_fixture_loop_scope = config.getini("asyncio_default_fixture_loop_scope")
  222. default_test_loop_scope = _get_default_test_loop_scope(config)
  223. header = [
  224. f"mode={mode}",
  225. f"debug={debug}",
  226. f"asyncio_default_fixture_loop_scope={default_fixture_loop_scope}",
  227. f"asyncio_default_test_loop_scope={default_test_loop_scope}",
  228. ]
  229. return [
  230. "asyncio: " + ", ".join(header),
  231. ]
  232. def _fixture_synchronizer(
  233. fixturedef: FixtureDef, runner: Runner, request: FixtureRequest
  234. ) -> Callable:
  235. """Returns a synchronous function evaluating the specified fixture."""
  236. fixture_function = resolve_fixture_function(fixturedef, request)
  237. if inspect.isasyncgenfunction(fixturedef.func):
  238. return _wrap_asyncgen_fixture(fixture_function, runner, request) # type: ignore[arg-type]
  239. elif inspect.iscoroutinefunction(fixturedef.func):
  240. return _wrap_async_fixture(fixture_function, runner, request) # type: ignore[arg-type]
  241. else:
  242. return fixturedef.func
  243. AsyncGenFixtureParams = ParamSpec("AsyncGenFixtureParams")
  244. AsyncGenFixtureYieldType = TypeVar("AsyncGenFixtureYieldType")
  245. def _wrap_asyncgen_fixture(
  246. fixture_function: Callable[
  247. AsyncGenFixtureParams, AsyncGeneratorType[AsyncGenFixtureYieldType, Any]
  248. ],
  249. runner: Runner,
  250. request: FixtureRequest,
  251. ) -> Callable[AsyncGenFixtureParams, AsyncGenFixtureYieldType]:
  252. @functools.wraps(fixture_function)
  253. def _asyncgen_fixture_wrapper(
  254. *args: AsyncGenFixtureParams.args,
  255. **kwargs: AsyncGenFixtureParams.kwargs,
  256. ):
  257. gen_obj = fixture_function(*args, **kwargs)
  258. async def setup():
  259. res = await gen_obj.__anext__()
  260. return res
  261. context = contextvars.copy_context()
  262. result = runner.run(setup(), context=context)
  263. reset_contextvars = _apply_contextvar_changes(context)
  264. def finalizer() -> None:
  265. """Yield again, to finalize."""
  266. async def async_finalizer() -> None:
  267. try:
  268. await gen_obj.__anext__()
  269. except StopAsyncIteration:
  270. pass
  271. else:
  272. msg = "Async generator fixture didn't stop."
  273. msg += "Yield only once."
  274. raise ValueError(msg)
  275. runner.run(async_finalizer(), context=context)
  276. if reset_contextvars is not None:
  277. reset_contextvars()
  278. request.addfinalizer(finalizer)
  279. return result
  280. return _asyncgen_fixture_wrapper
  281. AsyncFixtureParams = ParamSpec("AsyncFixtureParams")
  282. AsyncFixtureReturnType = TypeVar("AsyncFixtureReturnType")
  283. def _wrap_async_fixture(
  284. fixture_function: Callable[
  285. AsyncFixtureParams, CoroutineType[Any, Any, AsyncFixtureReturnType]
  286. ],
  287. runner: Runner,
  288. request: FixtureRequest,
  289. ) -> Callable[AsyncFixtureParams, AsyncFixtureReturnType]:
  290. @functools.wraps(fixture_function)
  291. def _async_fixture_wrapper(
  292. *args: AsyncFixtureParams.args,
  293. **kwargs: AsyncFixtureParams.kwargs,
  294. ):
  295. async def setup():
  296. res = await fixture_function(*args, **kwargs)
  297. return res
  298. context = contextvars.copy_context()
  299. result = runner.run(setup(), context=context)
  300. # Copy the context vars modified by the setup task into the current
  301. # context, and (if needed) add a finalizer to reset them.
  302. #
  303. # Note that this is slightly different from the behavior of a non-async
  304. # fixture, which would rely on the fixture author to add a finalizer
  305. # to reset the variables. In this case, the author of the fixture can't
  306. # write such a finalizer because they have no way to capture the Context
  307. # in which the setup function was run, so we need to do it for them.
  308. reset_contextvars = _apply_contextvar_changes(context)
  309. if reset_contextvars is not None:
  310. request.addfinalizer(reset_contextvars)
  311. return result
  312. return _async_fixture_wrapper
  313. def _apply_contextvar_changes(
  314. context: contextvars.Context,
  315. ) -> Callable[[], None] | None:
  316. """
  317. Copy contextvar changes from the given context to the current context.
  318. If any contextvars were modified by the fixture, return a finalizer that
  319. will restore them.
  320. """
  321. context_tokens = []
  322. for var in context:
  323. try:
  324. if var.get() is context.get(var):
  325. # This variable is not modified, so leave it as-is.
  326. continue
  327. except LookupError:
  328. # This variable isn't yet set in the current context at all.
  329. pass
  330. token = var.set(context.get(var))
  331. context_tokens.append((var, token))
  332. if not context_tokens:
  333. return None
  334. def restore_contextvars():
  335. while context_tokens:
  336. (var, token) = context_tokens.pop()
  337. var.reset(token)
  338. return restore_contextvars
  339. class PytestAsyncioFunction(Function):
  340. """Base class for all test functions managed by pytest-asyncio."""
  341. @classmethod
  342. def item_subclass_for(cls, item: Function, /) -> type[PytestAsyncioFunction] | None:
  343. """
  344. Returns a subclass of PytestAsyncioFunction if there is a specialized subclass
  345. for the specified function item.
  346. Return None if no specialized subclass exists for the specified item.
  347. """
  348. for subclass in cls.__subclasses__():
  349. if subclass._can_substitute(item):
  350. return subclass
  351. return None
  352. @classmethod
  353. def _from_function(cls, function: Function, /) -> Function:
  354. """
  355. Instantiates this specific PytestAsyncioFunction type from the specified
  356. Function item.
  357. """
  358. assert function.get_closest_marker("asyncio")
  359. assert function.parent is not None
  360. subclass_instance = cls.from_parent(
  361. function.parent,
  362. name=function.name,
  363. callspec=getattr(function, "callspec", None),
  364. callobj=function.obj,
  365. fixtureinfo=function._fixtureinfo,
  366. keywords=function.keywords,
  367. originalname=function.originalname,
  368. )
  369. subclass_instance.own_markers = function.own_markers
  370. assert subclass_instance.own_markers == function.own_markers
  371. return subclass_instance
  372. @staticmethod
  373. def _can_substitute(item: Function) -> bool:
  374. """Returns whether the specified function can be replaced by this class"""
  375. raise NotImplementedError()
  376. def setup(self) -> None:
  377. runner_fixture_id = f"_{self._loop_scope}_scoped_runner"
  378. if runner_fixture_id not in self.fixturenames:
  379. self.fixturenames.append(runner_fixture_id)
  380. return super().setup()
  381. def runtest(self) -> None:
  382. runner_fixture_id = f"_{self._loop_scope}_scoped_runner"
  383. runner = self._request.getfixturevalue(runner_fixture_id)
  384. context = contextvars.copy_context()
  385. synchronized_obj = _synchronize_coroutine(
  386. getattr(*self._synchronization_target_attr), runner, context
  387. )
  388. with MonkeyPatch.context() as c:
  389. c.setattr(*self._synchronization_target_attr, synchronized_obj)
  390. super().runtest()
  391. @functools.cached_property
  392. def _loop_scope(self) -> _ScopeName:
  393. """
  394. Return the scope of the asyncio event loop this item is run in.
  395. The effective scope is determined lazily. It is identical to to the
  396. `loop_scope` value of the closest `asyncio` pytest marker. If no such
  397. marker is present, the the loop scope is determined by the configuration
  398. value of `asyncio_default_test_loop_scope`, instead.
  399. """
  400. marker = self.get_closest_marker("asyncio")
  401. assert marker is not None
  402. default_loop_scope = _get_default_test_loop_scope(self.config)
  403. return _get_marked_loop_scope(marker, default_loop_scope)
  404. @property
  405. def _synchronization_target_attr(self) -> tuple[object, str]:
  406. """
  407. Return the coroutine that needs to be synchronized during the test run.
  408. This method is inteded to be overwritten by subclasses when they need to apply
  409. the coroutine synchronizer to a value that's different from self.obj
  410. e.g. the AsyncHypothesisTest subclass.
  411. """
  412. return self, "obj"
  413. class Coroutine(PytestAsyncioFunction):
  414. """Pytest item created by a coroutine"""
  415. @staticmethod
  416. def _can_substitute(item: Function) -> bool:
  417. func = item.obj
  418. return inspect.iscoroutinefunction(func)
  419. class AsyncGenerator(PytestAsyncioFunction):
  420. """Pytest item created by an asynchronous generator"""
  421. @staticmethod
  422. def _can_substitute(item: Function) -> bool:
  423. func = item.obj
  424. return inspect.isasyncgenfunction(func)
  425. @classmethod
  426. def _from_function(cls, function: Function, /) -> Function:
  427. async_gen_item = super()._from_function(function)
  428. unsupported_item_type_message = (
  429. f"Tests based on asynchronous generators are not supported. "
  430. f"{function.name} will be ignored."
  431. )
  432. async_gen_item.warn(PytestCollectionWarning(unsupported_item_type_message))
  433. async_gen_item.add_marker(
  434. pytest.mark.xfail(run=False, reason=unsupported_item_type_message)
  435. )
  436. return async_gen_item
  437. class AsyncStaticMethod(PytestAsyncioFunction):
  438. """
  439. Pytest item that is a coroutine or an asynchronous generator
  440. decorated with staticmethod
  441. """
  442. @staticmethod
  443. def _can_substitute(item: Function) -> bool:
  444. func = item.obj
  445. return isinstance(func, staticmethod) and _is_coroutine_or_asyncgen(
  446. func.__func__
  447. )
  448. class AsyncHypothesisTest(PytestAsyncioFunction):
  449. """
  450. Pytest item that is coroutine or an asynchronous generator decorated by
  451. @hypothesis.given.
  452. """
  453. def setup(self) -> None:
  454. if not getattr(self.obj, "hypothesis", False) and getattr(
  455. self.obj, "is_hypothesis_test", False
  456. ):
  457. pytest.fail(
  458. f"test function `{self!r}` is using Hypothesis, but pytest-asyncio "
  459. "only works with Hypothesis 3.64.0 or later."
  460. )
  461. return super().setup()
  462. @staticmethod
  463. def _can_substitute(item: Function) -> bool:
  464. func = item.obj
  465. return (
  466. getattr(func, "is_hypothesis_test", False) # type: ignore[return-value]
  467. and getattr(func, "hypothesis", None)
  468. and inspect.iscoroutinefunction(func.hypothesis.inner_test)
  469. )
  470. @property
  471. def _synchronization_target_attr(self) -> tuple[object, str]:
  472. return self.obj.hypothesis, "inner_test"
  473. # The function name needs to start with "pytest_"
  474. # see https://github.com/pytest-dev/pytest/issues/11307
  475. @pytest.hookimpl(specname="pytest_pycollect_makeitem", hookwrapper=True)
  476. def pytest_pycollect_makeitem_convert_async_functions_to_subclass(
  477. collector: pytest.Module | pytest.Class, name: str, obj: object
  478. ) -> Generator[None, pluggy.Result, None]:
  479. """
  480. Converts coroutines and async generators collected as pytest.Functions
  481. to AsyncFunction items.
  482. """
  483. hook_result = yield
  484. try:
  485. node_or_list_of_nodes: (
  486. pytest.Item | pytest.Collector | list[pytest.Item | pytest.Collector] | None
  487. ) = hook_result.get_result()
  488. except BaseException as e:
  489. hook_result.force_exception(e)
  490. return
  491. if not node_or_list_of_nodes:
  492. return
  493. if isinstance(node_or_list_of_nodes, Sequence):
  494. node_iterator = iter(node_or_list_of_nodes)
  495. else:
  496. # Treat single node as a single-element iterable
  497. node_iterator = iter((node_or_list_of_nodes,))
  498. updated_node_collection = []
  499. for node in node_iterator:
  500. updated_item = node
  501. if isinstance(node, Function):
  502. specialized_item_class = PytestAsyncioFunction.item_subclass_for(node)
  503. if specialized_item_class:
  504. if _get_asyncio_mode(
  505. node.config
  506. ) == Mode.AUTO and not node.get_closest_marker("asyncio"):
  507. node.add_marker("asyncio")
  508. if node.get_closest_marker("asyncio"):
  509. updated_item = specialized_item_class._from_function(node)
  510. updated_node_collection.append(updated_item)
  511. hook_result.force_result(updated_node_collection)
  512. @contextlib.contextmanager
  513. def _temporary_event_loop_policy(policy: AbstractEventLoopPolicy) -> Iterator[None]:
  514. old_loop_policy = _get_event_loop_policy()
  515. try:
  516. old_loop = _get_event_loop_no_warn()
  517. except RuntimeError:
  518. old_loop = None
  519. _set_event_loop_policy(policy)
  520. try:
  521. yield
  522. finally:
  523. _set_event_loop_policy(old_loop_policy)
  524. _set_event_loop(old_loop)
  525. def _get_event_loop_policy() -> AbstractEventLoopPolicy:
  526. with warnings.catch_warnings():
  527. warnings.simplefilter("ignore", DeprecationWarning)
  528. return asyncio.get_event_loop_policy()
  529. def _set_event_loop_policy(policy: AbstractEventLoopPolicy) -> None:
  530. with warnings.catch_warnings():
  531. warnings.simplefilter("ignore", DeprecationWarning)
  532. asyncio.set_event_loop_policy(policy)
  533. def _get_event_loop_no_warn(
  534. policy: AbstractEventLoopPolicy | None = None,
  535. ) -> asyncio.AbstractEventLoop:
  536. with warnings.catch_warnings():
  537. warnings.simplefilter("ignore", DeprecationWarning)
  538. if policy is not None:
  539. return policy.get_event_loop()
  540. else:
  541. return asyncio.get_event_loop()
  542. def _set_event_loop(loop: AbstractEventLoop | None) -> None:
  543. with warnings.catch_warnings():
  544. warnings.simplefilter("ignore", DeprecationWarning)
  545. asyncio.set_event_loop(loop)
  546. @pytest.hookimpl(tryfirst=True, hookwrapper=True)
  547. def pytest_pyfunc_call(pyfuncitem: Function) -> object | None:
  548. """Pytest hook called before a test case is run."""
  549. if pyfuncitem.get_closest_marker("asyncio") is not None:
  550. if is_async_test(pyfuncitem):
  551. asyncio_mode = _get_asyncio_mode(pyfuncitem.config)
  552. for fixname, fixtures in pyfuncitem._fixtureinfo.name2fixturedefs.items():
  553. # name2fixturedefs is a dict between fixture name and a list of matching
  554. # fixturedefs. The last entry in the list is closest and the one used.
  555. func = fixtures[-1].func
  556. if (
  557. asyncio_mode == Mode.STRICT
  558. and _is_coroutine_or_asyncgen(func)
  559. and not _is_asyncio_fixture_function(func)
  560. ):
  561. warnings.warn(
  562. PytestDeprecationWarning(
  563. f"asyncio test {pyfuncitem.name!r} requested async "
  564. "@pytest.fixture "
  565. f"{fixname!r} in strict mode. "
  566. "You might want to use @pytest_asyncio.fixture or switch "
  567. "to auto mode. "
  568. "This will become an error in future versions of "
  569. "pytest-asyncio."
  570. ),
  571. stacklevel=1,
  572. )
  573. # no stacklevel points at the users code, so we set stacklevel=1
  574. # so it at least indicates that it's the plugin complaining.
  575. # Pytest gives the test file & name in the warnings summary at least
  576. else:
  577. pyfuncitem.warn(
  578. pytest.PytestWarning(
  579. f"The test {pyfuncitem} is marked with '@pytest.mark.asyncio' "
  580. "but it is not an async function. "
  581. "Please remove the asyncio mark. "
  582. "If the test is not marked explicitly, "
  583. "check for global marks applied via 'pytestmark'."
  584. )
  585. )
  586. yield
  587. return None
  588. def _synchronize_coroutine(
  589. func: Callable[..., CoroutineType],
  590. runner: asyncio.Runner,
  591. context: contextvars.Context,
  592. ):
  593. """
  594. Return a sync wrapper around a coroutine executing it in the
  595. specified runner and context.
  596. """
  597. @functools.wraps(func)
  598. def inner(*args, **kwargs):
  599. coro = func(*args, **kwargs)
  600. runner.run(coro, context=context)
  601. return inner
  602. @pytest.hookimpl(wrapper=True)
  603. def pytest_fixture_setup(fixturedef: FixtureDef, request) -> object | None:
  604. asyncio_mode = _get_asyncio_mode(request.config)
  605. if not _is_asyncio_fixture_function(fixturedef.func):
  606. if asyncio_mode == Mode.STRICT:
  607. # Ignore async fixtures without explicit asyncio mark in strict mode
  608. # This applies to pytest_trio fixtures, for example
  609. return (yield)
  610. if not _is_coroutine_or_asyncgen(fixturedef.func):
  611. return (yield)
  612. default_loop_scope = request.config.getini("asyncio_default_fixture_loop_scope")
  613. loop_scope = (
  614. getattr(fixturedef.func, "_loop_scope", None)
  615. or default_loop_scope
  616. or fixturedef.scope
  617. )
  618. runner_fixture_id = f"_{loop_scope}_scoped_runner"
  619. runner = request.getfixturevalue(runner_fixture_id)
  620. synchronizer = _fixture_synchronizer(fixturedef, runner, request)
  621. _make_asyncio_fixture_function(synchronizer, loop_scope)
  622. with MonkeyPatch.context() as c:
  623. c.setattr(fixturedef, "func", synchronizer)
  624. hook_result = yield
  625. return hook_result
  626. _DUPLICATE_LOOP_SCOPE_DEFINITION_ERROR = """\
  627. An asyncio pytest marker defines both "scope" and "loop_scope", \
  628. but it should only use "loop_scope".
  629. """
  630. _MARKER_SCOPE_KWARG_DEPRECATION_WARNING = """\
  631. The "scope" keyword argument to the asyncio marker has been deprecated. \
  632. Please use the "loop_scope" argument instead.
  633. """
  634. def _get_marked_loop_scope(
  635. asyncio_marker: Mark, default_loop_scope: _ScopeName
  636. ) -> _ScopeName:
  637. assert asyncio_marker.name == "asyncio"
  638. if asyncio_marker.args or (
  639. asyncio_marker.kwargs and set(asyncio_marker.kwargs) - {"loop_scope", "scope"}
  640. ):
  641. raise ValueError("mark.asyncio accepts only a keyword argument 'loop_scope'.")
  642. if "scope" in asyncio_marker.kwargs:
  643. if "loop_scope" in asyncio_marker.kwargs:
  644. raise pytest.UsageError(_DUPLICATE_LOOP_SCOPE_DEFINITION_ERROR)
  645. warnings.warn(PytestDeprecationWarning(_MARKER_SCOPE_KWARG_DEPRECATION_WARNING))
  646. scope = asyncio_marker.kwargs.get("loop_scope") or asyncio_marker.kwargs.get(
  647. "scope"
  648. )
  649. if scope is None:
  650. scope = default_loop_scope
  651. assert scope in {"function", "class", "module", "package", "session"}
  652. return scope
  653. def _get_default_test_loop_scope(config: Config) -> Any:
  654. return config.getini("asyncio_default_test_loop_scope")
  655. _RUNNER_TEARDOWN_WARNING = """\
  656. An exception occurred during teardown of an asyncio.Runner. \
  657. The reason is likely that you closed the underlying event loop in a test, \
  658. which prevents the cleanup of asynchronous generators by the runner.
  659. This warning will become an error in future versions of pytest-asyncio. \
  660. Please ensure that your tests don't close the event loop. \
  661. Here is the traceback of the exception triggered during teardown:
  662. %s
  663. """
  664. def _create_scoped_runner_fixture(scope: _ScopeName) -> Callable:
  665. @pytest.fixture(
  666. scope=scope,
  667. name=f"_{scope}_scoped_runner",
  668. )
  669. def _scoped_runner(
  670. event_loop_policy,
  671. request: FixtureRequest,
  672. ) -> Iterator[Runner]:
  673. new_loop_policy = event_loop_policy
  674. debug_mode = _get_asyncio_debug(request.config)
  675. with _temporary_event_loop_policy(new_loop_policy):
  676. runner = Runner(debug=debug_mode).__enter__()
  677. try:
  678. yield runner
  679. except Exception as e:
  680. runner.__exit__(type(e), e, e.__traceback__)
  681. else:
  682. with warnings.catch_warnings():
  683. warnings.filterwarnings(
  684. "ignore", ".*BaseEventLoop.shutdown_asyncgens.*", RuntimeWarning
  685. )
  686. try:
  687. runner.__exit__(None, None, None)
  688. except RuntimeError:
  689. warnings.warn(
  690. _RUNNER_TEARDOWN_WARNING % traceback.format_exc(),
  691. RuntimeWarning,
  692. )
  693. return _scoped_runner
  694. for scope in Scope:
  695. globals()[f"_{scope.value}_scoped_runner"] = _create_scoped_runner_fixture(
  696. scope.value
  697. )
  698. @pytest.fixture(scope="session", autouse=True)
  699. def event_loop_policy() -> AbstractEventLoopPolicy:
  700. """Return an instance of the policy used to create asyncio event loops."""
  701. return _get_event_loop_policy()
  702. def is_async_test(item: Item) -> TypeIs[PytestAsyncioFunction]:
  703. """Returns whether a test item is a pytest-asyncio test"""
  704. return isinstance(item, PytestAsyncioFunction)
  705. def _unused_port(socket_type: int) -> int:
  706. """Find an unused localhost port from 1024-65535 and return it."""
  707. with contextlib.closing(socket.socket(type=socket_type)) as sock:
  708. sock.bind(("127.0.0.1", 0))
  709. return sock.getsockname()[1]
  710. @pytest.fixture
  711. def unused_tcp_port() -> int:
  712. return _unused_port(socket.SOCK_STREAM)
  713. @pytest.fixture
  714. def unused_udp_port() -> int:
  715. return _unused_port(socket.SOCK_DGRAM)
  716. @pytest.fixture(scope="session")
  717. def unused_tcp_port_factory() -> Callable[[], int]:
  718. """A factory function, producing different unused TCP ports."""
  719. produced = set()
  720. def factory():
  721. """Return an unused port."""
  722. port = _unused_port(socket.SOCK_STREAM)
  723. while port in produced:
  724. port = _unused_port(socket.SOCK_STREAM)
  725. produced.add(port)
  726. return port
  727. return factory
  728. @pytest.fixture(scope="session")
  729. def unused_udp_port_factory() -> Callable[[], int]:
  730. """A factory function, producing different unused UDP ports."""
  731. produced = set()
  732. def factory():
  733. """Return an unused port."""
  734. port = _unused_port(socket.SOCK_DGRAM)
  735. while port in produced:
  736. port = _unused_port(socket.SOCK_DGRAM)
  737. produced.add(port)
  738. return port
  739. return factory