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
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