address already in use in async client tests in python

Question

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!

Answer 1

For address already in use, check the current running processes by

ps -ax | grep <your app/app port>

In this, you will notice that the process already exists. Pls. kill that process using sudo kill -9 <process_id>

and now restart your service. I am not sure about async tests.