Merge "Support error handling for delete commands in volume v1"

Author: Jenkins

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 @@ def get_services(services=None, count=2):
         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):

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 @@ def test_qos_create_with_properties(self):
 
 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):