如何模拟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']

并测试:

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

那么模拟异步上下文管理器的正确方法是什么？

Answer 1

在您的链接中,有一个编辑:

编辑:本文中提到的GitHub问题已经解决,从版本0.11.1开始,asynctest支持开箱即用的异步上下文管理器.

因为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.

Answer 2

自 2020 年以来尚未asynctest收到任何更新，并且不断收到以下弃用通知：

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 的规范设置为异步函数将导致调用后返回协程对象。

所以你可以轻松地使用以下内容：

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

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}