TypeError：“协程”对象不支持 pytest_asyncio 中的异步上下文管理器协议

Question

我正在尝试对使用 aiohttp.ClientSession 执行 HTTPS 请求的类执行异步测试。 我遇到的问题是，当我调用 async with session.get 作为响应时，会引发以下异常

TypeError：“协程”对象不支持异步上下文管理器协议。

这是我的代码：

class URLFetcher:
    def __init__(self, proxies=None, user_agents=None, rate_limit=1, cache=None, authentication: Authentication = None, session_manager=None):
        self.proxies = proxies or []
        self.user_agents = user_agents or [self.default_user_agent()]
        self.rate_limiter = RateLimiter(rate_limit)
        self.authentication = authentication
        self.session_manager = session_manager or SessionManager()
        self.cache = cache

    async def fetch(self, url, retries=3):
        cache = await self._pre_flight(url)
        if cache:
            return cache

        try:
            async with self.session_manager as session:
                async with session.get(
                    url,
                    headers=self.headers,
                    proxy=self.proxy,
                    timeout=10
                ) as response:
                    response.raise_for_status()
                    content = await response.text()

                    if self.cache:
                        self.cache.set(url, content)

                    return content
        except aiohttp.ClientError as e:
            logger.error(f"HTTP error for URL {url}: {e}")
            if retries > 0:
                await asyncio.sleep(2 ** (3 - retries))
                return await self.fetch(url, retries - 1)
            else:
                raise e

这是测试：

@pytest_asyncio.fixture
async def url_fetcher():
    fetcher = URLFetcher()
    yield fetcher
    await fetcher.session_manager.close()


@pytest.mark.asyncio
async def test_fetch_success(url_fetcher):
    # Mock the response
    mock_response = AsyncMock()
    mock_response.status = 200
    mock_response.text.return_value = 'Mock Content'

    # Mock the session.get methods
    mock_get = AsyncMock(return_value=mock_response)

    # Patch the ClientSession.get method
    with patch('aiohttp.ClientSession.get', new=mock_get):
        content = await url_fetcher.fetch('http://example.com')

        assert content == 'Mock Content'
        mock_get.assert_called_once_with(
            'http://example.com',
            headers=url_fetcher.headers,
            proxy=url_fetcher.proxy,
            timeout=10
        )

我正在尝试为我的班级模拟异步测试。

Answer 1

修补使用上下文管理器的对象时，请用

contextlib.nullcontext

包装模拟。

import contextlib
from unittest import mock

import pytest


@pytest.mark.asyncio
async def test_aiohttp_client():
    mock_session = mock.Mock(
        spec=aiohttp.ClientSession,
        **{'get.return_value': contextlib.nullcontext(mock.Mock())}
    )

    with mock.patch(
            target='aiohttp.ClientSession',
            return_value=contextlib.nullcontext(mock_session)
    ):
        async with aiohttp.ClientSession() as session:
            url = 'https://example.com'
            async with session.get(url=url):
                mock_session.get.assert_called_once_with(url=url)