python - pytest: use the same TCP server between tests - Stack Overflow

I wanted to create a small test program using pytest to setup a TCP server that sends some data and checks the response. When using @pytest.fixture() without defining a scope, the program works fine and the test passes. However this creates a new TCP server for each test which can take quite a bit of time. When I put @pytest.fixture(scope="module") there are errors. Does anyone have any experience with this? Or how this can be implemented another way?

#tcp_server.py

import asyncio


class TCPServer:
    def __init__(self, host, port):
        self.host = host
        self.port = port
        self.last_received_data = None
        self.header_function = None
        self.data_received_event = asyncio.Event()

    async def start(self):
        server = await asyncio.start_server(self.handle_client, self.host, self.port)
        print(f"Server started at {self.host}:{self.port}")
        async with server:
            await server.serve_forever()

    async def handle_client(self, reader, writer):
        header = self.header_function()

        writer.write(header)
        await writer.drain()

        self.last_received_data = await reader.read(1024)
        self.data_received_event.set()
        print(f"Received from client: {self.last_received_data}")
        writer.close()

    def set_header_function(self, func):
        self.header_function = func

And here is the code for the test program using pytest

#test_tcp.py

@pytest.fixture(scope="module")
async def tcp_server():
    server = TCPServer('localhost', 3102)
    task = asyncio.create_task(server.start())
    yield server
    task.cancel()


async def test_motor_identification_dataFL(tcp_server):
    tcp_server.set_header_function(lambda: create_motor_identification_header(110))

    try:
        await asyncio.wait_for(tcp_server.data_received_event.wait(), timeout=10.0)
    except asyncio.TimeoutError:
        pytest.fail("Timeout waiting for the server to receive data")

    assert tcp_server.last_received_data == b'\x00\x06'

Here are the errors:

C:\BEP\Software\Envs\pythontests\lib\site-packages\pytest_asyncio\plugin.py:207: PytestDeprecationWarning: The configuration option "asyncio_default_fixture_loop_scope" is unset.
The event loop scope for asynchronous fixtures will default to the fixture caching scope. Future versions of pytest-asyncio will default the loop scope for asynchronous fixtures to function scope. Set the default fixture loop scope explicitly in order to avoid unexpected behavior in the future. Valid fixture loop scopes are: "function", "class", "module", "package", "session"

  warnings.warn(PytestDeprecationWarning(_DEFAULT_FIXTURE_LOOP_SCOPE_UNSET))
================================================================== test session starts ==================================================================
platform win32 -- Python 3.9.13, pytest-8.3.4, pluggy-1.5.0
rootdir: C:\BEP\Software\Projects\Customer\MercedesRbWebApp\Tests
plugins: asyncio-0.25.2
asyncio: mode=auto, asyncio_default_fixture_loop_scope=None
collected 1 item

test_tcp.py F                                                                                                                                      [100%]

======================================================================= FAILURES ========================================================================
___________________________________________________________ test_motor_identification_dataFL ____________________________________________________________

tcp_server = <tcp_server.TCPServer object at 0x00000290851F79D0>

    async def test_motor_identification_dataFL(tcp_server):
        tcp_server.set_header_function(lambda: create_motor_identification_header(110))

        try:
>           await asyncio.wait_for(tcp_server.data_received_event.wait(), timeout=10.0)

test_tcp.py:20:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
C:\BEP\System\Python\Python39\lib\asyncio\tasks.py:479: in wait_for
    return fut.result()
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

self = <asyncio.locks.Event object at 0x00000290851F7DF0 [unset]>

    async def wait(self):
        """Block until the internal flag is true.

        If the internal flag is true on entry, return True
        immediately.  Otherwise, block until another coroutine calls
        set() to set the flag to true, then return True.
        """
        if self._value:
            return True

        fut = self._loop.create_future()
        self._waiters.append(fut)
        try:
>           await fut
E           RuntimeError: Task <Task pending name='Task-5' coro=<Event.wait() running at C:\BEP\System\Python\Python39\lib\asyncio\locks.py:226> cb=[_release_waiter(<Future pendi...085203C10>()]>)() at C:\BEP\System\Python\Python39\lib\asyncio\tasks.py:416]> got Future <Future pending> attached to a different loop

