asyncio.start_server 没有在 pytest-asyncio 测试中运行答案

【问题标题】：asyncio.start_server not running inside pytest-asyncio testasyncio.start_server 没有在 pytest-asyncio 测试中运行
【发布时间】：2018-03-21 16:14:53
【问题描述】：

使用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

【问题讨论】：

标签： python pytest python-asyncio

【解决方案1】：

你不能像这样混合阻塞和非阻塞代码。您的 recv() 调用阻塞了整个线程并阻止您的异步服务器运行。您还必须使您的客户端异步。

【讨论】：