asyncio.start_server 不是运行在 pytest-asyncio 测试中

Question

当使用 asyncio.start_server() 测试服务器程序时，该方法似乎没有调用第一个 client_connected_cb 方法。当程序在测试之外独立运行时，它会按预期工作。我怀疑这与没有正确使用事件循环有关。

为了测试，我有一些非异步代码连接到套接字服务器，它在监听套接字时无限期挂起。

为什么事件循环不处理单元测试内部的服务器代码？

测试：

@pytest.mark.asyncio
async def test_connect(server, connection):
    await server.start()
    connection.connect()
    connection.join_lobby('John')
    assert server.lobby.user_count() == 1

服务器：

async def start(self):
    loop = asyncio.get_event_loop()
    coro = asyncio.start_server(self.new_client, self.HOST, self.PORT, loop = loop)
    self.server_coro = await loop.create_task(coro)
    print('Serving on {}'.format(self.server_coro.sockets[0].getsockname()))

连接码：

def connect(self):
    self.s.connect((self.HOST, self.PORT))
    print('Just connected')

def join_lobby(self, name):
    self.s.send(b'Some data')
    print('Just tried to join lobby')
    msg = self.s.recv(BUFFER_SIZE)
    print(msg)

注释掉s.recv行后会显示如下输出。当 s.recv 未被注释时，测试挂起（因为服务器从不发送数据）：

Serving on ('0.0.0.0', 8888)
Just connected
Just tried to join lobby

Answer 1

您不能像那样混合阻塞和 non-blocking 代码。您的 recv() 调用阻塞了整个线程并使您的异步服务器无法访问运行。您还必须使客户端异步。

asyncio.start_server 不是运行在 pytest-asyncio 测试中

asyncio.start_server not running inside pytest-asyncio test

python

pytest

python-asyncio

asyncio.start_server 不是 运行 在 pytest-asyncio 测试中

asyncio.start_server not running inside pytest-asyncio test

python

pytest

python-asyncio

asyncio.start_server 不是运行在 pytest-asyncio 测试中