问题描述
我一直在尝试学习 asyncio，但找不到任何创建可用于测试服务器代码的 pytest 固定装置的示例.一旦服务器启动，我猜它会阻止其他所有内容，因此测试永远不会运行.pytest-asyncio 有没有办法在单独的线程中运行夹具?还是我需要自己编写线程代码?或者，还有更好的方法?下面是一些我一直在搞乱的代码.这是从官方 使用流的 TCP 回显服务器 带有 pytest 固定装置的文档并在最后进行测试:
I've been trying to learn asyncio, and I can't find any examples of creating a pytest fixture I can use to test my server code. As soon as the server starts, I guess it blocks everything else, so the tests never run. Does pytest-asyncio have a way to run a fixture in a separate thread or something? Or do I need to write the thread code myself? Or is there a better way? Below is some code I've been messing with. It's a direct copy and paste from the official TCP echo server using streams documentation with a pytest fixture and test at the end:
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
推荐答案
如果你特别想要 session 作用域，你可能在使用 pytest-asyncio 合作调度服务器方面不走运.如果你愿意接受 function 范围，我已经让它工作了.当然，这意味着您的服务器将在每次测试时启动和停止，这对于这里的微不足道的回显服务器来说并不是很多开销，但可能是您的实际服务器，无论它是什么.这是您的示例的改编版，对我有用.
If you specifically want session scope, you may be out of luck in terms of cooperatively scheduled servers with pytest-asyncio. If you're willing to settle for function scope, I've gotten it to work. Of course, this means your server will be started and stopped for each test, which isn't a lot of overhead for the trivial echo server here, but may be for your actual server, whatever that may be. Here's an adaptation of your example that works for me.
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%，但如果你想真的确定，你会做这样的事情:
The astute reader will notice the asyncio.sleep(0.01) in the server fixture. I don't know whether the non-determinism is inherent in the asyncio implementation, or specific to pytest's use of it, but without that sleep about 20% of the time (on my machine, naturally) the server will not have started listening before the test tries to connect to it, meaning the test then fails with ConnectionRefusedError. I played around with it quite a bit... spinning the event loop once(via loop._run_once()) doesn't guarantee the server will be listening. Sleeping for 0.001s still fails about 1% of the time. Sleeping for 0.01s seems to pass 100% over 1,000 runs, but if you want to be really sure, you'd do something like this:
# 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 秒后超时).这就是我在真实"场景中的做法.测试.
This will keep trying to connect until it succeeds (or, very unlikely, times out after 5 seconds) before running the test. This is how I'm doing it in my "real" tests.
现在，关于范围.从查看源代码来看，看起来 pytest-asyncio 已经决定 event_loop 是一个函数范围的装置.我尝试编写自己的模块/会话范围版本，但他们在内部使用它在自己的事件循环中安排每个测试(大概是为了防止测试以某种方式相互影响).所以除非你想放弃 pytest-asyncio 并自己动手"测试工具作为异步协程运行测试，我认为你在更大的范围内很不走运.
Now, about the scope. From looking at the source, it looks like pytest-asyncio has decided that event_loop is a function-scoped fixture. I tried writing my own module/session scoped version of it, but they're using it internally to schedule each test on its own event loop (presumably to prevent tests from somehow stepping on each other). So unless you want to give up on pytest-asyncio and "roll your own" test harness to run tests as async coroutines, I think you're pretty much out of luck on the larger scopes.
FWIW，在我想出这个合作解决方案之前，我尝试了后台线程"，模块范围的解决方案，这有点痛苦.首先，您的服务器需要一种方法来执行线程安全、干净关闭、可从您的夹具触发，该夹具本身将在主线程上运行.其次，(这对你来说可能无关紧要，但对我来说确实如此)调试绝对是令人抓狂的.遵循(众所周知的)线程"已经够难了.在单个 OS 线程中运行的单个事件循环上执行协程.试图在两个线程中解决这个问题，每个线程都有自己的事件循环，但在任何给定时间只有一个停止......好吧，这很困难.基本场景是这样的:我有一个包含一百个测试的文件.我运行它.约 50 次测试失败.这很奇怪，我只更改了一件小事……我可以在控制台输出中看到回溯，在服务器代码深处引发了异常.没问题，我会在那里放一个断点.在调试器中再次运行.执行在断点处停止.伟大的！好的，现在，是 50 个测试中的哪一个触发了这个错误?哦！我不知道，因为调试器中只停止了后台线程.我最终找出错误，修复它，再次运行，并且 100% 的测试通过.嗯?哦...是的...因为服务器在整个会话中运行，并且其内部状态被一个测试打乱，某些其他测试会在该打乱之后失败.
FWIW, I tried the "background thread", module-scoped solution before I figured out this cooperative solution, and it was a bit of a pain. First, your server needs a way to do a thread-safe, clean shutdown, trigger-able from your fixture that will, itself, be running on the main thread. Second, (and this may not matter to you, but it certainly did to me) debugging was absolutely maddening. It's hard enough to follow the (proverbial) "thread" of coroutine execution on a single event loop running in a single OS thread. Trying to work that out across two threads, each with their own event loop, but only one of which stops at any given time... well, it's difficult. The basic scenario was like this: I had a file with a hundred tests in it. I run it. ~50 tests fail. That's odd, I only changed one little thing... I can see the backtrace in the console output, something is raising an exception deep inside the server code. No problem, I'll put a breakpoint there. Run again in the debugger. Execution stops at the breakpoint. Great! OK, now, which of the 50 tests is it that triggered this error? Oh! I can't know because only the background thread is stopped in the debugger. I eventually figure out the bug, fix it, run again, and 100% of tests pass. Huh? Oh... yeah... because the server runs across the whole session, and was having its internal state scrambled by one test, certain other tests would fail after that scrambling.
长话短说，后台线程/范围更广的解决方案是可能的，但没有这么好.第二个教训是，您实际上可能想要一个针对每个测试的服务器/功能范围的装置，以便您的测试彼此隔离.
Long story short, the background thread/broader scoped solution is possible, but not as nice as this. The second lesson is that you actually probably want a server-per-test/function-scoped fixture, so that your tests are isolated from one another.
顺便说一句:作为一个测试书呆子，我连这样做的想法都很挣扎(在 pytest 中端到端测试客户端和服务器).正如我在最初的评论中所说，它并不是真正的单元测试".在这一点上，它是集成测试"，因此单元测试框架没有设置为开箱即用，这并不奇怪.幸运的是，对于我所有的疑虑，这样做已经帮助我找到(并修复)到目前为止可能有十几个错误，我真的很高兴我可以在无头测试工具中找到/复制，而不是通过编写一堆 selenium 脚本或更糟，手动点击网页.由于服务器在单个线程中与测试协同运行，因此使用调试器甚至非常容易.玩得开心！
As an aside: Being a bit of a testing nerd, I struggled with the idea of even doing this (testing client and server end-to-end in pytest). As I said in my initial comment, it isn't really "unit testing" any more at this point, it's "integration testing", so it's not all that surprising that a unit testing framework isn't set up to do it very well right out of the box. Fortunately, for all my doubts, doing this has helped me to find (and fix) probably a dozen bugs so far that I'm really glad I can find/replicate in a headless test harness, and not by writing a bunch of selenium scripts or worse, manually clicking around on a web page. And with the server running cooperatively with the tests in a single thread, it's even pretty easy to use the debugger. Have fun!
这篇关于如何为 asyncio 流服务器编写 pytest 夹具?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持！