使用Pytest测试AioHTTP代码时遇AttributeError: __aenter__错误求助
解决异步logs函数测试中的AttributeError: __aenter__问题
问题背景
测试异步logs函数时,在session.get的async with语句处触发AttributeError: __aenter__错误。该函数通过Docker Unix socket获取容器日志并输出到控制台,测试时使用pytest的monkeypatch替换aiohttp.ClientSession.get方法,但mock对象不符合异步上下文管理器的要求。
错误原因
async with语句要求操作的对象必须实现异步上下文管理器协议,即拥有__aenter__和__aexit__异步方法。当前的MockResponse只是普通类,既没有这两个方法,也无法模拟ClientResponse的异步迭代器行为(resp.content需要是异步可迭代对象)。
修复方案
修改MockResponse类,使其满足以下两个要求:
- 实现异步上下文管理器方法
__aenter__和__aexit__ - 让
content成为异步迭代器,支持async for遍历
另外,原logs函数无返回值,测试时改为捕获控制台输出,验证日志内容是否正确。
完整修复后的测试代码
import pytest import aiohttp from typing import AsyncGenerator from io import StringIO async def logs(cont, name): conn = aiohttp.UnixConnector(path="/var/run/docker.sock") async with aiohttp.ClientSession(connector=conn) as session: async with session.get(f"http://xx/containers/{cont}/logs?follow=1&stdout=1") as resp: async for line in resp.content: print(f"{name}: {line.decode('utf-8').strip()}") class MockResponse: def __init__(self): self.content = self._async_content_generator() async def __aenter__(self): return self async def __aexit__(self, exc_type, exc, tb): pass async def _async_content_generator(self) -> AsyncGenerator[bytes, None]: for line in ["one", "two", "three"]: yield line.encode('utf-8') def __aiter__(self): return self.content @pytest.mark.asyncio async def test_logs_normal(monkeypatch, capsys): name = "reddis" cont = 14322 def mock_get(*args, **kwargs): return MockResponse() monkeypatch.setattr(aiohttp.ClientSession, "get", mock_get) await logs(cont=cont, name=name) # 捕获控制台输出并验证 captured = capsys.readouterr() assert f"{name}: one" in captured.out
说明
MockResponse新增__aenter__和__aexit__方法,满足异步上下文管理器要求_async_content_generator生成异步迭代器,模拟resp.content的行为- 使用
capsysfixture捕获控制台输出,验证logs函数的输出内容(符合原函数无返回值的特性) - 修复了
monkeypatch替换get方法的逻辑,确保返回的MockResponse能被async with正确处理
内容的提问来源于stack exchange,提问作者Maksim Smolovsky
相关产品推荐
相关产品推荐

