From 5e2690644674c8de3079cea2a0fd3b2e01256bfd Mon Sep 17 00:00:00 2001 From: Yuriy Nesenenko Date: Mon, 10 Aug 2015 19:30:20 +0300 Subject: [PATCH] Add tests for python-cinderclient Add tests for cinder volume creation and deletion by ID and by Name. Add test for cinder show volume output. Add method to wait for given volume status. Add method to check that volume deleted. Add method to check that volume is not deleted (for negative cases). Add method for volume deletion. Add method to get Volume properties from CLI output. Change-Id: I1e1f9bea2afc63e408a88e64fc1b6b6ee9f4e79b Implements: partial blueprint cinderclient-functional-tests Co-Authored-by: Kyrylo Romanenko --- cinderclient/tests/functional/base.py | 90 ++++++++++++++++++++++- cinderclient/tests/functional/test_cli.py | 57 ++++++++++++++ 2 files changed, 144 insertions(+), 3 deletions(-) create mode 100644 cinderclient/tests/functional/test_cli.py diff --git a/cinderclient/tests/functional/base.py b/cinderclient/tests/functional/base.py index b8d593904..e2b1b676b 100644 --- a/cinderclient/tests/functional/base.py +++ b/cinderclient/tests/functional/base.py @@ -11,11 +11,12 @@ # under the License. import os +import time -from six.moves import configparser +import six from tempest_lib.cli import base from tempest_lib.cli import output_parser - +from tempest_lib import exceptions _CREDS_FILE = 'functional_creds.conf' @@ -36,7 +37,7 @@ def credentials(): tenant_name = os.environ.get('OS_TENANT_NAME') auth_url = os.environ.get('OS_AUTH_URL') - config = configparser.RawConfigParser() + config = six.moves.configparser.RawConfigParser() if config.read(_CREDS_FILE): username = username or config.get('admin', 'user') password = password or config.get('admin', 'pass') @@ -95,3 +96,86 @@ class ClientTestBase(base.ClientTestBase): for item in items: for field in field_names: self.assertIn(field, item) + + def assert_volume_details_rows(self, items): + """Check presence of common volume properties. + + :param items: volume properties + """ + values = ('attachments', 'availability_zone', 'bootable', 'created_at', + 'description', 'encrypted', 'id', 'metadata', 'name', 'size', + 'status', 'user_id', 'volume_type') + + for value in values: + self.assertIn(value, items) + + def wait_for_volume_status(self, volume_id, status, timeout=60): + """Wait until volume reaches given status. + + :param volume_id: uuid4 id of given volume + :param status: expected status of volume + :param timeout: timeout in seconds + """ + start_time = time.time() + while time.time() - start_time < timeout: + if status in self.cinder('show', params=volume_id): + break + else: + self.fail("Volume %s did not reach status %s after %d seconds." + % (volume_id, status, timeout)) + + def check_volume_not_deleted(self, volume_id): + """Check that volume exists. + + :param volume_id: uuid4 id of given volume + """ + self.assertTrue(self.cinder('show', params=volume_id)) + + def check_volume_deleted(self, volume_id, timeout=60): + """Check that volume deleted successfully. + + :param timeout: + :param volume_id: uuid4 id of given volume + :param timeout: timeout in seconds + """ + try: + start_time = time.time() + while time.time() - start_time < timeout: + if volume_id not in self.cinder('show', params=volume_id): + break + except exceptions.CommandFailed: + pass + else: + self.fail("Volume %s not deleted after %d seconds." + % (volume_id, timeout)) + + def volume_create(self, params): + """Create volume. + + :param params: parameters to cinder command + :return: volume dictionary + """ + output = self.cinder('create', params=params) + volume = self._get_property_from_output(output) + self.addCleanup(self.volume_delete, volume['id']) + self.wait_for_volume_status(volume['id'], 'available') + return volume + + def volume_delete(self, volume_id): + """Delete specified volume by ID. + + :param volume_id: uuid4 id of given volume + """ + if volume_id in self.cinder('list'): + self.cinder('delete', params=volume_id) + + def _get_property_from_output(self, output): + """Create a dictionary from an output + + :param output: the output of the cmd + """ + obj = {} + items = self.parser.listing(output) + for item in items: + obj[item['Property']] = six.text_type(item['Value']) + return obj diff --git a/cinderclient/tests/functional/test_cli.py b/cinderclient/tests/functional/test_cli.py new file mode 100644 index 000000000..b4beae509 --- /dev/null +++ b/cinderclient/tests/functional/test_cli.py @@ -0,0 +1,57 @@ +# 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 cinderclient.tests.functional import base + + +class CinderClientTests(base.ClientTestBase): + """Basic test for cinder client. + + Check of base cinder commands. + """ + def test_volume_create_delete_id(self): + """Create and delete a volume by ID.""" + volume = self.volume_create(params='1') + self.assert_volume_details_rows(volume.keys()) + self.volume_delete(volume['id']) + self.check_volume_deleted(volume['id']) + + def test_volume_create_delete_name(self): + """Create and delete a volume by name.""" + volume = self.volume_create(params='1 --name TestVolumeNamedCreate') + + self.cinder('delete', params='TestVolumeNamedCreate') + self.check_volume_deleted(volume['id']) + + def test_volume_show(self): + """Show volume details.""" + volume = self.volume_create(params='1 --name TestVolumeShow') + output = self.cinder('show', params='TestVolumeShow') + volume = self._get_property_from_output(output) + self.assertEqual('TestVolumeShow', volume['name']) + self.assert_volume_details_rows(volume.keys()) + + self.volume_delete(volume['id']) + self.check_volume_deleted(volume['id']) + + def test_volume_extend(self): + """Extend a volume size.""" + volume = self.volume_create(params='1 --name TestVolumeExtend') + self.cinder('extend', params="%s %s" % (volume['id'], 2)) + self.wait_for_volume_status(volume['id'], 'available') + output = self.cinder('show', params=volume['id']) + volume = self._get_property_from_output(output) + self.assertEqual('2', volume['size']) + + self.volume_delete(volume['id']) + self.check_volume_deleted(volume['id'])