openstack/python-openstackclient
Code Issues Proposed changes

volume: Migrate 'snapshot set', 'snapshot unset' to SDK

Browse Source 
Change-Id: Id34d460c8c5656bf43f48717b13a002508562e4e
Signed-off-by: Stephen Finucane <stephenfin@redhat.com>
This commit is contained in:
Stephen Finucane
2025-03-27 23:08:22 +00:00
parent e0020aec6a
commit 3c6fa42642
4 changed files with 203 additions and 189 deletions
Download Patch File Download Diff File Expand all files Collapse all files

156
openstackclient/tests/unit/volume/v2/test_volume_snapshot.py
View File

@@ -519,16 +519,18 @@ class TestVolumeSnapshotList(TestVolumeSnapshot):
) )
class TestVolumeSnapshotSet(TestVolumeSnapshot): class TestVolumeSnapshotSet(volume_fakes.TestVolume):
snapshot = volume_fakes.create_one_snapshot()
def setUp(self): def setUp(self):
super().setUp() super().setUp()
self.snapshots_mock.get.return_value = self.snapshot self.snapshot = sdk_fakes.generate_fake_resource(
self.snapshots_mock.set_metadata.return_value = None _snapshot.Snapshot, metadata={'foo': 'bar'}
self.snapshots_mock.update.return_value = None )
# Get the command object to mock self.volume_sdk_client.find_snapshot.return_value = self.snapshot
self.volume_sdk_client.delete_snapshot_metadata.return_value = None
self.volume_sdk_client.set_snapshot_metadata.return_value = None
self.volume_sdk_client.update_snapshot.return_value = None
self.cmd = volume_snapshot.SetVolumeSnapshot(self.app, None) self.cmd = volume_snapshot.SetVolumeSnapshot(self.app, None)
def test_snapshot_set_no_option(self): def test_snapshot_set_no_option(self):
@@ -541,11 +543,14 @@ class TestVolumeSnapshotSet(TestVolumeSnapshot):
parsed_args = self.check_parser(self.cmd, arglist, verifylist) parsed_args = self.check_parser(self.cmd, arglist, verifylist)
result = self.cmd.take_action(parsed_args) 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) self.assertIsNone(result)
self.volume_sdk_client.find_snapshot.assert_called_once_with(
parsed_args.snapshot, ignore_missing=False
)
self.volume_sdk_client.reset_snapshot_status.assert_not_called()
self.volume_sdk_client.update_snapshot.assert_not_called()
self.volume_sdk_client.set_snapshot_metadata.assert_not_called()
def test_snapshot_set_name_and_property(self): def test_snapshot_set_name_and_property(self):
arglist = [ arglist = [
@@ -557,26 +562,22 @@ class TestVolumeSnapshotSet(TestVolumeSnapshot):
"foo=foo", "foo=foo",
self.snapshot.id, self.snapshot.id,
] ]
new_property = {"x": "y", "foo": "foo"}
verifylist = [ verifylist = [
("name", "new_snapshot"), ("name", "new_snapshot"),
("property", new_property), ("properties", {"x": "y", "foo": "foo"}),
("snapshot", self.snapshot.id), ("snapshot", self.snapshot.id),
] ]
parsed_args = self.check_parser(self.cmd, arglist, verifylist) parsed_args = self.check_parser(self.cmd, arglist, verifylist)
result = self.cmd.take_action(parsed_args) 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) self.assertIsNone(result)
self.volume_sdk_client.update_snapshot.assert_called_with(
self.snapshot.id, name="new_snapshot"
)
self.volume_sdk_client.set_snapshot_metadata.assert_called_with(
self.snapshot.id, x="y", foo="foo"
)
def test_snapshot_set_with_no_property(self): def test_snapshot_set_with_no_property(self):
arglist = [ arglist = [
@@ -590,14 +591,17 @@ class TestVolumeSnapshotSet(TestVolumeSnapshot):
parsed_args = self.check_parser(self.cmd, arglist, verifylist) parsed_args = self.check_parser(self.cmd, arglist, verifylist)
result = self.cmd.take_action(parsed_args) 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) self.assertIsNone(result)
self.volume_sdk_client.find_snapshot.assert_called_once_with(
parsed_args.snapshot, ignore_missing=False
)
self.volume_sdk_client.reset_snapshot_status.assert_not_called()
self.volume_sdk_client.update_snapshot.assert_not_called()
self.volume_sdk_client.set_snapshot_metadata.assert_not_called()
self.volume_sdk_client.delete_snapshot_metadata.assert_called_with(
self.snapshot.id, keys=["foo"]
)
def test_snapshot_set_with_no_property_and_property(self): def test_snapshot_set_with_no_property_and_property(self):
arglist = [ arglist = [
@@ -608,22 +612,26 @@ class TestVolumeSnapshotSet(TestVolumeSnapshot):
] ]
verifylist = [ verifylist = [
("no_property", True), ("no_property", True),
("property", {"foo_1": "bar_1"}), ("properties", {"foo_1": "bar_1"}),
("snapshot", self.snapshot.id), ("snapshot", self.snapshot.id),
] ]
parsed_args = self.check_parser(self.cmd, arglist, verifylist) parsed_args = self.check_parser(self.cmd, arglist, verifylist)
result = self.cmd.take_action(parsed_args) 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) self.assertIsNone(result)
self.volume_sdk_client.find_snapshot.assert_called_once_with(
parsed_args.snapshot, ignore_missing=False
)
self.volume_sdk_client.reset_snapshot_status.assert_not_called()
self.volume_sdk_client.update_snapshot.assert_not_called()
self.volume_sdk_client.delete_snapshot_metadata.assert_called_with(
self.snapshot.id, keys=["foo"]
)
self.volume_sdk_client.set_snapshot_metadata.assert_called_once_with(
self.snapshot.id,
foo_1="bar_1",
)
def test_snapshot_set_state_to_error(self): def test_snapshot_set_state_to_error(self):
arglist = ["--state", "error", self.snapshot.id] arglist = ["--state", "error", self.snapshot.id]
@@ -632,30 +640,32 @@ class TestVolumeSnapshotSet(TestVolumeSnapshot):
result = self.cmd.take_action(parsed_args) result = self.cmd.take_action(parsed_args)
self.snapshots_mock.reset_state.assert_called_with( self.assertIsNone(result)
self.volume_sdk_client.reset_snapshot_status.assert_called_with(
self.snapshot.id, "error" self.snapshot.id, "error"
) )
self.assertIsNone(result)
def test_volume_set_state_failed(self): def test_volume_set_state_failed(self):
self.snapshots_mock.reset_state.side_effect = exceptions.CommandError() self.volume_sdk_client.reset_snapshot_status.side_effect = (
exceptions.CommandError()
)
arglist = ['--state', 'error', self.snapshot.id] arglist = ['--state', 'error', self.snapshot.id]
verifylist = [('state', 'error'), ('snapshot', self.snapshot.id)] verifylist = [('state', 'error'), ('snapshot', self.snapshot.id)]
parsed_args = self.check_parser(self.cmd, arglist, verifylist) parsed_args = self.check_parser(self.cmd, arglist, verifylist)
try:
self.cmd.take_action(parsed_args) exc = self.assertRaises(
self.fail('CommandError should be raised.') exceptions.CommandError, self.cmd.take_action, parsed_args
except exceptions.CommandError as e: )
self.assertEqual( self.assertEqual('One or more of the set operations failed', str(exc))
'One or more of the set operations failed', str(e) self.volume_sdk_client.reset_snapshot_status.assert_called_once_with(
)
self.snapshots_mock.reset_state.assert_called_once_with(
self.snapshot.id, 'error' self.snapshot.id, 'error'
) )
def test_volume_set_name_and_state_failed(self): def test_volume_set_name_and_state_failed(self):
self.snapshots_mock.reset_state.side_effect = exceptions.CommandError() self.volume_sdk_client.reset_snapshot_status.side_effect = (
exceptions.CommandError()
)
arglist = [ arglist = [
'--state', '--state',
'error', 'error',
@@ -668,22 +678,19 @@ class TestVolumeSnapshotSet(TestVolumeSnapshot):
("name", "new_snapshot"), ("name", "new_snapshot"),
('snapshot', self.snapshot.id), ('snapshot', self.snapshot.id),
] ]
parsed_args = self.check_parser(self.cmd, arglist, verifylist) parsed_args = self.check_parser(self.cmd, arglist, verifylist)
try:
self.cmd.take_action(parsed_args) exc = self.assertRaises(
self.fail('CommandError should be raised.') exceptions.CommandError,
except exceptions.CommandError as e: self.cmd.take_action,
self.assertEqual( parsed_args,
'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.assertEqual('One or more of the set operations failed', str(exc))
self.volume_sdk_client.update_snapshot.assert_called_once_with(
self.snapshot.id, name="new_snapshot"
)
self.volume_sdk_client.reset_snapshot_status.assert_called_once_with(
self.snapshot.id, 'error' self.snapshot.id, 'error'
) )
@@ -732,15 +739,14 @@ class TestVolumeSnapshotShow(TestVolumeSnapshot):
self.assertCountEqual(self.data, data) self.assertCountEqual(self.data, data)
class TestVolumeSnapshotUnset(TestVolumeSnapshot): class TestVolumeSnapshotUnset(volume_fakes.TestVolume):
snapshot = volume_fakes.create_one_snapshot()
def setUp(self): def setUp(self):
super().setUp() super().setUp()
self.snapshots_mock.get.return_value = self.snapshot self.snapshot = sdk_fakes.generate_fake_resource(_snapshot.Snapshot)
self.snapshots_mock.delete_metadata.return_value = None self.volume_sdk_client.find_snapshot.return_value = self.snapshot
# Get the command object to mock self.volume_sdk_client.delete_snapshot_metadata.return_value = None
self.cmd = volume_snapshot.UnsetVolumeSnapshot(self.app, None) self.cmd = volume_snapshot.UnsetVolumeSnapshot(self.app, None)
def test_snapshot_unset(self): def test_snapshot_unset(self):
@@ -750,7 +756,7 @@ class TestVolumeSnapshotUnset(TestVolumeSnapshot):
self.snapshot.id, self.snapshot.id,
] ]
verifylist = [ verifylist = [
("property", ["foo"]), ("properties", ["foo"]),
("snapshot", self.snapshot.id), ("snapshot", self.snapshot.id),
] ]
@@ -758,7 +764,7 @@ class TestVolumeSnapshotUnset(TestVolumeSnapshot):
result = self.cmd.take_action(parsed_args) result = self.cmd.take_action(parsed_args)
self.snapshots_mock.delete_metadata.assert_called_with(
self.snapshot.id, ["foo"]
)
self.assertIsNone(result) self.assertIsNone(result)
self.volume_sdk_client.delete_snapshot_metadata.assert_called_with(
self.snapshot.id, keys=["foo"]
)

