How to mock an aiohttp get request?

[ghuname]

Is there a simple example how to mock an aiohttp get request?
For example, how to write pytest for RestClient.get_employees()? What if endpoints are class's or method's parameters? I would like to mock session.get('https://dummy.restapiexample.com/public/api/v1/employees') in order to produce fixed json as an output.

class Employees(BaseModel):
    all_employees: Dict
    single_employee: Dict


class RestClient:

    async def get_employees(self, employee_id: int) -> Employees:
        async with aiohttp.ClientSession() as session:
            t1 = asyncio.create_task(session.get('https://dummy.restapiexample.com/public/api/v1/employees'))
            t2 = asyncio.create_task(
                session.get(f'https://dummy.restapiexample.com/public/api/v1/employee/{employee_id}'))

            # all employees json: {"status":"success","data":[{"id":1,"employee_name":"Tiger Nixon","employee_salary":320800,"employee_age":61,"profile_image":""},{"id":2,"employee_name":"Garrett Winters","employee_salary":170750,"employee_age":63,"profile_image":""},{"id":3,"employee_name":"Ashton Cox","employee_salary":86000,"employee_age":66,"profile_image":""},{"id":4,"employee_name":"Cedric Kelly","employee_salary":433060,"employee_age":22,"profile_image":""},{"id":5,"employee_name":"Airi Satou","employee_salary":162700,"employee_age":33,"profile_image":""},{"id":6,"employee_name":"Brielle Williamson","employee_salary":372000,"employee_age":61,"profile_image":""},{"id":7,"employee_name":"Herrod Chandler","employee_salary":137500,"employee_age":59,"profile_image":""},{"id":8,"employee_name":"Rhona Davidson","employee_salary":327900,"employee_age":55,"profile_image":""},{"id":9,"employee_name":"Colleen Hurst","employee_salary":205500,"employee_age":39,"profile_image":""},{"id":10,"employee_name":"Sonya Frost","employee_salary":103600,"employee_age":23,"profile_image":""},{"id":11,"employee_name":"Jena Gaines","employee_salary":90560,"employee_age":30,"profile_image":""},{"id":12,"employee_name":"Quinn Flynn","employee_salary":342000,"employee_age":22,"profile_image":""},{"id":13,"employee_name":"Charde Marshall","employee_salary":470600,"employee_age":36,"profile_image":""},{"id":14,"employee_name":"Haley Kennedy","employee_salary":313500,"employee_age":43,"profile_image":""},{"id":15,"employee_name":"Tatyana Fitzpatrick","employee_salary":385750,"employee_age":19,"profile_image":""},{"id":16,"employee_name":"Michael Silva","employee_salary":198500,"employee_age":66,"profile_image":""},{"id":17,"employee_name":"Paul Byrd","employee_salary":725000,"employee_age":64,"profile_image":""},{"id":18,"employee_name":"Gloria Little","employee_salary":237500,"employee_age":59,"profile_image":""},{"id":19,"employee_name":"Bradley Greer","employee_salary":132000,"employee_age":41,"profile_image":""},{"id":20,"employee_name":"Dai Rios","employee_salary":217500,"employee_age":35,"profile_image":""},{"id":21,"employee_name":"Jenette Caldwell","employee_salary":345000,"employee_age":30,"profile_image":""},{"id":22,"employee_name":"Yuri Berry","employee_salary":675000,"employee_age":40,"profile_image":""},{"id":23,"employee_name":"Caesar Vance","employee_salary":106450,"employee_age":21,"profile_image":""},{"id":24,"employee_name":"Doris Wilder","employee_salary":85600,"employee_age":23,"profile_image":""}],"message":"Successfully! All records has been fetched."}
            # single employee json {"status":"success","data":{"id":1,"employee_name":"Tiger Nixon","employee_salary":320800,"employee_age":61,"profile_image":""},"message":"Successfully! Record has been fetched."}

            t1_resp, t2_resp = await asyncio.gather(t1, t2)
            t1_dict, t2_dict = await asyncio.gather(t1_resp.json(), t2_resp.json())

        return Employees(
            all_employees=t1_dict,
            single_employee=t2_dict
        )


async def main():
    client = RestClient()
    return await client.get_employees(employee_id=1)


if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    print(loop.run_until_complete(main()))

Output:

