mocked_aiohttp.py

import aiohttp
import asyncio

url = 'https://httpbin.org/get'


class HttpClient:
    def __init__(self):
        self.session = aiohttp.ClientSession()

    def __del__(self):
        loop = asyncio.get_event_loop()
        loop.create_task(self.session.close())

    async def get(self, url, **kwargs):
        return await self.session.get(url, **kwargs)


async def test_original_client():
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.json()


async def test_new_client():
    client = HttpClient()
    response = await client.get(url)
    return await response.json()


class MockedResponse:
    async def json(self):
        return {'url': url}


class MockedRequestContext:
    async def __aenter__(self):
        return MockedResponse()

    async def __aexit__(self, exc_type, exc_value, traceback):
        return self

    async def __call__(self, *args, **kwargs):
        return MockedResponse()

    def __await__(self):
        return self.__aenter__().__await__()


class MockedSession:
    async def __aenter__(self):
        return self

    async def __aexit__(self, exc_type, exc_value, traceback):
        return False

    async def close(self):
        pass

    def get(self, url):
        return MockedRequestContext()


async def test_original_mocked_client():
    async with MockedSession() as session:
        async with session.get(url) as response:
            return await response.json()


async def test_new_mocked_client():
    client = HttpClient()
    await client.session.close() ## Mocking purpose only
    client.session = MockedSession() ## Mocking purpose only
    response = await client.get(url)
    return await response.json()


loop = asyncio.get_event_loop()

result = loop.run_until_complete(test_original_client())
print('original:', result['url'] == url)

result = loop.run_until_complete(test_new_client())
print('new:', result['url'] == url)

result = loop.run_until_complete(test_original_mocked_client())
print('original mocked:', result['url'] == url)

result = loop.run_until_complete(test_new_mocked_client())
print('new mocked:', result['url'] == url)