adds tests for the pagination feature

Author: sufyankhanrao

apimatic_core/pagination/paginated_data.py

@@ -57,8 +57,7 @@ def __init__(self, api_call, paginated_items_converter):
             self._api_call.global_configuration.get_http_client_configuration().http_callback or HttpCallContext()
         _http_client_configuration = self._api_call.global_configuration.get_http_client_configuration().clone(
             http_callback=self._http_call_context)
-        self._global_configuration = self._api_call.global_configuration.clone_with(
-            http_client_configuration=_http_client_configuration)
+        self._global_configuration = self._api_call.global_configuration.clone_with()
         self._paged_response = None
         self._items = []
         self._page_size = 0
@@ -82,7 +81,8 @@ def __next__(self):
             return item
 
         self._paged_response = self._fetch_next_page()
-        self._items = self._paginated_items_converter(self._paged_response.body)
+        self._items = self._paginated_items_converter(
+            self._paged_response.body) if self._paged_response else []
         if not self._items:
             raise StopIteration
         self._page_size, self._current_index = len(self._items), 0
@@ -102,7 +102,8 @@ def pages(self):
 
         while True:
             paginated_data._paged_response = paginated_data._fetch_next_page()
-            paginated_data._items = self._paginated_items_converter(paginated_data._paged_response.body)
+            paginated_data._items = self._paginated_items_converter(
+                paginated_data._paged_response.body) if paginated_data._paged_response else []
             if not paginated_data._items:
                 break
             paginated_data._page_size = len(paginated_data._items)
@@ -132,7 +133,7 @@ def _fetch_next_page(self):
             ).execute()
             return pagination_strategy.apply_metadata_wrapper(response)
 
-        return []
+        return None
 
     def _get_new_self_instance(self):
         """

apimatic_core/pagination/pagination_strategy.py

@@ -82,8 +82,32 @@ def get_updated_request_builder(request_builder, input_pointer, offset):
             header_params = ApiHelper.update_entry_by_json_pointer(
                 header_params.copy(), field_path, offset, inplace=True)
 
-        return request_builder.clone_with(
-            template_params=template_params,
-            query_params=query_params,
-            header_params=header_params
-        )
+        return request_builder.clone_with(template_params=template_params, query_params=query_params,
+                                          header_params=header_params)
+
+    @staticmethod
+    def _get_initial_request_param_value(request_builder, input_pointer, default=0):
+        """
+        Extracts the initial pagination offset value from the request builder using the specified JSON pointer.
+
+        Args:
+            request_builder: The request builder containing path, query, and header parameters.
+            input_pointer (str): JSON pointer indicating which parameter to extract.
+            default (int, optional): The value to return if the parameter is not found. Defaults to 0.
+
+        Returns:
+            int: The initial offset value from the specified parameter, or default if not found.
+        """
+        path_prefix, field_path = ApiHelper.split_into_parts(input_pointer)
+
+        if path_prefix == "$request.path":
+            value = ApiHelper.get_value_by_json_pointer(request_builder.template_params, field_path)
+            return int(value) if value is not None else default
+        elif path_prefix == "$request.query":
+            value = ApiHelper.get_value_by_json_pointer(request_builder.query_params, field_path)
+            return int(value) if value is not None else default
+        elif path_prefix == "$request.headers":
+            value = ApiHelper.get_value_by_json_pointer(request_builder.header_params, field_path)
+            return int(value) if value is not None else default
+
+        return default

apimatic_core/pagination/strategies/link_pagination.py

@@ -61,9 +61,7 @@ def apply(self, paginated_data):
         updated_query_params = request_builder.query_params.copy()
         updated_query_params.update(query_params)
 
-        return request_builder.clone_with(
-            query_params=updated_query_params
-        )
+        return request_builder.clone_with(query_params=updated_query_params)
 
     def apply_metadata_wrapper(self, paged_response):
         """

apimatic_core/pagination/strategies/offset_pagination.py

@@ -49,7 +49,7 @@ def apply(self, paginated_data):
         last_page_size = paginated_data.page_size
         # The last response is none which means this is going to be the 1st page
         if last_response is None:
-            self._offset = self._get_initial_offset(request_builder)
+            self._offset = self._get_initial_request_param_value(request_builder, self._input)
             return request_builder
 
         self._offset += last_page_size