158
openstackclient/tests/unit/volume/v3/test_volume_snapshot.py
View File

@@ -566,15 +566,16 @@ class TestVolumeSnapshotList(TestVolumeSnapshot):
) )
class TestVolumeSnapshotSet(TestVolumeSnapshot): class TestVolumeSnapshotSet(volume_fakes_v3.TestVolume):
snapshot = volume_fakes.create_one_snapshot()
def setUp(self): def setUp(self):
super().setUp() super().setUp()
self.snapshots_mock.get.return_value = self.snapshot self.snapshot = sdk_fakes.generate_fake_resource(
self.snapshots_mock.set_metadata.return_value = None _snapshot.Snapshot, metadata={'foo': 'bar'}
self.snapshots_mock.update.return_value = None )
self.volume_sdk_client.find_snapshot.return_value = self.snapshot
self.volume_sdk_client.set_snapshot_metadata.return_value = None
self.volume_sdk_client.update_snapshot.return_value = None
# Get the command object to mock # Get the command object to mock
self.cmd = volume_snapshot.SetVolumeSnapshot(self.app, None) self.cmd = volume_snapshot.SetVolumeSnapshot(self.app, None)
@@ -588,11 +589,14 @@ class TestVolumeSnapshotSet(TestVolumeSnapshot):
parsed_args = self.check_parser(self.cmd, arglist, verifylist) parsed_args = self.check_parser(self.cmd, arglist, verifylist)
result = self.cmd.take_action(parsed_args) 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) self.assertIsNone(result)
self.volume_sdk_client.find_snapshot.assert_called_once_with(
parsed_args.snapshot, ignore_missing=False
)
self.volume_sdk_client.reset_snapshot_status.assert_not_called()
self.volume_sdk_client.update_snapshot.assert_not_called()
self.volume_sdk_client.set_snapshot_metadata.assert_not_called()
def test_snapshot_set_name_and_property(self): def test_snapshot_set_name_and_property(self):
arglist = [ arglist = [
@@ -604,26 +608,22 @@ class TestVolumeSnapshotSet(TestVolumeSnapshot):
"foo=foo", "foo=foo",
self.snapshot.id, self.snapshot.id,
] ]
new_property = {"x": "y", "foo": "foo"}
verifylist = [ verifylist = [
("name", "new_snapshot"), ("name", "new_snapshot"),
("property", new_property), ("properties", {"x": "y", "foo": "foo"}),
("snapshot", self.snapshot.id), ("snapshot", self.snapshot.id),
] ]
parsed_args = self.check_parser(self.cmd, arglist, verifylist) parsed_args = self.check_parser(self.cmd, arglist, verifylist)
result = self.cmd.take_action(parsed_args) 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) self.assertIsNone(result)
self.volume_sdk_client.update_snapshot.assert_called_with(
self.snapshot.id, name="new_snapshot"
)
self.volume_sdk_client.set_snapshot_metadata.assert_called_with(
self.snapshot.id, x="y", foo="foo"
)
def test_snapshot_set_with_no_property(self): def test_snapshot_set_with_no_property(self):
arglist = [ arglist = [
@@ -637,14 +637,17 @@ class TestVolumeSnapshotSet(TestVolumeSnapshot):
parsed_args = self.check_parser(self.cmd, arglist, verifylist) parsed_args = self.check_parser(self.cmd, arglist, verifylist)
result = self.cmd.take_action(parsed_args) 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) self.assertIsNone(result)
self.volume_sdk_client.find_snapshot.assert_called_once_with(
parsed_args.snapshot, ignore_missing=False
)
self.volume_sdk_client.reset_snapshot_status.assert_not_called()
self.volume_sdk_client.update_snapshot.assert_not_called()
self.volume_sdk_client.set_snapshot_metadata.assert_not_called()
self.volume_sdk_client.delete_snapshot_metadata.assert_called_with(
self.snapshot.id, keys=["foo"]
)
def test_snapshot_set_with_no_property_and_property(self): def test_snapshot_set_with_no_property_and_property(self):
arglist = [ arglist = [
@@ -655,22 +658,26 @@ class TestVolumeSnapshotSet(TestVolumeSnapshot):
] ]
verifylist = [ verifylist = [
("no_property", True), ("no_property", True),
("property", {"foo_1": "bar_1"}), ("properties", {"foo_1": "bar_1"}),
("snapshot", self.snapshot.id), ("snapshot", self.snapshot.id),
] ]
parsed_args = self.check_parser(self.cmd, arglist, verifylist) parsed_args = self.check_parser(self.cmd, arglist, verifylist)
result = self.cmd.take_action(parsed_args) 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) self.assertIsNone(result)
self.volume_sdk_client.find_snapshot.assert_called_once_with(
parsed_args.snapshot, ignore_missing=False
)
self.volume_sdk_client.reset_snapshot_status.assert_not_called()
self.volume_sdk_client.update_snapshot.assert_not_called()
self.volume_sdk_client.delete_snapshot_metadata.assert_called_with(
self.snapshot.id, keys=["foo"]
)
self.volume_sdk_client.set_snapshot_metadata.assert_called_once_with(
self.snapshot.id,
foo_1="bar_1",
)
def test_snapshot_set_state_to_error(self): def test_snapshot_set_state_to_error(self):
arglist = ["--state", "error", self.snapshot.id] arglist = ["--state", "error", self.snapshot.id]
@@ -679,30 +686,34 @@ class TestVolumeSnapshotSet(TestVolumeSnapshot):
result = self.cmd.take_action(parsed_args) result = self.cmd.take_action(parsed_args)
self.snapshots_mock.reset_state.assert_called_with( self.assertIsNone(result)
self.volume_sdk_client.reset_snapshot_status.assert_called_with(
self.snapshot.id, "error" self.snapshot.id, "error"
) )
self.assertIsNone(result)
def test_volume_set_state_failed(self): def test_volume_set_state_failed(self):
self.snapshots_mock.reset_state.side_effect = exceptions.CommandError() self.volume_sdk_client.reset_snapshot_status.side_effect = (
exceptions.CommandError()
)
arglist = ['--state', 'error', self.snapshot.id] arglist = ['--state', 'error', self.snapshot.id]
verifylist = [('state', 'error'), ('snapshot', self.snapshot.id)] verifylist = [('state', 'error'), ('snapshot', self.snapshot.id)]
parsed_args = self.check_parser(self.cmd, arglist, verifylist) parsed_args = self.check_parser(self.cmd, arglist, verifylist)
try:
self.cmd.take_action(parsed_args) exc = self.assertRaises(
self.fail('CommandError should be raised.') exceptions.CommandError,
except exceptions.CommandError as e: self.cmd.take_action,
self.assertEqual( parsed_args,
'One or more of the set operations failed', str(e) )
)
self.snapshots_mock.reset_state.assert_called_once_with( self.assertEqual('One or more of the set operations failed', str(exc))
self.volume_sdk_client.reset_snapshot_status.assert_called_once_with(
self.snapshot.id, 'error' self.snapshot.id, 'error'
) )
def test_volume_set_name_and_state_failed(self): def test_volume_set_name_and_state_failed(self):
self.snapshots_mock.reset_state.side_effect = exceptions.CommandError() self.volume_sdk_client.reset_snapshot_status.side_effect = (
exceptions.CommandError()
)
arglist = [ arglist = [
'--state', '--state',
'error', 'error',
@@ -715,22 +726,19 @@ class TestVolumeSnapshotSet(TestVolumeSnapshot):
("name", "new_snapshot"), ("name", "new_snapshot"),
('snapshot', self.snapshot.id), ('snapshot', self.snapshot.id),
] ]
parsed_args = self.check_parser(self.cmd, arglist, verifylist) parsed_args = self.check_parser(self.cmd, arglist, verifylist)
try:
self.cmd.take_action(parsed_args) exc = self.assertRaises(
self.fail('CommandError should be raised.') exceptions.CommandError,
except exceptions.CommandError as e: self.cmd.take_action,
self.assertEqual( parsed_args,
'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.assertEqual('One or more of the set operations failed', str(exc))
self.volume_sdk_client.update_snapshot.assert_called_once_with(
self.snapshot.id, name="new_snapshot"
)
self.volume_sdk_client.reset_snapshot_status.assert_called_once_with(
self.snapshot.id, 'error' self.snapshot.id, 'error'
) )
@@ -779,15 +787,14 @@ class TestVolumeSnapshotShow(TestVolumeSnapshot):
self.assertCountEqual(self.data, data) self.assertCountEqual(self.data, data)
class TestVolumeSnapshotUnset(TestVolumeSnapshot): class TestVolumeSnapshotUnset(volume_fakes_v3.TestVolume):
snapshot = volume_fakes.create_one_snapshot()
def setUp(self): def setUp(self):
super().setUp() super().setUp()
self.snapshots_mock.get.return_value = self.snapshot self.snapshot = sdk_fakes.generate_fake_resource(_snapshot.Snapshot)
self.snapshots_mock.delete_metadata.return_value = None self.volume_sdk_client.find_snapshot.return_value = self.snapshot
# Get the command object to mock self.volume_sdk_client.delete_snapshot_metadata.return_value = None
self.cmd = volume_snapshot.UnsetVolumeSnapshot(self.app, None) self.cmd = volume_snapshot.UnsetVolumeSnapshot(self.app, None)
def test_snapshot_unset(self): def test_snapshot_unset(self):
@@ -797,15 +804,14 @@ class TestVolumeSnapshotUnset(TestVolumeSnapshot):
self.snapshot.id, self.snapshot.id,
] ]
verifylist = [ verifylist = [
("property", ["foo"]), ("properties", ["foo"]),
("snapshot", self.snapshot.id), ("snapshot", self.snapshot.id),
] ]
parsed_args = self.check_parser(self.cmd, arglist, verifylist) parsed_args = self.check_parser(self.cmd, arglist, verifylist)
result = self.cmd.take_action(parsed_args) result = self.cmd.take_action(parsed_args)
self.snapshots_mock.delete_metadata.assert_called_with(
self.snapshot.id, ["foo"]
)
self.assertIsNone(result) self.assertIsNone(result)
self.volume_sdk_client.delete_snapshot_metadata.assert_called_with(
self.snapshot.id, keys=["foo"]
)

