如何模拟 "async with" 语句?
How to mock "async with" statements?
我正在尝试为使用“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
__aexit__
也需要异步 (needs to return an awaitable):
async def __aexit__(self, exc_type, exc_val, exc_tb):
pass
如果不是异步,它会返回 None
而不是协程,因此它会引发错误,至于我创建的非常具有误导性的错误消息 this issue 以指出错误消息需要固定。
我正在尝试为使用“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
__aexit__
也需要异步 (needs to return an awaitable):
async def __aexit__(self, exc_type, exc_val, exc_tb):
pass
如果不是异步,它会返回 None
而不是协程,因此它会引发错误,至于我创建的非常具有误导性的错误消息 this issue 以指出错误消息需要固定。