Add commands for "consistency group snapshot"

Add commands:
    consistency group snapshot create
    consistency group snapshot delete
    consistency group snapshot list
    consistency group snapshot show
in volume v2 (v2 only)

Change-Id: Ib4115f8ff00fb5aa8194588223032657eb1346b5
Closes-Bug: #1642238
Implements: bp cinder-command-support

3 files changed, 619 insertions, 0 deletions

diff --git a/openstackclient/tests/unit/volume/v2/fakes.py b/openstackclient/tests/unit/volume/v2/fakes.py
index 5e1d16e1..70234f43 100644
--- a/openstackclient/tests/unit/volume/v2/fakes.py
+++ b/openstackclient/tests/unit/volume/v2/fakes.py
@@ -224,6 +224,8 @@ class FakeVolumeClient(object):
         self.quota_classes.resource_class = fakes.FakeResource(None, {})
         self.consistencygroups = mock.Mock()
         self.consistencygroups.resource_class = fakes.FakeResource(None, {})
+        self.cgsnapshots = mock.Mock()
+        self.cgsnapshots.resource_class = fakes.FakeResource(None, {})
         self.auth_token = kwargs['token']
         self.management_url = kwargs['endpoint']
 
@@ -548,6 +550,82 @@ class FakeConsistencyGroup(object):
         return consistency_groups
 
 
+class FakeConsistencyGroupSnapshot(object):
+    """Fake one or more consistency group snapshot."""
+
+    @staticmethod
+    def create_one_consistency_group_snapshot(attrs=None):
+        """Create a fake consistency group snapshot.
+
+        :param Dictionary attrs:
+            A dictionary with all attributes
+        :return:
+            A FakeResource object with id, name, description, etc.
+        """
+        attrs = attrs or {}
+
+        # Set default attributes.
+        consistency_group_snapshot_info = {
+            "id": 'id-' + uuid.uuid4().hex,
+            "name": 'backup-name-' + uuid.uuid4().hex,
+            "description": 'description-' + uuid.uuid4().hex,
+            "status": "error",
+            "consistencygroup_id": 'consistency-group-id' + uuid.uuid4().hex,
+            "created_at": 'time-' + uuid.uuid4().hex,
+        }
+
+        # Overwrite default attributes.
+        consistency_group_snapshot_info.update(attrs)
+
+        consistency_group_snapshot = fakes.FakeResource(
+            info=copy.deepcopy(consistency_group_snapshot_info),
+            loaded=True)
+        return consistency_group_snapshot
+
+    @staticmethod
+    def create_consistency_group_snapshots(attrs=None, count=2):
+        """Create multiple fake consistency group snapshots.
+
+        :param Dictionary attrs:
+            A dictionary with all attributes
+        :param int count:
+            The number of consistency group snapshots to fake
+        :return:
+            A list of FakeResource objects faking the
+            consistency group snapshots
+        """
+        consistency_group_snapshots = []
+        for i in range(0, count):
+            consistency_group_snapshot = (
+                FakeConsistencyGroupSnapshot.
+                create_one_consistency_group_snapshot(attrs)
+            )
+            consistency_group_snapshots.append(consistency_group_snapshot)
+
+        return consistency_group_snapshots
+
+    @staticmethod
+    def get_consistency_group_snapshots(snapshots=None, count=2):
+        """Get an iterable MagicMock object with a list of faked cgsnapshots.
+
+        If consistenct group snapshots list is provided, then initialize
+        the Mock object with the list. Otherwise create one.
+
+        :param List snapshots:
+            A list of FakeResource objects faking consistency group snapshots
+        :param Integer count:
+            The number of consistency group snapshots to be faked
+        :return
+            An iterable Mock object with side_effect set to a list of faked
+            consistency groups
+        """
+        if snapshots is None:
+            snapshots = (FakeConsistencyGroupSnapshot.
+                         create_consistency_group_snapshots(count))
+
+        return mock.Mock(side_effect=snapshots)
+
+
 class FakeExtension(object):
     """Fake one or more extension."""
 