@@ -67,25 +67,3 @@ def apply_metadata_wrapper(self, page_response):
             The result of the metadata wrapper callable with the page response and offset.
         """
         return self._metadata_wrapper(page_response, self._offset)
-
-    def _get_initial_offset(self, request_builder):
-        """
-        Determines the initial offset value for pagination by extracting it from the request builder
-        based on the configured JSON pointer input.
-
-        Args:
-            request_builder: The request builder containing path, query, and header parameters.
-
-        Returns:
-            int: The initial offset value extracted from the appropriate request parameter, or 0 if not found.
-        """
-        path_prefix, field_path = ApiHelper.split_into_parts(self._input)
-
-        if path_prefix == "$request.path":
-            return int(ApiHelper.get_value_by_json_pointer(request_builder.template_params, field_path))
-        elif path_prefix == "$request.query":
-            return int(ApiHelper.get_value_by_json_pointer(request_builder.query_params, field_path))
-        elif path_prefix == "$request.headers":
-            return int(ApiHelper.get_value_by_json_pointer(request_builder.header_params, field_path))
-
-        return 0

apimatic_core/pagination/strategies/page_pagination.py

@@ -44,7 +44,7 @@ def apply(self, paginated_data):
         last_page_size = paginated_data.page_size
         # The last response is none which means this is going to be the 1st page
         if last_response is None:
-            self._page_number = self._get_initial_page_offset(request_builder)
+            self._page_number = self._get_initial_request_param_value(request_builder, self._input, 1)
             return request_builder
         self._page_number += 1 if last_page_size > 0 else 0
 
@@ -61,27 +61,3 @@ def apply_metadata_wrapper(self, paged_response):
             The result of the metadata wrapper with the paged response and current page number.
         """
         return self._metadata_wrapper(paged_response, self._page_number)
-
-    def _get_initial_page_offset(self, request_builder):
-        """
-        Determines the initial page offset for pagination by extracting the value from the request builder
-        based on the configured JSON pointer. Returns 1 if the value is missing or invalid.
-
-        Args:
-            request_builder: The request builder containing path, query, and header parameters.
-
-        Returns:
-            int: The initial page offset value.
-        """
-        path_prefix, field_path = ApiHelper.split_into_parts(self._input)
-
-        try:
-            if path_prefix == "$request.path":
-                return int(ApiHelper.get_value_by_json_pointer(request_builder.template_params, field_path))
-            elif path_prefix == "$request.query":
-                return int(ApiHelper.get_value_by_json_pointer(request_builder.query_params, field_path))
-            elif path_prefix == "$request.headers":
-                return int(ApiHelper.get_value_by_json_pointer(request_builder.header_params, field_path))
-        except ValueError:
-            pass
-        return 1

apimatic_core/request_builder.py

@@ -230,7 +230,7 @@ def apply_auth(self, auth_managers, http_request):
                 raise AuthValidationException(self._auth.error_message)
 
     def clone_with(
-            self, template_params=None, header_params=None, query_params=None, form_params=None
+            self, template_params=None, header_params=None, query_params=None
     ):
         """
         Clone the current instance with the given parameters.
@@ -251,7 +251,7 @@ def clone_with(
         new_instance._template_params = template_params or self._template_params
         new_instance._header_params = header_params or self._header_params
         new_instance._query_params = query_params or self._query_params
-        new_instance._form_params = form_params or self._form_params
+        new_instance._form_params = self._form_params
         new_instance._additional_form_params = self._additional_form_params
         new_instance._additional_query_params = self._additional_query_params
         new_instance._multipart_params = self._multipart_params

apimatic_core/utilities/api_helper.py

@@ -818,7 +818,10 @@ def get_value_by_json_pointer(dictionary, pointer):
         Raises:
             JsonPointerException: If the pointer does not resolve to a value.
         """
-        return resolve_pointer(dictionary, pointer)
+        try:
+            return resolve_pointer(dictionary, pointer)
+        except JsonPointerException as jpe:
+            return None
 
     class CustomDate(object):
 

test-requirements.txt

@@ -1,3 +1,4 @@
 pytest~=7.2.2
+pytest-mock~=3.14.0
 coverage~=7.2.2
 testfixtures~=8.2.0