openstack/python-tackerclient
Code Issues Proposed changes

Merge "Support listing all records at once with paging"

Browse Source
This commit is contained in:
Zuul 2022-09-01 01:03:51 +00:00 committed by Gerrit Code Review
commit 3da27cc89a
4 changed files with 160 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

43
tackerclient/tests/unit/osc/v1/test_vnflcm.py
View File

@ -13,6 +13,7 @@
# License for the specific language governing permissions and limitations
# under the License.
import copy
from io import StringIO
import os
import sys
@ -182,6 +183,48 @@ class TestListVnfLcm(TestVnfLcm):
actual_columns)
self.assertCountEqual(expected_data, list(data))
def test_take_action_with_pagination(self):
next_links_num = 3
parsed_args = self.check_parser(self.list_vnf_instance, [], [])
path = os.path.join(self.url, 'vnflcm/v1/vnf_instances')
links = [0] * next_links_num
link_headers = [0] * next_links_num
for i in range(next_links_num):
links[i] = (
'{base_url}?nextpage_opaque_marker={vnf_instance_id}'.format(
base_url=path,
vnf_instance_id=self.vnf_instances[i]['id']))
link_headers[i] = copy.deepcopy(self.header)
link_headers[i]['Link'] = '<{link_url}>; rel="next"'.format(
link_url=links[i])
self.requests_mock.register_uri(
'GET', path, json=[self.vnf_instances[0]], headers=link_headers[0])
self.requests_mock.register_uri(
'GET', links[0], json=[self.vnf_instances[1]],
headers=link_headers[1])
self.requests_mock.register_uri(
'GET', links[1], json=[self.vnf_instances[2]],
headers=link_headers[2])
self.requests_mock.register_uri(
'GET', links[2], json=[], headers=self.header)
actual_columns, data = self.list_vnf_instance.take_action(parsed_args)
headers, columns = tacker_osc_utils.get_column_definitions(
vnflcm._attr_map, long_listing=True)
expected_data = []
for vnf_instance_obj in self.vnf_instances:
expected_data.append(vnflcm_fakes.get_vnflcm_data(
vnf_instance_obj, columns=columns, list_action=True))
self.assertCountEqual(_get_columns_vnflcm(action='list'),
actual_columns)
self.assertCountEqual(expected_data, list(data))
class TestInstantiateVnfLcm(TestVnfLcm):

45
tackerclient/tests/unit/osc/v1/test_vnflcm_op_occs.py
View File

@ -10,6 +10,7 @@
# License for the specific language governing permissions and limitations
# under the License.
import copy
from io import StringIO
import os
import sys
@ -529,6 +530,50 @@ class TestListVnfLcmOp(TestVnfLcm):
actual_columns)
self.assertListItemsEqual(expected_data, list(data))
def test_take_action_with_pagination(self):
next_links_num = 3
vnflcm_op_occs_obj = vnflcm_op_occs_fakes.create_vnflcm_op_occs(
count=next_links_num)
parsed_args = self.check_parser(self.list_vnflcm_op_occ, [], [])
path = os.path.join(self.url, 'vnflcm/v1/vnf_lcm_op_occs')
links = [0] * next_links_num
link_headers = [0] * next_links_num
for i in range(next_links_num):
links[i] = (
'{base_url}?nextpage_opaque_marker={vnflcm_op_occ_id}'.format(
base_url=path,
vnflcm_op_occ_id=vnflcm_op_occs_obj[i]['id']))
link_headers[i] = copy.deepcopy(self.header)
link_headers[i]['Link'] = '<{link_url}>; rel="next"'.format(
link_url=links[i])
self.requests_mock.register_uri(
'GET', path, json=[vnflcm_op_occs_obj[0]], headers=link_headers[0])
self.requests_mock.register_uri(
'GET', links[0], json=[vnflcm_op_occs_obj[1]],
headers=link_headers[1])
self.requests_mock.register_uri(
'GET', links[1], json=[vnflcm_op_occs_obj[2]],
headers=link_headers[2])
self.requests_mock.register_uri(
'GET', links[2], json=[], headers=self.header)
actual_columns, data = self.list_vnflcm_op_occ.take_action(parsed_args)
headers, columns = tacker_osc_utils.get_column_definitions(
self.list_vnflcm_op_occ.get_attributes(), long_listing=True)
expected_data = []
for vnflcm_op_occ_obj_idx in vnflcm_op_occs_obj:
expected_data.append(vnflcm_op_occs_fakes.get_vnflcm_op_occ_data(
vnflcm_op_occ_obj_idx, columns=columns))
self.assertCountEqual(_get_columns_vnflcm_op_occs(action='list'),
actual_columns)
self.assertCountEqual(expected_data, list(data))
class TestShowVnfLcmOp(TestVnfLcm):

