我在模拟aiohttp.client.ClientSession.get上下文管理器时遇到了一些麻烦。我找到了一些文章,这是一个似乎可行的示例:文章1
I have some troubles with mocking aiohttp.client.ClientSession.get context manager. I found some articles and here is one example that seems was working: article 1
所以我要测试的代码为:
So my code that I want to test:
async_app。 py
import random from aiohttp.client import ClientSession async def get_random_photo_url(): while True: async with ClientSession() as session: async with session.get('random.photos') as resp: json = await resp.json() photos = json['photos'] if not photos: continue return random.choice(photos)['img_src']并测试:
test_async_app.py
from asynctest import CoroutineMock, MagicMock, patch from asynctest import TestCase as TestCaseAsync from async_app import get_random_photo_url class AsyncContextManagerMock(MagicMock): async def __aenter__(self): return self.aenter async def __aexit__(self, *args): pass class TestAsyncExample(TestCaseAsync): @patch('aiohttp.client.ClientSession.get', new_callable=AsyncContextManagerMock) async def test_call_api_again_if_photos_not_found(self, mock_get): mock_get.return_value.aenter.json = CoroutineMock(side_effect=[{'photos': []}, {'photos': [{'img_src': 'a.jpg'}]}]) image_url = await get_random_photo_url() assert mock_get.call_count == 2 assert mock_get.return_value.aenter.json.call_count == 2 assert image_url == 'a.jpg'运行测试时,出现错误:
When I'm running test, I'm getting an error:
(test-0zFWLpVX) ➜ test python -m unittest test_async_app.py -v test_call_api_again_if_photos_not_found (test_async_app.TestAsyncExample) ... ERROR ====================================================================== ERROR: test_call_api_again_if_photos_not_found (test_async_app.TestAsyncExample) ---------------------------------------------------------------------- Traceback (most recent call last): File "/home/kamyanskiy/.local/share/virtualenvs/test-0zFWLpVX/lib/python3.6/site-packages/asynctest/case.py", line 294, in run self._run_test_method(testMethod) File "/home/kamyanskiy/.local/share/virtualenvs/test-0zFWLpVX/lib/python3.6/site-packages/asynctest/case.py", line 351, in _run_test_method self.loop.run_until_complete(result) File "/home/kamyanskiy/.local/share/virtualenvs/test-0zFWLpVX/lib/python3.6/site-packages/asynctest/case.py", line 221, in wrapper return method(*args, **kwargs) File "/usr/lib/python3.6/asyncio/base_events.py", line 467, in run_until_complete return future.result() File "/home/kamyanskiy/.local/share/virtualenvs/test-0zFWLpVX/lib/python3.6/site-packages/asynctest/_awaitable.py", line 21, in wrapper return await coroutine(*args, **kwargs) File "/home/kamyanskiy/.local/share/virtualenvs/test-0zFWLpVX/lib/python3.6/site-packages/asynctest/mock.py", line 588, in __next__ return self.gen.send(None) File "/home/kamyanskiy/work/test/test_async_app.py", line 23, in test_call_api_again_if_photos_not_found image_url = await get_random_photo_url() File "/home/kamyanskiy/work/test/async_app.py", line 9, in get_random_photo_url json = await resp.json() TypeError: object MagicMock can't be used in 'await' expression ---------------------------------------------------------------------- Ran 1 test in 0.003s FAILED (errors=1)所以我尝试调试-这是我所看到的:
So I've tried to debug - here is what I can see:
> /home/kamyanskiy/work/test/async_app.py(10)get_random_photo_url() 9 import ipdb; ipdb.set_trace() ---> 10 json = await resp.json() 11 photos = json['photos'] ipdb> resp.__aenter__() <generator object CoroutineMock._mock_call.<locals>.<lambda> at 0x7effad980048> ipdb> resp.aenter <MagicMock name='get().__aenter__().aenter' id='139636643357584'> ipdb> resp.__aenter__().json() *** AttributeError: 'generator' object has no attribute 'json' ipdb> resp.__aenter__() <generator object CoroutineMock._mock_call.<locals>.<lambda> at 0x7effad912468> ipdb> resp.json() <MagicMock name='get().__aenter__().json()' id='139636593767928'> ipdb> session <aiohttp.client.ClientSession object at 0x7effb15548d0> ipdb> next(resp.__aenter__()) TypeError: object MagicMock can't be used in 'await' expression那么模拟异步上下文管理器的正确方法是什么?
So what is proper way to mock async context manager ?
推荐答案在您的链接中,有一个编辑:
In your link, there is an edit:
编辑:A 本文中提到的GitHub问题已解决,由于$ 0.1版本的 asynctest支持的异步上下文管理器。
A GitHub issue mentioned in this post has been resolved and as of version 0.11.1 asynctest supports asynchronous context managers out of the box.
由于 asynctest == 0.11.1 ,它被更改了,一个有效的例子是:
Since asynctest==0.11.1, it was changed, a working example is:
import random from aiohttp import ClientSession from asynctest import CoroutineMock, patch async def get_random_photo_url(): while True: async with ClientSession() as session: async with session.get('random.photos') as resp: json = await resp.json() photos = json['photos'] if not photos: continue return random.choice(photos)['img_src'] @patch('aiohttp.ClientSession.get') async def test_call_api_again_if_photos_not_found(mock_get): mock_get.return_value.__aenter__.return_value.json = CoroutineMock(side_effect=[ {'photos': []}, {'photos': [{'img_src': 'a.jpg'}]} ]) image_url = await get_random_photo_url() assert mock_get.call_count == 2 assert mock_get.return_value.__aenter__.return_value.json.call_count == 2 assert image_url == 'a.jpg'关键的问题是您需要正确地模拟函数 json 为默认情况下,它是一个 MagicMock 实例。要访问此功能,您需要 mock_get.return_value .__ aenter __。return_value.json 。
The critical problem is that you need to correctly mock function json as by default it is a MagicMock instance. To get access to this function, you need mock_get.return_value.__aenter__.return_value.json.
更多推荐
如何模拟aiohttp.client.ClientSession.get异步上下文管理器
发布评论