I have a question about async client tests. Here is my code about my test
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()
I can run the tests under TestSocketClient individually and they will pass individually. But when I run the test suites together with pytest -n auto, The latter test case will raise error while attempting to bind on address ('0.0.0.0', 2056): address already in use. My question is how to make the test suites pass without address the allocation issues so that they can run successfully in CI process. I will be of great appreciation if there is some more valuable suggestion in writing async tests(for example, what should I assert if I only want to test a request the client would like to send to the server? Should I assert the message received on the server-side or just write something like assert_called_once on the client-side). Thanks in advance!
Updates:
I finally solved the problem with port increment in different tests like below
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()
...
Hope this could be helpful!
