{"id":22643,"date":"2024-09-12T10:32:29","date_gmt":"2024-09-12T14:32:29","guid":{"rendered":"https:\/\/qxf2.com\/blog\/?p=22643"},"modified":"2024-09-12T10:32:29","modified_gmt":"2024-09-12T14:32:29","slug":"asynchronous-api-automation-testing-using-qxf2s-framework","status":"publish","type":"post","link":"https:\/\/qxf2.com\/blog\/asynchronous-api-automation-testing-using-qxf2s-framework\/","title":{"rendered":"Asynchronous API Automation testing using Qxf2’s Framework"},"content":{"rendered":"

You can now create asynchronous API automation test using Qxf2’s API Automation Framework<\/a>.<\/p>\n

In this post we will go about explaining:
\n– Why we need Async?
\n– How we modified our Synchronous framework to support Asynchronous HTTP calls
\n– How to create an Async API Automation test using our framework
\n– Cases where Async is not the right fit
\n
\nNote:<\/strong> This post is mostly about how a few changes to our framework helped make it support running tests asynchronously but if you have not used our API framework and have no idea on how it works you can still follow along to know how to create non-blocking<\/a> functions for blocking<\/a> functions and how to create coroutine test functions for pytest<\/a>.<\/p>\n

\n

Why Async?<\/h3>\n

Modern computers equipped with multi cpu cores can perform multiple tasks at the same time. When running an I\/O bound<\/a> task computers sometime wait on the I\/O operation<\/code> idly. Async<\/a> helps reduce this idle wait time by running another task during the wait. Running steps asynchronously in most cases can also help reduce execution time drastically. When we ran 300 API requests synchronously and asynchronously using our framework, we noticed the asynchronous steps ran more than 10x<\/code> faster. We find making asynchronous calls for setup steps to be especially useful. <\/p>\n

\r\n2024-08-28 14:44:12 | WARNING | tests.test_api_sync_example| The total number of sync requests is 300\r\n2024-08-28 14:44:12 | WARNING | tests.test_api_sync_example| The time taken to complete sync call is 90.46587109565735\r\n2024-08-28 14:44:13 | WARNING | test_api_async_example     | The total number of async requests is 300\r\n2024-08-28 14:44:13 | WARNING | test_api_async_example     | The time taken to complete async call is 7.610826253890991\r\n<\/pre>\n
\n
Changes we made to our Framework:<\/h3>\n
We have added an image below implementing our framework against a sample Cars-API<\/a> application to show you have our Player-Interface API Automation framework works.<\/p>\n
\"API<\/a><\/p>\n
The following layers form the crux of the framework:
\n1. Endpoints layer
\n2. Interface layer
\n3. Player layer
\nThe API endpoints in the application are abstracted in the Endpoints layer<\/a>(all the *_API_Endpoints.py modules), the Interface layer<\/a>(the API_Interface.py module) collects all the API endpoint abstractions & the Player layer<\/a>(API_Player.py module) houses the business logic to validate the application.
\nThe Base_API<\/a> file that the endpoints abstractions extend to make the HTTP calls using requests<\/a> module only used blocking functions to make GET, POST, PUT & DELETE requests so to start supporting Async API automation test we added non blocking functions to make those requests.<\/p>\n
Blocking GET HTTP method:<\/h5>\n
\r\ndef get(self, url, headers=None):\r\n    \"Run HTTP Get request against an url\"\r\n    headers = headers if headers else {}\r\n    response = None\r\n    try:\r\n        response = requests.get(url=url,headers=headers)\r\n    except Exception as generalexcep:\r\n        print(f\"Unable to run GET request against {url} due to {generalexcep}\")\r\n    return response\r\n<\/pre>\n
Disclaimer: <\/strong>The code in this blog post is intended for demonstration purposes only. We recommend following better coding standards for real-world applications.<\/p>\n
Non-blocking GET HTTP method for the previous synchoronous method:<\/h5>\n
\r\nasync def async_get(self, url, headers={}):\r\n    \"Run the blocking GET method in a thread\"\r\n    response = await asyncio.to_thread(self.get, url, headers)\r\n    return response\r\n<\/pre>\n
Although running code on a separate thread is not asynchronous but using asyncio.to_thread<\/code> allows us to await on the result asynchronously.
\nWe used asyncio.to_thread<\/code> to add similar non-blocking<\/code> functions for the corresponding blocking functions in the *_API_Endpoints, API_Player modules too.<\/p>\n
\n
Create an Async API Automation test<\/h3>\n
A simple test cannot be used to call the coroutines<\/a> we added in the API_Player module. The test function needs to be a couroutine too. pytest-asyncio<\/a> plugin allows creating coroutine test functions.<\/p>\n
\r\n@pytest.mark.asyncio\r\nasync def test_api_async_example(test_api_obj):\r\n    \"Run api test\"\r\n<\/pre>\n
The @pytest.mark.asyncio<\/code> is required for pytest to collect the coroutine as a test.
\nThe test scenarios are then added as tasks inside the coroutine test function, we used asyncio.TaskGroup<\/a> object to group the test scenarios and run them asynchronously.<\/p>\n
\r\nasync with asyncio.TaskGroup() as group:\r\n    get_cars = group.create_task(test_api_obj.async_get_cars(auth_details))\r\n    add_new_car = group.create_task(test_api_obj.async_add_car(car_details=car_details,\r\n                                                               auth_details=auth_details))\r\n    get_car = group.create_task(test_api_obj.async_get_car(auth_details=auth_details,\r\n                                                           car_name=existing_car,\r\n                                                           brand=brand))\r\n<\/pre>\n
Putting it all together, this is how our sample Async API Automation test for the Cars-API app looks:<\/p>\n
\r\n\"\"\"\r\nAPI Async EXAMPLE TEST\r\nThis test collects tasks using asyncio.TaskGroup object \\\r\nand runs these scenarios asynchronously:\r\n1. Get the list of cars\r\n2. Add a new car\r\n3. Get a specifi car from the cars list\r\n4. Get the registered cars\r\n\"\"\"\r\n\r\nimport asyncio\r\nimport os\r\nimport sys\r\nimport pytest\r\nsys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))\r\nfrom conf import api_example_conf\r\n\r\n@pytest.mark.asyncio\r\n# Skip running the test if Python version < 3.11\r\n@pytest.mark.skipif(sys.version_info < (3,11),\r\n                    reason=\"requires Python3.11 or higher\")\r\nasync def test_api_async_example(test_api_obj):\r\n    \"Run api test\"\r\n    try:\r\n        expected_pass = 0\r\n        actual_pass = -1\r\n\r\n        # set authentication details\r\n        username = api_example_conf.user_name\r\n        password = api_example_conf.password\r\n        auth_details = test_api_obj.set_auth_details(username, password)\r\n\r\n        # Get an existing car detail from conf\r\n        existing_car = api_example_conf.car_name_1\r\n        brand = api_example_conf.brand\r\n        # Get a new car detail from conf\r\n        car_details = api_example_conf.car_details\r\n\r\n        async with asyncio.TaskGroup() as group:\r\n            get_cars = group.create_task(test_api_obj.async_get_cars(auth_details))\r\n            add_new_car = group.create_task(test_api_obj.async_add_car(car_details=car_details,\r\n                                                                       auth_details=auth_details))\r\n            get_car = group.create_task(test_api_obj.async_get_car(auth_details=auth_details,\r\n                                                                   car_name=existing_car,\r\n                                                                   brand=brand))\r\n            get_reg_cars = group.create_task(test_api_obj.async_get_registered_cars(auth_details=auth_details))\r\n\r\n        test_api_obj.log_result(get_cars.result(),\r\n                                positive=\"Successfully obtained the list of cars\",\r\n                                negative=\"Failed to get the cars\")\r\n        test_api_obj.log_result(add_new_car.result(),\r\n                                positive=f\"Successfully added new car {car_details}\",\r\n                                negative=\"Failed to add a new car\")\r\n        test_api_obj.log_result(get_car.result(),\r\n                                positive=f\"Successfully obtained a car - {existing_car}\",\r\n                                negative=\"Failed to add a new car\")\r\n        test_api_obj.log_result(get_reg_cars.result(),\r\n                                positive=\"Successfully obtained registered cars\",\r\n                                negative=\"Failed to get registered cars\")\r\n        # write out test summary\r\n        expected_pass = test_api_obj.total\r\n        actual_pass = test_api_obj.passed\r\n        test_api_obj.write_test_summary()\r\n        # Assertion\r\n        assert expected_pass == actual_pass,f\"Test failed: {__file__}\"\r\n\r\n    except Exception as e:\r\n        test_api_obj.write(f\"Exception when trying to run test: {__file__}\")\r\n        test_api_obj.write(f\"Python says: {str(e)}\")\r\n\r\n<\/pre>\n
We have used @pytest.mark.skipif<\/code> marker to skip running the test for Python versions less than 3.11, the TaskGroup<\/code> object api is available on versions 3.11 and later.<\/p>\n
\n
Scenarios where Async API test is not the right choice<\/h3>\n
Though Async API Automation test can help validate an app and reduce the time it takes to run the validations, there are cases where making HTTP calls asynchronously might not be the right approach. Here are a few scenarios where we think synchronous test is a better choice:
\n- Test where a few scenarios need to run in an order are not a right fit for asynchronous testing, because the order of execution of the coroutines collected as tasks in the test cannot be controlled, hence is susceptible to race condition<\/a>.
\n- Test with very few scenarios are better off being run synchronously, because the overhead required to collect and run very less scenarios asynchronously minimises its advantage.
\n- Projects that cannot be maintained regularly. The asyncio Python module has gone through drastic changes with each Python version and hence warrant frequent maintenance.<\/p>\n
\n
References<\/h3>\n
1. How do you test APIs that use asynchronous communication?<\/a>
\n2. Using asyncio.to_thread to create non blocking functions<\/a>
\n3. pytest-asyncio<\/a>
\n4. asyncio Task Groups<\/a><\/p>\n
\n
Hire testers from Qxf2<\/h3>\n
We at Qxf2 services<\/a> make a dedicated effort to constantly evaluate our Automation framework against new frameworks to identify and implement new useful features. You can find our open-source framework on GitHub here - Qxf2 Page Object Model Framework<\/a>.
\nOur expertise extend beyond maintaining our own framework, we have used various other frameworks at clients and even helped maintain a few of them. If you are looking for testers to help create automation tests using our framework or maintain your own automation framework contact Qxf2<\/a>.<\/p>\n
\n","protected":false},"excerpt":{"rendered":"
You can now create asynchronous API automation test using Qxf2’s API Automation Framework. In this post we will go about explaining: – Why we need Async? – How we modified our Synchronous framework to support Asynchronous HTTP calls – How to create an Async API Automation test using our framework – Cases where Async is not the right fit Note: […]<\/p>\n","protected":false},"author":9,"featured_media":0,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[43,322,107,180],"tags":[],"class_list":["post-22643","post","type-post","status-publish","format-standard","hentry","category-api-testing","category-asyncio","category-pytest","category-requests"],"_links":{"self":[{"href":"https:\/\/qxf2.com\/blog\/wp-json\/wp\/v2\/posts\/22643","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/qxf2.com\/blog\/wp-json\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/qxf2.com\/blog\/wp-json\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/qxf2.com\/blog\/wp-json\/wp\/v2\/users\/9"}],"replies":[{"embeddable":true,"href":"https:\/\/qxf2.com\/blog\/wp-json\/wp\/v2\/comments?post=22643"}],"version-history":[{"count":53,"href":"https:\/\/qxf2.com\/blog\/wp-json\/wp\/v2\/posts\/22643\/revisions"}],"predecessor-version":[{"id":22805,"href":"https:\/\/qxf2.com\/blog\/wp-json\/wp\/v2\/posts\/22643\/revisions\/22805"}],"wp:attachment":[{"href":"https:\/\/qxf2.com\/blog\/wp-json\/wp\/v2\/media?parent=22643"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/qxf2.com\/blog\/wp-json\/wp\/v2\/categories?post=22643"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/qxf2.com\/blog\/wp-json\/wp\/v2\/tags?post=22643"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}