From f63934ffaf9495a51dc6d6b522ee79bba2537ec7 Mon Sep 17 00:00:00 2001 From: "Dominik S. Buse" Date: Tue, 2 Oct 2018 21:18:34 +0200 Subject: [PATCH 1/3] Add unused port helpers for UDP Extends the unused_tcp_port and unused_tcp_port_factory mechanisms for UDP ports. --- README.rst | 5 ++ pytest_asyncio/plugin.py | 35 +++++++++++--- tests/test_simple.py | 100 ++++++++++++++++++++++++++++++++++++++- 3 files changed, 133 insertions(+), 7 deletions(-) diff --git a/README.rst b/README.rst index 8c36da8a..972c1269 100644 --- a/README.rst +++ b/README.rst @@ -107,6 +107,11 @@ when several unused TCP ports are required in a test. port1, port2 = unused_tcp_port_factory(), unused_tcp_port_factory() ... +``unused_udp_port`` and ``unused_udp_port_factory`` +~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ +Work just like their TCP counterparts but return unused UDP ports. + + Async fixtures ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ Asynchronous fixtures are defined just like ordinary pytest fixtures, except they should be coroutines or asynchronous generators. diff --git a/pytest_asyncio/plugin.py b/pytest_asyncio/plugin.py index 584efbf4..bafde516 100644 --- a/pytest_asyncio/plugin.py +++ b/pytest_asyncio/plugin.py @@ -174,16 +174,21 @@ def event_loop(request): loop.close() -def _unused_tcp_port(): - """Find an unused localhost TCP port from 1024-65535 and return it.""" - with contextlib.closing(socket.socket()) as sock: +def _unused_port(socket_type=socket.SOCK_STREAM): + """Find an unused localhost port from 1024-65535 and return it.""" + with contextlib.closing(socket.socket(type=socket_type)) as sock: sock.bind(('127.0.0.1', 0)) return sock.getsockname()[1] @pytest.fixture def unused_tcp_port(): - return _unused_tcp_port() + return _unused_port() + + +@pytest.fixture +def unused_udp_port(): + return _unused_port(socket.SOCK_DGRAM) @pytest.fixture @@ -193,10 +198,28 @@ def unused_tcp_port_factory(): def factory(): """Return an unused port.""" - port = _unused_tcp_port() + port = _unused_port() + + while port in produced: + port = _unused_port() + + produced.add(port) + + return port + return factory + + +@pytest.fixture +def unused_udp_port_factory(): + """A factory function, producing different unused UDP ports.""" + produced = set() + + def factory(): + """Return an unused port.""" + port = _unused_port(socket.SOCK_DGRAM) while port in produced: - port = _unused_tcp_port() + port = _unused_port(socket.SOCK_DGRAM) produced.add(port) diff --git a/tests/test_simple.py b/tests/test_simple.py index f9e3992f..fc47b19f 100644 --- a/tests/test_simple.py +++ b/tests/test_simple.py @@ -57,6 +57,35 @@ async def closer(_, writer): await server1.wait_closed() +@pytest.mark.asyncio +async def test_unused_udp_port_fixture(unused_udp_port, event_loop): + """Test the unused TCP port fixture.""" + + class Closer: + def connection_made(self, transport): + pass + + def connection_lost(self, *arg, **kwd): + pass + + transport1, _ = await event_loop.create_datagram_endpoint( + Closer, + local_addr=('127.0.0.1', unused_udp_port), + reuse_port=False, + reuse_address=False, + ) + + with pytest.raises(IOError): + await event_loop.create_datagram_endpoint( + Closer, + local_addr=('127.0.0.1', unused_udp_port), + reuse_port=False, + reuse_address=False, + ) + + transport1.abort() + + @pytest.mark.asyncio async def test_unused_port_factory_fixture(unused_tcp_port_factory, event_loop): """Test the unused TCP port factory fixture.""" @@ -91,6 +120,54 @@ async def closer(_, writer): await server3.wait_closed() +@pytest.mark.asyncio +async def test_unused_udp_port_factory_fixture(unused_udp_port_factory, + event_loop): + """Test the unused UDP port factory fixture.""" + + class Closer: + def connection_made(self, transport): + pass + + def connection_lost(self, *arg, **kwd): + pass + + port1, port2, port3 = (unused_udp_port_factory(), unused_udp_port_factory(), + unused_udp_port_factory()) + + transport1, _ = await event_loop.create_datagram_endpoint( + Closer, + local_addr=('127.0.0.1', port1), + reuse_port=False, + reuse_address=False, + ) + transport2, _ = await event_loop.create_datagram_endpoint( + Closer, + local_addr=('127.0.0.1', port2), + reuse_port=False, + reuse_address=False, + ) + transport3, _ = await event_loop.create_datagram_endpoint( + Closer, + local_addr=('127.0.0.1', port3), + reuse_port=False, + reuse_address=False, + ) + + for port in port1, port2, port3: + with pytest.raises(IOError): + await event_loop.create_datagram_endpoint( + Closer, + local_addr=('127.0.0.1', port), + reuse_port=False, + reuse_address=False, + ) + + transport1.abort() + transport2.abort() + transport3.abort() + + def test_unused_port_factory_duplicate(unused_tcp_port_factory, monkeypatch): """Test correct avoidance of duplicate ports.""" counter = 0 @@ -104,13 +181,34 @@ def mock_unused_tcp_port(): else: return 10000 + counter - monkeypatch.setattr(pytest_asyncio.plugin, '_unused_tcp_port', + monkeypatch.setattr(pytest_asyncio.plugin, '_unused_port', mock_unused_tcp_port) assert unused_tcp_port_factory() == 10000 assert unused_tcp_port_factory() > 10000 +def test_unused_udp_port_factory_duplicate(unused_udp_port_factory, + monkeypatch): + """Test correct avoidance of duplicate UDP ports.""" + counter = 0 + + def mock_unused_udp_port(_ignored): + """Force some duplicate ports.""" + nonlocal counter + counter += 1 + if counter < 5: + return 10000 + else: + return 10000 + counter + + monkeypatch.setattr(pytest_asyncio.plugin, '_unused_port', + mock_unused_udp_port) + + assert unused_udp_port_factory() == 10000 + assert unused_udp_port_factory() > 10000 + + class Test: """Test that asyncio marked functions work in test methods.""" From 0a4ebb9d0301d66215f87213c42df52fe3d779be Mon Sep 17 00:00:00 2001 From: Andrew Svetlov Date: Fri, 7 Jan 2022 11:49:39 +0200 Subject: [PATCH 2/3] Update pytest_asyncio/plugin.py --- pytest_asyncio/plugin.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pytest_asyncio/plugin.py b/pytest_asyncio/plugin.py index bafde516..689c84c5 100644 --- a/pytest_asyncio/plugin.py +++ b/pytest_asyncio/plugin.py @@ -209,7 +209,7 @@ def factory(): return factory -@pytest.fixture +@pytest.fixture(scope=session) def unused_udp_port_factory(): """A factory function, producing different unused UDP ports.""" produced = set() From 1e3905aae3d805510c6f45d8b6be167a7e02d251 Mon Sep 17 00:00:00 2001 From: Andrew Svetlov Date: Fri, 7 Jan 2022 14:02:03 +0200 Subject: [PATCH 3/3] Add changenote --- README.rst | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/README.rst b/README.rst index 015c55b9..7330e1a9 100644 --- a/README.rst +++ b/README.rst @@ -171,7 +171,7 @@ Note about unittest ------------------- Test classes subclassing the standard `unittest `__ library are not supported, users -are recommended to use `unitest.IsolatedAsyncioTestCase `__ +are recommended to use `unitest.IsolatedAsyncioTestCase `__ or an async framework such as `asynctest `__. Changelog @@ -182,6 +182,7 @@ Changelog - Drop support for Python 3.6 - Fixed an issue when pytest-asyncio was used in combination with `flaky` or inherited asynchronous Hypothesis tests. `#178 `_ `#231 `_ - Added `flaky `_ to test dependencies +- Added ``unused_udp_port`` and ``unused_udp_port_factory`` fixtures (similar to ``unused_tcp_port`` and ``unused_tcp_port_factory`` counterparts. `#99 `_ 0.16.0 (2021-10-16) ~~~~~~~~~~~~~~~~~~~