Pytest: используйте один и тот же TCP-сервер между тестами ⇐ Python

[Anonymous]

Я хотел создать небольшую тестовую программу, используя pytest для настройки TCP-сервера, который отправляет некоторые данные и проверяет ответ. При использовании @pytest.fixture() без определения области программа работает нормально, и тест проходит. Однако при этом для каждого теста создается новый TCP-сервер, что может занять довольно много времени. Когда я ставлю @pytest.fixture(scope="module"), возникают ошибки. Есть ли у кого-нибудь опыт в этом вопросе? Или как это можно реализовать по-другому?
#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 =

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 =

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 )() at C:\BEP\System\Python\Python39\lib\asyncio\tasks.py:416]> got Future 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

Подробнее здесь: https://stackoverflow.com/questions/793 ... ween-tests