Merge "Support for snapshot force delete"

cinderclient/tests/unit/v2/fakes.py

@@ -356,6 +356,8 @@ class FakeHTTPClient(base_client.HTTPClient):
             assert 'status' in body['os-reset_status']
         elif action == 'os-update_snapshot_status':
             assert 'status' in body['os-update_snapshot_status']
+        elif action == 'os-force_delete':
+            assert body[action] is None
         elif action == 'os-unmanage':
             assert body[action] is None
         else:

cinderclient/tests/unit/v2/test_shell.py

@@ -946,17 +946,44 @@ class ShellTest(utils.TestCase):
         self.assert_called('POST', '/volumes/1234/action', body=expected)
 
     def test_snapshot_delete(self):
+        """Tests delete snapshot without force parameter"""
         self.run_command('snapshot-delete 1234')
         self.assert_called('DELETE', '/snapshots/1234')
 
+    def test_snapshot_delete_multiple(self):
+        """Tests delete multiple snapshots without force parameter"""
+        self.run_command('snapshot-delete 5678 1234')
+        self.assert_called_anytime('DELETE', '/snapshots/5678')
+        self.assert_called('DELETE', '/snapshots/1234')
+
+    def test_force_snapshot_delete(self):
+        """Tests delete snapshot with default force parameter value(True)"""
+        self.run_command('snapshot-delete 1234 --force')
+        expected_body = {'os-force_delete': None}
+        self.assert_called('POST',
+                           '/snapshots/1234/action',
+                           expected_body)
+
+    def test_force_snapshot_delete_multiple(self):
+        """
+        Tests delete multiple snapshots with force parameter
+
+        Snapshot delete with force parameter allows deleting snapshot of a
+        volume when its status is other than "available" or "error".
+        """
+        self.run_command('snapshot-delete 5678 1234 --force')
+        expected_body = {'os-force_delete': None}
+        self.assert_called_anytime('POST',
+                                   '/snapshots/5678/action',
+                                   expected_body)
+        self.assert_called_anytime('POST',
+                                   '/snapshots/1234/action',
+                                   expected_body)
+
     def test_quota_delete(self):
         self.run_command('quota-delete 1234')
         self.assert_called('DELETE', '/os-quota-sets/1234')
 
-    def test_snapshot_delete_multiple(self):
-        self.run_command('snapshot-delete 5678')
-        self.assert_called('DELETE', '/snapshots/5678')
-
     def test_volume_manage(self):
         self.run_command('manage host1 some_fake_name '
                          '--name foo --description bar '

cinderclient/v3/shell.py

@@ -784,13 +784,19 @@ def do_snapshot_create(cs, args):
 @utils.arg('snapshot',
            metavar='<snapshot>', nargs='+',
            help='Name or ID of the snapshot(s) to delete.')
+@utils.arg('--force',
+           action="store_true",
+           help='Allows deleting snapshot of a volume '
+           'when its status is other than "available" or "error". '
+           'Default=False.')
 @utils.service_type('volumev3')
 def do_snapshot_delete(cs, args):
     """Removes one or more snapshots."""
     failure_count = 0
+
     for snapshot in args.snapshot:
         try:
-            _find_volume_snapshot(cs, snapshot).delete()
+            _find_volume_snapshot(cs, snapshot).delete(args.force)
         except Exception as e:
             failure_count += 1
             print("Delete for snapshot %s failed: %s" % (snapshot, e))

cinderclient/v3/volume_snapshots.py

@@ -25,9 +25,9 @@ class Snapshot(base.Resource):
     def __repr__(self):
         return "<Snapshot: %s>" % self.id
 
-    def delete(self):
+    def delete(self, force=False):
         """Delete this snapshot."""
-        return self.manager.delete(self)
+        return self.manager.delete(self, force)
 
     def update(self, **kwargs):
         """Update the name or description for this snapshot."""
@@ -118,11 +118,15 @@ class SnapshotManager(base.ManagerWithFind):
                                    limit=limit, sort=sort)
         return self._list(url, resource_type, limit=limit)
 
-    def delete(self, snapshot):
+    def delete(self, snapshot, force=False):
         """Delete a snapshot.
 
         :param snapshot: The :class:`Snapshot` to delete.
+        :param force: Allow delete in state other than error or available.
         """
+        if force:
+            return self._action('os-force_delete', snapshot)
+        else:
             return self._delete("/snapshots/%s" % base.getid(snapshot))
 
     def update(self, snapshot, **kwargs):