aiohttp_utils.py 文件源码

def create_response(method, url, content, loop=None):
    loop = loop or asyncio.get_event_loop()

    response = aiohttp.ClientResponse(method.lower(), URL(url))

    def side_effect(*args, **kwargs):
        fut = loop.create_future()
        if isinstance(content, str):
            fut.set_result(content.encode())
        else:
            fut.set_result(content)
        return fut

    response.content = mock.Mock()
    response.content.read.side_effect = side_effect

    return response