Merge "3PAR: Adding volume checks to manage snapshot API"

cinder/tests/unit/test_hpe3par.py

@@ -3460,14 +3460,17 @@ class HPE3PARBaseDriver(object):
 
         new_comment = Comment({
             "display_name": "snap",
-            "volume_name": "volume-007dbfce-7579-40bc-8f90-a20b3902283e",
+            "volume_name": self.VOLUME_NAME,
-            "volume_id": "007dbfce-7579-40bc-8f90-a20b3902283e",
+            "volume_id": self.VOLUME_ID,
             "description": "",
         })
+
+        volume = {'id': self.VOLUME_ID}
+
         snapshot = {
             'display_name': None,
-            'id': '007dbfce-7579-40bc-8f90-a20b3902283e',
+            'id': self.SNAPSHOT_ID,
-            'volume_id': self.VOLUME_ID,
+            'volume': volume,
         }
 
         mock_client.getVolume.return_value = {
@@ -3493,7 +3496,6 @@ class HPE3PARBaseDriver(object):
                                        {'newName': oss_matcher,
                                         'comment': new_comment}),
             ]
-
             mock_client.assert_has_calls(
                 self.standard_login +
                 expected +
@@ -3503,10 +3505,12 @@ class HPE3PARBaseDriver(object):
     def test_manage_existing_snapshot_invalid_parent(self):
         mock_client = self.setup_driver()
 
+        volume = {'id': self.VOLUME_ID}
+
         snapshot = {
             'display_name': None,
             'id': '007dbfce-7579-40bc-8f90-a20b3902283e',
-            'volume_id': self.VOLUME_ID,
+            'volume': volume,
         }
 
         mock_client.getVolume.return_value = {
@@ -3536,6 +3540,38 @@ class HPE3PARBaseDriver(object):
                 expected +
                 self.standard_logout)
 
+    def test_manage_existing_snapshot_failed_over_volume(self):
+        mock_client = self.setup_driver()
+
+        volume = {
+            'id': self.VOLUME_ID,
+            'replication_status': 'failed-over',
+        }
+
+        snapshot = {
+            'display_name': None,
+            'id': '007dbfce-7579-40bc-8f90-a20b3902283e',
+            'volume': volume,
+        }
+
+        mock_client.getVolume.return_value = {
+            "comment": "{'display_name': 'snap'}",
+            'copyOf': self.VOLUME_NAME_3PAR,
+        }
+
+        with mock.patch.object(hpecommon.HPE3PARCommon,
+                               '_create_client') as mock_create_client:
+            mock_create_client.return_value = mock_client
+            common = self.driver._login()
+
+            ums_matcher = common._get_3par_ums_name(snapshot['id'])
+            existing_ref = {'source-name': ums_matcher}
+
+            self.assertRaises(exception.InvalidInput,
+                              self.driver.manage_existing_snapshot,
+                              snapshot=snapshot,
+                              existing_ref=existing_ref)
+
     def test_manage_existing_get_size(self):
         mock_client = self.setup_driver()
         mock_client.getVolume.return_value = {'sizeMiB': 2048}
@@ -3731,6 +3767,22 @@ class HPE3PARBaseDriver(object):
                 expected +
                 self.standard_logout)
 
+    def test_unmanage_snapshot_failed_over_volume(self):
+        mock_client = self.setup_driver()
+
+        volume = {'replication_status': 'failed-over', }
+        snapshot = {'id': self.SNAPSHOT_ID,
+                    'display_name': 'fake_snap',
+                    'volume': volume, }
+
+        with mock.patch.object(hpecommon.HPE3PARCommon,
+                               '_create_client') as mock_create_client:
+            mock_create_client.return_value = mock_client
+
+            self.assertRaises(exception.SnapshotIsBusy,
+                              self.driver.unmanage_snapshot,
+                              snapshot=snapshot)
+
     def test__safe_hostname(self):
         long_hostname = "abc123abc123abc123abc123abc123abc123"
         fixed_hostname = "abc123abc123abc123abc123abc123a"

cinder/volume/drivers/hpe/hpe_3par_common.py

@@ -223,10 +223,11 @@ class HPE3PARCommon(object):
         3.0.7 - Enable standard capabilities based on 3PAR licenses
         3.0.8 - Optimize array ID retrieval
         3.0.9 - Bump minimum API version for volume replication
+        3.0.10 - Added additional volumes checks to the manage snapshot API
 
     """
 
-    VERSION = "3.0.9"
+    VERSION = "3.0.10"
 
     stats = {}
 
@@ -793,6 +794,15 @@ class HPE3PARCommon(object):
         existing_ref is a dictionary of the form:
         {'source-name': <name of the snapshot>}
         """
+        # Potential parent volume for the snapshot
+        volume = snapshot['volume']
+
+        # Do not allow for managing of snapshots for 'failed-over' volumes.
+        if volume.get('replication_status') == 'failed-over':
+            err = (_("Managing of snapshots to failed-over volumes is "
+                     "not allowed."))
+            raise exception.InvalidInput(reason=err)
+
         target_snap_name = self._get_existing_volume_ref_name(existing_ref,
                                                               is_snapshot=True)
 
@@ -806,7 +816,7 @@ class HPE3PARCommon(object):
             raise exception.InvalidInput(reason=err)
 
         # Make sure the snapshot is being associated with the correct volume.
-        parent_vol_name = self._get_3par_vol_name(snapshot['volume_id'])
+        parent_vol_name = self._get_3par_vol_name(volume['id'])
         if parent_vol_name != snap['copyOf']:
             err = (_("The provided snapshot '%s' is not a snapshot of "
                      "the provided volume.") % target_snap_name)
@@ -830,8 +840,8 @@ class HPE3PARCommon(object):
 
         # Generate the new snapshot information based on the new ID.
         new_snap_name = self._get_3par_snap_name(snapshot['id'])
-        new_comment['volume_id'] = snapshot['id']
+        new_comment['volume_id'] = volume['id']
-        new_comment['volume_name'] = 'volume-' + snapshot['id']
+        new_comment['volume_name'] = 'volume-' + volume['id']
         if snapshot.get('display_description', None):
             new_comment['description'] = snapshot['display_description']
         else:
@@ -923,6 +933,18 @@ class HPE3PARCommon(object):
 
     def unmanage_snapshot(self, snapshot):
         """Removes the specified snapshot from Cinder management."""
+        # Parent volume for the snapshot
+        volume = snapshot['volume']
+
+        # Do not allow unmanaging of snapshots from 'failed-over' volumes.
+        if volume.get('replication_status') == 'failed-over':
+            err = (_("Unmanaging of snapshots from failed-over volumes is "
+                     "not allowed."))
+            LOG.error(err)
+            # TODO(leeantho) Change this exception to Invalid when the volume
+            # manager supports handling that.
+            raise exception.SnapshotIsBusy(snapshot_name=snapshot['id'])
+
         # Rename the snapshots's name to ums-* format so that it can be
         # easily found later.
         snap_name = self._get_3par_snap_name(snapshot['id'])