2017-03-24 5 views
8

"async with"文(この場合はaioredisの接続プール)を使用するメソッドのテストを作成しようとしていますが、私はredisへの接続を擬似したいのですが、ここで"async with"ステートメントを模擬する方法は?

は私がこれまで持っているものです。

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

答えて

5

__aexit__もする必要があります非同期(needs to return an awaitable)

それは私がエラーメッセージを修正する必要があることを指摘して this issueを作成している非常に誤解を招くようなエラーメッセージのようなエラーが発生しますので、それは代わりにコルーチンの Noneを返す非同期(async)されることなく
async def __aexit__(self, exc_type, exc_val, exc_tb): 
    pass 

関連する問題