aiohttp_utils.py 文件源码

def mock_session(response, session=None, mock_object=None):
    """
    :param aiohttp.ClientSession session:
    :param aiohttp.ClientResponse|list[aiohttp.ClientResponse] response:
    """
    session = session or aiohttp.ClientSession()
    request = session._request

    session.mock = mock_object or mock.Mock()
    if isinstance(response, (list, tuple)):
        session.mock.side_effect = response
    else:
        session.mock.return_value = response

    async def _request(*args, **kwargs):
        return session.mock(*args, **kwargs)

    try:
        with mock.patch.object(session, '_request') as request_mock:
            request_mock.side_effect = _request
            yield session
    finally:
        delattr(session, 'mock')