diff --git a/openstackclient/tests/unit/volume/v3/test_volume_snapshot.py b/openstackclient/tests/unit/volume/v3/test_volume_snapshot.py index 4c990d840a..cc0381a8cf 100644 --- a/openstackclient/tests/unit/volume/v3/test_volume_snapshot.py +++ b/openstackclient/tests/unit/volume/v3/test_volume_snapshot.py @@ -13,9 +13,12 @@ from unittest import mock +from osc_lib.cli import format_columns from osc_lib import exceptions from osc_lib import utils +from openstackclient.tests.unit.identity.v3 import fakes as project_fakes +from openstackclient.tests.unit import utils as test_utils from openstackclient.tests.unit.volume.v2 import fakes as volume_fakes from openstackclient.tests.unit.volume.v3 import fakes as volume_fakes_v3 from openstackclient.volume.v3 import volume_snapshot @@ -27,10 +30,163 @@ class TestVolumeSnapshot(volume_fakes_v3.TestVolume): self.snapshots_mock = self.volume_client.volume_snapshots self.snapshots_mock.reset_mock() + self.volumes_mock = self.volume_client.volumes + self.volumes_mock.reset_mock() + self.project_mock = self.identity_client.projects + self.project_mock.reset_mock() self.volume_sdk_client.unmanage_snapshot.return_value = None +class TestVolumeSnapshotCreate(TestVolumeSnapshot): + columns = ( + 'created_at', + 'description', + 'id', + 'name', + 'properties', + 'size', + 'status', + 'volume_id', + ) + + def setUp(self): + super().setUp() + + self.volume = volume_fakes.create_one_volume() + self.new_snapshot = volume_fakes.create_one_snapshot( + attrs={'volume_id': self.volume.id} + ) + + self.data = ( + self.new_snapshot.created_at, + self.new_snapshot.description, + self.new_snapshot.id, + self.new_snapshot.name, + format_columns.DictColumn(self.new_snapshot.metadata), + self.new_snapshot.size, + self.new_snapshot.status, + self.new_snapshot.volume_id, + ) + + self.volumes_mock.get.return_value = self.volume + self.snapshots_mock.create.return_value = self.new_snapshot + self.snapshots_mock.manage.return_value = self.new_snapshot + # Get the command object to test + self.cmd = volume_snapshot.CreateVolumeSnapshot(self.app, None) + + def test_snapshot_create(self): + arglist = [ + "--volume", + self.new_snapshot.volume_id, + "--description", + self.new_snapshot.description, + "--force", + '--property', + 'Alpha=a', + '--property', + 'Beta=b', + self.new_snapshot.name, + ] + verifylist = [ + ("volume", self.new_snapshot.volume_id), + ("description", self.new_snapshot.description), + ("force", True), + ('property', {'Alpha': 'a', 'Beta': 'b'}), + ("snapshot_name", self.new_snapshot.name), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + columns, data = self.cmd.take_action(parsed_args) + + self.snapshots_mock.create.assert_called_with( + self.new_snapshot.volume_id, + force=True, + name=self.new_snapshot.name, + description=self.new_snapshot.description, + metadata={'Alpha': 'a', 'Beta': 'b'}, + ) + self.assertEqual(self.columns, columns) + self.assertEqual(self.data, data) + + def test_snapshot_create_without_name(self): + arglist = [ + "--volume", + self.new_snapshot.volume_id, + ] + verifylist = [ + ("volume", self.new_snapshot.volume_id), + ] + self.assertRaises( + test_utils.ParserException, + self.check_parser, + self.cmd, + arglist, + verifylist, + ) + + def test_snapshot_create_without_volume(self): + arglist = [ + "--description", + self.new_snapshot.description, + "--force", + self.new_snapshot.name, + ] + verifylist = [ + ("description", self.new_snapshot.description), + ("force", True), + ("snapshot_name", self.new_snapshot.name), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + columns, data = self.cmd.take_action(parsed_args) + + self.volumes_mock.get.assert_called_once_with(self.new_snapshot.name) + self.snapshots_mock.create.assert_called_once_with( + self.new_snapshot.volume_id, + force=True, + name=self.new_snapshot.name, + description=self.new_snapshot.description, + metadata=None, + ) + self.assertEqual(self.columns, columns) + self.assertEqual(self.data, data) + + def test_snapshot_create_with_remote_source(self): + arglist = [ + '--remote-source', + 'source-name=test_source_name', + '--remote-source', + 'source-id=test_source_id', + '--volume', + self.new_snapshot.volume_id, + self.new_snapshot.name, + ] + ref_dict = { + 'source-name': 'test_source_name', + 'source-id': 'test_source_id', + } + verifylist = [ + ('remote_source', ref_dict), + ('volume', self.new_snapshot.volume_id), + ("snapshot_name", self.new_snapshot.name), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + columns, data = self.cmd.take_action(parsed_args) + + self.snapshots_mock.manage.assert_called_with( + volume_id=self.new_snapshot.volume_id, + ref=ref_dict, + name=self.new_snapshot.name, + description=None, + metadata=None, + ) + self.snapshots_mock.create.assert_not_called() + self.assertEqual(self.columns, columns) + self.assertEqual(self.data, data) + + class TestVolumeSnapshotDelete(TestVolumeSnapshot): snapshots = volume_fakes.create_snapshots(count=2) @@ -158,3 +314,476 @@ class TestVolumeSnapshotDelete(TestVolumeSnapshot): calls.append(mock.call(s.id)) self.volume_sdk_client.unmanage_snapshot.assert_has_calls(calls) self.assertIsNone(result) + + +class TestVolumeSnapshotList(TestVolumeSnapshot): + volume = volume_fakes.create_one_volume() + project = project_fakes.FakeProject.create_one_project() + snapshots = volume_fakes.create_snapshots( + attrs={'volume_id': volume.name}, count=3 + ) + + columns = ["ID", "Name", "Description", "Status", "Size"] + columns_long = columns + ["Created At", "Volume", "Properties"] + + data = [] + for s in snapshots: + data.append( + ( + s.id, + s.name, + s.description, + s.status, + s.size, + ) + ) + data_long = [] + for s in snapshots: + data_long.append( + ( + s.id, + s.name, + s.description, + s.status, + s.size, + s.created_at, + volume_snapshot.VolumeIdColumn( + s.volume_id, volume_cache={volume.id: volume} + ), + format_columns.DictColumn(s.metadata), + ) + ) + + def setUp(self): + super().setUp() + + self.volumes_mock.list.return_value = [self.volume] + self.volumes_mock.get.return_value = self.volume + self.project_mock.get.return_value = self.project + self.snapshots_mock.list.return_value = self.snapshots + # Get the command to test + self.cmd = volume_snapshot.ListVolumeSnapshot(self.app, None) + + def test_snapshot_list_without_options(self): + arglist = [] + verifylist = [('all_projects', False), ('long', False)] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + columns, data = self.cmd.take_action(parsed_args) + + self.snapshots_mock.list.assert_called_once_with( + limit=None, + marker=None, + search_opts={ + 'all_tenants': False, + 'name': None, + 'status': None, + 'project_id': None, + 'volume_id': None, + }, + ) + self.assertEqual(self.columns, columns) + self.assertEqual(self.data, list(data)) + + def test_snapshot_list_with_options(self): + arglist = [ + "--long", + "--limit", + "2", + "--project", + self.project.id, + "--marker", + self.snapshots[0].id, + ] + verifylist = [ + ("long", True), + ("limit", 2), + ("project", self.project.id), + ("marker", self.snapshots[0].id), + ('all_projects', False), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + columns, data = self.cmd.take_action(parsed_args) + + self.snapshots_mock.list.assert_called_once_with( + limit=2, + marker=self.snapshots[0].id, + search_opts={ + 'all_tenants': True, + 'project_id': self.project.id, + 'name': None, + 'status': None, + 'volume_id': None, + }, + ) + self.assertEqual(self.columns_long, columns) + self.assertEqual(self.data_long, list(data)) + + def test_snapshot_list_all_projects(self): + arglist = [ + '--all-projects', + ] + verifylist = [('long', False), ('all_projects', True)] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + columns, data = self.cmd.take_action(parsed_args) + + self.snapshots_mock.list.assert_called_once_with( + limit=None, + marker=None, + search_opts={ + 'all_tenants': True, + 'name': None, + 'status': None, + 'project_id': None, + 'volume_id': None, + }, + ) + self.assertEqual(self.columns, columns) + self.assertEqual(self.data, list(data)) + + def test_snapshot_list_name_option(self): + arglist = [ + '--name', + self.snapshots[0].name, + ] + verifylist = [ + ('all_projects', False), + ('long', False), + ('name', self.snapshots[0].name), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + columns, data = self.cmd.take_action(parsed_args) + + self.snapshots_mock.list.assert_called_once_with( + limit=None, + marker=None, + search_opts={ + 'all_tenants': False, + 'name': self.snapshots[0].name, + 'status': None, + 'project_id': None, + 'volume_id': None, + }, + ) + self.assertEqual(self.columns, columns) + self.assertEqual(self.data, list(data)) + + def test_snapshot_list_status_option(self): + arglist = [ + '--status', + self.snapshots[0].status, + ] + verifylist = [ + ('all_projects', False), + ('long', False), + ('status', self.snapshots[0].status), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + columns, data = self.cmd.take_action(parsed_args) + + self.snapshots_mock.list.assert_called_once_with( + limit=None, + marker=None, + search_opts={ + 'all_tenants': False, + 'name': None, + 'status': self.snapshots[0].status, + 'project_id': None, + 'volume_id': None, + }, + ) + self.assertEqual(self.columns, columns) + self.assertEqual(self.data, list(data)) + + def test_snapshot_list_volumeid_option(self): + arglist = [ + '--volume', + self.volume.id, + ] + verifylist = [ + ('all_projects', False), + ('long', False), + ('volume', self.volume.id), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + columns, data = self.cmd.take_action(parsed_args) + + self.snapshots_mock.list.assert_called_once_with( + limit=None, + marker=None, + search_opts={ + 'all_tenants': False, + 'name': None, + 'status': None, + 'project_id': None, + 'volume_id': self.volume.id, + }, + ) + self.assertEqual(self.columns, columns) + self.assertEqual(self.data, list(data)) + + def test_snapshot_list_negative_limit(self): + arglist = [ + "--limit", + "-2", + ] + verifylist = [ + ("limit", -2), + ] + self.assertRaises( + test_utils.ParserException, + self.check_parser, + self.cmd, + arglist, + verifylist, + ) + + +class TestVolumeSnapshotSet(TestVolumeSnapshot): + snapshot = volume_fakes.create_one_snapshot() + + def setUp(self): + super().setUp() + + self.snapshots_mock.get.return_value = self.snapshot + self.snapshots_mock.set_metadata.return_value = None + self.snapshots_mock.update.return_value = None + # Get the command object to mock + self.cmd = volume_snapshot.SetVolumeSnapshot(self.app, None) + + def test_snapshot_set_no_option(self): + arglist = [ + self.snapshot.id, + ] + verifylist = [ + ("snapshot", self.snapshot.id), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + result = self.cmd.take_action(parsed_args) + self.snapshots_mock.get.assert_called_once_with(parsed_args.snapshot) + self.assertNotCalled(self.snapshots_mock.reset_state) + self.assertNotCalled(self.snapshots_mock.update) + self.assertNotCalled(self.snapshots_mock.set_metadata) + self.assertIsNone(result) + + def test_snapshot_set_name_and_property(self): + arglist = [ + "--name", + "new_snapshot", + "--property", + "x=y", + "--property", + "foo=foo", + self.snapshot.id, + ] + new_property = {"x": "y", "foo": "foo"} + verifylist = [ + ("name", "new_snapshot"), + ("property", new_property), + ("snapshot", self.snapshot.id), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + result = self.cmd.take_action(parsed_args) + + kwargs = { + "name": "new_snapshot", + } + self.snapshots_mock.update.assert_called_with( + self.snapshot.id, **kwargs + ) + self.snapshots_mock.set_metadata.assert_called_with( + self.snapshot.id, new_property + ) + self.assertIsNone(result) + + def test_snapshot_set_with_no_property(self): + arglist = [ + "--no-property", + self.snapshot.id, + ] + verifylist = [ + ("no_property", True), + ("snapshot", self.snapshot.id), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + result = self.cmd.take_action(parsed_args) + self.snapshots_mock.get.assert_called_once_with(parsed_args.snapshot) + self.assertNotCalled(self.snapshots_mock.reset_state) + self.assertNotCalled(self.snapshots_mock.update) + self.assertNotCalled(self.snapshots_mock.set_metadata) + self.snapshots_mock.delete_metadata.assert_called_with( + self.snapshot.id, ["foo"] + ) + self.assertIsNone(result) + + def test_snapshot_set_with_no_property_and_property(self): + arglist = [ + "--no-property", + "--property", + "foo_1=bar_1", + self.snapshot.id, + ] + verifylist = [ + ("no_property", True), + ("property", {"foo_1": "bar_1"}), + ("snapshot", self.snapshot.id), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + result = self.cmd.take_action(parsed_args) + self.snapshots_mock.get.assert_called_once_with(parsed_args.snapshot) + self.assertNotCalled(self.snapshots_mock.reset_state) + self.assertNotCalled(self.snapshots_mock.update) + self.snapshots_mock.delete_metadata.assert_called_with( + self.snapshot.id, ["foo"] + ) + self.snapshots_mock.set_metadata.assert_called_once_with( + self.snapshot.id, {"foo_1": "bar_1"} + ) + self.assertIsNone(result) + + def test_snapshot_set_state_to_error(self): + arglist = ["--state", "error", self.snapshot.id] + verifylist = [("state", "error"), ("snapshot", self.snapshot.id)] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + result = self.cmd.take_action(parsed_args) + + self.snapshots_mock.reset_state.assert_called_with( + self.snapshot.id, "error" + ) + self.assertIsNone(result) + + def test_volume_set_state_failed(self): + self.snapshots_mock.reset_state.side_effect = exceptions.CommandError() + arglist = ['--state', 'error', self.snapshot.id] + verifylist = [('state', 'error'), ('snapshot', self.snapshot.id)] + + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + try: + self.cmd.take_action(parsed_args) + self.fail('CommandError should be raised.') + except exceptions.CommandError as e: + self.assertEqual( + 'One or more of the set operations failed', str(e) + ) + self.snapshots_mock.reset_state.assert_called_once_with( + self.snapshot.id, 'error' + ) + + def test_volume_set_name_and_state_failed(self): + self.snapshots_mock.reset_state.side_effect = exceptions.CommandError() + arglist = [ + '--state', + 'error', + "--name", + "new_snapshot", + self.snapshot.id, + ] + verifylist = [ + ('state', 'error'), + ("name", "new_snapshot"), + ('snapshot', self.snapshot.id), + ] + + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + try: + self.cmd.take_action(parsed_args) + self.fail('CommandError should be raised.') + except exceptions.CommandError as e: + self.assertEqual( + 'One or more of the set operations failed', str(e) + ) + kwargs = { + "name": "new_snapshot", + } + self.snapshots_mock.update.assert_called_once_with( + self.snapshot.id, **kwargs + ) + self.snapshots_mock.reset_state.assert_called_once_with( + self.snapshot.id, 'error' + ) + + +class TestVolumeSnapshotShow(TestVolumeSnapshot): + columns = ( + 'created_at', + 'description', + 'id', + 'name', + 'properties', + 'size', + 'status', + 'volume_id', + ) + + def setUp(self): + super().setUp() + + self.snapshot = volume_fakes.create_one_snapshot() + + self.data = ( + self.snapshot.created_at, + self.snapshot.description, + self.snapshot.id, + self.snapshot.name, + format_columns.DictColumn(self.snapshot.metadata), + self.snapshot.size, + self.snapshot.status, + self.snapshot.volume_id, + ) + + self.snapshots_mock.get.return_value = self.snapshot + # Get the command object to test + self.cmd = volume_snapshot.ShowVolumeSnapshot(self.app, None) + + def test_snapshot_show(self): + arglist = [self.snapshot.id] + verifylist = [("snapshot", self.snapshot.id)] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + columns, data = self.cmd.take_action(parsed_args) + self.snapshots_mock.get.assert_called_with(self.snapshot.id) + + self.assertEqual(self.columns, columns) + self.assertCountEqual(self.data, data) + + +class TestVolumeSnapshotUnset(TestVolumeSnapshot): + snapshot = volume_fakes.create_one_snapshot() + + def setUp(self): + super().setUp() + + self.snapshots_mock.get.return_value = self.snapshot + self.snapshots_mock.delete_metadata.return_value = None + # Get the command object to mock + self.cmd = volume_snapshot.UnsetVolumeSnapshot(self.app, None) + + def test_snapshot_unset(self): + arglist = [ + "--property", + "foo", + self.snapshot.id, + ] + verifylist = [ + ("property", ["foo"]), + ("snapshot", self.snapshot.id), + ] + + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + result = self.cmd.take_action(parsed_args) + + self.snapshots_mock.delete_metadata.assert_called_with( + self.snapshot.id, ["foo"] + ) + self.assertIsNone(result) diff --git a/openstackclient/volume/v3/volume_snapshot.py b/openstackclient/volume/v3/volume_snapshot.py index bb7f5a660e..b0b1c0f554 100644 --- a/openstackclient/volume/v3/volume_snapshot.py +++ b/openstackclient/volume/v3/volume_snapshot.py @@ -14,17 +14,144 @@ """Volume v3 snapshot action implementations""" +import copy +import functools import logging +from cliff import columns as cliff_columns +from osc_lib.cli import format_columns +from osc_lib.cli import parseractions from osc_lib.command import command from osc_lib import exceptions from osc_lib import utils +from openstackclient.common import pagination from openstackclient.i18n import _ +from openstackclient.identity import common as identity_common LOG = logging.getLogger(__name__) +class VolumeIdColumn(cliff_columns.FormattableColumn): + """Formattable column for volume ID column. + + Unlike the parent FormattableColumn class, the initializer of the + class takes volume_cache as the second argument. + osc_lib.utils.get_item_properties instantiate cliff FormattableColumn + object with a single parameter "column value", so you need to pass + a partially initialized class like + ``functools.partial(VolumeIdColumn, volume_cache)``. + """ + + def __init__(self, value, volume_cache=None): + super().__init__(value) + self._volume_cache = volume_cache or {} + + def human_readable(self): + """Return a volume name if available + + :rtype: either the volume ID or name + """ + volume_id = self._value + volume = volume_id + if volume_id in self._volume_cache.keys(): + volume = self._volume_cache[volume_id].name + return volume + + +class CreateVolumeSnapshot(command.ShowOne): + _description = _("Create new volume snapshot") + + def get_parser(self, prog_name): + parser = super().get_parser(prog_name) + parser.add_argument( + "snapshot_name", + metavar="<snapshot-name>", + help=_("Name of the new snapshot"), + ) + parser.add_argument( + "--volume", + metavar="<volume>", + help=_( + "Volume to snapshot (name or ID) (default is <snapshot-name>)" + ), + ) + parser.add_argument( + "--description", + metavar="<description>", + help=_("Description of the snapshot"), + ) + parser.add_argument( + "--force", + action="store_true", + default=False, + help=_( + "Create a snapshot attached to an instance. Default is False" + ), + ) + parser.add_argument( + "--property", + metavar="<key=value>", + action=parseractions.KeyValueAction, + help=_( + "Set a property to this snapshot " + "(repeat option to set multiple properties)" + ), + ) + parser.add_argument( + "--remote-source", + metavar="<key=value>", + action=parseractions.KeyValueAction, + help=_( + "The attribute(s) of the existing remote volume snapshot " + "(admin required) (repeat option to specify multiple " + "attributes) e.g.: '--remote-source source-name=test_name " + "--remote-source source-id=test_id'" + ), + ) + return parser + + def take_action(self, parsed_args): + volume_client = self.app.client_manager.volume + volume = parsed_args.volume + if not parsed_args.volume: + volume = parsed_args.snapshot_name + volume_id = utils.find_resource(volume_client.volumes, volume).id + if parsed_args.remote_source: + # Create a new snapshot from an existing remote snapshot source + if parsed_args.force: + msg = _( + "'--force' option will not work when you create " + "new volume snapshot from an existing remote " + "volume snapshot" + ) + LOG.warning(msg) + snapshot = volume_client.volume_snapshots.manage( + volume_id=volume_id, + ref=parsed_args.remote_source, + name=parsed_args.snapshot_name, + description=parsed_args.description, + metadata=parsed_args.property, + ) + else: + # create a new snapshot from scratch + snapshot = volume_client.volume_snapshots.create( + volume_id, + force=parsed_args.force, + name=parsed_args.snapshot_name, + description=parsed_args.description, + metadata=parsed_args.property, + ) + snapshot._info.update( + { + 'properties': format_columns.DictColumn( + snapshot._info.pop('metadata') + ) + } + ) + return zip(*sorted(snapshot._info.items())) + + class DeleteVolumeSnapshot(command.Command): _description = _("Delete volume snapshot(s)") @@ -96,3 +223,316 @@ class DeleteVolumeSnapshot(command.Command): 'total': total, } raise exceptions.CommandError(msg) + + +class ListVolumeSnapshot(command.Lister): + _description = _("List volume snapshots") + + def get_parser(self, prog_name): + parser = super().get_parser(prog_name) + parser.add_argument( + '--all-projects', + action='store_true', + default=False, + help=_('Include all projects (admin only)'), + ) + parser.add_argument( + '--project', + metavar='<project>', + help=_('Filter results by project (name or ID) (admin only)'), + ) + identity_common.add_project_domain_option_to_parser(parser) + parser.add_argument( + '--long', + action='store_true', + default=False, + help=_('List additional fields in output'), + ) + parser.add_argument( + '--name', + metavar='<name>', + default=None, + help=_('Filters results by a name.'), + ) + 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( + '--volume', + metavar='<volume>', + default=None, + help=_('Filters results by a volume (name or ID).'), + ) + pagination.add_marker_pagination_option_to_parser(parser) + return parser + + def take_action(self, parsed_args): + volume_client = self.app.client_manager.volume + identity_client = self.app.client_manager.identity + + if parsed_args.long: + columns = [ + 'ID', + 'Name', + 'Description', + 'Status', + 'Size', + 'Created At', + 'Volume ID', + 'Metadata', + ] + column_headers = copy.deepcopy(columns) + column_headers[6] = 'Volume' + column_headers[7] = 'Properties' + else: + columns = ['ID', 'Name', 'Description', 'Status', 'Size'] + column_headers = copy.deepcopy(columns) + + # Cache the volume list + volume_cache = {} + try: + for s in volume_client.volumes.list(): + volume_cache[s.id] = s + except Exception: # noqa: S110 + # Just forget it if there's any trouble + pass + _VolumeIdColumn = functools.partial( + VolumeIdColumn, volume_cache=volume_cache + ) + + volume_id = None + if parsed_args.volume: + volume_id = utils.find_resource( + volume_client.volumes, parsed_args.volume + ).id + + project_id = None + if parsed_args.project: + project_id = identity_common.find_project( + identity_client, + parsed_args.project, + parsed_args.project_domain, + ).id + + # set value of 'all_tenants' when using project option + all_projects = ( + True if parsed_args.project else parsed_args.all_projects + ) + + search_opts = { + 'all_tenants': all_projects, + 'project_id': project_id, + 'name': parsed_args.name, + 'status': parsed_args.status, + 'volume_id': volume_id, + } + + data = volume_client.volume_snapshots.list( + search_opts=search_opts, + marker=parsed_args.marker, + limit=parsed_args.limit, + ) + return ( + column_headers, + ( + utils.get_item_properties( + s, + columns, + formatters={ + 'Metadata': format_columns.DictColumn, + 'Volume ID': _VolumeIdColumn, + }, + ) + for s in data + ), + ) + + +class SetVolumeSnapshot(command.Command): + _description = _("Set volume snapshot properties") + + def get_parser(self, prog_name): + parser = super().get_parser(prog_name) + parser.add_argument( + 'snapshot', + metavar='<snapshot>', + help=_('Snapshot to modify (name or ID)'), + ) + parser.add_argument( + '--name', metavar='<name>', help=_('New snapshot name') + ) + parser.add_argument( + '--description', + metavar='<description>', + help=_('New snapshot description'), + ) + parser.add_argument( + "--no-property", + dest="no_property", + action="store_true", + help=_( + "Remove all properties from <snapshot> " + "(specify both --no-property and --property to " + "remove the current properties before setting " + "new properties.)" + ), + ) + parser.add_argument( + '--property', + metavar='<key=value>', + action=parseractions.KeyValueAction, + help=_( + 'Property to add/change for this snapshot ' + '(repeat option to set multiple properties)' + ), + ) + parser.add_argument( + '--state', + metavar='<state>', + choices=[ + 'available', + 'error', + 'creating', + 'deleting', + 'error_deleting', + ], + help=_( + 'New snapshot state. ("available", "error", "creating", ' + '"deleting", or "error_deleting") (admin only) ' + '(This option simply changes the state of the snapshot ' + 'in the database with no regard to actual status, ' + 'exercise caution when using)' + ), + ) + return parser + + def take_action(self, parsed_args): + volume_client = self.app.client_manager.volume + snapshot = utils.find_resource( + volume_client.volume_snapshots, parsed_args.snapshot + ) + + result = 0 + if parsed_args.no_property: + try: + key_list = snapshot.metadata.keys() + volume_client.volume_snapshots.delete_metadata( + snapshot.id, + list(key_list), + ) + except Exception as e: + LOG.error(_("Failed to clean snapshot properties: %s"), e) + result += 1 + + if parsed_args.property: + try: + volume_client.volume_snapshots.set_metadata( + snapshot.id, parsed_args.property + ) + except Exception as e: + LOG.error(_("Failed to set snapshot property: %s"), e) + result += 1 + + if parsed_args.state: + try: + volume_client.volume_snapshots.reset_state( + snapshot.id, parsed_args.state + ) + except Exception as e: + LOG.error(_("Failed to set snapshot state: %s"), e) + result += 1 + + kwargs = {} + if parsed_args.name: + kwargs['name'] = parsed_args.name + if parsed_args.description: + kwargs['description'] = parsed_args.description + if kwargs: + try: + volume_client.volume_snapshots.update(snapshot.id, **kwargs) + except Exception as e: + LOG.error( + _("Failed to update snapshot name or description: %s"), + e, + ) + result += 1 + + if result > 0: + raise exceptions.CommandError( + _("One or more of the set operations failed") + ) + + +class ShowVolumeSnapshot(command.ShowOne): + _description = _("Display volume snapshot details") + + def get_parser(self, prog_name): + parser = super().get_parser(prog_name) + parser.add_argument( + "snapshot", + metavar="<snapshot>", + help=_("Snapshot to display (name or ID)"), + ) + return parser + + def take_action(self, parsed_args): + volume_client = self.app.client_manager.volume + snapshot = utils.find_resource( + volume_client.volume_snapshots, parsed_args.snapshot + ) + snapshot._info.update( + { + 'properties': format_columns.DictColumn( + snapshot._info.pop('metadata') + ) + } + ) + return zip(*sorted(snapshot._info.items())) + + +class UnsetVolumeSnapshot(command.Command): + _description = _("Unset volume snapshot properties") + + def get_parser(self, prog_name): + parser = super().get_parser(prog_name) + parser.add_argument( + 'snapshot', + metavar='<snapshot>', + help=_('Snapshot to modify (name or ID)'), + ) + parser.add_argument( + '--property', + metavar='<key>', + action='append', + default=[], + help=_( + 'Property to remove from snapshot ' + '(repeat option to remove multiple properties)' + ), + ) + return parser + + def take_action(self, parsed_args): + volume_client = self.app.client_manager.volume + snapshot = utils.find_resource( + volume_client.volume_snapshots, parsed_args.snapshot + ) + + if parsed_args.property: + volume_client.volume_snapshots.delete_metadata( + snapshot.id, + parsed_args.property, + ) diff --git a/setup.cfg b/setup.cfg index 519a371cbb..5c6449d5f2 100644 --- a/setup.cfg +++ b/setup.cfg @@ -780,12 +780,12 @@ openstack.volume.v3 = block_storage_resource_filter_list = openstackclient.volume.v3.block_storage_resource_filter:ListBlockStorageResourceFilter block_storage_resource_filter_show = openstackclient.volume.v3.block_storage_resource_filter:ShowBlockStorageResourceFilter - volume_snapshot_create = openstackclient.volume.v2.volume_snapshot:CreateVolumeSnapshot + volume_snapshot_create = openstackclient.volume.v3.volume_snapshot:CreateVolumeSnapshot 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 - volume_snapshot_unset = openstackclient.volume.v2.volume_snapshot:UnsetVolumeSnapshot + volume_snapshot_list = openstackclient.volume.v3.volume_snapshot:ListVolumeSnapshot + volume_snapshot_set = openstackclient.volume.v3.volume_snapshot:SetVolumeSnapshot + volume_snapshot_show = openstackclient.volume.v3.volume_snapshot:ShowVolumeSnapshot + volume_snapshot_unset = openstackclient.volume.v3.volume_snapshot:UnsetVolumeSnapshot volume_type_create = openstackclient.volume.v3.volume_type:CreateVolumeType volume_type_delete = openstackclient.volume.v3.volume_type:DeleteVolumeType