C:\BEP\System\Python\Python39\lib\asyncio\locks.py:226: RuntimeError
================================================================ short test summary info ================================================================
FAILED test_tcp.py::test_motor_identification_dataFL - RuntimeError: Task <Task pending name='Task-5' coro=<Event.wait() running at C:\BEP\System\Python\Python39\lib\asyncio\locks.py:226> cb=[_release_wai...
=================================================================== 1 failed in 0.06s ===================================================================

I wanted to create a small test program using pytest to setup a TCP server that sends some data and checks the response. When using @pytest.fixture() without defining a scope, the program works fine and the test passes. However this creates a new TCP server for each test which can take quite a bit of time. When I put @pytest.fixture(scope="module") there are errors. Does anyone have any experience with this? Or how this can be implemented another way?

#tcp_server.py

import asyncio


class TCPServer:
    def __init__(self, host, port):
        self.host = host
        self.port = port
        self.last_received_data = None
        self.header_function = None
        self.data_received_event = asyncio.Event()

    async def start(self):
        server = await asyncio.start_server(self.handle_client, self.host, self.port)
        print(f"Server started at {self.host}:{self.port}")
        async with server:
            await server.serve_forever()

    async def handle_client(self, reader, writer):
        header = self.header_function()

        writer.write(header)
        await writer.drain()

        self.last_received_data = await reader.read(1024)
        self.data_received_event.set()
        print(f"Received from client: {self.last_received_data}")
        writer.close()

    def set_header_function(self, func):
        self.header_function = func

And here is the code for the test program using pytest

#test_tcp.py

@pytest.fixture(scope="module")
async def tcp_server():
    server = TCPServer('localhost', 3102)
    task = asyncio.create_task(server.start())
    yield server
    task.cancel()


async def test_motor_identification_dataFL(tcp_server):
    tcp_server.set_header_function(lambda: create_motor_identification_header(110))

    try:
        await asyncio.wait_for(tcp_server.data_received_event.wait(), timeout=10.0)
    except asyncio.TimeoutError:
        pytest.fail("Timeout waiting for the server to receive data")

    assert tcp_server.last_received_data == b'\x00\x06'

Here are the errors:

C:\BEP\Software\Envs\pythontests\lib\site-packages\pytest_asyncio\plugin.py:207: PytestDeprecationWarning: The configuration option "asyncio_default_fixture_loop_scope" is unset.
The event loop scope for asynchronous fixtures will default to the fixture caching scope. Future versions of pytest-asyncio will default the loop scope for asynchronous fixtures to function scope. Set the default fixture loop scope explicitly in order to avoid unexpected behavior in the future. Valid fixture loop scopes are: "function", "class", "module", "package", "session"

  warnings.warn(PytestDeprecationWarning(_DEFAULT_FIXTURE_LOOP_SCOPE_UNSET))
================================================================== test session starts ==================================================================
platform win32 -- Python 3.9.13, pytest-8.3.4, pluggy-1.5.0
rootdir: C:\BEP\Software\Projects\Customer\MercedesRbWebApp\Tests
plugins: asyncio-0.25.2
asyncio: mode=auto, asyncio_default_fixture_loop_scope=None
collected 1 item

test_tcp.py F                                                                                                                                      [100%]

======================================================================= FAILURES ========================================================================
___________________________________________________________ test_motor_identification_dataFL ____________________________________________________________

tcp_server = <tcp_server.TCPServer object at 0x00000290851F79D0>

    async def test_motor_identification_dataFL(tcp_server):
        tcp_server.set_header_function(lambda: create_motor_identification_header(110))

        try:
>           await asyncio.wait_for(tcp_server.data_received_event.wait(), timeout=10.0)

test_tcp.py:20:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
C:\BEP\System\Python\Python39\lib\asyncio\tasks.py:479: in wait_for
    return fut.result()
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

self = <asyncio.locks.Event object at 0x00000290851F7DF0 [unset]>

    async def wait(self):
        """Block until the internal flag is true.

        If the internal flag is true on entry, return True
        immediately.  Otherwise, block until another coroutine calls
        set() to set the flag to true, then return True.
        """
        if self._value:
            return True

        fut = self._loop.create_future()
        self._waiters.append(fut)
        try:
>           await fut
E           RuntimeError: Task <Task pending name='Task-5' coro=<Event.wait() running at C:\BEP\System\Python\Python39\lib\asyncio\locks.py:226> cb=[_release_waiter(<Future pendi...085203C10>()]>)() at C:\BEP\System\Python\Python39\lib\asyncio\tasks.py:416]> got Future <Future pending> attached to a different loop

C:\BEP\System\Python\Python39\lib\asyncio\locks.py:226: RuntimeError
================================================================ short test summary info ================================================================
FAILED test_tcp.py::test_motor_identification_dataFL - RuntimeError: Task <Task pending name='Task-5' coro=<Event.wait() running at C:\BEP\System\Python\Python39\lib\asyncio\locks.py:226> cb=[_release_wai...
=================================================================== 1 failed in 0.06s ===================================================================

• python
• pytest
• python-asyncio
• pytest-asyncio

Share Improve this question Follow asked Jan 8 at 13:26 Quinn Van den BrouckeQuinn Van den Broucke 3366 bronze badges 5

• Try to change the asyncio_default_fixture_loop_scope to "module". Docs here: pytest-asyncio.readthedocs.io/en/latest/reference/… – VPfB Commented Jan 11 at 20:31
• I added asyncio_default_fixture_loop_scope = module in the pytest.ini file, but it still gives the same error: got Future <Future pending> attached to a different loop – Quinn Van den Broucke Commented Jan 13 at 7:34
• I tried for a while to find out what's wrong. I did not make any progress. The loop running the fixture differs from the loop running the test and I don't know why. After reading some issues at the project's github page I'm afraid there are bugs in pytest-asyncio. Several users complained about issues like this: github.com/pytest-dev/pytest-asyncio/issues/868 – VPfB Commented Jan 14 at 11:54
• Thank you for providing more information. I'll keep an eye on the pytest-asyncio repo and its issues. Just to be complete when people are reading this, I'm using pytest-asyncio verion 0.25.2 – Quinn Van den Broucke Commented Jan 14 at 12:36
• Version 26.0 has now two loop scope config settings, maybe it's worth to give it a try. I have no time to test it myself. pytest-asyncio.readthedocs.io/en/stable/reference/… – VPfB Commented Apr 15 at 20:33

Add a comment  | 

1 Answer 1

Sorted by: Reset to default 0

You'll have to use pytest-asyncio fixture. The loop scope must be a higher level than the fixture scope (session > module > function)

Documentation: https://pytest-asyncio.readthedocs.io/en/latest/reference/decorators/index.html#decorators-pytest-asyncio-fixture

Define the fixture like this:

@pytest_asyncio.fixture(loop_scope="session", scope="module")
async def tcp_server():
    ...

and when marking the tests as async tests, use the loop_scope="session" too:

pytestmark = pytest.mark.asyncio(loop_scope="session")

or

@pytest.mark.asyncio(loop_scope="session")
async def test_motor_identification_dataFL(tcp_server):
    ...