Merge branch 'master' into cheaper-no-other-refs

Author: graingert

docs/api.rst

@@ -200,6 +200,7 @@ Testing and debugging
 ---------------------
 
 .. autoclass:: anyio.TaskInfo
+.. autoclass:: anyio.pytest_plugin.FreePortFactory
 .. autofunction:: anyio.get_current_task
 .. autofunction:: anyio.get_running_tasks
 .. autofunction:: anyio.wait_all_tasks_blocked

docs/conf.py

@@ -28,7 +28,7 @@
 exclude_patterns = ["_build"]
 pygments_style = "sphinx"
 autodoc_default_options = {"members": True, "show-inheritance": True}
-autodoc_mock_imports = ["_typeshed"]
+autodoc_mock_imports = ["_typeshed", "pytest", "_pytest"]
 todo_include_todos = False
 
 html_theme = "sphinx_rtd_theme"

docs/testing.rst

@@ -140,6 +140,83 @@ scoped::
         yield
         await server.shutdown()
 
+Built-in utility fixtures
+-------------------------
+
+Some useful pytest fixtures are provided to make testing network services easier:
+
+* ``free_tcp_port_factory``: session scoped fixture returning a callable
+  (:class:`~pytest_plugin.FreePortFactory`) that generates unused TCP port numbers
+* ``free_udp_port_factory``: session scoped fixture returning a callable
+  (:class:`~pytest_plugin.FreePortFactory`) that generates unused UDP port numbers
+* ``free_tcp_port``: function level fixture that invokes the ``free_tcp_port_factory``
+  fixture to generate a free TCP port number
+* ``free_udp_port``: function level fixture that invokes the ``free_udp_port_factory``
+  fixture to generate a free UDP port number
+
+The use of these fixtures, in place of hard-coded ports numbers, will avoid errors due
+to a port already being allocated. In particular, they are a must for running multiple
+instances of the same test suite concurrently, either via ``pytest-xdist`` or ``tox`` or
+similar tools which can run the test suite in multiple interpreters in parallel.
+
+For example, you could set up a network listener in an ephemeral port and then connect
+to it::
+
+    from anyio import connect_tcp, create_task_group, create_tcp_listener
+    from anyio.abc import SocketStream
+
+
+    async def test_echo(free_tcp_port: int) -> None:
+        async def handle(client_stream: SocketStream) -> None:
+            async with client_stream:
+                payload = await client_stream.receive()
+                await client_stream.send(payload[::-1])
+
+        async with (
+            await create_tcp_listener(local_port=free_tcp_port) as listener,
+            create_task_group() as tg
+        ):
+            tg.start_soon(listener.serve, handle)
+
+            async with await connect_tcp("127.0.0.1", free_tcp_port) as stream:
+                await stream.send(b"hello")
+                assert await stream.receive() == b"olleh"
+
+            tg.cancel_scope.cancel()
+
+.. warning:::: It is possible in rare cases, particularly in local development, that
+    another process could bind to the port returned by one of these fixtures before your
+    code can do the same, leading to an :exc:`OSError` with the ``EADDRINUSE`` code. It
+    is advisable to just rerun the test if this happens.
+
+This is mostly useful with APIs that don't natively offer any way to bind to ephemeral
+ports (and retrieve those ports after binding). If you're working with AnyIO's own APIs,
+however, you could make use of this native capability::
+
+    from anyio import connect_tcp, create_task_group, create_tcp_listener
+    from anyio.abc import SocketAttribute, SocketStream
+
+    async def test_echo() -> None:
+        async def handle(client_stream: SocketStream) -> None:
+            async with client_stream:
+                payload = await client_stream.receive()
+                await client_stream.send(payload[::-1])
+
+        async with (
+            await create_tcp_listener(local_host="127.0.0.1") as listener,
+            create_task_group() as tg
+        ):
+            tg.start_soon(listener.serve, handle)
+            port = listener.extra(SocketAttribute.local_port)
+
+            async with await connect_tcp("127.0.0.1", port) as stream:
+                await stream.send(b"hello")
+                assert await stream.receive() == b"olleh"
+
+            tg.cancel_scope.cancel()
+
+.. versionadded:: 4.9.0
+
 Technical details
 -----------------
 

