如果您特别需要
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
而失败。我做了很多尝试...通过一次旋转事件循环(via
loop._run_once()
)不能保证服务器将开始监听。睡眠
0.001s
仍然有大约1%的失败率。睡眠
0.01s
似乎在1000次运行中总是成功的,但如果你想要真正地确保,你可以这样做:
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
是一个函数作用域的fixture。我尝试编写自己的模块/会话作用域版本,但他们在内部使用它来为每个测试安排自己的事件循环(可能是为了防止测试相互干扰)。因此,除非你想放弃pytest-asyncio
并“滚动自己的”测试工具来运行异步协程测试,否则你在更大的范围内就基本上没有机会了。
FWIW,在找到这个合作解决方案之前,我尝试了“后台线程”,模块作用域的解决方案,而且这是有点棘手的。首先,您的服务器需要一种线程安全,干净的关闭方式,并从您的fixture触发它,这本身就在主线程上运行。其次,(也许对您来说这无所谓,但对我来说绝对很重要)调试非常令人发狂。跨越两个线程,每个线程都有自己的事件循环,但只有一个线程停止在任何给定时间...好吧,这很难。基本情况是这样的:我有一个文件里有一百个测试。我运行它。大约50个测试失败了。这很奇怪,我只改了一点点东西...我可以看到控制台输出中的回溯,某些东西在服务器代码深处引发了异常。没问题,我会在那里设置断点。在调试器中再次运行。执行停在断点处。太棒了!现在,哪个测试触发了这个错误?哦!我不知道,因为只有后台线程在调试器中停在那里。我最终找出了错误,修复了它,再次运行,在100%的测试通过。什么?哦...对了...因为服务器跨越整个会话运行,并且由于一个测试而被其内部状态改变,某些其他测试也会在那种混乱之后失败。
长话短说,后台线程/更广泛的作用域解决方案是可行的,但不如这个好。第二个教训是,您实际上可能希望一个服务器-每个测试/函数作用域fixture,以使您的测试相互隔离。
顺带提一下:作为一个测试迷,我对于在pytest
中测试客户端和服务端的端到端测试有些困难。正如我在最初的评论中所说,这已经不再是"单元测试"了,而是"集成测试",因此并不奇怪一个单元测试框架没有很好地设置它。幸运的是,尽管我怀疑这样做,但这已经帮助我发现(并修复)目前为止可能有几十个缺陷,我非常高兴可以在无头测试工具中找到/复制这些缺陷,而不是编写一堆selenium
脚本或更糟糕的是手动在网页上点击。而且,由于服务器与测试在单线程中协同运行,使用调试器甚至变得相当容易。祝你愉快!