pythonpython-asynciopytest-aiohttp

How to mock the asyncio.run method?


I need to mock a result of an async function which uses inside another function calling asyncio.run(func(*args,**kwargs))

Example code

async def get_responses(
    **kwargs
) -> list:
   
    start_time = time.monotonic()

    responses: list = []
    conn = aiohttp.TCPConnector(limit=50)
    async with aiohttp.ClientSession(connector=conn) as session:
        tasks = []
        for contact_id in contact_ids:
            tasks.append(
                asyncio.ensure_future(
                    x.get_async_contract(
                        session=session,
                        contact_id=contact_id,
                    ),
                ),
            )
   

        await asyncio.gather(*tasks)

    return responses

I call this function inside another function like this

       
def get_contacts(**kwargs):
         try:
            responses = asyncio.run(
                get_responses(
                    **kwargs
                ),
            )

        except ServerProblem:
            raise

I need to mock the result of `get_responses`.

Tried several solutions using pytest but not seems to be working.

Thanks


Solution

  • You can mock the result of asyncio.run() instead of get_responses.

    with patch('asyncio.run') as run_mock:
        run_mock.return_value = ['responses']
        # Rest of test