如何模拟aiohttp.client.ClientSession.get异步上下文管理器 [英] How to mock aiohttp.client.ClientSession.get async context manager
问题描述
我在模拟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 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异步上下文管理器的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！
