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

问题描述 投票：0回答：4

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

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

python python-asyncio python-unittest aiohttp asynctest
4个回答
35
投票

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

编辑：本文中提到的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
。

9
投票

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'
7
投票

无需安装任何框架即可测试

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}
3
投票

以@Sraw 的答案为基础：

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

    with open('tests/mock_response.json', 'r') as f:
        mock_response = 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 = mock_response

这对我有用。

最新问题
© www.soinside.com 2019 - 2024. All rights reserved.