39
openstackclient/volume/v2/volume_snapshot.py
View File

@@ -411,6 +411,7 @@ class SetVolumeSnapshot(command.Command):
'--property', '--property',
metavar='<key=value>', metavar='<key=value>',
action=parseractions.KeyValueAction, action=parseractions.KeyValueAction,
dest='properties',
help=_( help=_(
'Property to add/change for this snapshot ' 'Property to add/change for this snapshot '
'(repeat option to set multiple properties)' '(repeat option to set multiple properties)'
@@ -437,27 +438,26 @@ class SetVolumeSnapshot(command.Command):
return parser return parser
def take_action(self, parsed_args): def take_action(self, parsed_args):
volume_client = self.app.client_manager.volume volume_client = self.app.client_manager.sdk_connection.volume
snapshot = utils.find_resource(
volume_client.volume_snapshots, parsed_args.snapshot snapshot = volume_client.find_snapshot(
parsed_args.snapshot, ignore_missing=False
) )
result = 0 result = 0
if parsed_args.no_property: if parsed_args.no_property:
try: try:
key_list = snapshot.metadata.keys() volume_client.delete_snapshot_metadata(
volume_client.volume_snapshots.delete_metadata( snapshot.id, keys=list(snapshot.metadata)
snapshot.id,
list(key_list),
) )
except Exception as e: except Exception as e:
LOG.error(_("Failed to clean snapshot properties: %s"), e) LOG.error(_("Failed to clean snapshot properties: %s"), e)
result += 1 result += 1
if parsed_args.property: if parsed_args.properties:
try: try:
volume_client.volume_snapshots.set_metadata( volume_client.set_snapshot_metadata(
snapshot.id, parsed_args.property snapshot.id, **parsed_args.properties
) )
except Exception as e: except Exception as e:
LOG.error(_("Failed to set snapshot property: %s"), e) LOG.error(_("Failed to set snapshot property: %s"), e)
@@ -465,7 +465,7 @@ class SetVolumeSnapshot(command.Command):
if parsed_args.state: if parsed_args.state:
try: try:
volume_client.volume_snapshots.reset_state( volume_client.reset_snapshot_status(
snapshot.id, parsed_args.state snapshot.id, parsed_args.state
) )
except Exception as e: except Exception as e:
@@ -479,7 +479,7 @@ class SetVolumeSnapshot(command.Command):
kwargs['description'] = parsed_args.description kwargs['description'] = parsed_args.description
if kwargs: if kwargs:
try: try:
volume_client.volume_snapshots.update(snapshot.id, **kwargs) volume_client.update_snapshot(snapshot.id, **kwargs)
except Exception as e: except Exception as e:
LOG.error( LOG.error(
_("Failed to update snapshot name or description: %s"), _("Failed to update snapshot name or description: %s"),
@@ -535,6 +535,7 @@ class UnsetVolumeSnapshot(command.Command):
metavar='<key>', metavar='<key>',
action='append', action='append',
default=[], default=[],
dest='properties',
help=_( help=_(
'Property to remove from snapshot ' 'Property to remove from snapshot '
'(repeat option to remove multiple properties)' '(repeat option to remove multiple properties)'
@@ -543,13 +544,13 @@ class UnsetVolumeSnapshot(command.Command):
return parser return parser
def take_action(self, parsed_args): def take_action(self, parsed_args):
volume_client = self.app.client_manager.volume volume_client = self.app.client_manager.sdk_connection.volume
snapshot = utils.find_resource(
volume_client.volume_snapshots, parsed_args.snapshot snapshot = volume_client.find_snapshot(
parsed_args.snapshot, ignore_missing=False
) )
if parsed_args.property: if parsed_args.properties:
volume_client.volume_snapshots.delete_metadata( volume_client.delete_snapshot_metadata(
snapshot.id, snapshot.id, keys=parsed_args.properties
parsed_args.property,
) )

39
openstackclient/volume/v3/volume_snapshot.py
View File

@@ -429,6 +429,7 @@ class SetVolumeSnapshot(command.Command):
'--property', '--property',
metavar='<key=value>', metavar='<key=value>',
action=parseractions.KeyValueAction, action=parseractions.KeyValueAction,
dest='properties',
help=_( help=_(
'Property to add/change for this snapshot ' 'Property to add/change for this snapshot '
'(repeat option to set multiple properties)' '(repeat option to set multiple properties)'
@@ -455,27 +456,26 @@ class SetVolumeSnapshot(command.Command):
return parser return parser
def take_action(self, parsed_args): def take_action(self, parsed_args):
volume_client = self.app.client_manager.volume volume_client = self.app.client_manager.sdk_connection.volume
snapshot = utils.find_resource(
volume_client.volume_snapshots, parsed_args.snapshot snapshot = volume_client.find_snapshot(
parsed_args.snapshot, ignore_missing=False
) )
result = 0 result = 0
if parsed_args.no_property: if parsed_args.no_property:
try: try:
key_list = snapshot.metadata.keys() volume_client.delete_snapshot_metadata(
volume_client.volume_snapshots.delete_metadata( snapshot.id, keys=list(snapshot.metadata)
snapshot.id,
list(key_list),
) )
except Exception as e: except Exception as e:
LOG.error(_("Failed to clean snapshot properties: %s"), e) LOG.error(_("Failed to clean snapshot properties: %s"), e)
result += 1 result += 1
if parsed_args.property: if parsed_args.properties:
try: try:
volume_client.volume_snapshots.set_metadata( volume_client.set_snapshot_metadata(
snapshot.id, parsed_args.property snapshot.id, **parsed_args.properties
) )
except Exception as e: except Exception as e:
LOG.error(_("Failed to set snapshot property: %s"), e) LOG.error(_("Failed to set snapshot property: %s"), e)
@@ -483,7 +483,7 @@ class SetVolumeSnapshot(command.Command):
if parsed_args.state: if parsed_args.state:
try: try:
volume_client.volume_snapshots.reset_state( volume_client.reset_snapshot_status(
snapshot.id, parsed_args.state snapshot.id, parsed_args.state
) )
except Exception as e: except Exception as e:
@@ -497,7 +497,7 @@ class SetVolumeSnapshot(command.Command):
kwargs['description'] = parsed_args.description kwargs['description'] = parsed_args.description
if kwargs: if kwargs:
try: try:
volume_client.volume_snapshots.update(snapshot.id, **kwargs) volume_client.update_snapshot(snapshot.id, **kwargs)
except Exception as e: except Exception as e:
LOG.error( LOG.error(
_("Failed to update snapshot name or description: %s"), _("Failed to update snapshot name or description: %s"),
@@ -551,6 +551,7 @@ class UnsetVolumeSnapshot(command.Command):
parser.add_argument( parser.add_argument(
'--property', '--property',
metavar='<key>', metavar='<key>',
dest='properties',
action='append', action='append',
default=[], default=[],
help=_( help=_(
@@ -561,13 +562,13 @@ class UnsetVolumeSnapshot(command.Command):
return parser return parser
def take_action(self, parsed_args): def take_action(self, parsed_args):
volume_client = self.app.client_manager.volume volume_client = self.app.client_manager.sdk_connection.volume
snapshot = utils.find_resource(
volume_client.volume_snapshots, parsed_args.snapshot snapshot = volume_client.find_snapshot(
parsed_args.snapshot, ignore_missing=False
) )
if parsed_args.property: if parsed_args.properties:
volume_client.volume_snapshots.delete_metadata( volume_client.delete_snapshot_metadata(
snapshot.id, snapshot.id, keys=parsed_args.properties
parsed_args.property,
) )