diff options
Diffstat (limited to 'openstackclient/tests')
| -rw-r--r-- | openstackclient/tests/unit/volume/v1/fakes.py | 156 | ||||
| -rw-r--r-- | openstackclient/tests/unit/volume/v1/test_qos_specs.py | 77 | ||||
| -rw-r--r-- | openstackclient/tests/unit/volume/v1/test_volume.py | 103 |
3 files changed, 321 insertions, 15 deletions
diff --git a/openstackclient/tests/unit/volume/v1/fakes.py b/openstackclient/tests/unit/volume/v1/fakes.py index c6fee7d1..5c1f3b12 100644 --- a/openstackclient/tests/unit/volume/v1/fakes.py +++ b/openstackclient/tests/unit/volume/v1/fakes.py @@ -13,7 +13,10 @@ # under the License. # +import copy import mock +import random +import uuid from openstackclient.tests.unit import fakes from openstackclient.tests.unit.identity.v2_0 import fakes as identity_fakes @@ -234,6 +237,159 @@ class FakeService(object): return mock.MagicMock(side_effect=services) +class FakeQos(object): + """Fake one or more Qos specification.""" + + @staticmethod + def create_one_qos(attrs=None): + """Create a fake Qos specification. + + :param Dictionary attrs: + A dictionary with all attributes + :return: + A FakeResource object with id, name, consumer, etc. + """ + attrs = attrs or {} + + # Set default attributes. + qos_info = { + "id": 'qos-id-' + uuid.uuid4().hex, + "name": 'qos-name-' + uuid.uuid4().hex, + "consumer": 'front-end', + "specs": {"foo": "bar", "iops": "9001"}, + } + + # Overwrite default attributes. + qos_info.update(attrs) + + qos = fakes.FakeResource( + info=copy.deepcopy(qos_info), + loaded=True) + return qos + + @staticmethod + def create_qoses(attrs=None, count=2): + """Create multiple fake Qos specifications. + + :param Dictionary attrs: + A dictionary with all attributes + :param int count: + The number of Qos specifications to fake + :return: + A list of FakeResource objects faking the Qos specifications + """ + qoses = [] + for i in range(0, count): + qos = FakeQos.create_one_qos(attrs) + qoses.append(qos) + + return qoses + + @staticmethod + def get_qoses(qoses=None, count=2): + """Get an iterable MagicMock object with a list of faked qoses. + + If qoses list is provided, then initialize the Mock object with the + list. Otherwise create one. + + :param List volumes: + A list of FakeResource objects faking qoses + :param Integer count: + The number of qoses to be faked + :return + An iterable Mock object with side_effect set to a list of faked + qoses + """ + if qoses is None: + qoses = FakeQos.create_qoses(count) + + return mock.MagicMock(side_effect=qoses) + + +class FakeVolume(object): + """Fake one or more volumes.""" + + @staticmethod + def create_one_volume(attrs=None): + """Create a fake volume. + + :param Dictionary attrs: + A dictionary with all attributes of volume + :return: + A FakeResource object with id, name, status, etc. + """ + attrs = attrs or {} + + # Set default attribute + volume_info = { + 'id': 'volume-id' + uuid.uuid4().hex, + 'name': 'volume-name' + uuid.uuid4().hex, + 'description': 'description' + uuid.uuid4().hex, + 'status': random.choice(['available', 'in_use']), + 'size': random.randint(1, 20), + 'volume_type': + random.choice(['fake_lvmdriver-1', 'fake_lvmdriver-2']), + 'bootable': + random.randint(0, 1), + 'metadata': { + 'key' + uuid.uuid4().hex: 'val' + uuid.uuid4().hex, + 'key' + uuid.uuid4().hex: 'val' + uuid.uuid4().hex, + 'key' + uuid.uuid4().hex: 'val' + uuid.uuid4().hex}, + 'snapshot_id': random.randint(1, 5), + 'availability_zone': 'zone' + uuid.uuid4().hex, + 'attachments': [{ + 'device': '/dev/' + uuid.uuid4().hex, + 'server_id': uuid.uuid4().hex, + }, ], + } + + # Overwrite default attributes if there are some attributes set + volume_info.update(attrs) + + volume = fakes.FakeResource( + None, + volume_info, + loaded=True) + return volume + + @staticmethod + def create_volumes(attrs=None, count=2): + """Create multiple fake volumes. + + :param Dictionary attrs: + A dictionary with all attributes of volume + :param Integer count: + The number of volumes to be faked + :return: + A list of FakeResource objects + """ + volumes = [] + for n in range(0, count): + volumes.append(FakeVolume.create_one_volume(attrs)) + + return volumes + + @staticmethod + def get_volumes(volumes=None, count=2): + """Get an iterable MagicMock object with a list of faked volumes. + + If volumes list is provided, then initialize the Mock object with the + list. Otherwise create one. + + :param List volumes: + A list of FakeResource objects faking volumes + :param Integer count: + The number of volumes to be faked + :return + An iterable Mock object with side_effect set to a list of faked + volumes + """ + if volumes is None: + volumes = FakeVolume.create_volumes(count) + + return mock.MagicMock(side_effect=volumes) + + class FakeImagev1Client(object): def __init__(self, **kwargs): diff --git a/openstackclient/tests/unit/volume/v1/test_qos_specs.py b/openstackclient/tests/unit/volume/v1/test_qos_specs.py index 7b87ccb3..81680ab4 100644 --- a/openstackclient/tests/unit/volume/v1/test_qos_specs.py +++ b/openstackclient/tests/unit/volume/v1/test_qos_specs.py @@ -14,7 +14,10 @@ # import copy +import mock +from mock import call +from osc_lib import exceptions from osc_lib import utils from openstackclient.tests.unit import fakes @@ -188,62 +191,106 @@ class TestQosCreate(TestQos): class TestQosDelete(TestQos): + qos_specs = volume_fakes.FakeQos.create_qoses(count=2) + def setUp(self): super(TestQosDelete, self).setUp() - self.qos_mock.get.return_value = fakes.FakeResource( - None, - copy.deepcopy(volume_fakes.QOS), - loaded=True, - ) - + self.qos_mock.get = ( + volume_fakes.FakeQos.get_qoses(self.qos_specs)) # Get the command object to test self.cmd = qos_specs.DeleteQos(self.app, None) def test_qos_delete_with_id(self): arglist = [ - volume_fakes.qos_id + self.qos_specs[0].id ] verifylist = [ - ('qos_specs', [volume_fakes.qos_id]) + ('qos_specs', [self.qos_specs[0].id]) ] parsed_args = self.check_parser(self.cmd, arglist, verifylist) result = self.cmd.take_action(parsed_args) - self.qos_mock.delete.assert_called_with(volume_fakes.qos_id, False) + self.qos_mock.delete.assert_called_with(self.qos_specs[0].id, False) self.assertIsNone(result) def test_qos_delete_with_name(self): arglist = [ - volume_fakes.qos_name + self.qos_specs[0].name ] verifylist = [ - ('qos_specs', [volume_fakes.qos_name]) + ('qos_specs', [self.qos_specs[0].name]) ] parsed_args = self.check_parser(self.cmd, arglist, verifylist) result = self.cmd.take_action(parsed_args) - self.qos_mock.delete.assert_called_with(volume_fakes.qos_id, False) + self.qos_mock.delete.assert_called_with(self.qos_specs[0].id, False) self.assertIsNone(result) def test_qos_delete_with_force(self): arglist = [ '--force', - volume_fakes.qos_id + self.qos_specs[0].id ] verifylist = [ ('force', True), - ('qos_specs', [volume_fakes.qos_id]) + ('qos_specs', [self.qos_specs[0].id]) ] parsed_args = self.check_parser(self.cmd, arglist, verifylist) result = self.cmd.take_action(parsed_args) - self.qos_mock.delete.assert_called_with(volume_fakes.qos_id, True) + self.qos_mock.delete.assert_called_with(self.qos_specs[0].id, True) self.assertIsNone(result) + def test_delete_multiple_qoses(self): + arglist = [] + for q in self.qos_specs: + arglist.append(q.id) + verifylist = [ + ('qos_specs', arglist), + ] + + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + result = self.cmd.take_action(parsed_args) + + calls = [] + for q in self.qos_specs: + calls.append(call(q.id, False)) + self.qos_mock.delete.assert_has_calls(calls) + self.assertIsNone(result) + + def test_delete_multiple_qoses_with_exception(self): + arglist = [ + self.qos_specs[0].id, + 'unexist_qos', + ] + verifylist = [ + ('qos_specs', arglist), + ] + + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + find_mock_result = [self.qos_specs[0], exceptions.CommandError] + with mock.patch.object(utils, 'find_resource', + side_effect=find_mock_result) as find_mock: + try: + self.cmd.take_action(parsed_args) + self.fail('CommandError should be raised.') + except exceptions.CommandError as e: + self.assertEqual( + '1 of 2 QoS specifications failed to delete.', str(e)) + + find_mock.assert_any_call(self.qos_mock, self.qos_specs[0].id) + find_mock.assert_any_call(self.qos_mock, 'unexist_qos') + + self.assertEqual(2, find_mock.call_count) + self.qos_mock.delete.assert_called_once_with( + self.qos_specs[0].id, False + ) + class TestQosDisassociate(TestQos): diff --git a/openstackclient/tests/unit/volume/v1/test_volume.py b/openstackclient/tests/unit/volume/v1/test_volume.py index f90566fd..6a860fd0 100644 --- a/openstackclient/tests/unit/volume/v1/test_volume.py +++ b/openstackclient/tests/unit/volume/v1/test_volume.py @@ -15,6 +15,10 @@ import copy import mock +from mock import call + +from osc_lib import exceptions +from osc_lib import utils from openstackclient.tests.unit import fakes from openstackclient.tests.unit.identity.v2_0 import fakes as identity_fakes @@ -43,6 +47,14 @@ class TestVolume(volume_fakes.TestVolumev1): self.images_mock = self.app.client_manager.image.images self.images_mock.reset_mock() + def setup_volumes_mock(self, count): + volumes = volume_fakes.FakeVolume.create_volumes(count=count) + + self.volumes_mock.get = volume_fakes.FakeVolume.get_volumes( + volumes, + 0) + return volumes + # TODO(dtroyer): The volume create tests are incomplete, only the minimal # options and the options that require additional processing @@ -397,6 +409,97 @@ class TestVolumeCreate(TestVolume): self.assertEqual(self.datalist, data) +class TestVolumeDelete(TestVolume): + + def setUp(self): + super(TestVolumeDelete, self).setUp() + + self.volumes_mock.delete.return_value = None + + # Get the command object to mock + self.cmd = volume.DeleteVolume(self.app, None) + + def test_volume_delete_one_volume(self): + volumes = self.setup_volumes_mock(count=1) + + arglist = [ + volumes[0].id + ] + verifylist = [ + ("force", False), + ("volumes", [volumes[0].id]), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + result = self.cmd.take_action(parsed_args) + + self.volumes_mock.delete.assert_called_once_with(volumes[0].id) + self.assertIsNone(result) + + def test_volume_delete_multi_volumes(self): + volumes = self.setup_volumes_mock(count=3) + + arglist = [v.id for v in volumes] + verifylist = [ + ('force', False), + ('volumes', arglist), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + result = self.cmd.take_action(parsed_args) + + calls = [call(v.id) for v in volumes] + self.volumes_mock.delete.assert_has_calls(calls) + self.assertIsNone(result) + + def test_volume_delete_multi_volumes_with_exception(self): + volumes = self.setup_volumes_mock(count=2) + + arglist = [ + volumes[0].id, + 'unexist_volume', + ] + verifylist = [ + ('force', False), + ('volumes', arglist), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + find_mock_result = [volumes[0], exceptions.CommandError] + with mock.patch.object(utils, 'find_resource', + side_effect=find_mock_result) as find_mock: + try: + self.cmd.take_action(parsed_args) + self.fail('CommandError should be raised.') + except exceptions.CommandError as e: + self.assertEqual('1 of 2 volumes failed to delete.', + str(e)) + + find_mock.assert_any_call(self.volumes_mock, volumes[0].id) + find_mock.assert_any_call(self.volumes_mock, 'unexist_volume') + + self.assertEqual(2, find_mock.call_count) + self.volumes_mock.delete.assert_called_once_with(volumes[0].id) + + def test_volume_delete_with_force(self): + volumes = self.setup_volumes_mock(count=1) + + arglist = [ + '--force', + volumes[0].id, + ] + verifylist = [ + ('force', True), + ('volumes', [volumes[0].id]), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + result = self.cmd.take_action(parsed_args) + + self.volumes_mock.force_delete.assert_called_once_with(volumes[0].id) + self.assertIsNone(result) + + class TestVolumeList(TestVolume): columns = ( |