地址已在 python 中的异步客户端测试中使用

Question

我有关于异步客户端测试的问题。这是我的测试代码

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

一起运行测试套件时，后一个测试用例将引发

error while attempting to bind on address ('0.0.0.0', 2056): address already in use

。我的问题是如何在不解决分配问题的情况下使测试套件通过，以便它们可以在 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()
         ...

希望这会有所帮助！

Answer 1

对于

address already in use

，通过

检查当前正在运行的进程

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

在此，您会注意到该进程已经存在。请。使用

sudo kill -9 <process_id>

终止该进程

现在重新启动您的服务。我不确定异步测试。

Answer 2

发生这种情况是因为 pytest 中的会话是针对每个工作线程的，因此如果您运行更多工作线程（通过

-n auto

执行此操作），每个工作线程都会初始化固定装置，即使它们是使用会话范围定义的。

这意味着您必须注意绑定到隔离端口。 这比你的发现更简洁：

@pytest.fixture(scope="session")
def fixt_free_port():
    """Find a free port dynamically."""

    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
        s.bind(("", 0))
        free_port = s.getsockname()[1]

        return free_port

然后在其他装置中，您可以通过这种方式将某些服务绑定到自定义端口：

@pytest.fixture
def fixt_server(fixt_free_port):
    ...

更好地使用固定装置，这是配置和设置测试的地方。 有了它们，您就可以在测试主体中仅保留测试逻辑。