45
tackerclient/tests/unit/osc/v1/test_vnflcm_subsc.py
View File

@ -13,6 +13,7 @@
# License for the specific language governing permissions and limitations
# under the License.
import copy
import os
import sys
@ -99,6 +100,50 @@ class TestListLccnSubscription(test_vnflcm.TestVnfLcm):
actual_columns)
self.assertCountEqual(expected_data, list(data))
def test_take_action_with_pagination(self):
next_links_num = 3
path = os.path.join(self.url, 'vnflcm/v1/subscriptions')
parsed_args = self.check_parser(self.list_subscription, [], [])
links = [0] * next_links_num
link_headers = [0] * next_links_num
for i in range(next_links_num):
links[i] = (
'{base_url}?nextpage_opaque_marker={subscription_id}'.format(
base_url=path,
subscription_id=self.subscriptions[i]['id']))
link_headers[i] = copy.deepcopy(self.header)
link_headers[i]['Link'] = '<{link_url}>; rel="next"'.format(
link_url=links[i])
self.requests_mock.register_uri(
'GET', path, json=[self.subscriptions[0]], headers=link_headers[0])
self.requests_mock.register_uri(
'GET', links[0], json=[self.subscriptions[1]],
headers=link_headers[1])
self.requests_mock.register_uri(
'GET', links[1], json=[self.subscriptions[2]],
headers=link_headers[2])
self.requests_mock.register_uri(
'GET', links[2], json=[], headers=self.header)
actual_columns, data = self.list_subscription.take_action(parsed_args)
headers, columns = tacker_osc_utils.get_column_definitions(
self.list_subscription.get_attributes(), long_listing=True)
expected_data = []
for subscription_obj in self.subscriptions:
expected_data.append(vnflcm_subsc_fakes.get_subscription_data(
subscription_obj, columns=columns, list_action=True))
self.assertCountEqual(_get_columns_vnflcm_subsc(action='list'),
actual_columns)
self.assertCountEqual(expected_data, list(data))
class TestShowLccnSubscription(test_vnflcm.TestVnfLcm):

36
tackerclient/v1_0/client.py
View File

@ -16,6 +16,7 @@
#
import logging
import re
import time
import requests
@ -180,6 +181,8 @@ class ClientBase(object):
self.format = 'json'
self.action_prefix = "/v%s" % (self.version)
self.retry_interval = 1
self.rel = None
self.params = None
def _handle_fault_response(self, status_code, response_body):
# Create exception with HTTP status code and message
@ -246,6 +249,19 @@ class ClientBase(object):
else:
self.format = 'json'
url = None
rel = None
link = resp.headers.get('Link', None)
if link is not None:
url = re.findall('<(.*)>', link)[0]
rel = re.findall('rel="(.*)"', link)[0]
if rel == 'next':
self.rel = 'next'
query_str = urlparse.urlparse(url).query
self.params = urlparse.parse_qs(query_str)
status_code = resp.status_code
if status_code in (requests.codes.ok,
requests.codes.created,
@ -379,21 +395,23 @@ class ClientBase(object):
linkrel = 'next'
next = True
while next:
self.rel = None
res = self.get(path, headers=headers, params=params)
yield res
next = False
try:
# TODO(tpatil): Handle pagination for list data type
# once it's supported by tacker.
if type(res) is list:
break
for link in res['%s_links' % collection]:
if link['rel'] == linkrel:
query_str = urlparse.urlparse(link['href']).query
params = urlparse.parse_qs(query_str)
if self.rel == 'next':
params = self.params
next = True
break
else:
for link in res['%s_links' % collection]:
if link['rel'] == linkrel:
query_str = urlparse.urlparse(link['href']).query
params = urlparse.parse_qs(query_str)
next = True
break
except KeyError:
break