CODEWITHSUNDEEP
pythonpython-3.xunit-testingpython-asyncio
lukeg
lukeg

Reputation: 4389

Testing and mocking asynchronous code that uses async with statement

I have a simple asynchronous functions that fetches JSON from the web:

async def fetch (session : aiohttp.ClientSession, url : str, pageSize : int = 100, pageNumber : int = 0):
    async with session.get(url, params={'pageSize' : f"{pageSize}", 'pageNumber' : f"{pageNumber}"}) as response:
        return await response.json()

How do I write unit tests for it? I am learning how to do that in Python currently and I have settled down on using unittest and aiounittest modules. The scaffolding that I use for testing is as follows:

class AsyncTest(aiounittest.AsyncTestCase):
    async def test_fetch(self):
        mocked_session = ... #no idea how to create
        json = await fetch(mocked_session, "url_for_mock", 10, 0)

I think I would have to somehow mock the session argument using unittest.Mock() functionality to return some harcoded json data but I am a little lost on how to do this so it works inside async with statement. I have never used an unit testing library in Python and an example would be very helpful here. I am using Python 3.8 and aiohttp for networking.

Upvotes: 3

Views: 3300

Answers (1)

lukeg
lukeg

Reputation: 4389

I was able to mock the asynchronous resource manager by implementing my own version of __aenter__ magic method and using AsyncMock to mock relevant async methods. The proof of concept test that mocks network calls looks like follows:

class AsyncTest(aiounittest.AsyncTestCase):
    async def test_fetch(self):

        request_mock = AsyncMock()
        request_mock.__aenter__.return_value = request_mock
        request_mock.json.return_value = { 'hello' : 'world'}

        session = Mock()
        session.get.return_value = request_mock

        json = await fetch(session, "url_for_mock", 10, 0)

Upvotes: 4

Related Questions