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

admin管理员组

文章数量:1128428

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 ===================================================================

本文标签： pythonpytest use the same TCP server between testsStack Overflow