Merge pull request #14 from pytest-dev/unused_tcp_port_factory

Unused TCP port factory fixture. Bump to 0.2.0.

Author: nicoddemus

README.rst

@@ -30,6 +30,7 @@ Features
 --------
 
 - fixtures for creating and injecting versions of the asyncio event loop
+- fixtures for injecting unused tcp ports
 - pytest markers for treating tests as asyncio coroutines
 
 
@@ -72,9 +73,20 @@ The ``event_loop_process_pool`` fixture is almost identical to the
 
 ``unused_tcp_port``
 ~~~~~~~~~~~~~~~~~~~
-Finds and yields an unused TCP port on the localhost interface. Useful for
+Finds and yields a single unused TCP port on the localhost interface. Useful for
 binding temporary test servers.
 
+``unused_tcp_port_factory``
+~~~~~~~~~~~~~~~~~~~~~~~~~~~
+A callable which returns a different unused TCP port each invocation. Useful
+when several unused TCP ports are required in a test.
+
+.. code-block:: python
+
+    def a_test(unused_tcp_port_factory):
+        port1, port2 = unused_tcp_port_factory(), unused_tcp_port_factory()
+        ...
+
 Markers
 -------
 

pytest_asyncio/__init__.py

@@ -1 +1 @@
-__version__ = '0.1.3'
+__version__ = '0.2.0'

pytest_asyncio/plugin.py

@@ -88,3 +88,20 @@ def unused_tcp_port():
     with closing(socket.socket()) as sock:
         sock.bind(('127.0.0.1', 0))
         return sock.getsockname()[1]
+
+
+@pytest.fixture
+def unused_tcp_port_factory():
+    """A factory function, producing different unused TCP ports."""
+    produced = set()
+
+    def factory():
+        port = unused_tcp_port()
+
+        while port in produced:
+            port = unused_tcp_port()
+
+        produced.add(port)
+
+        return port
+    return factory

tests/test_simple.py

@@ -3,6 +3,8 @@
 import os
 import pytest
 
+import pytest_asyncio.plugin
+
 
 @asyncio.coroutine
 def async_coro(loop):
@@ -71,6 +73,61 @@ def closer(_, writer):
     yield from server1.wait_closed()
 
 
+@pytest.mark.asyncio
+def test_unused_port_factory_fixture(unused_tcp_port_factory, event_loop):
+    """Test the unused TCP port factory fixture."""
+
+    @asyncio.coroutine
+    def closer(_, writer):
+        writer.close()
+
+    port1, port2, port3 = (unused_tcp_port_factory(), unused_tcp_port_factory(),
+                           unused_tcp_port_factory())
+
+    server1 = yield from asyncio.start_server(closer, host='localhost',
+                                              port=port1,
+                                              loop=event_loop)
+    server2 = yield from asyncio.start_server(closer, host='localhost',
+                                              port=port2,
+                                              loop=event_loop)
+    server3 = yield from asyncio.start_server(closer, host='localhost',
+                                              port=port3,
+                                              loop=event_loop)
+
+    for port in port1, port2, port3:
+        with pytest.raises(IOError):
+            yield from asyncio.start_server(closer, host='localhost',
+                                            port=port,
+                                            loop=event_loop)
+
+    server1.close()
+    yield from server1.wait_closed()
+    server2.close()
+    yield from server2.wait_closed()
+    server3.close()
+    yield from server3.wait_closed()
+
+
+def test_unused_port_factory_duplicate(unused_tcp_port_factory, monkeypatch):
+    """Test correct avoidance of duplicate ports."""
+    counter = 0
+
+    def mock_unused_tcp_port():
+        """Force some duplicate ports."""
+        nonlocal counter
+        counter += 1
+        if counter < 5:
+            return 10000
+        else:
+            return 10000 + counter
+
+    monkeypatch.setattr(pytest_asyncio.plugin, 'unused_tcp_port',
+                        mock_unused_tcp_port)
+
+    assert unused_tcp_port_factory() == 10000
+    assert unused_tcp_port_factory() > 10000
+
+
 class Test:
     """Test that asyncio marked functions work in test methods."""