docs/versionhistory.rst

@@ -5,6 +5,16 @@ This library adheres to `Semantic Versioning 2.0 <http://semver.org/>`_.
 
 **UNRELEASED**
 
+- Added 4 new fixtures for the AnyIO ``pytest`` plugin:
+
+  * ``free_tcp_port_factory``: session scoped fixture returning a callable that
+    generates unused TCP port numbers
+  * ``free_udp_port_factory``: session scoped fixture returning a callable that
+    generates unused UDP port numbers
+  * ``free_tcp_port``: function scoped fixture that invokes the
+    ``free_tcp_port_factory`` fixture to generate a free TCP port number
+  * ``free_udp_port``: function scoped fixture that invokes the
+    ``free_udp_port_factory`` fixture to generate a free UDP port number
 - Added ``stdin`` argument to ``anyio.run_process()`` akin to what
   ``anyio.open_process()``, ``asyncio.create_subprocess_…()``, ``trio.run_process()``,
   and ``subprocess.run()`` already accept (PR by @jmehnle)

src/anyio/pytest_plugin.py

@@ -1,7 +1,8 @@
 from __future__ import annotations
 
+import socket
 import sys
-from collections.abc import Generator, Iterator
+from collections.abc import Callable, Generator, Iterator
 from contextlib import ExitStack, contextmanager
 from inspect import isasyncgenfunction, iscoroutinefunction, ismethod
 from typing import Any, cast
@@ -188,3 +189,84 @@ def anyio_backend_options(anyio_backend: Any) -> dict[str, Any]:
         return {}
     else:
         return anyio_backend[1]
+
+
+class FreePortFactory:
+    """
+    Manages port generation based on specified socket kind, ensuring no duplicate
+    ports are generated.
+
+    This class provides functionality for generating available free ports on the
+    system. It is initialized with a specific socket kind and can generate ports
+    for given address families while avoiding reuse of previously generated ports.
+
+    Users should not instantiate this class directly, but use the
+    ``free_tcp_port_factory`` and ``free_udp_port_factory`` fixtures instead. For simple
+    uses cases, ``free_tcp_port`` and ``free_udp_port`` can be used instead.
+    """
+
+    def __init__(self, kind: socket.SocketKind) -> None:
+        self._kind = kind
+        self._generated = set[int]()
+
+    @property
+    def kind(self) -> socket.SocketKind:
+        """
+        The type of socket connection (e.g., :data:`~socket.SOCK_STREAM` or
+        :data:`~socket.SOCK_DGRAM`) used to bind for checking port availability
+
+        """
+        return self._kind
+
+    def __call__(self, family: socket.AddressFamily | None = None) -> int:
+        """
+        Return an unbound port for the given address family.
+
+        :param family: if omitted, both IPv4 and IPv6 addresses will be tried
+        :return: a port number
+
+        """
+        if family is not None:
+            families = [family]
+        else:
+            families = [socket.AF_INET]
+            if socket.has_ipv6:
+                families.append(socket.AF_INET6)
+
+        while True:
+            port = 0
+            with ExitStack() as stack:
+                for family in families:
+                    sock = stack.enter_context(socket.socket(family, self._kind))
+                    addr = "::1" if family == socket.AF_INET6 else "127.0.0.1"
+                    try:
+                        sock.bind((addr, port))
+                    except OSError:
+                        break
+
+                    if not port:
+                        port = sock.getsockname()[1]
+                else:
+                    if port not in self._generated:
+                        self._generated.add(port)
+                        return port
+
+
+@pytest.fixture(scope="session")
+def free_tcp_port_factory() -> FreePortFactory:
+    return FreePortFactory(socket.SOCK_STREAM)
+
+
+@pytest.fixture(scope="session")
+def free_udp_port_factory() -> FreePortFactory:
+    return FreePortFactory(socket.SOCK_DGRAM)
+
+
+@pytest.fixture
+def free_tcp_port(free_tcp_port_factory: Callable[[], int]) -> int:
+    return free_tcp_port_factory()
+
+
+@pytest.fixture
+def free_udp_port(free_udp_port_factory: Callable[[], int]) -> int:
+    return free_udp_port_factory()

