5

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

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
4

1 回答 1

3

如果您特别想要session范围，那么在与pytest-asyncio. 如果你愿意接受function范围，我已经让它工作了。当然，这意味着您的服务器将在每次测试时启动和停止，这对于这里的微不足道的 echo 服务器来说并不是很多开销，但可能适用于您的实际服务器，无论它是什么。这是对您的示例的改编，对我有用。

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% 的时间。Sleeping for0.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，在我想出这个合作解决方案之前，我尝试了“后台线程”，模块范围的解决方案，这有点痛苦。首先，您的服务器需要一种方法来执行线程安全、干净关闭、可从您的夹具触发，该夹具本身将在主线程上运行。其次，（这对你来说可能无关紧要，但对我来说确实如此）调试绝对令人抓狂。在单个 OS 线程中运行的单个事件循环上遵循协程执行的（众所周知的）“线程”已经足够困难了。尝试跨两个线程解决这个问题，每个线程都有自己的事件循环，但在任何给定时间只有一个停止......嗯，这很困难。基本场景是这样的：我有一个包含一百个测试的文件。我运行它。约 50 次测试失败。这很奇怪，我只改变了一点点...... 我可以在控制台输出中看到回溯，在服务器代码深处引发异常。没问题，我会在那里放一个断点。在调试器中再次运行。执行在断点处停止。伟大的！好的，现在，触发此错误的 50 个测试中的哪一个？哦！我不知道，因为只有后台线程在调试器中停止。我最终找出了这个错误，修复了它，再次运行，并且 100% 的测试通过了。嗯？哦...是的...因为服务器在整个会话中运行，并且其内部状态被一个测试加扰，因此在加扰之后某些其他测试会失败。触发此错误的 50 个测试中的哪一个？哦！我不知道，因为只有后台线程在调试器中停止。我最终找出了这个错误，修复了它，再次运行，并且 100% 的测试通过了。嗯？哦...是的...因为服务器在整个会话中运行，并且其内部状态被一个测试加扰，因此在加扰之后某些其他测试会失败。触发此错误的 50 个测试中的哪一个？哦！我不知道，因为只有后台线程在调试器中停止。我最终找出了这个错误，修复了它，再次运行，并且 100% 的测试通过了。嗯？哦...是的...因为服务器在整个会话中运行，并且其内部状态被一个测试加扰，因此在加扰之后某些其他测试会失败。

长话短说，后台线程/范围更广的解决方案是可能的，但不如这个好。第二个教训是，您实际上可能想要一个 server-per-test/function-scoped fixture，以便您的测试彼此隔离。

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

于 2020-09-17T12:23:52.463 回答