volume: Add v3-specific volume snapshot module
Change-Id: I23026abbb909c082fbc0fe0c9b2efcc89f4d464a Signed-off-by: Stephen Finucane <stephenfin@redhat.com>
This commit is contained in:
Stephen Finucane
@ -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)
|
||||
|
||||
@ -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,
|
||||
)
|
||||
|
||||
10
setup.cfg
10
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
|
||||
|
||||
Reference in New Issue
Block a user