tests/test_pytest_plugin.py

@@ -1,15 +1,22 @@
 from __future__ import annotations
 
+import socket
+from collections.abc import Sequence
+
 import pytest
 from _pytest.logging import LogCaptureFixture
 from _pytest.pytester import Pytester
 
 from anyio import get_all_backends
+from anyio.pytest_plugin import FreePortFactory
 
-pytestmark = pytest.mark.filterwarnings(
-    "ignore:The TerminalReporter.writer attribute is deprecated"
-    ":pytest.PytestDeprecationWarning:"
-)
+pytestmark = [
+    pytest.mark.filterwarnings(
+        "ignore:The TerminalReporter.writer attribute is deprecated"
+        ":pytest.PytestDeprecationWarning:"
+    ),
+    pytest.mark.anyio,
+]
 
 pytest_args = "-v", "-p", "anyio", "-p", "no:asyncio", "-p", "no:trio"
 
@@ -561,3 +568,62 @@ async def test_params(fixt):
 
     result = testdir.runpytest(*pytest_args)
     result.assert_outcomes(passed=len(get_all_backends()) * 2)
+
+
+class TestFreePortFactory:
+    @pytest.fixture(scope="class")
+    def families(self) -> Sequence[tuple[socket.AddressFamily, str]]:
+        from .test_sockets import has_ipv6
+
+        families: list[tuple[socket.AddressFamily, str]] = [
+            (socket.AF_INET, "127.0.0.1")
+        ]
+        if has_ipv6:
+            families.append((socket.AF_INET6, "::1"))
+
+        return families
+
+    async def test_tcp_factory(
+        self,
+        families: Sequence[tuple[socket.AddressFamily, str]],
+        free_tcp_port_factory: FreePortFactory,
+    ) -> None:
+        generated_ports = {free_tcp_port_factory() for _ in range(5)}
+        assert all(isinstance(port, int) for port in generated_ports)
+        assert len(generated_ports) == 5
+        for port in generated_ports:
+            for family, addr in families:
+                with socket.socket(family, socket.SOCK_STREAM) as sock:
+                    try:
+                        sock.bind((addr, port))
+                    except OSError:
+                        pass
+
+    async def test_udp_factory(
+        self,
+        families: Sequence[tuple[socket.AddressFamily, str]],
+        free_udp_port_factory: FreePortFactory,
+    ) -> None:
+        generated_ports = {free_udp_port_factory() for _ in range(5)}
+        assert all(isinstance(port, int) for port in generated_ports)
+        assert len(generated_ports) == 5
+        for port in generated_ports:
+            for family, addr in families:
+                with socket.socket(family, socket.SOCK_DGRAM) as sock:
+                    sock.bind((addr, port))
+
+    async def test_free_tcp_port(
+        self, families: Sequence[tuple[socket.AddressFamily, str]], free_tcp_port: int
+    ) -> None:
+        assert isinstance(free_tcp_port, int)
+        for family, addr in families:
+            with socket.socket(family, socket.SOCK_STREAM) as sock:
+                sock.bind((addr, free_tcp_port))
+
+    async def test_free_udp_port(
+        self, families: Sequence[tuple[socket.AddressFamily, str]], free_udp_port: int
+    ) -> None:
+        assert isinstance(free_udp_port, int)
+        for family, addr in families:
+            with socket.socket(family, socket.SOCK_DGRAM) as sock:
+                sock.bind((addr, free_udp_port))