diff --git a/openstackclient/tests/unit/volume/v2/test_consistency_group_snapshot.py b/openstackclient/tests/unit/volume/v2/test_consistency_group_snapshot.py
new file mode 100644
index 00000000..3bfe93df
--- /dev/null
+++ b/openstackclient/tests/unit/volume/v2/test_consistency_group_snapshot.py
@@ -0,0 +1,351 @@
+#
+#   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 mock import call
+
+from openstackclient.tests.unit.volume.v2 import fakes as volume_fakes
+from openstackclient.volume.v2 import consistency_group_snapshot
+
+
+class TestConsistencyGroupSnapshot(volume_fakes.TestVolume):
+
+    def setUp(self):
+        super(TestConsistencyGroupSnapshot, self).setUp()
+
+        # Get a shortcut to the TransferManager Mock
+        self.cgsnapshots_mock = (
+            self.app.client_manager.volume.cgsnapshots)
+        self.cgsnapshots_mock.reset_mock()
+        self.consistencygroups_mock = (
+            self.app.client_manager.volume.consistencygroups)
+        self.consistencygroups_mock.reset_mock()
+
+
+class TestConsistencyGroupSnapshotCreate(TestConsistencyGroupSnapshot):
+
+    _consistency_group_snapshot = (
+        volume_fakes.
+        FakeConsistencyGroupSnapshot.
+        create_one_consistency_group_snapshot()
+    )
+    consistency_group = (
+        volume_fakes.FakeConsistencyGroup.create_one_consistency_group())
+
+    columns = (
+        'consistencygroup_id',
+        'created_at',
+        'description',
+        'id',
+        'name',
+        'status',
+    )
+    data = (
+        _consistency_group_snapshot.consistencygroup_id,
+        _consistency_group_snapshot.created_at,
+        _consistency_group_snapshot.description,
+        _consistency_group_snapshot.id,
+        _consistency_group_snapshot.name,
+        _consistency_group_snapshot.status,
+    )
+
+    def setUp(self):
+        super(TestConsistencyGroupSnapshotCreate, self).setUp()
+        self.cgsnapshots_mock.create.return_value = (
+            self._consistency_group_snapshot)
+        self.consistencygroups_mock.get.return_value = (
+            self.consistency_group)
+
+        # Get the command object to test
+        self.cmd = (consistency_group_snapshot.
+                    CreateConsistencyGroupSnapshot(self.app, None))
+
+    def test_consistency_group_snapshot_create(self):
+        arglist = [
+            '--consistency-group', self.consistency_group.id,
+            '--description', self._consistency_group_snapshot.description,
+            self._consistency_group_snapshot.name,
+        ]
+        verifylist = [
+            ('consistency_group', self.consistency_group.id),
+            ('description', self._consistency_group_snapshot.description),
+            ('snapshot_name', self._consistency_group_snapshot.name),
+        ]
+        parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+        columns, data = self.cmd.take_action(parsed_args)
+
+        self.consistencygroups_mock.get.assert_called_once_with(
+            self.consistency_group.id)
+        self.cgsnapshots_mock.create.assert_called_once_with(
+            self.consistency_group.id,
+            name=self._consistency_group_snapshot.name,
+            description=self._consistency_group_snapshot.description,
+        )
+
+        self.assertEqual(self.columns, columns)
+        self.assertEqual(self.data, data)
+
+    def test_consistency_group_snapshot_create_no_consistency_group(self):
+        arglist = [
+            '--description', self._consistency_group_snapshot.description,
+            self._consistency_group_snapshot.name,
+        ]
+        verifylist = [
+            ('description', self._consistency_group_snapshot.description),
+            ('snapshot_name', self._consistency_group_snapshot.name),
+        ]
+        parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+        columns, data = self.cmd.take_action(parsed_args)
+
+        self.consistencygroups_mock.get.assert_called_once_with(
+            self._consistency_group_snapshot.name)
+        self.cgsnapshots_mock.create.assert_called_once_with(
+            self.consistency_group.id,
+            name=self._consistency_group_snapshot.name,
+            description=self._consistency_group_snapshot.description,
+        )
+
+        self.assertEqual(self.columns, columns)
+        self.assertEqual(self.data, data)
+
+
+class TestConsistencyGroupSnapshotDelete(TestConsistencyGroupSnapshot):
+
+    consistency_group_snapshots = (
+        volume_fakes.FakeConsistencyGroupSnapshot.
+        create_consistency_group_snapshots(count=2)
+    )
+
+    def setUp(self):
+        super(TestConsistencyGroupSnapshotDelete, self).setUp()
+
+        self.cgsnapshots_mock.get = (
+            volume_fakes.FakeConsistencyGroupSnapshot.
+            get_consistency_group_snapshots(self.consistency_group_snapshots)
+        )
+        self.cgsnapshots_mock.delete.return_value = None
+
+        # Get the command object to mock
+        self.cmd = (consistency_group_snapshot.
+                    DeleteConsistencyGroupSnapshot(self.app, None))
+
+    def test_consistency_group_snapshot_delete(self):
+        arglist = [
+            self.consistency_group_snapshots[0].id
+        ]
+        verifylist = [
+            ("consistency_group_snapshot",
+             [self.consistency_group_snapshots[0].id])
+        ]
+        parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+        result = self.cmd.take_action(parsed_args)
+
+        self.cgsnapshots_mock.delete.assert_called_once_with(
+            self.consistency_group_snapshots[0].id)
+        self.assertIsNone(result)
+
+    def test_multiple_consistency_group_snapshots_delete(self):
+        arglist = []
+        for c in self.consistency_group_snapshots:
+            arglist.append(c.id)
+        verifylist = [
+            ('consistency_group_snapshot', arglist),
+        ]
+
+        parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+        result = self.cmd.take_action(parsed_args)
+
+        calls = []
+        for c in self.consistency_group_snapshots:
+            calls.append(call(c.id))
+        self.cgsnapshots_mock.delete.assert_has_calls(calls)
+        self.assertIsNone(result)
+
+
+class TestConsistencyGroupSnapshotList(TestConsistencyGroupSnapshot):
+
+    consistency_group_snapshots = (
+        volume_fakes.FakeConsistencyGroupSnapshot.
+        create_consistency_group_snapshots(count=2)
+    )
+    consistency_group = (
+        volume_fakes.FakeConsistencyGroup.create_one_consistency_group()
+    )
+
+    columns = [
+        'ID',
+        'Status',
+        'Name',
+    ]
+    columns_long = [
+        'ID',
+        'Status',
+        'ConsistencyGroup ID',
+        'Name',
+        'Description',
+        'Created At',
+    ]
+    data = []
+    for c in consistency_group_snapshots:
+        data.append((
+            c.id,
+            c.status,
+            c.name,
+        ))
+    data_long = []
+    for c in consistency_group_snapshots:
+        data_long.append((
+            c.id,
+            c.status,
+            c.consistencygroup_id,
+            c.name,
+            c.description,
+            c.created_at,
+        ))
+
+    def setUp(self):
+        super(TestConsistencyGroupSnapshotList, self).setUp()
+
+        self.cgsnapshots_mock.list.return_value = (
+            self.consistency_group_snapshots)
+        self.consistencygroups_mock.get.return_value = self.consistency_group
+        # Get the command to test
+        self.cmd = (
+            consistency_group_snapshot.
+            ListConsistencyGroupSnapshot(self.app, None)
+        )
+
+    def test_consistency_group_snapshot_list_without_options(self):
+        arglist = []
+        verifylist = [
+            ("all_projects", False),
+            ("long", False),
+            ("status", None),
+            ("consistency_group", None),
+        ]
+
+        parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+        columns, data = self.cmd.take_action(parsed_args)
+
+        search_opts = {
+            'all_tenants': False,
+            'status': None,
+            'consistencygroup_id': None,
+        }
+        self.cgsnapshots_mock.list.assert_called_once_with(
+            detailed=True, search_opts=search_opts)
+        self.assertEqual(self.columns, columns)
+        self.assertEqual(self.data, list(data))
+
+    def test_consistency_group_snapshot_list_with_long(self):
+        arglist = [
+            "--long",
+        ]
+        verifylist = [
+            ("all_projects", False),
+            ("long", True),
+            ("status", None),
+            ("consistency_group", None),
+        ]
+
+        parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+        columns, data = self.cmd.take_action(parsed_args)
+
+        search_opts = {
+            'all_tenants': False,
+            'status': None,
+            'consistencygroup_id': None,
+        }
+        self.cgsnapshots_mock.list.assert_called_once_with(
+            detailed=True, search_opts=search_opts)
+        self.assertEqual(self.columns_long, columns)
+        self.assertEqual(self.data_long, list(data))
+
+    def test_consistency_group_snapshot_list_with_options(self):
+        arglist = [
+            "--all-project",
+            "--status", self.consistency_group_snapshots[0].status,
+            "--consistency-group", self.consistency_group.id,
+        ]
+        verifylist = [
+            ("all_projects", True),
+            ("long", False),
+            ("status", self.consistency_group_snapshots[0].status),
+            ("consistency_group", self.consistency_group.id),
+        ]
+
+        parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+        columns, data = self.cmd.take_action(parsed_args)
+
+        search_opts = {
+            'all_tenants': True,
+            'status': self.consistency_group_snapshots[0].status,
+            'consistencygroup_id': self.consistency_group.id,
+        }
+        self.consistencygroups_mock.get.assert_called_once_with(
+            self.consistency_group.id)
+        self.cgsnapshots_mock.list.assert_called_once_with(
+            detailed=True, search_opts=search_opts)
+        self.assertEqual(self.columns, columns)
+        self.assertEqual(self.data, list(data))
+
+
+class TestConsistencyGroupSnapshotShow(TestConsistencyGroupSnapshot):
+
+    _consistency_group_snapshot = (
+        volume_fakes.
+        FakeConsistencyGroupSnapshot.
+        create_one_consistency_group_snapshot()
+    )
+
+    columns = (
+        'consistencygroup_id',
+        'created_at',
+        'description',
+        'id',
+        'name',
+        'status',
+    )
+    data = (
+        _consistency_group_snapshot.consistencygroup_id,
+        _consistency_group_snapshot.created_at,
+        _consistency_group_snapshot.description,
+        _consistency_group_snapshot.id,
+        _consistency_group_snapshot.name,
+        _consistency_group_snapshot.status,
+    )
+
+    def setUp(self):
+        super(TestConsistencyGroupSnapshotShow, self).setUp()
+
+        self.cgsnapshots_mock.get.return_value = (
+            self._consistency_group_snapshot)
+        self.cmd = (consistency_group_snapshot.
+                    ShowConsistencyGroupSnapshot(self.app, None))
+
+    def test_consistency_group_snapshot_show(self):
+        arglist = [
+            self._consistency_group_snapshot.id
+        ]
+        verifylist = [
+            ("consistency_group_snapshot", self._consistency_group_snapshot.id)
+        ]
+        parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+        columns, data = self.cmd.take_action(parsed_args)
+        self.cgsnapshots_mock.get.assert_called_once_with(
+            self._consistency_group_snapshot.id)
+        self.assertEqual(self.columns, columns)
+        self.assertEqual(self.data, data)
diff --git a/openstackclient/volume/v2/consistency_group_snapshot.py b/openstackclient/volume/v2/consistency_group_snapshot.py
new file mode 100644
index 00000000..540deb01
--- /dev/null
+++ b/openstackclient/volume/v2/consistency_group_snapshot.py
@@ -0,0 +1,190 @@
+#
+#   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 v2 consistency group snapshot action implementations"""
+
+import logging
+
+from osc_lib.command import command
+from osc_lib import exceptions
+from osc_lib import utils
+import six
+
+from openstackclient.i18n import _
+
+
+LOG = logging.getLogger(__name__)
+
+
+class CreateConsistencyGroupSnapshot(command.ShowOne):
+    _description = _("Create new consistency group snapshot.")
+
+    def get_parser(self, prog_name):
+        parser = super(
+            CreateConsistencyGroupSnapshot, self).get_parser(prog_name)
+        parser.add_argument(
+            "snapshot_name",
+            metavar="<snapshot-name>",
+            nargs="?",
+            help=_("Name of new consistency group snapshot (default to None)")
+        )
+        parser.add_argument(
+            "--consistency-group",
+            metavar="<consistency-group>",
+            help=_("Consistency group to snapshot (name or ID) "
+                   "(default to be the same as <snapshot-name>)")
+        )
+        parser.add_argument(
+            "--description",
+            metavar="<description>",
+            help=_("Description of this consistency group snapshot")
+        )
+        return parser
+
+    def take_action(self, parsed_args):
+        volume_client = self.app.client_manager.volume
+        consistency_group = parsed_args.consistency_group
+        if not parsed_args.consistency_group:
+            # If "--consistency-group" not specified, then consistency_group
+            # will be the same as the new consistency group snapshot name
+            consistency_group = parsed_args.snapshot_name
+        consistency_group_id = utils.find_resource(
+            volume_client.consistencygroups,
+            consistency_group).id
+        consistency_group_snapshot = volume_client.cgsnapshots.create(
+            consistency_group_id,
+            name=parsed_args.snapshot_name,
+            description=parsed_args.description,
+        )
+
+        return zip(*sorted(six.iteritems(consistency_group_snapshot._info)))
+
+
+class DeleteConsistencyGroupSnapshot(command.Command):
+    _description = _("Delete consistency group snapshot(s).")
+
+    def get_parser(self, prog_name):
+        parser = super(
+            DeleteConsistencyGroupSnapshot, self).get_parser(prog_name)
+        parser.add_argument(
+            "consistency_group_snapshot",
+            metavar="<consistency-group-snapshot>",
+            nargs="+",
+            help=_("Consistency group snapshot(s) to delete (name or ID)")
+        )
+        return parser
+
+    def take_action(self, parsed_args):
+        volume_client = self.app.client_manager.volume
+        result = 0
+
+        for snapshot in parsed_args.consistency_group_snapshot:
+            try:
+                snapshot_id = utils.find_resource(volume_client.cgsnapshots,
+                                                  snapshot).id
+
+                volume_client.cgsnapshots.delete(snapshot_id)
+            except Exception as e:
+                result += 1
+                LOG.error(_("Failed to delete consistency group snapshot "
+                            "with name or ID '%(snapshot)s': %(e)s")
+                          % {'snapshot': snapshot, 'e': e})
+
+        if result > 0:
+            total = len(parsed_args.consistency_group_snapshot)
+            msg = (_("%(result)s of %(total)s consistency group snapshots "
+                   "failed to delete.") % {'result': result, 'total': total})
+            raise exceptions.CommandError(msg)
+
+
+class ListConsistencyGroupSnapshot(command.Lister):
+    _description = _("List consistency group snapshots.")
+
+    def get_parser(self, prog_name):
+        parser = super(
+            ListConsistencyGroupSnapshot, self).get_parser(prog_name)
+        parser.add_argument(
+            '--all-projects',
+            action="store_true",
+            help=_('Show detail for all projects (admin only) '
+                   '(defaults to False)')
+        )
+        parser.add_argument(
+            '--long',
+            action="store_true",
+            help=_('List additional fields in output')
+        )
+        parser.add_argument(
+            '--status',
+            metavar="<status>",
+            choices=['available', 'error', 'creating', 'deleting',
+                     'error-deleting'],
+            help=_('Filters results by a status ("available", "error", '
+                   '"creating", "deleting" or "error_deleting")')
+        )
+        parser.add_argument(
+            '--consistency-group',
+            metavar="<consistency-group>",
+            help=_('Filters results by a consistency group (name or ID)')
+        )
+        return parser
+
+    def take_action(self, parsed_args):
+        if parsed_args.long:
+            columns = ['ID', 'Status', 'ConsistencyGroup ID',
+                       'Name', 'Description', 'Created At']
+        else:
+            columns = ['ID', 'Status', 'Name']
+        volume_client = self.app.client_manager.volume
+        consistency_group_id = None
+        if parsed_args.consistency_group:
+            consistency_group_id = utils.find_resource(
+                volume_client.consistencygroups,
+                parsed_args.consistency_group,
+            ).id
+        search_opts = {
+            'all_tenants': parsed_args.all_projects,
+            'status': parsed_args.status,
+            'consistencygroup_id': consistency_group_id,
+        }
+        consistency_group_snapshots = volume_client.cgsnapshots.list(
+            detailed=True,
+            search_opts=search_opts,
+        )
+
+        return (columns, (
+            utils.get_item_properties(
+                s, columns)
+            for s in consistency_group_snapshots))
+
+
+class ShowConsistencyGroupSnapshot(command.ShowOne):
+    _description = _("Display consistency group snapshot details")
+
+    def get_parser(self, prog_name):
+        parser = super(
+            ShowConsistencyGroupSnapshot, self).get_parser(prog_name)
+        parser.add_argument(
+            "consistency_group_snapshot",
+            metavar="<consistency-group-snapshot>",
+            help=_("Consistency group snapshot to display (name or ID)")
+        )
+        return parser
+
+    def take_action(self, parsed_args):
+        volume_client = self.app.client_manager.volume
+        consistency_group_snapshot = utils.find_resource(
+            volume_client.cgsnapshots,
+            parsed_args.consistency_group_snapshot)
+        return zip(*sorted(six.iteritems(consistency_group_snapshot._info)))