pytest_plugin.py 12 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363
  1. from __future__ import annotations
  2. import dataclasses
  3. import socket
  4. import sys
  5. from collections.abc import Callable, Generator, Iterator
  6. from contextlib import ExitStack, contextmanager
  7. from inspect import isasyncgenfunction, iscoroutinefunction, ismethod
  8. from typing import Any, cast
  9. import pytest
  10. from _pytest.fixtures import FuncFixtureInfo, SubRequest
  11. from _pytest.outcomes import Exit
  12. from _pytest.python import CallSpec2
  13. from _pytest.scope import Scope
  14. from . import get_available_backends
  15. from ._core._eventloop import (
  16. current_async_library,
  17. get_async_backend,
  18. reset_current_async_library,
  19. set_current_async_library,
  20. )
  21. from ._core._exceptions import iterate_exceptions
  22. from .abc import TestRunner
  23. if sys.version_info < (3, 11):
  24. from exceptiongroup import ExceptionGroup
  25. _current_runner: TestRunner | None = None
  26. _runner_stack: ExitStack | None = None
  27. _runner_leases = 0
  28. def extract_backend_and_options(backend: object) -> tuple[str, dict[str, Any]]:
  29. if isinstance(backend, str):
  30. return backend, {}
  31. elif isinstance(backend, tuple) and len(backend) == 2:
  32. if isinstance(backend[0], str) and isinstance(backend[1], dict):
  33. return cast(tuple[str, dict[str, Any]], backend)
  34. raise TypeError("anyio_backend must be either a string or tuple of (string, dict)")
  35. @contextmanager
  36. def get_runner(
  37. backend_name: str, backend_options: dict[str, Any]
  38. ) -> Iterator[TestRunner]:
  39. global _current_runner, _runner_leases, _runner_stack
  40. if _current_runner is None:
  41. asynclib = get_async_backend(backend_name)
  42. _runner_stack = ExitStack()
  43. if current_async_library() is None:
  44. # Since we're in control of the event loop, we can cache the name of the
  45. # async library
  46. token = set_current_async_library(backend_name)
  47. _runner_stack.callback(reset_current_async_library, token)
  48. backend_options = backend_options or {}
  49. _current_runner = _runner_stack.enter_context(
  50. asynclib.create_test_runner(backend_options)
  51. )
  52. _runner_leases += 1
  53. try:
  54. yield _current_runner
  55. finally:
  56. _runner_leases -= 1
  57. if not _runner_leases:
  58. assert _runner_stack is not None
  59. _runner_stack.close()
  60. _runner_stack = _current_runner = None
  61. def pytest_addoption(parser: pytest.Parser) -> None:
  62. parser.addini(
  63. "anyio_mode",
  64. default="strict",
  65. help='AnyIO plugin mode (either "strict" or "auto")',
  66. )
  67. def pytest_configure(config: pytest.Config) -> None:
  68. config.addinivalue_line(
  69. "markers",
  70. "anyio: mark the (coroutine function) test to be run asynchronously via anyio.",
  71. )
  72. if (
  73. config.getini("anyio_mode") == "auto"
  74. and config.pluginmanager.has_plugin("asyncio")
  75. and config.getini("asyncio_mode") == "auto"
  76. ):
  77. config.issue_config_time_warning(
  78. pytest.PytestConfigWarning(
  79. "AnyIO auto mode has been enabled together with pytest-asyncio auto "
  80. "mode. This may cause unexpected behavior."
  81. ),
  82. 1,
  83. )
  84. @pytest.hookimpl(hookwrapper=True)
  85. def pytest_fixture_setup(fixturedef: Any, request: Any) -> Generator[Any]:
  86. def wrapper(anyio_backend: Any, request: SubRequest, **kwargs: Any) -> Any:
  87. # Rebind any fixture methods to the request instance
  88. if (
  89. request.instance
  90. and ismethod(func)
  91. and type(func.__self__) is type(request.instance)
  92. ):
  93. local_func = func.__func__.__get__(request.instance)
  94. else:
  95. local_func = func
  96. backend_name, backend_options = extract_backend_and_options(anyio_backend)
  97. if has_backend_arg:
  98. kwargs["anyio_backend"] = anyio_backend
  99. if has_request_arg:
  100. kwargs["request"] = request
  101. with get_runner(backend_name, backend_options) as runner:
  102. if isasyncgenfunction(local_func):
  103. yield from runner.run_asyncgen_fixture(local_func, kwargs)
  104. else:
  105. yield runner.run_fixture(local_func, kwargs)
  106. # Only apply this to coroutine functions and async generator functions in requests
  107. # that involve the anyio_backend fixture
  108. func = fixturedef.func
  109. if isasyncgenfunction(func) or iscoroutinefunction(func):
  110. if "anyio_backend" in request.fixturenames:
  111. fixturedef.func = wrapper
  112. original_argname = fixturedef.argnames
  113. if not (has_backend_arg := "anyio_backend" in fixturedef.argnames):
  114. fixturedef.argnames += ("anyio_backend",)
  115. if not (has_request_arg := "request" in fixturedef.argnames):
  116. fixturedef.argnames += ("request",)
  117. try:
  118. return (yield)
  119. finally:
  120. fixturedef.func = func
  121. fixturedef.argnames = original_argname
  122. return (yield)
  123. @pytest.hookimpl(tryfirst=True)
  124. def pytest_pycollect_makeitem(
  125. collector: pytest.Module | pytest.Class, name: str, obj: object
  126. ) -> None:
  127. if collector.istestfunction(obj, name):
  128. inner_func = obj.hypothesis.inner_test if hasattr(obj, "hypothesis") else obj
  129. if iscoroutinefunction(inner_func):
  130. anyio_auto_mode = collector.config.getini("anyio_mode") == "auto"
  131. marker = collector.get_closest_marker("anyio")
  132. own_markers = getattr(obj, "pytestmark", ())
  133. if (
  134. anyio_auto_mode
  135. or marker
  136. or any(marker.name == "anyio" for marker in own_markers)
  137. ):
  138. pytest.mark.usefixtures("anyio_backend")(obj)
  139. def pytest_collection_finish(session: pytest.Session) -> None:
  140. for i, item in reversed(list(enumerate(session.items))):
  141. if (
  142. isinstance(item, pytest.Function)
  143. and iscoroutinefunction(item.function)
  144. and item.get_closest_marker("anyio") is not None
  145. and "anyio_backend" not in item.fixturenames
  146. ):
  147. new_items = []
  148. try:
  149. cs_fields = {f.name for f in dataclasses.fields(CallSpec2)}
  150. except TypeError:
  151. cs_fields = set()
  152. for param_index, backend in enumerate(get_available_backends()):
  153. if "_arg2scope" in cs_fields: # pytest >= 8
  154. callspec = CallSpec2(
  155. params={"anyio_backend": backend},
  156. indices={"anyio_backend": param_index},
  157. _arg2scope={"anyio_backend": Scope.Module},
  158. _idlist=[backend],
  159. marks=[],
  160. )
  161. else: # pytest 7.x
  162. callspec = CallSpec2( # type: ignore[call-arg]
  163. funcargs={},
  164. params={"anyio_backend": backend},
  165. indices={"anyio_backend": param_index},
  166. arg2scope={"anyio_backend": Scope.Module},
  167. idlist=[backend],
  168. marks=[],
  169. )
  170. fi = item._fixtureinfo
  171. new_names_closure = list(fi.names_closure)
  172. if "anyio_backend" not in new_names_closure:
  173. new_names_closure.append("anyio_backend")
  174. new_fixtureinfo = FuncFixtureInfo(
  175. argnames=fi.argnames,
  176. initialnames=fi.initialnames,
  177. names_closure=new_names_closure,
  178. name2fixturedefs=fi.name2fixturedefs,
  179. )
  180. new_item = pytest.Function.from_parent(
  181. item.parent,
  182. name=f"{item.originalname}[{backend}]",
  183. callspec=callspec,
  184. callobj=item.obj,
  185. fixtureinfo=new_fixtureinfo,
  186. keywords=item.keywords,
  187. originalname=item.originalname,
  188. )
  189. new_items.append(new_item)
  190. session.items[i : i + 1] = new_items
  191. @pytest.hookimpl(tryfirst=True)
  192. def pytest_pyfunc_call(pyfuncitem: Any) -> bool | None:
  193. def run_with_hypothesis(**kwargs: Any) -> None:
  194. with get_runner(backend_name, backend_options) as runner:
  195. runner.run_test(original_func, kwargs)
  196. backend = pyfuncitem.funcargs.get("anyio_backend")
  197. if backend:
  198. backend_name, backend_options = extract_backend_and_options(backend)
  199. if hasattr(pyfuncitem.obj, "hypothesis"):
  200. # Wrap the inner test function unless it's already wrapped
  201. original_func = pyfuncitem.obj.hypothesis.inner_test
  202. if original_func.__qualname__ != run_with_hypothesis.__qualname__:
  203. if iscoroutinefunction(original_func):
  204. pyfuncitem.obj.hypothesis.inner_test = run_with_hypothesis
  205. return None
  206. if iscoroutinefunction(pyfuncitem.obj):
  207. funcargs = pyfuncitem.funcargs
  208. testargs = {arg: funcargs[arg] for arg in pyfuncitem._fixtureinfo.argnames}
  209. with get_runner(backend_name, backend_options) as runner:
  210. try:
  211. runner.run_test(pyfuncitem.obj, testargs)
  212. except ExceptionGroup as excgrp:
  213. for exc in iterate_exceptions(excgrp):
  214. if isinstance(exc, (Exit, KeyboardInterrupt, SystemExit)):
  215. raise exc from excgrp
  216. raise
  217. return True
  218. return None
  219. @pytest.fixture(scope="module", params=get_available_backends())
  220. def anyio_backend(request: Any) -> Any:
  221. return request.param
  222. @pytest.fixture
  223. def anyio_backend_name(anyio_backend: Any) -> str:
  224. if isinstance(anyio_backend, str):
  225. return anyio_backend
  226. else:
  227. return anyio_backend[0]
  228. @pytest.fixture
  229. def anyio_backend_options(anyio_backend: Any) -> dict[str, Any]:
  230. if isinstance(anyio_backend, str):
  231. return {}
  232. else:
  233. return anyio_backend[1]
  234. class FreePortFactory:
  235. """
  236. Manages port generation based on specified socket kind, ensuring no duplicate
  237. ports are generated.
  238. This class provides functionality for generating available free ports on the
  239. system. It is initialized with a specific socket kind and can generate ports
  240. for given address families while avoiding reuse of previously generated ports.
  241. Users should not instantiate this class directly, but use the
  242. ``free_tcp_port_factory`` and ``free_udp_port_factory`` fixtures instead. For simple
  243. uses cases, ``free_tcp_port`` and ``free_udp_port`` can be used instead.
  244. """
  245. def __init__(self, kind: socket.SocketKind) -> None:
  246. self._kind = kind
  247. self._generated = set[int]()
  248. @property
  249. def kind(self) -> socket.SocketKind:
  250. """
  251. The type of socket connection (e.g., :data:`~socket.SOCK_STREAM` or
  252. :data:`~socket.SOCK_DGRAM`) used to bind for checking port availability
  253. """
  254. return self._kind
  255. def __call__(self, family: socket.AddressFamily | None = None) -> int:
  256. """
  257. Return an unbound port for the given address family.
  258. :param family: if omitted, both IPv4 and IPv6 addresses will be tried
  259. :return: a port number
  260. """
  261. if family is not None:
  262. families = [family]
  263. else:
  264. families = [socket.AF_INET]
  265. if socket.has_ipv6:
  266. families.append(socket.AF_INET6)
  267. while True:
  268. port = 0
  269. with ExitStack() as stack:
  270. for family in families:
  271. sock = stack.enter_context(socket.socket(family, self._kind))
  272. addr = "::1" if family == socket.AF_INET6 else "127.0.0.1"
  273. try:
  274. sock.bind((addr, port))
  275. except OSError:
  276. break
  277. if not port:
  278. port = sock.getsockname()[1]
  279. else:
  280. if port not in self._generated:
  281. self._generated.add(port)
  282. return port
  283. @pytest.fixture(scope="session")
  284. def free_tcp_port_factory() -> FreePortFactory:
  285. return FreePortFactory(socket.SOCK_STREAM)
  286. @pytest.fixture(scope="session")
  287. def free_udp_port_factory() -> FreePortFactory:
  288. return FreePortFactory(socket.SOCK_DGRAM)
  289. @pytest.fixture
  290. def free_tcp_port(free_tcp_port_factory: Callable[[], int]) -> int:
  291. return free_tcp_port_factory()
  292. @pytest.fixture
  293. def free_udp_port(free_udp_port_factory: Callable[[], int]) -> int:
  294. return free_udp_port_factory()