单元测试异步方法python

Question

尝试了解异步方法的单元测试

假设我有这个我想测试

async def run(self):
    while True:
        async with DBHook(mssql_conn_id=self.db_conn).get_cursor() as cursor:
            cursor.execute(self.sql)
            rows = cursor.fetchone()
            if rows[0] > 0:
                yield TriggerEvent(True)
            await asyncio.sleep(self.sleep_interval)

当然是在课堂上

现在，我想断言

asyncio is called when number of rows affected by fetchone is zero

所以，我正在尝试这样写测试（也尝试了其他变体）

import aiounittest
import mock


import mymodule as st


class SQLTriggerTests(aiounittest.AsyncTestCase):

    @mock.patch("mymodule.DBHook")
    @mock.patch("mymodule.asyncio")
    async def test_run(self, mock_asyncio, mock_MsSqlIntegratedHook):
        my_obj = mymodule.classname(sql="select blabla", mssql_conn_id="db")  
        conn=mock_MsSqlIntegratedHook.return_value.get_cursor.return_value.__enter__.return_value
        conn.fetchone.return_value= (0,)
        res = my_obj.run()
        mock_asyncio.sleep.assert_called()

断言失败说 asyncio 没有被调用。

我也尝试在调用 my_obj.run() 时使用 await，但这给了我一个错误提示

can't be awaited on a generator object.

我应该如何正确测试它。 Python版本：3.9

Answer 1

你还没有执行协程；你只是 created 它。由于您正在定义测试协程，因此您只需要使用

await

.

class SQLTriggerTests(aiounittest.AsyncTestCase):

    @mock.patch("mymodule.DBHook")
    @mock.patch("mymodule.asyncio")
    async def test_run(self, mock_asyncio, mock_MsSqlIntegratedHook):
        my_obj = mymodule.classname(sql="select blabla", mssql_conn_id="db")  
        conn = mock_MsSqlIntegratedHook.return_value.get_cursor.return_value.__enter__.return_value
        conn.fetchone.return_value = (0,)
        await my_obj.run()
        mock_asyncio.sleep.assert_called()