我有一个关于异步客户端测试的问题。这是我的测试代码
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()
...
希望这会有所帮助！