Merge "Move cinder tests into unversioned path - part2"

This commit is contained in:
Jenkins
2017-04-06 23:58:49 +00:00
committed by Gerrit Code Review
5 changed files with 225 additions and 299 deletions
+166 -1
View File
@@ -13,8 +13,11 @@
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import operator
import operator
import random
from six.moves.urllib import parse
from testtools import matchers
from tempest.api.volume import base
@@ -59,6 +62,10 @@ class VolumesListTestJSON(base.BaseVolumeTest):
def resource_setup(cls):
super(VolumesListTestJSON, cls).resource_setup()
cls.name = cls.VOLUME_FIELDS[1]
existing_volumes = cls.volumes_client.list_volumes()['volumes']
cls.volume_id_list = [vol['id'] for vol in existing_volumes]
# Create 3 test volumes
cls.volume_list = []
cls.metadata = {'Type': 'work'}
@@ -66,6 +73,7 @@ class VolumesListTestJSON(base.BaseVolumeTest):
volume = cls.create_volume(metadata=cls.metadata)
volume = cls.volumes_client.show_volume(volume['id'])['volume']
cls.volume_list.append(volume)
cls.volume_id_list.append(volume['id'])
def _list_by_param_value_and_assert(self, params, with_detail=False):
"""list or list_details with given params and validates result"""
@@ -217,3 +225,160 @@ class VolumesListTestJSON(base.BaseVolumeTest):
params = {self.name: volume[self.name],
'status': 'available'}
self._list_by_param_value_and_assert(params, with_detail=True)
@decorators.idempotent_id('2a7064eb-b9c3-429b-b888-33928fc5edd3')
def test_volume_list_details_with_multiple_params(self):
# List volumes detail using combined condition
def _list_details_with_multiple_params(limit=2,
status='available',
sort_dir='asc',
sort_key='id'):
params = {'limit': limit,
'status': status,
'sort_dir': sort_dir,
'sort_key': sort_key
}
fetched_volume = self.volumes_client.list_volumes(
detail=True, params=params)['volumes']
self.assertEqual(limit, len(fetched_volume),
"The count of volumes is %s, expected:%s " %
(len(fetched_volume), limit))
self.assertEqual(status, fetched_volume[0]['status'])
self.assertEqual(status, fetched_volume[1]['status'])
val0 = fetched_volume[0][sort_key]
val1 = fetched_volume[1][sort_key]
if sort_dir == 'asc':
self.assertLess(val0, val1,
"list is not in asc order with sort_key: %s."
" %s" % (sort_key, fetched_volume))
elif sort_dir == 'desc':
self.assertGreater(val0, val1,
"list is not in desc order with sort_key: "
"%s. %s" % (sort_key, fetched_volume))
_list_details_with_multiple_params()
_list_details_with_multiple_params(sort_dir='desc')
def _test_pagination(self, resource, ids=None, limit=1, **kwargs):
"""Check list pagination functionality for a resource.
This method requests the list of resources and follows pagination
links.
If an iterable is supplied in ids it will check that all ids are
retrieved and that only those are listed, that we will get a next
link for an empty page if the number of items is divisible by used
limit (this is expected behavior).
We can specify number of items per request using limit argument.
"""
# Get list method for the type of resource from the client
client = getattr(self, resource + '_client')
method = getattr(client, 'list_' + resource)
# Include limit in params for list request
params = kwargs.pop('params', {})
params['limit'] = limit
# Store remaining items we are expecting from list
if ids is not None:
remaining = list(ids)
else:
remaining = None
# Mark that the current iteration is not from a 'next' link
next = None
while True:
# Get a list page
response = method(params=params, **kwargs)
# If we have to check ids
if remaining is not None:
# Confirm we receive expected number of elements
num_expected = min(len(remaining), limit)
self.assertEqual(num_expected, len(response[resource]),
'Requested %(#expect)d but got %(#received)d '
% {'#expect': num_expected,
'#received': len(response[resource])})
# For each received element
for element in response[resource]:
element_id = element['id']
# Check it's one of expected ids
self.assertIn(element_id,
ids,
'Id %(id)s is not in expected ids %(ids)s' %
{'id': element_id, 'ids': ids})
# If not in remaining, we have received it twice
self.assertIn(element_id,
remaining,
'Id %s was received twice' % element_id)
# We no longer expect it
remaining.remove(element_id)
# If the current iteration is from a 'next' link, check that the
# absolute url is the same as the one used for this request
if next:
self.assertEqual(next, response.response['content-location'])
# Get next from response
next = None
for link in response.get(resource + '_links', ()):
if link['rel'] == 'next':
next = link['href']
break
# Check if we have next and we shouldn't or the other way around
if remaining is not None:
if remaining or (num_expected and len(ids) % limit == 0):
self.assertIsNotNone(next, 'Missing link to next page')
else:
self.assertIsNone(next, 'Unexpected link to next page')
# If we can follow to the next page, get params from url to make
# request in the form of a relative URL
if next:
params = parse.urlparse(next).query
# If cannot follow make sure it's because we have finished
else:
self.assertEqual([], remaining or [],
'No more pages reported, but still '
'missing ids %s' % remaining)
break
@decorators.idempotent_id('e9138a2c-f67b-4796-8efa-635c196d01de')
def test_volume_list_details_pagination(self):
self._test_pagination('volumes', ids=self.volume_id_list, detail=True)
@decorators.idempotent_id('af55e775-8e4b-4feb-8719-215c43b0238c')
def test_volume_list_pagination(self):
self._test_pagination('volumes', ids=self.volume_id_list, detail=False)
@decorators.idempotent_id('46eff077-100b-427f-914e-3db2abcdb7e2')
@decorators.skip_because(bug='1572765')
def test_volume_list_with_detail_param_marker(self):
# Choosing a random volume from a list of volumes for 'marker'
# parameter
random_volume = random.choice(self.volume_id_list)
params = {'marker': random_volume}
# Running volume list using marker parameter
vol_with_marker = self.volumes_client.list_volumes(
detail=True, params=params)['volumes']
# Fetching the index of the random volume from volume_id_list
index_marker = self.volume_id_list.index(random_volume)
# The expected list with marker parameter
verify_volume_list = self.volume_id_list[:index_marker]
failed_msg = "Failed to list volume details by marker"
# Validating the expected list is the same like the observed list
self.assertEqual(verify_volume_list,
map(lambda x: x['id'],
vol_with_marker[::-1]), failed_msg)
@@ -28,10 +28,14 @@ class VolumesSnapshotListTestJSON(base.BaseVolumeTest):
@classmethod
def resource_setup(cls):
super(VolumesSnapshotListTestJSON, cls).resource_setup()
cls.snapshot_id_list = []
volume_origin = cls.create_volume()
# Create snapshots with params
for _ in range(2):
cls.snapshot = cls.create_snapshot(volume_origin['id'])
for _ in range(3):
snapshot = cls.create_snapshot(volume_origin['id'])
cls.snapshot_id_list.append(snapshot['id'])
cls.snapshot = snapshot
def _list_by_param_values_and_assert(self, with_detail=False, **params):
"""list or list_details with given params and validates result."""
@@ -101,3 +105,56 @@ class VolumesSnapshotListTestJSON(base.BaseVolumeTest):
def test_snapshot_list_param_limit_equals_zero(self):
# List returns zero elements
self._list_snapshots_by_param_limit(limit=0, expected_elements=0)
def _list_snapshots_param_sort(self, sort_key, sort_dir):
"""list snapshots by sort param"""
snap_list = self.snapshots_client.list_snapshots(
sort_key=sort_key, sort_dir=sort_dir)['snapshots']
self.assertNotEmpty(snap_list)
if sort_key is 'display_name':
sort_key = 'name'
# Note: On Cinder API, 'display_name' works as a sort key
# on a request, a volume name appears as 'name' on the response.
# So Tempest needs to change the key name here for this inconsistent
# API behavior.
sorted_list = [snapshot[sort_key] for snapshot in snap_list]
msg = 'The list of snapshots was not sorted correctly.'
self.assertEqual(sorted(sorted_list, reverse=(sort_dir == 'desc')),
sorted_list, msg)
@decorators.idempotent_id('c5513ada-64c1-4d28-83b9-af3307ec1388')
def test_snapshot_list_param_sort_id_asc(self):
self._list_snapshots_param_sort(sort_key='id', sort_dir='asc')
@decorators.idempotent_id('8a7fe058-0b41-402a-8afd-2dbc5a4a718b')
def test_snapshot_list_param_sort_id_desc(self):
self._list_snapshots_param_sort(sort_key='id', sort_dir='desc')
@decorators.idempotent_id('4052c3a0-2415-440a-a8cc-305a875331b0')
def test_snapshot_list_param_sort_created_at_asc(self):
self._list_snapshots_param_sort(sort_key='created_at', sort_dir='asc')
@decorators.idempotent_id('dcbbe24a-f3c0-4ec8-9274-55d48db8d1cf')
def test_snapshot_list_param_sort_created_at_desc(self):
self._list_snapshots_param_sort(sort_key='created_at', sort_dir='desc')
@decorators.idempotent_id('d58b5fed-0c37-42d3-8c5d-39014ac13c00')
def test_snapshot_list_param_sort_name_asc(self):
self._list_snapshots_param_sort(sort_key='display_name',
sort_dir='asc')
@decorators.idempotent_id('96ba6f4d-1f18-47e1-b4bc-76edc6c21250')
def test_snapshot_list_param_sort_name_desc(self):
self._list_snapshots_param_sort(sort_key='display_name',
sort_dir='desc')
@decorators.idempotent_id('05489dde-44bc-4961-a1f5-3ce7ee7824f7')
def test_snapshot_list_param_marker(self):
# The list of snapshots should end before the provided marker
params = {'marker': self.snapshot_id_list[1]}
snap_list = self.snapshots_client.list_snapshots(**params)['snapshots']
fetched_list_id = [snap['id'] for snap in snap_list]
# Verify the list of snapshots ends before the provided
# marker(second snapshot), therefore only the first snapshot
# should displayed.
self.assertEqual(self.snapshot_id_list[:1], fetched_list_id)
View File
-203
View File
@@ -1,203 +0,0 @@
# Copyright 2012 OpenStack Foundation
# Copyright 2013 IBM Corp.
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import random
from six.moves.urllib import parse
from tempest.api.volume import base
from tempest.lib import decorators
class VolumesListTestJSON(base.BaseVolumeTest):
"""volumes tests.
This test creates a number of 1G volumes. To run successfully,
ensure that the backing file for the volume group that Nova uses
has space for at least 3 1G volumes!
If you are running a Devstack environment, ensure that the
VOLUME_BACKING_FILE_SIZE is at least 4G in your localrc
"""
@classmethod
def resource_setup(cls):
super(VolumesListTestJSON, cls).resource_setup()
# Create 3 test volumes
# NOTE(zhufl): When using pre-provisioned credentials, the project
# may have volumes other than those created below.
existing_volumes = cls.volumes_client.list_volumes()['volumes']
cls.volume_id_list = [vol['id'] for vol in existing_volumes]
for _ in range(3):
volume = cls.create_volume()
cls.volume_id_list.append(volume['id'])
@decorators.idempotent_id('2a7064eb-b9c3-429b-b888-33928fc5edd3')
def test_volume_list_details_with_multiple_params(self):
# List volumes detail using combined condition
def _list_details_with_multiple_params(limit=2,
status='available',
sort_dir='asc',
sort_key='id'):
params = {'limit': limit,
'status': status,
'sort_dir': sort_dir,
'sort_key': sort_key
}
fetched_volume = self.volumes_client.list_volumes(
detail=True, params=params)['volumes']
self.assertEqual(limit, len(fetched_volume),
"The count of volumes is %s, expected:%s " %
(len(fetched_volume), limit))
self.assertEqual(status, fetched_volume[0]['status'])
self.assertEqual(status, fetched_volume[1]['status'])
val0 = fetched_volume[0][sort_key]
val1 = fetched_volume[1][sort_key]
if sort_dir == 'asc':
self.assertLess(val0, val1,
"list is not in asc order with sort_key: %s."
" %s" % (sort_key, fetched_volume))
elif sort_dir == 'desc':
self.assertGreater(val0, val1,
"list is not in desc order with sort_key: "
"%s. %s" % (sort_key, fetched_volume))
_list_details_with_multiple_params()
_list_details_with_multiple_params(sort_dir='desc')
def _test_pagination(self, resource, ids=None, limit=1, **kwargs):
"""Check list pagination functionality for a resource.
This method requests the list of resources and follows pagination
links.
If an iterable is supplied in ids it will check that all ids are
retrieved and that only those are listed, that we will get a next
link for an empty page if the number of items is divisible by used
limit (this is expected behavior).
We can specify number of items per request using limit argument.
"""
# Get list method for the type of resource from the client
client = getattr(self, resource + '_client')
method = getattr(client, 'list_' + resource)
# Include limit in params for list request
params = kwargs.pop('params', {})
params['limit'] = limit
# Store remaining items we are expecting from list
if ids is not None:
remaining = list(ids)
else:
remaining = None
# Mark that the current iteration is not from a 'next' link
next = None
while True:
# Get a list page
response = method(params=params, **kwargs)
# If we have to check ids
if remaining is not None:
# Confirm we receive expected number of elements
num_expected = min(len(remaining), limit)
self.assertEqual(num_expected, len(response[resource]),
'Requested %(#expect)d but got %(#received)d '
% {'#expect': num_expected,
'#received': len(response[resource])})
# For each received element
for element in response[resource]:
element_id = element['id']
# Check it's one of expected ids
self.assertIn(element_id,
ids,
'Id %(id)s is not in expected ids %(ids)s' %
{'id': element_id, 'ids': ids})
# If not in remaining, we have received it twice
self.assertIn(element_id,
remaining,
'Id %s was received twice' % element_id)
# We no longer expect it
remaining.remove(element_id)
# If the current iteration is from a 'next' link, check that the
# absolute url is the same as the one used for this request
if next:
self.assertEqual(next, response.response['content-location'])
# Get next from response
next = None
for link in response.get(resource + '_links', ()):
if link['rel'] == 'next':
next = link['href']
break
# Check if we have next and we shouldn't or the other way around
if remaining is not None:
if remaining or (num_expected and len(ids) % limit == 0):
self.assertIsNotNone(next, 'Missing link to next page')
else:
self.assertIsNone(next, 'Unexpected link to next page')
# If we can follow to the next page, get params from url to make
# request in the form of a relative URL
if next:
params = parse.urlparse(next).query
# If cannot follow make sure it's because we have finished
else:
self.assertEqual([], remaining or [],
'No more pages reported, but still '
'missing ids %s' % remaining)
break
@decorators.idempotent_id('e9138a2c-f67b-4796-8efa-635c196d01de')
def test_volume_list_details_pagination(self):
self._test_pagination('volumes', ids=self.volume_id_list, detail=True)
@decorators.idempotent_id('af55e775-8e4b-4feb-8719-215c43b0238c')
def test_volume_list_pagination(self):
self._test_pagination('volumes', ids=self.volume_id_list, detail=False)
@decorators.idempotent_id('46eff077-100b-427f-914e-3db2abcdb7e2')
@decorators.skip_because(bug='1572765')
def test_volume_list_with_detail_param_marker(self):
# Choosing a random volume from a list of volumes for 'marker'
# parameter
random_volume = random.choice(self.volume_id_list)
params = {'marker': random_volume}
# Running volume list using marker parameter
vol_with_marker = self.volumes_client.list_volumes(
detail=True, params=params)['volumes']
# Fetching the index of the random volume from volume_id_list
index_marker = self.volume_id_list.index(random_volume)
# The expected list with marker parameter
verify_volume_list = self.volume_id_list[:index_marker]
failed_msg = "Failed to list volume details by marker"
# Validating the expected list is the same like the observed list
self.assertEqual(verify_volume_list,
map(lambda x: x['id'],
vol_with_marker[::-1]), failed_msg)
@@ -1,93 +0,0 @@
# Copyright 2016 Red Hat, Inc.
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
from tempest.api.volume import base
from tempest import config
from tempest.lib import decorators
CONF = config.CONF
class VolumesSnapshotListTestJSON(base.BaseVolumeTest):
@classmethod
def skip_checks(cls):
super(VolumesSnapshotListTestJSON, cls).skip_checks()
if not CONF.volume_feature_enabled.snapshot:
raise cls.skipException("Cinder volume snapshots are disabled")
@classmethod
def resource_setup(cls):
super(VolumesSnapshotListTestJSON, cls).resource_setup()
cls.snapshot_id_list = []
# Create a volume
volume_origin = cls.create_volume()
# Create 3 snapshots
for _ in range(3):
snapshot = cls.create_snapshot(volume_origin['id'])
cls.snapshot_id_list.append(snapshot['id'])
def _list_snapshots_param_sort(self, sort_key, sort_dir):
"""list snapshots by sort param"""
snap_list = self.snapshots_client.list_snapshots(
sort_key=sort_key, sort_dir=sort_dir)['snapshots']
self.assertNotEmpty(snap_list)
if sort_key is 'display_name':
sort_key = 'name'
# Note: On Cinder API, 'display_name' works as a sort key
# on a request, a volume name appears as 'name' on the response.
# So Tempest needs to change the key name here for this inconsistent
# API behavior.
sorted_list = [snapshot[sort_key] for snapshot in snap_list]
msg = 'The list of snapshots was not sorted correctly.'
self.assertEqual(sorted(sorted_list, reverse=(sort_dir == 'desc')),
sorted_list, msg)
@decorators.idempotent_id('c5513ada-64c1-4d28-83b9-af3307ec1388')
def test_snapshot_list_param_sort_id_asc(self):
self._list_snapshots_param_sort(sort_key='id', sort_dir='asc')
@decorators.idempotent_id('8a7fe058-0b41-402a-8afd-2dbc5a4a718b')
def test_snapshot_list_param_sort_id_desc(self):
self._list_snapshots_param_sort(sort_key='id', sort_dir='desc')
@decorators.idempotent_id('4052c3a0-2415-440a-a8cc-305a875331b0')
def test_snapshot_list_param_sort_created_at_asc(self):
self._list_snapshots_param_sort(sort_key='created_at', sort_dir='asc')
@decorators.idempotent_id('dcbbe24a-f3c0-4ec8-9274-55d48db8d1cf')
def test_snapshot_list_param_sort_created_at_desc(self):
self._list_snapshots_param_sort(sort_key='created_at', sort_dir='desc')
@decorators.idempotent_id('d58b5fed-0c37-42d3-8c5d-39014ac13c00')
def test_snapshot_list_param_sort_name_asc(self):
self._list_snapshots_param_sort(sort_key='display_name',
sort_dir='asc')
@decorators.idempotent_id('96ba6f4d-1f18-47e1-b4bc-76edc6c21250')
def test_snapshot_list_param_sort_name_desc(self):
self._list_snapshots_param_sort(sort_key='display_name',
sort_dir='desc')
@decorators.idempotent_id('05489dde-44bc-4961-a1f5-3ce7ee7824f7')
def test_snapshot_list_param_marker(self):
# The list of snapshots should end before the provided marker
params = {'marker': self.snapshot_id_list[1]}
snap_list = self.snapshots_client.list_snapshots(**params)['snapshots']
fetched_list_id = [snap['id'] for snap in snap_list]
# Verify the list of snapshots ends before the provided
# marker(second snapshot), therefore only the first snapshot
# should displayed.
self.assertEqual(self.snapshot_id_list[:1], fetched_list_id)