python中用于单元测试的模拟流式API

Question

我有一个调用流api的异步函数。为这个函数编写单元测试的最佳方法是什么？必须模拟api响应

我试过使用aiounittest并使用unittest中的mock。但是这会调用实际的api，而不是得到模拟的响应。还尝试了pytest.mark.asyncio注释，但这不断给我带来错误-从未等待协同程序。我已验证是否已安装pytest asyncio

我正在使用VS代码和Python 3.6.6

以下是相关的代码片段：

async def method1():
    response = requests.get(url=url, params=params, stream=True)
    for data in response.iter_lines():
        # processing logic here
        yield data

粘贴我尝试过的一些测试

def mocked_get(*args, **kwargs):
#implementation of mock

class TestClass (unittest.TestCase):
    @patch("requests.get", side_effect=mocked_get)
    async def test_method (self, mock_requests):
        resp = []
        async for data in method1:
            resp.append (data)
    
        #Also tried await method1
    
        assert resp
    

还尝试了TestClass类（aiounittest.AsyncTestCase）：

Answer 1

用^{}代替aiounittest

1. 用asynctest.TestCase替换unittest.TestCase
2. 将from unittest.mock import patch替换为from asynctest.mock import patch
3. async for data in method1:应该是async for data in method1():

import asynctest
from asynctest.mock import patch


class TestClass(asynctest.TestCase):
    @patch("requests.get", side_effect=mocked_get)
    async def test_method(self, mock_requests):
        resp = []
        async for data in method1():
            resp.append(data)

        assert resp