问题标签 [pytest-asyncio]

如何定义异步设备并在异步测试中使用它们？

以下代码都在同一个文件中，失败得很惨。测试运行程序是否明确调用了夹具并且没有等待？

生产

我知道我可能会做

但我想知道是否存在更简单/最优雅的方式。我在 pytest、pytest-asyncio、pytest-aiohttp 的项目页面中都找不到清晰简单的示例/解释。

我使用 Python 3.6.3、pytest 3.4.2、pytest-asyncio 0.8.0 和 pytest-aiohttp 0.3.0

非常感谢您的帮助

我目前正在用 pytest 编写一些异步测试，发现自己遇到了以下情况。

考虑我们有一个asyncio.Queue调用peer2_subscriber，我们想检查它是否收到某个消息（在触发某些操作后，为简洁起见省略）

现在，考虑我想测试另一个asyncio.Queue没有推动的东西。

我发现自己创建了这样一个辅助方法。

然后在测试中我写了类似的东西：

我想知道是否存在我缺少的现有模式？

Using Django and Channels 2, I have a consumer method that can can be accessed through channel groups and that may raise exceptions. Like this trivial one:

I want to test this method using pytest-asyncio. Since one can catch the exception of a coroutine with pytest.raises, I thought naively that something like this would be enough:

But the test fails in a pretty confusing way (truncated output):

So, what should I do? Is the raising of an exception from a consumer method a bad design?

使用 Python3.7，我有两个 py 脚本。server_execute.py 是接受一个 post 请求，一旦被接受，就会触发 server_scripts.py 异步启动测试脚本执行。

server_execute.py 脚本将接受一个发布请求，但我无法让它触发 server_scripts.py 异步函数来执行测试。我是发布请求和异步编程的新手......非常感谢任何帮助，谢谢！

server_execute.py

server_scripts.py

我实现了一段代码，它从一个队列中获取一个元素，并将相同的对象从队列列表中放入每个队列中。问题是，当我运行特定测试时，我遇到了ValueError: task_done() called too many times异常。此错误发生在测试代码中，而不是正在测试的代码中。

我正在使用asyncio.Queue和使用协程进行编程。我将每一个都Queue.get与一个完全匹配的Queue.task_done电话匹配。我正在用pytest测试代码。

我正在使用以下库：

• 蟒蛇 3.7
• pytest==3.10.0
• pytest-asyncio==0.9.0

我有两个文件：middleware.py包含我的类实现和test_middleware.py实现pytest测试。

文件middlware.py：

文件test_middleware.py：

在测试中，中间件应该从输入队列中获取元素并将它们放入列表中的 10 个队列中。它可以正常工作。

测试代码从 10 个队列中的每一个中获取所有元素，并检查它们是否存在于原始队列中。对于前 9 个队列，一切顺利，没有错误，但是当测试尝试从第十个列表中获取第一个元素时，ValueError会引发 a：

每个get匹配一个task_done. 我可以验证对文件进行以下修改test_middlware.py：

这样做我可以看到，即使有很多ValueError被提升，元素也会继续从队列中检索。测试成功：

为了确保测试使用了所有列表中的所有元素，我在测试结束时强制错误添加了一个错误断言：

结果输出显示从所有列表中检索到所有元素，但最后一个队列上的每个 task_done 都会生成一个ValueError.

问题是，我是否遗漏了什么，我的代码中有错误还是我发现了一个错误？

我正在尝试为 tornado_json Web 应用程序设置单元测试。我正在尝试测试一个 post 处理程序，但是我失败了，因为该fetch方法似乎返回了一个 _asyncio.Future 对象，该对象似乎永远不会完成/具有结果集。我试图发布代码摘要，目前我只是返回 ['test'] 项目。我查看了https://github.com/tornadoweb/tornado/issues/1154以及 tornado 文档。听起来我需要 self.stop 或 self.wait() 来完成任务，但我还没有弄清楚如何让它工作，或者这是否是解决方案。任何帮助将不胜感激。

@schema.validate( input_schema={ "type": "object", "properties": { "path": {"type": "string"} }, "required": ["path"] }, output_schema={ "type": "array", "items": { "properties": {"type": "string"} } } ) @coroutine def post(self): attributes = dict(self.body) path = attributes["path"] response = ["test"] return response

@gen_test def test_POST_method(self):
body = json.dumps({'path': 'bin'}) self.http_client.fetch(self.get_url('/api/listmyfiles'), method="POST", body=body ) response = self.wait() print(response.result()))

我得到的错误是： asyncio.base_futures.InvalidStateError: Result is not set.

为了在移动设备上运行并行测试，我使用 pytest + xdist。

我想运行与连接设备一样多的线程，仅此而已。

我将“-n 4”作为命令行参数传递，但如果我需要 2 个线程而不是 4 个线程，我无法在测试启动后更改它。

如何从我的测试代码中调整线程数？

我想从 aiohttp.ClientSession.get 方法模拟 json() 协程。它看起来返回一个异步生成器对象，这是我对如何在我的示例中模拟感到困惑的地方。这是我的代码：

这是我模拟 get 方法的测试用例：

我认为问题可能是 mock_response.json() 不可等待。在我的示例中，我无法从非异步函数调用 await，所以我对如何做到这一点感到困惑。我想将测试库保持在最低限度，即 pytest 和 pytest-asyncio 用于学习体验，并减少对 3rd 方库的依赖。

我正在寻找测试我的消费者，它使用scope['url_route']但使用HttpCommunicatoror ApplicationCommunicator，该参数未设置。如何设置此参数？测试文档非常有限，在此https://channels.readthedocs.io/en/latest/topics/testing.html上没有任何文档。

测试.py

我的应用程序.py

我正在测试一个可能会死锁的异步函数。我尝试添加一个固定装置以限制该功能在引发故障之前仅运行 5 秒，但到目前为止它还没有起作用。

设置：

代码：

--

编辑：米哈伊尔的解决方案工作正常。不过，我找不到将其合并到固定装置中的方法。