Added API tests for page_reverse and href next/previous links
Change-Id: I0e7834b600f73940af73043989492e9c1c188178
diff --git a/neutron/tests/tempest/api/base.py b/neutron/tests/tempest/api/base.py
index 31d6c3a..993e356 100644
--- a/neutron/tests/tempest/api/base.py
+++ b/neutron/tests/tempest/api/base.py
@@ -14,6 +14,7 @@
# under the License.
import functools
+import math
import netaddr
from tempest.lib.common.utils import data_utils
@@ -21,6 +22,7 @@
from tempest import test
from neutron.common import constants
+from neutron.common import utils
from neutron.tests.tempest.api import clients
from neutron.tests.tempest import config
from neutron.tests.tempest import exceptions
@@ -499,14 +501,31 @@
list_kwargs = {}
+ def assertSameOrder(self, original, actual):
+ # gracefully handle iterators passed
+ original = list(original)
+ actual = list(actual)
+ self.assertEqual(len(original), len(actual))
+ for expected, res in zip(original, actual):
+ self.assertEqual(expected['name'], res['name'])
+
+ @utils.classproperty
+ def plural_name(self):
+ return '%ss' % self.resource
+
def list_method(self, *args, **kwargs):
- method = getattr(self.client, 'list_%ss' % self.resource)
+ method = getattr(self.client, 'list_%s' % self.plural_name)
kwargs.update(self.list_kwargs)
return method(*args, **kwargs)
+ def get_bare_url(self, url):
+ base_url = self.client.base_url
+ self.assertTrue(url.startswith(base_url))
+ return url[len(base_url):]
+
@classmethod
def _extract_resources(cls, body):
- return body['%ss' % cls.resource]
+ return body[cls.plural_name]
def _test_list_sorts(self, direction):
sort_args = {
@@ -550,9 +569,7 @@
resources = self._extract_resources(body)
self.assertTrue(len(resources) >= len(self.resource_names))
- @_require_pagination
- @_require_sorting
- def _test_list_pagination_with_marker(self):
+ def _test_list_pagination_iteratively(self, lister):
# first, collect all resources for later comparison
sort_args = {
'sort_dir': constants.SORT_DIRECTION_ASC,
@@ -562,10 +579,19 @@
expected_resources = self._extract_resources(body)
self.assertNotEmpty(expected_resources)
+ resources = lister(
+ len(expected_resources), sort_args
+ )
+
+ # finally, compare that the list retrieved in one go is identical to
+ # the one containing pagination results
+ self.assertSameOrder(expected_resources, resources)
+
+ def _list_all_with_marker(self, niterations, sort_args):
# paginate resources one by one, using last fetched resource as a
# marker
resources = []
- for i in range(len(expected_resources)):
+ for i in range(niterations):
pagination_args = sort_args.copy()
pagination_args['limit'] = 1
if resources:
@@ -574,8 +600,119 @@
resources_ = self._extract_resources(body)
self.assertEqual(1, len(resources_))
resources.extend(resources_)
+ return resources
- # finally, compare that the list retrieved in one go is identical to
- # the one containing pagination results
- for expected, res in zip(expected_resources, resources):
- self.assertEqual(expected['name'], res['name'])
+ @_require_pagination
+ @_require_sorting
+ def _test_list_pagination_with_marker(self):
+ self._test_list_pagination_iteratively(self._list_all_with_marker)
+
+ def _list_all_with_hrefs(self, niterations, sort_args):
+ # paginate resources one by one, using next href links
+ resources = []
+ prev_links = {}
+
+ for i in range(niterations):
+ if prev_links:
+ uri = self.get_bare_url(prev_links['next'])
+ else:
+ uri = self.client.build_uri(
+ self.plural_name, limit=1, **sort_args)
+ prev_links, body = self.client.get_uri_with_links(
+ self.plural_name, uri
+ )
+ resources_ = self._extract_resources(body)
+ self.assertEqual(1, len(resources_))
+ resources.extend(resources_)
+
+ # The last element is empty and does not contain 'next' link
+ uri = self.get_bare_url(prev_links['next'])
+ prev_links, body = self.client.get_uri_with_links(
+ self.plural_name, uri
+ )
+ self.assertNotIn('next', prev_links)
+
+ # Now walk backwards and compare results
+ resources2 = []
+ for i in range(niterations):
+ uri = self.get_bare_url(prev_links['previous'])
+ prev_links, body = self.client.get_uri_with_links(
+ self.plural_name, uri
+ )
+ resources_ = self._extract_resources(body)
+ self.assertEqual(1, len(resources_))
+ resources2.extend(resources_)
+
+ self.assertSameOrder(resources, reversed(resources2))
+
+ return resources
+
+ @_require_pagination
+ @_require_sorting
+ def _test_list_pagination_with_href_links(self):
+ self._test_list_pagination_iteratively(self._list_all_with_hrefs)
+
+ @_require_pagination
+ @_require_sorting
+ def _test_list_pagination_page_reverse_with_href_links(
+ self, direction=constants.SORT_DIRECTION_ASC):
+ pagination_args = {
+ 'sort_dir': direction,
+ 'sort_key': 'name',
+ }
+ body = self.list_method(**pagination_args)
+ expected_resources = self._extract_resources(body)
+
+ page_size = 2
+ pagination_args['limit'] = page_size
+
+ prev_links = {}
+ resources = []
+ num_resources = len(expected_resources)
+ niterations = int(math.ceil(float(num_resources) / page_size))
+ for i in range(niterations):
+ if prev_links:
+ uri = self.get_bare_url(prev_links['previous'])
+ else:
+ uri = self.client.build_uri(
+ self.plural_name, page_reverse=True, **pagination_args)
+ prev_links, body = self.client.get_uri_with_links(
+ self.plural_name, uri
+ )
+ resources_ = self._extract_resources(body)
+ self.assertTrue(page_size >= len(resources_))
+ resources.extend(reversed(resources_))
+
+ self.assertSameOrder(expected_resources, reversed(resources))
+
+ @_require_pagination
+ @_require_sorting
+ def _test_list_pagination_page_reverse_asc(self):
+ self._test_list_pagination_page_reverse(
+ direction=constants.SORT_DIRECTION_ASC)
+
+ @_require_pagination
+ @_require_sorting
+ def _test_list_pagination_page_reverse_desc(self):
+ self._test_list_pagination_page_reverse(
+ direction=constants.SORT_DIRECTION_DESC)
+
+ def _test_list_pagination_page_reverse(self, direction):
+ pagination_args = {
+ 'sort_dir': direction,
+ 'sort_key': 'name',
+ 'limit': 3,
+ }
+ body = self.list_method(**pagination_args)
+ expected_resources = self._extract_resources(body)
+
+ pagination_args['limit'] -= 1
+ pagination_args['marker'] = expected_resources[-1]['id']
+ pagination_args['page_reverse'] = True
+ body = self.list_method(**pagination_args)
+
+ self.assertSameOrder(
+ # the last entry is not included in 2nd result when used as a
+ # marker
+ expected_resources[:-1],
+ self._extract_resources(body))