all_employees={'status': 'success', 'data': [{'id': 1, 'employee_name': 'Tiger Nixon', 'employee_salary': 320800, 'employee_age': 61, 'profile_image': ''}, {'id': 2, 'employee_name': 'Garrett Winters', 'employee_salary': 170750, 'employee_age': 63, 'profile_image': ''}, {'id': 3, 'employee_name': 'Ashton Cox', 'employee_salary': 86000, 'employee_age': 66, 'profile_image': ''}, {'id': 4, 'employee_name': 'Cedric Kelly', 'employee_salary': 433060, 'employee_age': 22, 'profile_image': ''}, {'id': 5, 'employee_name': 'Airi Satou', 'employee_salary': 162700, 'employee_age': 33, 'profile_image': ''}, {'id': 6, 'employee_name': 'Brielle Williamson', 'employee_salary': 372000, 'employee_age': 61, 'profile_image': ''}, {'id': 7, 'employee_name': 'Herrod Chandler', 'employee_salary': 137500, 'employee_age': 59, 'profile_image': ''}, {'id': 8, 'employee_name': 'Rhona Davidson', 'employee_salary': 327900, 'employee_age': 55, 'profile_image': ''}, {'id': 9, 'employee_name': 'Colleen Hurst', 'employee_salary': 205500, 'employee_age': 39, 'profile_image': ''}, {'id': 10, 'employee_name': 'Sonya Frost', 'employee_salary': 103600, 'employee_age': 23, 'profile_image': ''}, {'id': 11, 'employee_name': 'Jena Gaines', 'employee_salary': 90560, 'employee_age': 30, 'profile_image': ''}, {'id': 12, 'employee_name': 'Quinn Flynn', 'employee_salary': 342000, 'employee_age': 22, 'profile_image': ''}, {'id': 13, 'employee_name': 'Charde Marshall', 'employee_salary': 470600, 'employee_age': 36, 'profile_image': ''}, {'id': 14, 'employee_name': 'Haley Kennedy', 'employee_salary': 313500, 'employee_age': 43, 'profile_image': ''}, {'id': 15, 'employee_name': 'Tatyana Fitzpatrick', 'employee_salary': 385750, 'employee_age': 19, 'profile_image': ''}, {'id': 16, 'employee_name': 'Michael Silva', 'employee_salary': 198500, 'employee_age': 66, 'profile_image': ''}, {'id': 17, 'employee_name': 'Paul Byrd', 'employee_salary': 725000, 'employee_age': 64, 'profile_image': ''}, {'id': 18, 'employee_name': 'Gloria Little', 'employee_salary': 237500, 'employee_age': 59, 'profile_image': ''}, {'id': 19, 'employee_name': 'Bradley Greer', 'employee_salary': 132000, 'employee_age': 41, 'profile_image': ''}, {'id': 20, 'employee_name': 'Dai Rios', 'employee_salary': 217500, 'employee_age': 35, 'profile_image': ''}, {'id': 21, 'employee_name': 'Jenette Caldwell', 'employee_salary': 345000, 'employee_age': 30, 'profile_image': ''}, {'id': 22, 'employee_name': 'Yuri Berry', 'employee_salary': 675000, 'employee_age': 40, 'profile_image': ''}, {'id': 23, 'employee_name': 'Caesar Vance', 'employee_salary': 106450, 'employee_age': 21, 'profile_image': ''}, {'id': 24, 'employee_name': 'Doris Wilder', 'employee_salary': 85600, 'employee_age': 23, 'profile_image': ''}], 'message': 'Successfully! All records has been fetched.'} single_employee={'status': 'success', 'data': {'id': 1, 'employee_name': 'Tiger Nixon', 'employee_salary': 320800, 'employee_age': 61, 'profile_image': ''}, 'message': 'Successfully! Record has been fetched.'}

I tried to implement the solution from https://dino.codes/posts/mocking-asynchronous-functions-python/ but with no success.

[merwok]

Have you looked at the aioresponses module that was recommended in the other discussion?

The great thing with it is that it doesn’t care about the internal organization of your code (class or functions etc), it will intercept the requests being sent to reply with a pre-defined response. So it’s the best of mocking (controlling external behaviour) without the worst (being tied to internal code details and reproducing them)!

[Dreamsorcerer]

Your question was already answered: #2496 (comment)
Is there something else needed?

[ghuname]

Actually as I have provided a working code (server is alive and return result) I hoped that someone could add a test code.
I understand general testing but when async mocking is in question I am a beginner.
As I am using asyncio, I hope that it will not be in colision with aiohttp pytest plugin.

[merwok]

You have to read the docs for aioresponses.