21

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

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

4

3 回答 3

30

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

编辑：这篇文章中提到的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.

于 2018-02-13T09:04:43.760 回答
2

基于@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

这对我有用

于 2020-06-18T15:47:43.150 回答
1

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

因此，您可以轻松使用以下内容：

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'
于 2021-12-29T13:49:42.067 回答