如何编写pytest fixture以用于asyncio流服务器？

Question

我一直在尝试学习asyncio，但却找不到任何可以创建pytest fixture的示例来测试我的服务器代码。一旦服务器启动，它就会阻塞其他所有内容，因此测试永远不会运行。pytest-asyncio有一种方法可以在单独的线程或其他地方运行fixture吗？还是我需要自己编写线程代码？或者有更好的方法吗？以下是我一直在尝试的一些代码。它直接从官方的TCP echo server using streams文档中复制粘贴，并带有一个pytest fixture和测试：

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

- Mike Conigliaro

---

测试应该在单独的文件中，而不是与服务器运行的同一文件中。同时启动服务器和运行测试会有点混淆你想要实现的目标。 - gold_cy

我并不困惑。 :) 我想做同样的事情：在后台作为协程运行服务器，同时运行与服务器通信的异步测试。这可能不是“单元测试”，但是pytest的异步支持表明应该是可行的，更像是“集成测试”。 - ipmcc

@ipmcc 如果你有任何想法，请告诉我们。我想到的最好方法是在线程中运行服务器。它可以工作，但似乎应该有更好的方法。 - Mike Conigliaro

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而失败。我做了很多尝试...通过一次旋转事件循环(via loop._run_once())不能保证服务器将开始监听。睡眠0.001s仍然有大约1%的失败率。睡眠0.01s似乎在1000次运行中总是成功的，但如果你想要真正地确保，你可以这样做：

# 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是一个函数作用域的fixture。我尝试编写自己的模块/会话作用域版本，但他们在内部使用它来为每个测试安排自己的事件循环（可能是为了防止测试相互干扰）。因此，除非你想放弃pytest-asyncio并“滚动自己的”测试工具来运行异步协程测试，否则你在更大的范围内就基本上没有机会了。

FWIW，在找到这个合作解决方案之前，我尝试了“后台线程”，模块作用域的解决方案，而且这是有点棘手的。首先，您的服务器需要一种线程安全，干净的关闭方式，并从您的fixture触发它，这本身就在主线程上运行。其次，（也许对您来说这无所谓，但对我来说绝对很重要）调试非常令人发狂。跨越两个线程，每个线程都有自己的事件循环，但只有一个线程停止在任何给定时间...好吧，这很难。基本情况是这样的：我有一个文件里有一百个测试。我运行它。大约50个测试失败了。这很奇怪，我只改了一点点东西...我可以看到控制台输出中的回溯，某些东西在服务器代码深处引发了异常。没问题，我会在那里设置断点。在调试器中再次运行。执行停在断点处。太棒了！现在，哪个测试触发了这个错误？哦！我不知道，因为只有后台线程在调试器中停在那里。我最终找出了错误，修复了它，再次运行，在100%的测试通过。什么？哦...对了...因为服务器跨越整个会话运行，并且由于一个测试而被其内部状态改变，某些其他测试也会在那种混乱之后失败。

长话短说，后台线程/更广泛的作用域解决方案是可行的，但不如这个好。第二个教训是，您实际上可能希望一个服务器-每个测试/函数作用域fixture，以使您的测试相互隔离。

顺带提一下：作为一个测试迷，我对于在pytest中测试客户端和服务端的端到端测试有些困难。正如我在最初的评论中所说，这已经不再是"单元测试"了，而是"集成测试"，因此并不奇怪一个单元测试框架没有很好地设置它。幸运的是，尽管我怀疑这样做，但这已经帮助我发现（并修复）目前为止可能有几十个缺陷，我非常高兴可以在无头测试工具中找到/复制这些缺陷，而不是编写一堆selenium脚本或更糟糕的是手动在网页上点击。而且，由于服务器与测试在单线程中协同运行，使用调试器甚至变得相当容易。祝你愉快！

- ipmcc