python-3.x - 如何模拟 "async with"语句？

Question

我正在尝试为使用“async with”语句的方法(在本例中为 aioredis 的连接池)编写测试，我想模拟与 redis 的连接，但我无法弄清楚如何进行。
这是我到目前为止所拥有的:

from asyncio import Future
from unittest.mock import MagicMock

import pytest

# The thing i'm trying to test
async def set_value(redis, value):
    # Do things
    async with redis.get() as conn:
        await conn.set("key", value)

#My Mock classes
class MockRedis():
    def get(self):
        return MockAsyncPool()


class MockAsyncPool(MagicMock):
    async def __aenter__(self):
        conn = MagicMock()
        f = Future()
        f.set_result(True)
        conn.set = MagicMock(return_value=f)
        return conn

    def __aexit__(self, exc_type, exc_val, exc_tb):
        pass


# The actual test
@pytest.mark.asyncio
async def test_get_token():
    redis = MockRedis()

    token = await set_value(redis, 'something')
    assert token is not None

我运行它:

py.test path/to/file.py

我收到此错误:

> await conn.set("key", value)

E TypeError: object NoneType can't be used in 'await' expression

Answer 1

__aexit__还需要异步 (needs to return an awaitable) :

async def __aexit__(self, exc_type, exc_val, exc_tb):
    pass

没有异步它正在返回 None而不是协程，所以它会引发错误，至于我创建的非常具有误导性的错误消息 this issue指出需要修复错误消息。