Deprecate unused TCP and UDP port fixtures

changelog.d/1614.deprecated.rst

@@ -0,0 +1 @@
+The ``unused_tcp_port``, ``unused_tcp_port_factory``, ``unused_udp_port``, and ``unused_udp_port_factory`` fixtures are deprecated because the returned port can be claimed by another process before the test binds it. Bind to port 0 and read the assigned port from the server instead.

docs/how-to-guides/avoid_unused_port_fixtures.rst

@@ -0,0 +1,80 @@
+Avoid the unused port fixtures
+==============================
+
+The ``unused_tcp_port``, ``unused_tcp_port_factory``, ``unused_udp_port``, and
+``unused_udp_port_factory`` fixtures are deprecated because the port they
+return can be claimed by another process before the test binds it.
+
+Bind to port 0 instead and read the port assigned by the operating system.
+
+Migration
+---------
+
+Replace the fixtures with the patterns below:
+
+- ``unused_tcp_port`` / ``unused_udp_port``: bind the server or datagram
+  endpoint to port 0 and read the port assigned by the operating system.
+  The examples below show the full pattern for TCP and UDP.
+- ``unused_tcp_port_factory`` / ``unused_udp_port_factory``: create each
+  server or datagram endpoint on port 0 and read its assigned port. If the
+  port must be passed to another process before the server starts, bind the
+  socket in the test and pass the listening socket (or use a
+  ``socketpair``/pipe) instead of a bare port number; a bare port can always
+  be claimed before the other process binds it.
+
+The deprecated fixtures remain available and keep working, but they emit a
+``DeprecationWarning``. The warning and this guide describe the same
+migration; the guide exists because the fixtures are public API and the
+warning alone does not show the replacement pattern.
+
+TCP
+---
+
+.. code-block:: python
+
+    import asyncio
+
+    import pytest
+
+
+    async def handler(reader, writer):
+        writer.close()
+
+
+    @pytest.mark.asyncio
+    async def test_server():
+        server = await asyncio.start_server(handler, host="127.0.0.1", port=0)
+        port = server.sockets[0].getsockname()[1]
+        try:
+            reader, writer = await asyncio.open_connection("127.0.0.1", port)
+            writer.close()
+            await writer.wait_closed()
+        finally:
+            server.close()
+            await server.wait_closed()
+
+UDP
+---
+
+.. code-block:: python
+
+    import asyncio
+
+    import pytest
+
+
+    class Protocol(asyncio.DatagramProtocol):
+        pass
+
+
+    @pytest.mark.asyncio
+    async def test_datagram_endpoint():
+        loop = asyncio.get_running_loop()
+        transport, _ = await loop.create_datagram_endpoint(
+            Protocol, local_addr=("127.0.0.1", 0)
+        )
+        port = transport.get_extra_info("sockname")[1]
+        try:
+            transport.sendto(b"ping", ("127.0.0.1", port))
+        finally:
+            transport.close()

docs/how-to-guides/index.rst

@@ -19,5 +19,6 @@ How-To Guides
   parametrize_with_asyncio
   uvloop
   test_item_is_async
+  avoid_unused_port_fixtures
 
 This section of the documentation provides code snippets and recipes to accomplish specific tasks with pytest-asyncio.

docs/reference/fixtures/index.rst

@@ -2,6 +2,12 @@
 Fixtures
 ========
 
+.. warning::
+
+   The unused port fixtures are deprecated because the returned port can be
+   claimed before the test binds it. See
+   :doc:`../../how-to-guides/avoid_unused_port_fixtures`.
+
 unused_tcp_port
 ===============
 Finds and yields a single unused TCP port on the localhost interface. Useful for

pytest_asyncio/plugin.py

@@ -1073,19 +1073,32 @@ def _unused_port(socket_type: int) -> int:
         return sock.getsockname()[1]
 
 
+def _warn_deprecated_port_fixture(name: str) -> None:
+    warnings.warn(
+        f"The {name} fixture is deprecated because its port can be "
+        "claimed before the test binds it. Bind to port 0 and read the "
+        "assigned port from the server instead.",
+        DeprecationWarning,
+        stacklevel=2,
+    )
+
+
 @pytest.fixture
 def unused_tcp_port() -> int:
+    _warn_deprecated_port_fixture("unused_tcp_port")
     return _unused_port(socket.SOCK_STREAM)
 
 
 @pytest.fixture
 def unused_udp_port() -> int:
