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

Question

我在使用aiohttp.client.ClientSession.get上下文管理器进行模拟时遇到了一些问题。 我找到了一些文章，这里有一个似乎有效的示例：文章1

所以我想要测试的代码：

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']

And test:

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'

当我运行测试时，出现了错误：

(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)

所以我尝试进行调试 - 这是我能看到的：

> /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

那么，如何正确地模拟异步上下文管理器？

- Alexander Kamyanskiy

Answer 1

在您提供的链接中，有一个编辑：

编辑：本帖提到的GitHub问题已得到解决，asynctest版本0.11.1支持异步上下文管理器。

自从asynctest==0.11.1之后，它已经发生了变化，以下是一个可行的示例：

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。

- Sraw

---

你如何修补它以访问resp.status？因为在上下文中，我确实有if 200 <= resp.status <= 300，但它会抱怨在MagicMock和int之间不支持比较运算符。 - Hugo Sousa

Answer 2

“asynctest”自2020年以来没有更新，且一直收到以下弃用通知：

python3.9/site-packages/asynctest/mock.py:434
  python3.9/site-packages/asynctest/mock.py:434: DeprecationWarning: "@coroutine" decorator is deprecated since Python 3.8, use "async def" instead
    def wait(self, skip=0):

相反，可以使用MagicMock来模拟协程，正如文档中所提到的：

将Mock或MagicMock的规范设置为异步函数将导致在调用后返回协程对象。

因此，您可以轻松地使用以下方法：

from unittest.mock import MagicMock

@pytest.mark.asyncio
async def test_download():
    mock = aiohttp.ClientSession
    mock.get = MagicMock()
    mock.get.return_value.__aenter__.return_value.status = 200
    mock.get.return_value.__aenter__.return_value.text.return_value = 'test content'

    async with aiohttp.ClientSession() as session:
        async with session.get('http://test.com') as response:
            assert response.text() == 'test content'

- hhadi

Answer 3

您无需安装任何框架即可测试 aiohttp.ClientSession。

看看这个例子：

# Module A

import aiohttp


async def send_request():
    async with aiohttp.ClientSession() as session:
        async with session.post("https://example.com", json={"testing": True}) as response:
            if response.status != 200:
                print("Woops")
                return False

            print("YAY!")
            return True


# Module A test

import pytest
from types import SimpleNamespace
from unittest.mock import MagicMock, patch

# Do not forget to replace `path_to_module_a` with the module that imports aiohttp
import path_to_module_a.send_request

@patch("path_to_module_a.aiohttp.ClientSession")
@pytest.mark.asyncio
async def test_should_fail_to_send_request(mock: MagicMock):
    session = MagicMock()
    session.post.return_value.__aenter__.return_value = SimpleNamespace(status=500)

    mock.return_value.__aenter__.return_value = session

    response = await send_request()

    assert response == False
    assert session.post.call_args.kwargs["json"] == {"testing": True}

- Michael Pacheco

---

这是一个不错的解决方案，但是当send_request()方法调用'request.raise_for_status()'时，我该如何测试这种情况呢？ - mehekek

Answer 4

基于 @Sraw 的答案:

@pytest.mark.gen_test
@patch('application.adapters.http_retriever.aiohttp.ClientSession.get')
async def test_get_files(mock_get):

    with open('tests/unit/sample_csv_files/Katapult_mock_response.json','r') as f:
        json_dict = json.load(f)

    mock_get.return_value.__aenter__.return_value.json = CoroutineMock()
    mock_get.return_value.__aenter__.return_value.status = 200
    mock_get.return_value.__aenter__.return_value.json.return_value = json_dict

这对我有用

- Mickey Perlstein

---

非常有用且简单！解决了我的问题。 - Suyog Shimpi