Merge "Add DeleteVolumeSnapshot class to v3"

openstackclient/tests/unit/volume/v3/test_volume_snapshot.py

@@ -0,0 +1,113 @@
+#
+#   Licensed under the Apache License, Version 2.0 (the "License"); you may
+#   not use this file except in compliance with the License. You may obtain
+#   a copy of the License at
+#
+#        http://www.apache.org/licenses/LICENSE-2.0
+#
+#   Unless required by applicable law or agreed to in writing, software
+#   distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#   WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#   License for the specific language governing permissions and limitations
+#   under the License.
+
+from unittest import mock
+
+from osc_lib import exceptions
+from osc_lib import utils
+
+from openstackclient.tests.unit.volume.v2 import fakes as volume_fakes
+from openstackclient.volume.v3 import volume_snapshot
+
+
+class TestVolumeSnapshot(volume_fakes.TestVolume):
+    def setUp(self):
+        super().setUp()
+
+        self.snapshots_mock = self.volume_client.volume_snapshots
+        self.snapshots_mock.reset_mock()
+
+
+class TestVolumeSnapshotDelete(TestVolumeSnapshot):
+    snapshots = volume_fakes.create_snapshots(count=2)
+
+    def setUp(self):
+        super().setUp()
+
+        self.snapshots_mock.get = volume_fakes.get_snapshots(self.snapshots)
+        self.snapshots_mock.delete.return_value = None
+
+        # Get the command object to mock
+        self.cmd = volume_snapshot.DeleteVolumeSnapshot(self.app, None)
+
+    def test_snapshot_delete(self):
+        arglist = [self.snapshots[0].id]
+        verifylist = [("snapshots", [self.snapshots[0].id])]
+        parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+        result = self.cmd.take_action(parsed_args)
+
+        self.snapshots_mock.delete.assert_called_with(
+            self.snapshots[0].id, False
+        )
+        self.assertIsNone(result)
+
+    def test_snapshot_delete_with_force(self):
+        arglist = ['--force', self.snapshots[0].id]
+        verifylist = [('force', True), ("snapshots", [self.snapshots[0].id])]
+        parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+        result = self.cmd.take_action(parsed_args)
+
+        self.snapshots_mock.delete.assert_called_with(
+            self.snapshots[0].id, True
+        )
+        self.assertIsNone(result)
+
+    def test_delete_multiple_snapshots(self):
+        arglist = []
+        for s in self.snapshots:
+            arglist.append(s.id)
+        verifylist = [
+            ('snapshots', arglist),
+        ]
+
+        parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+        result = self.cmd.take_action(parsed_args)
+
+        calls = []
+        for s in self.snapshots:
+            calls.append(mock.call(s.id, False))
+        self.snapshots_mock.delete.assert_has_calls(calls)
+        self.assertIsNone(result)
+
+    def test_delete_multiple_snapshots_with_exception(self):
+        arglist = [
+            self.snapshots[0].id,
+            'unexist_snapshot',
+        ]
+        verifylist = [
+            ('snapshots', arglist),
+        ]
+
+        parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+        find_mock_result = [self.snapshots[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 snapshots failed to delete.', str(e))
+
+            find_mock.assert_any_call(
+                self.snapshots_mock, self.snapshots[0].id
+            )
+            find_mock.assert_any_call(self.snapshots_mock, 'unexist_snapshot')
+
+            self.assertEqual(2, find_mock.call_count)
+            self.snapshots_mock.delete.assert_called_once_with(
+                self.snapshots[0].id, False
+            )

openstackclient/volume/v3/volume_snapshot.py

@@ -0,0 +1,76 @@
+#
+#   Licensed under the Apache License, Version 2.0 (the "License"); you may
+#   not use this file except in compliance with the License. You may obtain
+#   a copy of the License at
+#
+#        http://www.apache.org/licenses/LICENSE-2.0
+#
+#   Unless required by applicable law or agreed to in writing, software
+#   distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#   WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#   License for the specific language governing permissions and limitations
+#   under the License.
+#
+
+"""Volume v3 snapshot action implementations"""
+
+import logging
+
+from osc_lib.command import command
+from osc_lib import exceptions
+from osc_lib import utils
+
+from openstackclient.i18n import _
+
+LOG = logging.getLogger(__name__)
+
+
+class DeleteVolumeSnapshot(command.Command):
+    _description = _("Delete volume snapshot(s)")
+
+    def get_parser(self, prog_name):
+        parser = super().get_parser(prog_name)
+        parser.add_argument(
+            "snapshots",
+            metavar="<snapshot>",
+            nargs="+",
+            help=_("Snapshot(s) to delete (name or ID)"),
+        )
+        parser.add_argument(
+            '--force',
+            action='store_true',
+            help=_(
+                "Attempt forced removal of snapshot(s), "
+                "regardless of state (defaults to False)"
+            ),
+        )
+        return parser
+
+    def take_action(self, parsed_args):
+        volume_client = self.app.client_manager.volume
+        result = 0
+
+        for i in parsed_args.snapshots:
+            try:
+                snapshot_id = utils.find_resource(
+                    volume_client.volume_snapshots, i
+                ).id
+                volume_client.volume_snapshots.delete(
+                    snapshot_id, parsed_args.force
+                )
+            except Exception as e:
+                result += 1
+                LOG.error(
+                    _(
+                        "Failed to delete snapshot with "
+                        "name or ID '%(snapshot)s': %(e)s"
+                    )
+                    % {'snapshot': i, 'e': e}
+                )
+
+        if result > 0:
+            total = len(parsed_args.snapshots)
+            msg = _(
+                "%(result)s of %(total)s snapshots failed " "to delete."
+            ) % {'result': result, 'total': total}
+            raise exceptions.CommandError(msg)

setup.cfg

@@ -825,7 +825,7 @@ openstack.volume.v3 =
     block_storage_resource_filter_show = openstackclient.volume.v3.block_storage_resource_filter:ShowBlockStorageResourceFilter
 
     volume_snapshot_create = openstackclient.volume.v2.volume_snapshot:CreateVolumeSnapshot
-    volume_snapshot_delete = openstackclient.volume.v2.volume_snapshot:DeleteVolumeSnapshot
+    volume_snapshot_delete = openstackclient.volume.v3.volume_snapshot:DeleteVolumeSnapshot
     volume_snapshot_list = openstackclient.volume.v2.volume_snapshot:ListVolumeSnapshot
     volume_snapshot_set = openstackclient.volume.v2.volume_snapshot:SetVolumeSnapshot
     volume_snapshot_show = openstackclient.volume.v2.volume_snapshot:ShowVolumeSnapshot