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