如果你特别想要session范围，您可能在与合作调度的服务器方面不走运pytest-asyncio。如果你愿意安于现状function范围，我已经让它发挥作用了。当然，这意味着您的服务器将在每次测试时启动和停止，这对于这里的微不足道的回显服务器来说并不是很多开销，但对于您的实际服务器来说可能是这样，无论它是什么。这是对您的示例的改编，对我有用。
HOST = "localhost"
@pytest.fixture()
def server(event_loop, unused_tcp_port):
cancel_handle = asyncio.ensure_future(main(unused_tcp_port), loop=event_loop)
event_loop.run_until_complete(asyncio.sleep(0.01))
try:
yield unused_tcp_port
finally:
cancel_handle.cancel()
async def handle_echo(reader, writer):
data = await reader.read(100)
message = data.decode()
addr = writer.get_extra_info('peername')
print(f"SERVER: Received {message!r} from {addr!r}")
writer.write(data)
await writer.drain()
print(f"SERVER: Sent: {message!r}")
writer.close()
print("SERVER: Closed the connection")
async def main(port):
server = await asyncio.start_server(handle_echo, HOST, port)
addr = server.sockets[0].getsockname()
print(f'SERVER: Serving on {addr[0:2]}')
async with server:
await server.serve_forever()
@pytest.mark.asyncio
async def test_something(server):
message = "Foobar!"
reader, writer = await asyncio.open_connection(HOST, server)
print(f'CLIENT: Sent {message!r}')
writer.write(message.encode())
await writer.drain()
data = await reader.read(100)
print(f'CLIENT: Received {data.decode()!r}')
print('CLIENT: Close the connection')
writer.close()
await writer.wait_closed()
精明的读者会注意到asyncio.sleep(0.01)在服务器固定装置中。我不知道非决定论是否是固有的asyncio实现，或者特定于 pytest 的使用，但没有那个sleep大约 20% 的情况下（当然，在我的机器上）服务器在测试尝试连接到服务器之前不会开始侦听，这意味着测试会失败ConnectionRefusedError。我玩了很多次......旋转事件循环一次（通过loop._run_once()）不保证服务器会监听。睡觉为0.001s仍然有大约 1% 的时间失败。睡觉为0.01s似乎在 1,000 次运行中通过了 100%，但如果你想成为really当然，你会做这样的事情：
# Replace `event_loop.run_until_complete(asyncio.sleep(0.01))` with this:
event_loop.run_until_complete(asyncio.wait_for(_async_wait_for_server(event_loop, HOST, unused_tcp_port), 5.0))
async def _async_wait_for_server(event_loop, addr, port):
while True:
a_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
try:
await event_loop.sock_connect(a_socket, (addr, port))
return
except ConnectionRefusedError:
await asyncio.sleep(0.001)
finally:
a_socket.close()
在运行测试之前，它将继续尝试连接，直到成功（或者不太可能在 5 秒后超时）。这就是我在“真实”测试中的做法。
现在，关于范围。从源码来看，看起来像pytest-asyncio已决定event_loop是一个函数范围的固定装置。我尝试编写自己的模块/会话范围版本，但他们在内部使用它来在自己的事件循环上安排每个测试（大概是为了防止测试以某种方式相互干扰）。所以除非你想放弃pytest-asyncio并“推出您自己的”测试工具来将测试作为异步协程运行，我认为您在更大的范围上几乎不走运。
FWIW，在找到这个协作解决方案之前，我尝试了“后台线程”、模块范围的解决方案，这有点痛苦。首先，您的服务器需要一种方法来执行线程安全、干净的关闭，并且可以从本身在主线程上运行的固定装置触发。其次，（这对你来说可能并不重要，但对我来说确实如此）调试绝对令人抓狂。在单个操作系统线程中运行的单个事件循环上跟踪协程执行的（众所周知的）“线程”已经足够困难了。尝试跨两个线程解决这个问题，每个线程都有自己的事件循环，但只有一个线程在任何给定时间停止......好吧，这很困难。基本场景是这样的：我有一个包含一百个测试的文件。我运行它。约 50 次测试失败。这很奇怪，我只改变了一件小事......我可以在控制台输出中看到回溯，有些东西在服务器代码深处引发了异常。没问题，我会在那里放置一个断点。在调试器中再次运行。执行在断点处停止。伟大的！好的，现在，50 个测试中的哪一个触发了此错误？哦！我不知道，因为调试器中只有后台线程停止。我最终找出了错误，修复了它，再次运行，测试 100% 通过。啊？哦...是的...因为服务器在整个会话中运行，并且其内部状态被一个测试打乱，所以某些其他测试在打乱之后会失败。
长话短说，后台线程/更广泛范围的解决方案是可能的，但没有那么好。第二个教训是你实际上可能want每个测试/功能范围的服务器固定装置，以便您的测试彼此隔离。
顺便说一句：作为一个有点测试的书呆子，我什至在做这个的想法上挣扎（测试客户端和服务器端对端）pytest）。正如我在最初的评论中所说，此时它不再是真正的“单元测试”，而是“集成测试”，因此单元测试框架没有设置得很好也就不足为奇了开箱即用。幸运的是，对于我的所有疑虑，这样做已经帮助我找到（并修复）了可能到目前为止的十几个错误，我真的很高兴我可以在无头测试工具中找到/复制，而不是通过编写一堆selenium脚本或更糟糕的是，手动单击网页。由于服务器与单个线程中的测试协同运行，因此使用调试器变得非常容易。玩得开心！
