pytest使用module作用域异步TCP服务器Fixture时遭遇RuntimeError问题求助

我想用pytest写一个小型测试程序，搭建一个TCP服务器发送数据并检查响应。当我使用不带作用域的@pytest.fixture()时，程序运行正常，测试也能通过，但这样每个测试都会创建一个新的TCP服务器，耗时有点久。可是当我改成@pytest.fixture(scope="module")之后，就出现了错误。有没有大佬碰到过类似的问题？或者有没有其他实现方式？

以下是我的TCP服务器代码：

#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

下面是我用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'

运行后出现的错误如下：

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

备注：内容来源于stack exchange，提问作者Quinn Van den Broucke