Add function to get all attachments in Cinder.API module

- Added function to get all attachments by instance or volume id from Cinder in Cinder.API
- Updated CinderFixture to add mock get_all_attachment functionality.
- Added unit tests get_all_attachments.

Related-Bug: 2019078

Change-Id: I8619d898f68250bf70a17b1e6b8b0c249245b43b

nova/tests/fixtures/cinder.py

@@ -162,6 +162,9 @@ class CinderFixture(fixtures.Fixture):
         self.useFixture(fixtures.MockPatch(
             'nova.volume.cinder.API.get_absolute_limits',
             side_effect=self.fake_get_absolute_limits, autospec=False))
+        self.useFixture(fixtures.MockPatch(
+            'nova.volume.cinder.API.attachment_get_all',
+            side_effect=self.fake_attachment_get_all, autospec=False))
 
     def _is_multiattach(self, volume_id):
         return volume_id in [
@@ -409,6 +412,23 @@ class CinderFixture(fixtures.Fixture):
         limits = {'totalSnapshotsUsed': 0, 'maxTotalSnapshots': -1}
         return limits
 
+    def fake_attachment_get_all(
+            self, context, instance_id=None, volume_id=None):
+        if not instance_id and not volume_id:
+            raise exception.InvalidRequest(
+                "Either instance or volume id must be passed.")
+
+        if volume_id in self.volume_to_attachment:
+            return self.volume_to_attachment[volume_id]
+
+        all_attachments = []
+        for _, attachments in self.volume_to_attachment.items():
+            all_attachments.extend(
+                [attach for attach in attachments.values()
+                 if instance_id == attach['instance_uuid']])
+
+        return all_attachments
+
     def volume_ids_for_instance(self, instance_uuid):
         for volume_id, attachments in self.volume_to_attachment.items():
             for attachment in attachments.values():
@@ -425,3 +445,24 @@ class CinderFixture(fixtures.Fixture):
                 if attachment['instance_uuid'] == instance_uuid:
                     attachment_ids.append(attachment['id'])
         return attachment_ids
+
+    def create_vol_attachment(self, volume_id, instance_id):
+        attachment_id = uuidutils.generate_uuid()
+        if self.attachment_error_id is not None:
+            attachment_id = self.attachment_error_id
+        attachment = {'id': attachment_id}
+        self.volume_to_attachment[volume_id][attachment_id] = {
+            'id': attachment_id,
+            'instance_uuid': instance_id,
+        }
+        return attachment
+
+    def get_vol_attachment(self, _id):
+        for _, attachments in self.volume_to_attachment.items():
+            for attachment_id in attachments:
+                if _id == attachment_id:
+                    # return because attachment id is unique
+                    return attachments[attachment_id]
+
+    def delete_vol_attachment(self, vol_id):
+        del self.volume_to_attachment[vol_id]

nova/tests/unit/volume/test_cinder.py

@@ -749,6 +749,44 @@ class CinderApiTestCase(test.NoDBTestCase):
                                                   skip_version_check=True)
         mock_attachment.show.assert_called_once_with(attachment_id)
 
+    @mock.patch('nova.volume.cinder.cinderclient')
+    def test_attachment_get_all_by_instance(self, mock_cinderclient):
+        mock_attachment = mock.MagicMock()
+        mock_cinderclient.return_value = \
+            mock.MagicMock(attachments=mock_attachment)
+
+        instance_id = uuids.instance_id
+        search_opts = {'instance_id': instance_id}
+        self.api.attachment_get_all(self.ctx, instance_id)
+        mock_cinderclient.assert_called_once_with(self.ctx, '3.44',
+                                                  skip_version_check=True)
+        mock_attachment.list.assert_called_once_with(search_opts=search_opts)
+
+    @mock.patch('nova.volume.cinder.cinderclient')
+    def test_attachment_get_all_by_volume(self, mock_cinderclient):
+        mock_attachment = mock.MagicMock()
+        mock_cinderclient.return_value = \
+            mock.MagicMock(attachments=mock_attachment)
+
+        volume_id = uuids.volume_id
+        search_opts = {'volume_id': volume_id}
+        self.api.attachment_get_all(self.ctx, volume_id=volume_id)
+        mock_cinderclient.assert_called_once_with(self.ctx, '3.44',
+                                                  skip_version_check=True)
+        mock_attachment.list.assert_called_once_with(search_opts=search_opts)
+
+    @mock.patch('nova.volume.cinder.cinderclient')
+    def test_attachment_get_all_failed(self, mock_cinderclient):
+        err = "Either instance or volume id must be passed."
+        mock_cinderclient.return_value.attachments.show.side_effect = (
+            exception.InvalidRequest(err))
+
+        ex = self.assertRaises(exception.InvalidRequest,
+                               self.api.attachment_get_all,
+                               self.ctx)
+
+        self.assertIn(err, str(ex))
+
     @mock.patch('nova.volume.cinder.cinderclient')
     def test_attachment_get_failed(self, mock_cinderclient):
         mock_cinderclient.return_value.attachments.show.side_effect = (

nova/volume/cinder.py

@@ -839,6 +839,39 @@ class API(object):
                            'msg': str(ex),
                            'code': getattr(ex, 'code', None)})
 
+    def attachment_get_all(self, context, instance_id=None, volume_id=None):
+        """Get all attchments by instance id or volume id
+
+        :param context: The nova request context.
+        :param instance_id: UUID of the instance attachment to get.
+        :param volume_id: UUID of the volume attachment to get.
+        :returns: a list of cinderclient.v3.attachments.VolumeAttachment
+            objects.
+        """
+        if not instance_id and not volume_id:
+            raise exception.InvalidRequest(
+                "Either instance or volume id must be passed.")
+
+        search_opts = {}
+
+        if instance_id:
+            search_opts['instance_id'] = instance_id
+        if volume_id:
+            search_opts['volume_id'] = volume_id
+
+        try:
+            attachments = cinderclient(
+                context, '3.44', skip_version_check=True).attachments.list(
+                    search_opts=search_opts)
+        except cinder_exception.ClientException as ex:
+            with excutils.save_and_reraise_exception():
+                LOG.error('Get all attachment failed. '
+                          'Error: %(msg)s Code: %(code)s',
+                          {'msg': str(ex),
+                           'code': getattr(ex, 'code', None)})
+        return [_translate_attachment_ref(
+            each.to_dict()) for each in attachments]
+
     @translate_attachment_exception
     def attachment_update(self, context, attachment_id, connector,
                           mountpoint=None):