Адрес уже используется в асинхронных клиентских тестах в Python ⇐ Python

[Anonymous]

У меня вопрос по поводу асинхронных клиентских тестов. Вот мой код моего теста

Код: Выделить всё

class TestSocketClient:
@classmethod
def setup_class(cls):
# enable parallel testing by adding the pytest worker id number to the default port
worker_id = os.environ.get("PYTEST_XDIST_WORKER", "gw0")
worker_number = int(worker_id[2:])

cls.mock_server = ServerMock()
cls.mock_server.port = ServerMock.port + worker_number

cls.username = os.environ.get("USERNAME", "")
cls.password = os.environ.get("PASSWORD", "")
cls.socket_client = SocketClient(username=cls.username, password=cls.password)
cls.socket_client.hostname = "0.0.0.0"
cls.socket_client.port = SocketClient.port + worker_number

@pytest.mark.asyncio
async def test_login(self):
await self.mock_server.start()
response = await self.socket_client.login()
assert response == actual_response
await self.socket_client.close()

@pytest.mark.asyncio
async def test_send_heartbeat(self):
await self.mock_server.start()
await self.socket_client.login()
await self.socket_client.send_heartbeat()
await self.socket_client.close()

Я могу запускать тесты в TestSocketClient индивидуально, и они пройдут индивидуально. Но когда я запускаю наборы тестов вместе с pytest -n auto, последний тестовый пример вызовет ошибку при попытке привязки по адресу ('0.0.0.0', 2056): адрес уже используется. Мой вопрос заключается в том, как обеспечить прохождение наборов тестов без решения проблем с распределением, чтобы они могли успешно работать в процессе CI. Я буду очень признателен, если появится более ценное предложение по написанию асинхронных тестов (например, что мне следует утверждать, если я хочу протестировать только запрос, который клиент хотел бы отправить на сервер? Должен ли я утверждать сообщение, полученное на на стороне сервера или просто напишите что-то вроде Assert_called_once на стороне клиента). Заранее спасибо!
Обновления:
Я наконец-то решил проблему с увеличением порта в различных тестах, как показано ниже.

Код: Выделить всё

class TestSocketClient:
ports_taken = set()

@classmethod
def setup_class(cls):

cls.mock_server = ServerMock()

cls.username = os.environ.get("USERNAME", "")
cls.password = os.environ.get("PASSWORD", "")
cls.socket_client = SocketClient(username=cls.username, password=cls.password)
cls.socket_client.hostname = "0.0.0.0"
cls.socket_client.port = cls.mock_server.port

def bump(self):
if len(self.ports_taken) == 0:
self.ports_taken.add(self.mock_server.port)
new_port = max(self.ports_taken) + 1
self.mock_server.port = new_port
self.socket_client.port = new_port
self.ports_taken.add(self.mock_server.port)

async def start(self):
self.bump()
try:
await self.mock_server.start()
except:
self.bump()
await self.mock_server.start()

@pytest.mark.asyncio
async def test_login(self):
await self.start()
...

Надеюсь, это будет полезно!

Подробнее здесь: https://stackoverflow.com/questions/647 ... -in-python