如何模拟aiohttp.client.ClientSession.get异步上下文管理器

本文介绍了如何模拟aiohttp.client.ClientSession.get异步上下文管理器的处理方法，对大家解决问题具有一定的参考价值，需要的朋友们下面随着小编来一起学习吧！ 问题描述

我在模拟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.