From a17aa53f685a04a7e4f323c389cbc0cba2c916ec Mon Sep 17 00:00:00 2001 From: Yuriy Nesenenko Date: Wed, 2 Sep 2015 19:07:04 +0300 Subject: [PATCH] Remove duplicate code in functional tests This patch allows to use a standard approach to volumes, snapshots and backups to avoid duplicating code. Change-Id: I0059fcb3bb8823105a7f960a30e70036bcbded8b --- cinderclient/tests/functional/base.py | 150 +++++++--------------- cinderclient/tests/functional/test_cli.py | 62 +++++---- 2 files changed, 82 insertions(+), 130 deletions(-) diff --git a/cinderclient/tests/functional/base.py b/cinderclient/tests/functional/base.py index 11e3072c8..fefa180e5 100644 --- a/cinderclient/tests/functional/base.py +++ b/cinderclient/tests/functional/base.py @@ -97,77 +97,15 @@ class ClientTestBase(base.ClientTestBase): for field in field_names: self.assertIn(field, item) - def assert_volume_details(self, items): - """Check presence of common volume properties. + def assert_object_details(self, expected, items): + """Check presence of common object properties. - :param items: volume properties + :param expected: expected object properties + :param items: object properties """ - values = ('attachments', 'availability_zone', 'bootable', 'created_at', - 'description', 'encrypted', 'id', 'metadata', 'name', 'size', - 'status', 'user_id', 'volume_type') - - for value in values: + for value in expected: 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 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 @@ -179,65 +117,67 @@ class ClientTestBase(base.ClientTestBase): obj[item['Property']] = six.text_type(item['Value']) return obj - def wait_for_snapshot_status(self, snapshot_id, status, timeout=60): - """Wait until snapshot reaches given status. + def object_cmd(self, object_name, cmd): + return (object_name + '-' + cmd if object_name != 'volume' else cmd) - :param snapshot_id: uuid4 id of given volume - :param status: expected snapshot's status + def wait_for_object_status(self, object_name, object_id, status, + timeout=60): + """Wait until object reaches given status. + + :param object_name: object name + :param object_id: uuid4 id of an object + :param status: expected status of an object :param timeout: timeout in seconds """ + cmd = self.object_cmd(object_name, 'show') start_time = time.time() while time.time() - start_time < timeout: - if status in self.cinder('snapshot-show', params=snapshot_id): + if status in self.cinder(cmd, params=object_id): break else: - self.fail("Snapshot %s did not reach status %s after %d seconds." - % (snapshot_id, status, timeout)) + self.fail("%s %s did not reach status %s after %d seconds." + % (object_name, object_id, status, timeout)) - def check_snapshot_deleted(self, snapshot_id, timeout=60): - """Check that snapshot deleted successfully. + def check_object_deleted(self, object_name, object_id, timeout=60): + """Check that volume deleted successfully. - :param snapshot_id: the given snapshot id + :param object_name: object name + :param object_id: uuid4 id of an object :param timeout: timeout in seconds """ + cmd = self.object_cmd(object_name, 'show') try: start_time = time.time() while time.time() - start_time < timeout: - if snapshot_id not in self.cinder('snapshot-show', - params=snapshot_id): + if object_id not in self.cinder(cmd, params=object_id): break except exceptions.CommandFailed: pass else: - self.fail("Snapshot %s has not deleted after %d seconds." - % (snapshot_id, timeout)) + self.fail("%s %s not deleted after %d seconds." + % (object_name, object_id, timeout)) - def assert_snapshot_details(self, items): - """Check presence of common volume snapshot properties. + def object_create(self, object_name, params): + """Create an object. - :param items: volume snapshot properties + :param object_name: object name + :param params: parameters to cinder command + :return: object dictionary """ - values = ('created_at', 'description', 'id', 'metadata', 'name', - 'size', 'status', 'volume_id') + cmd = self.object_cmd(object_name, 'create') + output = self.cinder(cmd, params=params) + object = self._get_property_from_output(output) + self.addCleanup(self.object_delete, object_name, object['id']) + self.wait_for_object_status(object_name, object['id'], 'available') + return object - for value in values: - self.assertIn(value, items) + def object_delete(self, object_name, object_id): + """Delete specified object by ID. - def snapshot_create(self, volume_id): - """Create a volume snapshot from the volume id. - - :param volume_id: the given volume id to create a snapshot + :param object_name: object name + :param object_id: uuid4 id of an object """ - output = self.cinder('snapshot-create', params=volume_id) - snapshot = self._get_property_from_output(output) - self.addCleanup(self.snapshot_delete, snapshot['id']) - self.wait_for_snapshot_status(snapshot['id'], 'available') - return snapshot - - def snapshot_delete(self, snapshot_id): - """Delete specified snapshot by ID. - - :param snapshot_id: the given snapshot id - """ - if snapshot_id in self.cinder('snapshot-list'): - self.cinder('snapshot-delete', params=snapshot_id) + cmd = self.object_cmd(object_name, 'list') + cmd_delete = self.object_cmd(object_name, 'delete') + if object_id in self.cinder(cmd): + self.cinder(cmd_delete, params=object_id) diff --git a/cinderclient/tests/functional/test_cli.py b/cinderclient/tests/functional/test_cli.py index 4767d9c3f..1e3a4d743 100644 --- a/cinderclient/tests/functional/test_cli.py +++ b/cinderclient/tests/functional/test_cli.py @@ -14,54 +14,66 @@ from cinderclient.tests.functional import base -class CinderClientTests(base.ClientTestBase): - """Basic test for cinder client. +class CinderVolumeTests(base.ClientTestBase): + """Check of base cinder volume commands.""" + + VOLUME_PROPERTY = ('attachments', 'availability_zone', 'bootable', + 'created_at', 'description', 'encrypted', 'id', + 'metadata', 'name', 'size', 'status', + 'user_id', 'volume_type') - 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(volume.keys()) - self.volume_delete(volume['id']) - self.check_volume_deleted(volume['id']) + volume = self.object_create('volume', params='1') + self.assert_object_details(self.VOLUME_PROPERTY, volume.keys()) + self.object_delete('volume', volume['id']) + self.check_object_deleted('volume', volume['id']) def test_volume_create_delete_name(self): """Create and delete a volume by name.""" - volume = self.volume_create(params='1 --name TestVolumeNamedCreate') + volume = self.object_create('volume', + params='1 --name TestVolumeNamedCreate') self.cinder('delete', params='TestVolumeNamedCreate') - self.check_volume_deleted(volume['id']) + self.check_object_deleted('volume', volume['id']) def test_volume_show(self): """Show volume details.""" - volume = self.volume_create(params='1 --name TestVolumeShow') + volume = self.object_create('volume', 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(volume.keys()) + self.assert_object_details(self.VOLUME_PROPERTY, volume.keys()) - self.volume_delete(volume['id']) - self.check_volume_deleted(volume['id']) + self.object_delete('volume', volume['id']) + self.check_object_deleted('volume', volume['id']) def test_volume_extend(self): """Extend a volume size.""" - volume = self.volume_create(params='1 --name TestVolumeExtend') + volume = self.object_create('volume', + params='1 --name TestVolumeExtend') self.cinder('extend', params="%s %s" % (volume['id'], 2)) - self.wait_for_volume_status(volume['id'], 'available') + self.wait_for_object_status('volume', 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']) + self.object_delete('volume', volume['id']) + self.check_object_deleted('volume', volume['id']) + + +class CinderSnapshotTests(base.ClientTestBase): + """Check of base cinder snapshot commands.""" + + SNAPSHOT_PROPERTY = ('created_at', 'description', 'metadata', 'id', + 'name', 'size', 'status', 'volume_id') def test_snapshot_create_and_delete(self): """Create a volume snapshot and then delete.""" - volume = self.volume_create(params='1') - snapshot = self.snapshot_create(volume['id']) - self.assert_snapshot_details(snapshot.keys()) - self.snapshot_delete(snapshot['id']) - self.check_snapshot_deleted(snapshot['id']) - self.volume_delete(volume['id']) - self.check_volume_deleted(volume['id']) + volume = self.object_create('volume', params='1') + snapshot = self.object_create('snapshot', params=volume['id']) + self.assert_object_details(self.SNAPSHOT_PROPERTY, snapshot.keys()) + self.object_delete('snapshot', snapshot['id']) + self.check_object_deleted('snapshot', snapshot['id']) + self.object_delete('volume', volume['id']) + self.check_object_deleted('volume', volume['id'])