+    _warn_deprecated_port_fixture("unused_udp_port")
     return _unused_port(socket.SOCK_DGRAM)
 
 
 @pytest.fixture(scope="session")
 def unused_tcp_port_factory() -> Callable[[], int]:
     """A factory function, producing different unused TCP ports."""
+    _warn_deprecated_port_fixture("unused_tcp_port_factory")
     produced = set()
 
     def factory():
@@ -1105,6 +1118,7 @@ def factory():
 @pytest.fixture(scope="session")
 def unused_udp_port_factory() -> Callable[[], int]:
     """A factory function, producing different unused UDP ports."""
+    _warn_deprecated_port_fixture("unused_udp_port_factory")
     produced = set()
 
     def factory():

tests/test_port_factories.py

@@ -2,12 +2,18 @@
 
 from textwrap import dedent
 
+import pytest
 from pytest import Pytester
 
 import pytest_asyncio.plugin
 
+pytestmark = pytest.mark.filterwarnings(
+    "ignore:The unused_.* fixture is deprecated:DeprecationWarning"
+)
+
 
 def test_unused_tcp_port_selects_unused_port(pytester: Pytester):
+    pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function")
     pytester.makepyfile(dedent("""\
             import asyncio
 
@@ -31,9 +37,17 @@ async def closer(_, writer):
                 await server1.wait_closed()
             """))
 
+    result = pytester.runpytest("-W", "ignore:The unused_:DeprecationWarning")
+    result.assert_outcomes(passed=1)
+
 
 def test_unused_udp_port_selects_unused_port(pytester: Pytester):
+    pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function")
     pytester.makepyfile(dedent("""\
+            import asyncio
+
+            import pytest
+
             @pytest.mark.asyncio
             async def test_unused_udp_port_fixture(unused_udp_port):
                 class Closer:
@@ -60,9 +74,17 @@ def connection_lost(self, *arg, **kwd):
                 transport1.abort()
             """))
 
+    result = pytester.runpytest("-W", "ignore:The unused_:DeprecationWarning")
+    result.assert_outcomes(passed=1)
+
 
 def test_unused_tcp_port_factory_selects_unused_port(pytester: Pytester):
+    pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function")
     pytester.makepyfile(dedent("""\
+            import asyncio
+
+            import pytest
+
             @pytest.mark.asyncio
             async def test_unused_port_factory_fixture(unused_tcp_port_factory):
                 async def closer(_, writer):
@@ -96,9 +118,17 @@ async def closer(_, writer):
                 await server3.wait_closed()
             """))
 
+    result = pytester.runpytest("-W", "ignore:The unused_:DeprecationWarning")
+    result.assert_outcomes(passed=1)
+
 
 def test_unused_udp_port_factory_selects_unused_port(pytester: Pytester):
+    pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function")
     pytester.makepyfile(dedent("""\
+            import asyncio
+
+            import pytest
+
             @pytest.mark.asyncio
             async def test_unused_udp_port_factory_fixture(unused_udp_port_factory):
                 class Closer:
@@ -144,6 +174,9 @@ def connection_lost(self, *arg, **kwd):
                 transport3.abort()
             """))
 
+    result = pytester.runpytest("-W", "ignore:The unused_:DeprecationWarning")
+    result.assert_outcomes(passed=1)
+
 
 def test_unused_port_factory_duplicate(unused_tcp_port_factory, monkeypatch):
     """Test correct avoidance of duplicate ports."""

tests/test_port_fixture_deprecation.py

@@ -0,0 +1,33 @@
+from __future__ import annotations
+
+from textwrap import dedent
+
+import pytest
+from pytest import Pytester
+
+
+@pytest.mark.parametrize(
+    "fixture_name",
+    [
+        "unused_tcp_port",
+        "unused_udp_port",
+        "unused_tcp_port_factory",
+        "unused_udp_port_factory",
+    ],
+)
+def test_port_fixtures_are_deprecated(pytester: Pytester, fixture_name: str) -> None:
+    pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function")
+    pytester.makepyfile(dedent(f"""\
+            import pytest
+
+            def test_deprecated(request):
+                with pytest.warns(
+                    DeprecationWarning,
+                    match="The {fixture_name} fixture is deprecated",
+                ):
+                    request.getfixturevalue("{fixture_name}")
+            """))
+
+    result = pytester.runpytest()
+
+    result.assert_outcomes(passed=1)