如何为异步流服务器编写 pytest 夹具？

Question

我一直在尝试学习 asyncio，但找不到任何创建可用于测试服务器代码的 pytest 夹具的示例。一旦服务器启动，我猜它会阻止其他一切，因此测试永远不会运行。pytest-asyncio 是否有办法在单独的线程中运行固定装置或其他东西？还是需要自己写线程代码？或者，还有更好的方法？下面是我一直在弄乱的一些代码。它是使用带有 pytest 固定装置的流文档从官方 TCP echo 服务器直接复制和粘贴并在最后进行测试：

import asyncio
import pytest


async def handle_echo(reader, writer):
    data = await reader.read(100)
    message = data.decode()
    addr = writer.get_extra_info('peername')

    print(f"Received {message!r} from {addr!r}")

    print(f"Send: {message!r}")
    writer.write(data)
    await writer.drain()

    print("Close the connection")
    writer.close()


async def main():
    server = await asyncio.start_server(
        handle_echo, '127.0.0.1', 8888)

    addr = server.sockets[0].getsockname()
    print(f'Serving on {addr}')

    async with server:
        await server.serve_forever()


@pytest.fixture(scope="session")
async def server():
    return await main()


@pytest.mark.asyncio
def test_something(server):
    assert False

Answer 1

如果您特别想要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%，但如果你想真正确定，你可以这样做：

# 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% 通过。啊？哦...是的...因为服务器在整个会话中运行，并且其内部状态被一个测试扰乱，所以某些其他测试在扰乱之后会失败。

长话短说，后台线程/更广泛范围的解决方案是可能的，但没有那么好。第二个教训是，您实际上可能需要一个每个测试/功能范围的服务器固定装置，以便您的测试彼此隔离。

顺便说一句：作为一个有点测试的书呆子，我什至在这样做的想法上挣扎（在 中测试客户端和服务器端对端pytest）。正如我在最初的评论中所说，此时它不再是真正的“单元测试”，而是“集成测试”，因此单元测试框架没有设置得很好也就不足为奇了开箱即用。幸运的是，对于我的所有疑问，这样做已经帮助我找到（并修复）了可能到目前为止的十几个错误，我真的很高兴我可以在无头测试工具中找到/复制，而不是通过编写一堆脚本selenium或更糟糕的是，手动点击网页。由于服务器与单个线程中的测试协同运行，因此使用调试器变得非常容易。玩得开心！