openstack/python-openstackclient
Code Issues Proposed changes

volume: Add v3-specific volume backup module

Browse Source 
This makes testing easier.

Change-Id: I71a13b34a85350af17612e12c03e6df8cb041f86
Signed-off-by: Stephen Finucane <stephenfin@redhat.com>
This commit is contained in:
Stephen Finucane 2024-05-07 15:00:23 +01:00
parent a6dcb773f1
commit ca91c826e3
7 changed files with 1696 additions and 600 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

29
openstackclient/tests/unit/volume/v2/fakes.py
View File

@ -20,7 +20,7 @@ import uuid
# FIXME(stephenfin): We are using v3 resource versions despite being v2 fakes
from cinderclient import api_versions
from openstack.block_storage.v2 import _proxy as block_storage_v2_proxy
from openstack.block_storage.v3 import backup as _backup
from openstack.block_storage.v2 import backup as _backup
from openstack.block_storage.v3 import capabilities as _capabilities
from openstack.block_storage.v3 import stats as _stats
from openstack.block_storage.v3 import volume as _volume
@ -514,34 +514,29 @@ def create_one_backup(attrs=None):
:param dict attrs:
A dictionary with all attributes
:return:
A FakeResource object with id, name, volume_id, etc.
:return: A fake
openstack.block_storage.v2.backup.Backup object
"""
attrs = attrs or {}
# Set default attributes.
backup_info = {
"availability_zone": 'zone' + uuid.uuid4().hex,
"container": 'container-' + uuid.uuid4().hex,
"created_at": 'time-' + uuid.uuid4().hex,
"data_timestamp": 'time-' + uuid.uuid4().hex,
"id": 'backup-id-' + uuid.uuid4().hex,
"encryption_key_id": None,
"description": 'description-' + uuid.uuid4().hex,
"fail_reason": "Service not found for creating backup.",
"has_dependent_backups": False,
"id": 'backup-id-' + uuid.uuid4().hex,
"is_incremental": False,
"metadata": {},
"project_id": uuid.uuid4().hex,
"updated_at": 'time-' + uuid.uuid4().hex,
"user_id": uuid.uuid4().hex,
"name": 'backup-name-' + uuid.uuid4().hex,
"volume_id": 'volume-id-' + uuid.uuid4().hex,
"snapshot_id": 'snapshot-id' + uuid.uuid4().hex,
"description": 'description-' + uuid.uuid4().hex,
"object_count": None,
"container": 'container-' + uuid.uuid4().hex,
"size": random.randint(1, 20),
"is_incremental": False,
"snapshot_id": 'snapshot-id' + uuid.uuid4().hex,
"status": "error",
"availability_zone": 'zone' + uuid.uuid4().hex,
"updated_at": 'time-' + uuid.uuid4().hex,
"volume_id": 'volume-id-' + uuid.uuid4().hex,
}
# Overwrite default attributes.
@ -558,8 +553,8 @@ def create_backups(attrs=None, count=2):
A dictionary with all attributes
:param int count:
The number of backups to fake
:return:
A list of FakeResource objects faking the backups
:return: A list of fake
openstack.block_storage.v2.backup.Backup objects
"""
backups = []
for i in range(0, count):

371
openstackclient/tests/unit/volume/v2/test_volume_backup.py
View File

@ -10,13 +10,9 @@
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
#
from unittest import mock
from unittest.mock import call
from cinderclient import api_versions
from openstack import utils as sdk_utils
from osc_lib import exceptions
from openstackclient.tests.unit.volume.v2 import fakes as volume_fakes
@ -37,29 +33,7 @@ class TestBackupLegacy(volume_fakes.TestVolume):
self.restores_mock.reset_mock()
class TestBackup(volume_fakes.TestVolume):
def setUp(self):
super().setUp()
patcher = mock.patch.object(
sdk_utils, 'supports_microversion', return_value=True
)
self.addCleanup(patcher.stop)
self.supports_microversion_mock = patcher.start()
self._set_mock_microversion(
self.app.client_manager.volume.api_version.get_string()
)
def _set_mock_microversion(self, mock_v):
"""Set a specific microversion for the mock supports_microversion()."""
self.supports_microversion_mock.reset_mock(return_value=True)
self.supports_microversion_mock.side_effect = (
lambda _, v: api_versions.APIVersion(v)
<= api_versions.APIVersion(mock_v)
)
class TestBackupCreate(TestBackup):
class TestBackupCreate(volume_fakes.TestVolume):
volume = volume_fakes.create_one_volume()
snapshot = volume_fakes.create_one_snapshot()
new_backup = volume_fakes.create_one_backup(
@ -126,104 +100,6 @@ class TestBackupCreate(TestBackup):
self.assertEqual(self.columns, columns)
self.assertEqual(self.data, data)
def test_backup_create_with_properties(self):
self._set_mock_microversion('3.43')
arglist = [
"--property",
"foo=bar",
"--property",
"wow=much-cool",
self.new_backup.volume_id,
]
verifylist = [
("properties", {"foo": "bar", "wow": "much-cool"}),
("volume", self.new_backup.volume_id),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
columns, data = self.cmd.take_action(parsed_args)
self.volume_sdk_client.create_backup.assert_called_with(
volume_id=self.new_backup.volume_id,
container=None,
name=None,
description=None,
force=False,
is_incremental=False,
metadata={"foo": "bar", "wow": "much-cool"},
)
self.assertEqual(self.columns, columns)
self.assertEqual(self.data, data)
def test_backup_create_with_properties_pre_v343(self):
self._set_mock_microversion('3.42')
arglist = [
"--property",
"foo=bar",
"--property",
"wow=much-cool",
self.new_backup.volume_id,
]
verifylist = [
("properties", {"foo": "bar", "wow": "much-cool"}),
("volume", self.new_backup.volume_id),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
exc = self.assertRaises(
exceptions.CommandError, self.cmd.take_action, parsed_args
)
self.assertIn("--os-volume-api-version 3.43 or greater", str(exc))
def test_backup_create_with_availability_zone(self):
self._set_mock_microversion('3.51')
arglist = [
"--availability-zone",
"my-az",
self.new_backup.volume_id,
]
verifylist = [
("availability_zone", "my-az"),
("volume", self.new_backup.volume_id),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
columns, data = self.cmd.take_action(parsed_args)
self.volume_sdk_client.create_backup.assert_called_with(
volume_id=self.new_backup.volume_id,
container=None,
name=None,
description=None,
force=False,
is_incremental=False,
availability_zone="my-az",
)
self.assertEqual(self.columns, columns)
self.assertEqual(self.data, data)
def test_backup_create_with_availability_zone_pre_v351(self):
self._set_mock_microversion('3.50')
arglist = [
"--availability-zone",
"my-az",
self.new_backup.volume_id,
]
verifylist = [
("availability_zone", "my-az"),
("volume", self.new_backup.volume_id),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
exc = self.assertRaises(
exceptions.CommandError, self.cmd.take_action, parsed_args
)
self.assertIn("--os-volume-api-version 3.51 or greater", str(exc))
def test_backup_create_without_name(self):
arglist = [
"--description",
@ -253,7 +129,7 @@ class TestBackupCreate(TestBackup):
self.assertEqual(self.data, data)
class TestBackupDelete(TestBackup):
class TestBackupDelete(volume_fakes.TestVolume):
backups = volume_fakes.create_backups(count=2)
def setUp(self):
@ -346,7 +222,7 @@ class TestBackupDelete(TestBackup):
)
class TestBackupList(TestBackup):
class TestBackupList(volume_fakes.TestVolume):
volume = volume_fakes.create_one_volume()
backups = volume_fakes.create_backups(
attrs={'volume_id': volume.name}, count=3
@ -479,7 +355,7 @@ class TestBackupList(TestBackup):
self.assertCountEqual(self.data_long, list(data))
class TestBackupRestore(TestBackup):
class TestBackupRestore(volume_fakes.TestVolume):
volume = volume_fakes.create_one_volume()
backup = volume_fakes.create_one_backup(
attrs={'volume_id': volume.id},
@ -592,91 +468,6 @@ class TestBackupSet(TestBackupLegacy):
# Get the command object to test
self.cmd = volume_backup.SetVolumeBackup(self.app, None)
def test_backup_set_name(self):
self.volume_client.api_version = api_versions.APIVersion('3.9')
arglist = [
'--name',
'new_name',
self.backup.id,
]
verifylist = [
('name', 'new_name'),
('backup', self.backup.id),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
# In base command class ShowOne in cliff, abstract method take_action()
# returns nothing
result = self.cmd.take_action(parsed_args)
self.backups_mock.update.assert_called_once_with(
self.backup.id, **{'name': 'new_name'}
)
self.assertIsNone(result)
def test_backup_set_name_pre_v39(self):
self.volume_client.api_version = api_versions.APIVersion('3.8')
arglist = [
'--name',
'new_name',
self.backup.id,
]
verifylist = [
('name', 'new_name'),
('backup', self.backup.id),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
exc = self.assertRaises(
exceptions.CommandError, self.cmd.take_action, parsed_args
)
self.assertIn("--os-volume-api-version 3.9 or greater", str(exc))
def test_backup_set_description(self):
self.volume_client.api_version = api_versions.APIVersion('3.9')
arglist = [
'--description',
'new_description',
self.backup.id,
]
verifylist = [
('name', None),
('description', 'new_description'),
('backup', self.backup.id),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
result = self.cmd.take_action(parsed_args)
# Set expected values
kwargs = {'description': 'new_description'}
self.backups_mock.update.assert_called_once_with(
self.backup.id, **kwargs
)
self.assertIsNone(result)
def test_backup_set_description_pre_v39(self):
self.volume_client.api_version = api_versions.APIVersion('3.8')
arglist = [
'--description',
'new_description',
self.backup.id,
]
verifylist = [
('name', None),
('description', 'new_description'),
('backup', self.backup.id),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
exc = self.assertRaises(
exceptions.CommandError, self.cmd.take_action, parsed_args
)
self.assertIn("--os-volume-api-version 3.9 or greater", str(exc))
def test_backup_set_state(self):
arglist = ['--state', 'error', self.backup.id]
verifylist = [('state', 'error'), ('backup', self.backup.id)]
@ -706,152 +497,8 @@ class TestBackupSet(TestBackupLegacy):
self.backup.id, 'error'
)
def test_backup_set_no_property(self):
self.volume_client.api_version = api_versions.APIVersion('3.43')
arglist = [
'--no-property',
self.backup.id,
]
verifylist = [
('no_property', True),
('backup', self.backup.id),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
result = self.cmd.take_action(parsed_args)
# Set expected values
kwargs = {
'metadata': {},
}
self.backups_mock.update.assert_called_once_with(
self.backup.id, **kwargs
)
self.assertIsNone(result)
def test_backup_set_no_property_pre_v343(self):
self.volume_client.api_version = api_versions.APIVersion('3.42')
arglist = [
'--no-property',
self.backup.id,
]
verifylist = [
('no_property', True),
('backup', self.backup.id),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
exc = self.assertRaises(
exceptions.CommandError, self.cmd.take_action, parsed_args
)
self.assertIn("--os-volume-api-version 3.43 or greater", str(exc))
def test_backup_set_property(self):
self.volume_client.api_version = api_versions.APIVersion('3.43')
arglist = [
'--property',
'foo=bar',
self.backup.id,
]
verifylist = [
('properties', {'foo': 'bar'}),
('backup', self.backup.id),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
result = self.cmd.take_action(parsed_args)
# Set expected values
kwargs = {
'metadata': {'wow': 'cool', 'foo': 'bar'},
}
self.backups_mock.update.assert_called_once_with(
self.backup.id, **kwargs
)
self.assertIsNone(result)
def test_backup_set_property_pre_v343(self):
self.volume_client.api_version = api_versions.APIVersion('3.42')
arglist = [
'--property',
'foo=bar',
self.backup.id,
]
verifylist = [
('properties', {'foo': 'bar'}),
('backup', self.backup.id),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
exc = self.assertRaises(
exceptions.CommandError, self.cmd.take_action, parsed_args
)
self.assertIn("--os-volume-api-version 3.43 or greater", str(exc))
class TestBackupUnset(TestBackupLegacy):
backup = volume_fakes.create_one_backup(
attrs={'metadata': {'foo': 'bar'}},
)
def setUp(self):
super().setUp()
self.backups_mock.get.return_value = self.backup
# Get the command object to test
self.cmd = volume_backup.UnsetVolumeBackup(self.app, None)
def test_backup_unset_property(self):
self.volume_client.api_version = api_versions.APIVersion('3.43')
arglist = [
'--property',
'foo',
self.backup.id,
]
verifylist = [
('properties', ['foo']),
('backup', self.backup.id),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
result = self.cmd.take_action(parsed_args)
# Set expected values
kwargs = {
'metadata': {},
}
self.backups_mock.update.assert_called_once_with(
self.backup.id, **kwargs
)
self.assertIsNone(result)
def test_backup_unset_property_pre_v343(self):
self.volume_client.api_version = api_versions.APIVersion('3.42')
arglist = [
'--property',
'foo',
self.backup.id,
]
verifylist = [
('properties', ['foo']),
('backup', self.backup.id),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
exc = self.assertRaises(
exceptions.CommandError, self.cmd.take_action, parsed_args
)
self.assertIn("--os-volume-api-version 3.43 or greater", str(exc))
class TestBackupShow(TestBackup):
class TestBackupShow(volume_fakes.TestVolume):
backup = volume_fakes.create_one_backup()
columns = (
@ -860,20 +507,16 @@ class TestBackupShow(TestBackup):
"created_at",
"data_timestamp",
"description",
"encryption_key_id",
"fail_reason",
"has_dependent_backups",
"id",
"is_incremental",
"metadata",
"name",
"object_count",
"project_id",
"size",
"snapshot_id",
"status",
"updated_at",
"user_id",
"volume_id",
)
data = (
@ -882,20 +525,16 @@ class TestBackupShow(TestBackup):
backup.created_at,
backup.data_timestamp,
backup.description,
backup.encryption_key_id,
backup.fail_reason,
backup.has_dependent_backups,
backup.id,
backup.is_incremental,
backup.metadata,
backup.name,
backup.object_count,
backup.project_id,
backup.size,
backup.snapshot_id,
backup.status,
backup.updated_at,
backup.user_id,
backup.volume_id,
)

84
openstackclient/tests/unit/volume/v3/fakes.py
View File

@ -17,6 +17,7 @@ import uuid
from cinderclient import api_versions
from openstack.block_storage.v3 import _proxy
from openstack.block_storage.v3 import availability_zone as _availability_zone
from openstack.block_storage.v3 import backup as _backup
from openstack.block_storage.v3 import extension as _extension
from openstack.block_storage.v3 import resource_filter as _filters
from openstack.block_storage.v3 import volume as _volume
@ -35,6 +36,8 @@ class FakeVolumeClient:
self.availability_zones = mock.Mock()
self.availability_zones.resource_class = fakes.FakeResource(None, {})
self.backups = mock.Mock()
self.backups.resource_class = fakes.FakeResource(None, {})
self.attachments = mock.Mock()
self.attachments.resource_class = fakes.FakeResource(None, {})
self.clusters = mock.Mock()
@ -53,6 +56,8 @@ class FakeVolumeClient:
self.quotas.resource_class = fakes.FakeResource(None, {})
self.resource_filters = mock.Mock()
self.resource_filters.resource_class = fakes.FakeResource(None, {})
self.restores = mock.Mock()
self.restores.resource_class = fakes.FakeResource(None, {})
self.volumes = mock.Mock()
self.volumes.resource_class = fakes.FakeResource(None, {})
self.volume_snapshots = mock.Mock()
@ -179,6 +184,85 @@ def create_one_extension(attrs=None):
return extension
def create_one_backup(attrs=None):
"""Create a fake backup.
:param dict attrs:
A dictionary with all attributes
:return: A fake
openstack.block_storage.v3.backup.Backup object
"""
attrs = attrs or {}
# Set default attributes.
backup_info = {
"availability_zone": 'zone' + uuid.uuid4().hex,
"container": 'container-' + uuid.uuid4().hex,
"created_at": 'time-' + uuid.uuid4().hex,
"data_timestamp": 'time-' + uuid.uuid4().hex,
"description": 'description-' + uuid.uuid4().hex,
"encryption_key_id": None,
"fail_reason": "Service not found for creating backup.",
"has_dependent_backups": False,
"id": 'backup-id-' + uuid.uuid4().hex,
"is_incremental": False,
"metadata": {},
"name": 'backup-name-' + uuid.uuid4().hex,
"object_count": None,
"project_id": uuid.uuid4().hex,
"size": random.randint(1, 20),
"snapshot_id": 'snapshot-id' + uuid.uuid4().hex,
"status": "error",
"updated_at": 'time-' + uuid.uuid4().hex,
"user_id": uuid.uuid4().hex,
"volume_id": 'volume-id-' + uuid.uuid4().hex,
}
# Overwrite default attributes.
backup_info.update(attrs)
backup = _backup.Backup(**backup_info)
return backup
def create_backups(attrs=None, count=2):
"""Create multiple fake backups.
:param dict attrs:
A dictionary with all attributes
:param int count:
The number of backups to fake
:return: A list of fake
openstack.block_storage.v3.backup.Backup objects
"""
backups = []
for i in range(0, count):
backup = create_one_backup(attrs)
backups.append(backup)
return backups
def get_backups(backups=None, count=2):
"""Get an iterable MagicMock object with a list of faked backups.
If backups list is provided, then initialize the Mock object with the
list. Otherwise create one.
:param List backups:
A list of FakeResource objects faking backups
:param Integer count:
The number of backups to be faked
:return
An iterable Mock object with side_effect set to a list of faked
backups
"""
if backups is None:
backups = create_backups(count)
return mock.Mock(side_effect=backups)
def create_one_cluster(attrs=None):
"""Create a fake service cluster.

918
openstackclient/tests/unit/volume/v3/test_volume_backup.py Normal file
View File

@ -0,0 +1,918 @@
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
#
from unittest import mock
from unittest.mock import call
from cinderclient import api_versions
from openstack import utils as sdk_utils
from osc_lib import exceptions
from openstackclient.tests.unit.volume.v3 import fakes as volume_fakes
from openstackclient.volume.v3 import volume_backup
class TestBackupLegacy(volume_fakes.TestVolume):
def setUp(self):
super().setUp()
self.backups_mock = self.volume_client.backups
self.backups_mock.reset_mock()
self.volumes_mock = self.volume_client.volumes
self.volumes_mock.reset_mock()
self.snapshots_mock = self.volume_client.volume_snapshots
self.snapshots_mock.reset_mock()
self.restores_mock = self.volume_client.restores
self.restores_mock.reset_mock()
class TestBackup(volume_fakes.TestVolume):
def setUp(self):
super().setUp()
patcher = mock.patch.object(
sdk_utils, 'supports_microversion', return_value=True
)
self.addCleanup(patcher.stop)
self.supports_microversion_mock = patcher.start()
self._set_mock_microversion(
self.app.client_manager.volume.api_version.get_string()
)
def _set_mock_microversion(self, mock_v):
"""Set a specific microversion for the mock supports_microversion()."""
self.supports_microversion_mock.reset_mock(return_value=True)
self.supports_microversion_mock.side_effect = (
lambda _, v: api_versions.APIVersion(v)
<= api_versions.APIVersion(mock_v)
)
class TestBackupCreate(TestBackup):
volume = volume_fakes.create_one_volume()
snapshot = volume_fakes.create_one_snapshot()
new_backup = volume_fakes.create_one_backup(
attrs={'volume_id': volume.id, 'snapshot_id': snapshot.id}
)
columns = (
'id',
'name',
'volume_id',
)
data = (
new_backup.id,
new_backup.name,
new_backup.volume_id,
)
def setUp(self):
super().setUp()
self.volume_sdk_client.find_volume.return_value = self.volume
self.volume_sdk_client.find_snapshot.return_value = self.snapshot
self.volume_sdk_client.create_backup.return_value = self.new_backup
# Get the command object to test
self.cmd = volume_backup.CreateVolumeBackup(self.app, None)
def test_backup_create(self):
arglist = [
"--name",
self.new_backup.name,
"--description",
self.new_backup.description,
"--container",
self.new_backup.container,
"--force",
"--incremental",
"--snapshot",
self.new_backup.snapshot_id,
self.new_backup.volume_id,
]
verifylist = [
("name", self.new_backup.name),
("description", self.new_backup.description),
("container", self.new_backup.container),
("force", True),
("incremental", True),
("snapshot", self.new_backup.snapshot_id),
("volume", self.new_backup.volume_id),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
columns, data = self.cmd.take_action(parsed_args)
self.volume_sdk_client.create_backup.assert_called_with(
volume_id=self.new_backup.volume_id,
container=self.new_backup.container,
name=self.new_backup.name,
description=self.new_backup.description,
force=True,
is_incremental=True,
snapshot_id=self.new_backup.snapshot_id,
)
self.assertEqual(self.columns, columns)
self.assertEqual(self.data, data)
def test_backup_create_with_properties(self):
self._set_mock_microversion('3.43')
arglist = [
"--property",
"foo=bar",
"--property",
"wow=much-cool",
self.new_backup.volume_id,
]
verifylist = [
("properties", {"foo": "bar", "wow": "much-cool"}),
("volume", self.new_backup.volume_id),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
columns, data = self.cmd.take_action(parsed_args)
self.volume_sdk_client.create_backup.assert_called_with(
volume_id=self.new_backup.volume_id,
container=None,
name=None,
description=None,
force=False,
is_incremental=False,
metadata={"foo": "bar", "wow": "much-cool"},
)
self.assertEqual(self.columns, columns)
self.assertEqual(self.data, data)
def test_backup_create_with_properties_pre_v343(self):
self._set_mock_microversion('3.42')
arglist = [
"--property",
"foo=bar",
"--property",
"wow=much-cool",
self.new_backup.volume_id,
]
verifylist = [
("properties", {"foo": "bar", "wow": "much-cool"}),
("volume", self.new_backup.volume_id),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
exc = self.assertRaises(
exceptions.CommandError, self.cmd.take_action, parsed_args
)
self.assertIn("--os-volume-api-version 3.43 or greater", str(exc))
def test_backup_create_with_availability_zone(self):
self._set_mock_microversion('3.51')
arglist = [
"--availability-zone",
"my-az",
self.new_backup.volume_id,
]
verifylist = [
("availability_zone", "my-az"),
("volume", self.new_backup.volume_id),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
columns, data = self.cmd.take_action(parsed_args)
self.volume_sdk_client.create_backup.assert_called_with(
volume_id=self.new_backup.volume_id,
container=None,
name=None,
description=None,
force=False,
is_incremental=False,
availability_zone="my-az",
)
self.assertEqual(self.columns, columns)
self.assertEqual(self.data, data)
def test_backup_create_with_availability_zone_pre_v351(self):
self._set_mock_microversion('3.50')
arglist = [
"--availability-zone",
"my-az",
self.new_backup.volume_id,
]
verifylist = [
("availability_zone", "my-az"),
("volume", self.new_backup.volume_id),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
exc = self.assertRaises(
exceptions.CommandError, self.cmd.take_action, parsed_args
)
self.assertIn("--os-volume-api-version 3.51 or greater", str(exc))
def test_backup_create_without_name(self):
arglist = [
"--description",
self.new_backup.description,
"--container",
self.new_backup.container,
self.new_backup.volume_id,
]
verifylist = [
("description", self.new_backup.description),
("container", self.new_backup.container),
("volume", self.new_backup.volume_id),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
columns, data = self.cmd.take_action(parsed_args)
self.volume_sdk_client.create_backup.assert_called_with(
volume_id=self.new_backup.volume_id,
container=self.new_backup.container,
name=None,
description=self.new_backup.description,
force=False,
is_incremental=False,
)
self.assertEqual(self.columns, columns)
self.assertEqual(self.data, data)
class TestBackupDelete(TestBackup):
backups = volume_fakes.create_backups(count=2)
def setUp(self):
super().setUp()
self.volume_sdk_client.find_backup = volume_fakes.get_backups(
self.backups
)
self.volume_sdk_client.delete_backup.return_value = None
# Get the command object to mock
self.cmd = volume_backup.DeleteVolumeBackup(self.app, None)
def test_backup_delete(self):
arglist = [self.backups[0].id]
verifylist = [("backups", [self.backups[0].id])]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
result = self.cmd.take_action(parsed_args)
self.volume_sdk_client.delete_backup.assert_called_with(
self.backups[0].id, ignore_missing=False, force=False
)
self.assertIsNone(result)
def test_backup_delete_with_force(self):
arglist = [
'--force',
self.backups[0].id,
]
verifylist = [('force', True), ("backups", [self.backups[0].id])]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
result = self.cmd.take_action(parsed_args)
self.volume_sdk_client.delete_backup.assert_called_with(
self.backups[0].id, ignore_missing=False, force=True
)
self.assertIsNone(result)
def test_delete_multiple_backups(self):
arglist = []
for b in self.backups:
arglist.append(b.id)
verifylist = [
('backups', arglist),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
result = self.cmd.take_action(parsed_args)
calls = []
for b in self.backups:
calls.append(call(b.id, ignore_missing=False, force=False))
self.volume_sdk_client.delete_backup.assert_has_calls(calls)
self.assertIsNone(result)
def test_delete_multiple_backups_with_exception(self):
arglist = [
self.backups[0].id,
'unexist_backup',
]
verifylist = [
('backups', arglist),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
find_mock_result = [self.backups[0], exceptions.CommandError]
self.volume_sdk_client.find_backup.side_effect = find_mock_result
try:
self.cmd.take_action(parsed_args)
self.fail('CommandError should be raised.')
except exceptions.CommandError as e:
self.assertEqual('1 of 2 backups failed to delete.', str(e))
self.volume_sdk_client.find_backup.assert_any_call(
self.backups[0].id, ignore_missing=False
)
self.volume_sdk_client.find_backup.assert_any_call(
'unexist_backup', ignore_missing=False
)
self.assertEqual(2, self.volume_sdk_client.find_backup.call_count)
self.volume_sdk_client.delete_backup.assert_called_once_with(
self.backups[0].id,
ignore_missing=False,
force=False,
)
class TestBackupList(TestBackup):
volume = volume_fakes.create_one_volume()
backups = volume_fakes.create_backups(
attrs={'volume_id': volume.name}, count=3
)
columns = (
'ID',
'Name',
'Description',
'Status',
'Size',
'Incremental',
)
columns_long = columns + (
'Availability Zone',
'Volume',
'Container',
)
data = []
for b in backups:
data.append(
(
b.id,
b.name,
b.description,
b.status,
b.size,
b.is_incremental,
)
)
data_long = []
for b in backups:
data_long.append(
(
b.id,
b.name,
b.description,
b.status,
b.size,
b.is_incremental,
b.availability_zone,
volume_backup.VolumeIdColumn(b.volume_id),
b.container,
)
)
def setUp(self):
super().setUp()
self.volume_sdk_client.volumes.return_value = [self.volume]
self.volume_sdk_client.backups.return_value = self.backups
self.volume_sdk_client.find_volume.return_value = self.volume
self.volume_sdk_client.find_backup.return_value = self.backups[0]
# Get the command to test
self.cmd = volume_backup.ListVolumeBackup(self.app, None)
def test_backup_list_without_options(self):
arglist = []
verifylist = [
("long", False),
("name", None),
("status", None),
("volume", None),
("marker", None),
("limit", None),
('all_projects', False),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
columns, data = self.cmd.take_action(parsed_args)
self.volume_sdk_client.find_volume.assert_not_called()
self.volume_sdk_client.find_backup.assert_not_called()
self.volume_sdk_client.backups.assert_called_with(
name=None,
status=None,
volume_id=None,
all_tenants=False,
marker=None,
limit=None,
)
self.assertEqual(self.columns, columns)
self.assertCountEqual(self.data, list(data))
def test_backup_list_with_options(self):
arglist = [
"--long",
"--name",
self.backups[0].name,
"--status",
"error",
"--volume",
self.volume.id,
"--marker",
self.backups[0].id,
"--all-projects",
"--limit",
"3",
]
verifylist = [
("long", True),
("name", self.backups[0].name),
("status", "error"),
("volume", self.volume.id),
("marker", self.backups[0].id),
('all_projects', True),
("limit", 3),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
columns, data = self.cmd.take_action(parsed_args)
self.volume_sdk_client.find_volume.assert_called_once_with(
self.volume.id, ignore_missing=False
)
self.volume_sdk_client.find_backup.assert_called_once_with(
self.backups[0].id, ignore_missing=False
)
self.volume_sdk_client.backups.assert_called_with(
name=self.backups[0].name,
status="error",
volume_id=self.volume.id,
all_tenants=True,
marker=self.backups[0].id,
limit=3,
)
self.assertEqual(self.columns_long, columns)
self.assertCountEqual(self.data_long, list(data))
class TestBackupRestore(TestBackup):
volume = volume_fakes.create_one_volume()
backup = volume_fakes.create_one_backup(
attrs={'volume_id': volume.id},
)
def setUp(self):
super().setUp()
self.volume_sdk_client.find_backup.return_value = self.backup
self.volume_sdk_client.find_volume.return_value = self.volume
self.volume_sdk_client.restore_backup.return_value = (
volume_fakes.create_one_volume(
{'id': self.volume['id']},
)
)
# Get the command object to mock
self.cmd = volume_backup.RestoreVolumeBackup(self.app, None)
def test_backup_restore(self):
self.volume_sdk_client.find_volume.side_effect = (
exceptions.CommandError()
)
arglist = [self.backup.id]
verifylist = [
("backup", self.backup.id),
("volume", None),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
result = self.cmd.take_action(parsed_args)
self.volume_sdk_client.restore_backup.assert_called_with(
self.backup.id,
volume_id=None,
name=None,
)
self.assertIsNotNone(result)
def test_backup_restore_with_volume(self):
self.volume_sdk_client.find_volume.side_effect = (
exceptions.CommandError()
)
arglist = [
self.backup.id,
self.backup.volume_id,
]
verifylist = [
("backup", self.backup.id),
("volume", self.backup.volume_id),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
result = self.cmd.take_action(parsed_args)
self.volume_sdk_client.restore_backup.assert_called_with(
self.backup.id,
volume_id=None,
name=self.backup.volume_id,
)
self.assertIsNotNone(result)
def test_backup_restore_with_volume_force(self):
arglist = [
"--force",
self.backup.id,
self.volume.name,
]
verifylist = [
("force", True),
("backup", self.backup.id),
("volume", self.volume.name),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
result = self.cmd.take_action(parsed_args)
self.volume_sdk_client.restore_backup.assert_called_with(
self.backup.id,
volume_id=self.volume.id,
name=None,
)
self.assertIsNotNone(result)
def test_backup_restore_with_volume_existing(self):
arglist = [
self.backup.id,
self.volume.name,
]
verifylist = [
("backup", self.backup.id),
("volume", self.volume.name),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
self.assertRaises(
exceptions.CommandError,
self.cmd.take_action,
parsed_args,
)
class TestBackupSet(TestBackupLegacy):
backup = volume_fakes.create_one_backup(
attrs={'metadata': {'wow': 'cool'}},
)
def setUp(self):
super().setUp()
self.backups_mock.get.return_value = self.backup
# Get the command object to test
self.cmd = volume_backup.SetVolumeBackup(self.app, None)
def test_backup_set_name(self):
self.volume_client.api_version = api_versions.APIVersion('3.9')
arglist = [
'--name',
'new_name',
self.backup.id,
]
verifylist = [
('name', 'new_name'),
('backup', self.backup.id),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
# In base command class ShowOne in cliff, abstract method take_action()
# returns nothing
result = self.cmd.take_action(parsed_args)
self.backups_mock.update.assert_called_once_with(
self.backup.id, **{'name': 'new_name'}
)
self.assertIsNone(result)
def test_backup_set_name_pre_v39(self):
self.volume_client.api_version = api_versions.APIVersion('3.8')
arglist = [
'--name',
'new_name',
self.backup.id,
]
verifylist = [
('name', 'new_name'),
('backup', self.backup.id),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
exc = self.assertRaises(
exceptions.CommandError, self.cmd.take_action, parsed_args
)
self.assertIn("--os-volume-api-version 3.9 or greater", str(exc))
def test_backup_set_description(self):
self.volume_client.api_version = api_versions.APIVersion('3.9')
arglist = [
'--description',
'new_description',
self.backup.id,
]
verifylist = [
('name', None),
('description', 'new_description'),
('backup', self.backup.id),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
result = self.cmd.take_action(parsed_args)
# Set expected values
kwargs = {'description': 'new_description'}
self.backups_mock.update.assert_called_once_with(
self.backup.id, **kwargs
)
self.assertIsNone(result)
def test_backup_set_description_pre_v39(self):
self.volume_client.api_version = api_versions.APIVersion('3.8')
arglist = [
'--description',
'new_description',
self.backup.id,
]
verifylist = [
('name', None),
('description', 'new_description'),
('backup', self.backup.id),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
exc = self.assertRaises(
exceptions.CommandError, self.cmd.take_action, parsed_args
)
self.assertIn("--os-volume-api-version 3.9 or greater", str(exc))
def test_backup_set_state(self):
arglist = ['--state', 'error', self.backup.id]
verifylist = [('state', 'error'), ('backup', self.backup.id)]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
result = self.cmd.take_action(parsed_args)
self.backups_mock.reset_state.assert_called_once_with(
self.backup.id, 'error'
)
self.assertIsNone(result)
def test_backup_set_state_failed(self):
self.backups_mock.reset_state.side_effect = exceptions.CommandError()
arglist = ['--state', 'error', self.backup.id]
verifylist = [('state', 'error'), ('backup', self.backup.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.backups_mock.reset_state.assert_called_with(
self.backup.id, 'error'
)
def test_backup_set_no_property(self):
self.volume_client.api_version = api_versions.APIVersion('3.43')
arglist = [
'--no-property',
self.backup.id,
]
verifylist = [
('no_property', True),
('backup', self.backup.id),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
result = self.cmd.take_action(parsed_args)
# Set expected values
kwargs = {
'metadata': {},
}
self.backups_mock.update.assert_called_once_with(
self.backup.id, **kwargs
)
self.assertIsNone(result)
def test_backup_set_no_property_pre_v343(self):
self.volume_client.api_version = api_versions.APIVersion('3.42')
arglist = [
'--no-property',
self.backup.id,
]
verifylist = [
('no_property', True),
('backup', self.backup.id),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
exc = self.assertRaises(
exceptions.CommandError, self.cmd.take_action, parsed_args
)
self.assertIn("--os-volume-api-version 3.43 or greater", str(exc))
def test_backup_set_property(self):
self.volume_client.api_version = api_versions.APIVersion('3.43')
arglist = [
'--property',
'foo=bar',
self.backup.id,
]
verifylist = [
('properties', {'foo': 'bar'}),
('backup', self.backup.id),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
result = self.cmd.take_action(parsed_args)
# Set expected values
kwargs = {
'metadata': {'wow': 'cool', 'foo': 'bar'},
}
self.backups_mock.update.assert_called_once_with(
self.backup.id, **kwargs
)
self.assertIsNone(result)
def test_backup_set_property_pre_v343(self):
self.volume_client.api_version = api_versions.APIVersion('3.42')
arglist = [
'--property',
'foo=bar',
self.backup.id,
]
verifylist = [
('properties', {'foo': 'bar'}),
('backup', self.backup.id),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
exc = self.assertRaises(
exceptions.CommandError, self.cmd.take_action, parsed_args
)
self.assertIn("--os-volume-api-version 3.43 or greater", str(exc))
class TestBackupUnset(TestBackupLegacy):
backup = volume_fakes.create_one_backup(
attrs={'metadata': {'foo': 'bar'}},
)
def setUp(self):
super().setUp()
self.backups_mock.get.return_value = self.backup
# Get the command object to test
self.cmd = volume_backup.UnsetVolumeBackup(self.app, None)
def test_backup_unset_property(self):
self.volume_client.api_version = api_versions.APIVersion('3.43')
arglist = [
'--property',
'foo',
self.backup.id,
]
verifylist = [
('properties', ['foo']),
('backup', self.backup.id),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
result = self.cmd.take_action(parsed_args)
# Set expected values
kwargs = {
'metadata': {},
}
self.backups_mock.update.assert_called_once_with(
self.backup.id, **kwargs
)
self.assertIsNone(result)
def test_backup_unset_property_pre_v343(self):
self.volume_client.api_version = api_versions.APIVersion('3.42')
arglist = [
'--property',
'foo',
self.backup.id,
]
verifylist = [
('properties', ['foo']),
('backup', self.backup.id),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
exc = self.assertRaises(
exceptions.CommandError, self.cmd.take_action, parsed_args
)
self.assertIn("--os-volume-api-version 3.43 or greater", str(exc))
class TestBackupShow(TestBackup):
backup = volume_fakes.create_one_backup()
columns = (
"availability_zone",
"container",
"created_at",
"data_timestamp",
"description",
"encryption_key_id",
"fail_reason",
"has_dependent_backups",
"id",
"is_incremental",
"metadata",
"name",
"object_count",
"project_id",
"size",
"snapshot_id",
"status",
"updated_at",
"user_id",
"volume_id",
)
data = (
backup.availability_zone,
backup.container,
backup.created_at,
backup.data_timestamp,
backup.description,
backup.encryption_key_id,
backup.fail_reason,
backup.has_dependent_backups,
backup.id,
backup.is_incremental,
backup.metadata,
backup.name,
backup.object_count,
backup.project_id,
backup.size,
backup.snapshot_id,
backup.status,
backup.updated_at,
backup.user_id,
backup.volume_id,
)
def setUp(self):
super().setUp()
self.volume_sdk_client.get_backup.return_value = self.backup
# Get the command object to test
self.cmd = volume_backup.ShowVolumeBackup(self.app, None)
def test_backup_show(self):
arglist = [self.backup.id]
verifylist = [("backup", self.backup.id)]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
columns, data = self.cmd.take_action(parsed_args)
self.volume_sdk_client.get_backup.assert_called_with(self.backup.id)
self.assertEqual(self.columns, columns)
self.assertEqual(self.data, data)

210
openstackclient/volume/v2/volume_backup.py
View File

@ -14,14 +14,10 @@
"""Volume v2 Backup action implementations"""
import copy
import functools
import logging
from cinderclient import api_versions
from cliff import columns as cliff_columns
from openstack import utils as sdk_utils
from osc_lib.cli import parseractions
from osc_lib.command import command
from osc_lib import exceptions
from osc_lib import utils
@ -104,26 +100,6 @@ class CreateVolumeBackup(command.ShowOne):
action='store_false',
help=_("Do not perform an incremental backup"),
)
parser.add_argument(
'--property',
metavar='<key=value>',
action=parseractions.KeyValueAction,
dest='properties',
help=_(
'Set a property on this backup '
'(repeat option to remove multiple values) '
'(supported by --os-volume-api-version 3.43 or above)'
),
)
parser.add_argument(
'--availability-zone',
metavar='<zone-name>',
help=_(
'AZ where the backup should be stored; by default it will be '
'the same as the source '
'(supported by --os-volume-api-version 3.51 or above)'
),
)
return parser
def take_action(self, parsed_args):
@ -142,26 +118,6 @@ class CreateVolumeBackup(command.ShowOne):
ignore_missing=False,
).id
if parsed_args.properties:
if not sdk_utils.supports_microversion(volume_client, '3.43'):
msg = _(
'--os-volume-api-version 3.43 or greater is required to '
'support the --property option'
)
raise exceptions.CommandError(msg)
kwargs['metadata'] = parsed_args.properties
if parsed_args.availability_zone:
if not sdk_utils.supports_microversion(volume_client, '3.51'):
msg = _(
'--os-volume-api-version 3.51 or greater is required to '
'support the --availability-zone option'
)
raise exceptions.CommandError(msg)
kwargs['availability_zone'] = parsed_args.availability_zone
columns = (
"id",
"name",
@ -279,20 +235,6 @@ class ListVolumeBackup(command.Lister):
default=False,
help=_('Include all projects (admin only)'),
)
# TODO(stephenfin): Add once we have an equivalent command for
# 'cinder list-filters'
# parser.add_argument(
# '--filter',
# metavar='<key=value>',
# action=parseractions.KeyValueAction,
# dest='filters',
# help=_(
# "Filter key and value pairs. Use 'foo' to "
# "check enabled filters from server. Use 'key~=value' for "
# "inexact filtering if the key supports "
# "(supported by --os-volume-api-version 3.33 or above)"
# ),
# )
return parser
def take_action(self, parsed_args):
@ -451,22 +393,6 @@ class SetVolumeBackup(command.Command):
metavar="<backup>",
help=_("Backup to modify (name or ID)"),
)
parser.add_argument(
'--name',
metavar='<name>',
help=_(
'New backup name'
'(supported by --os-volume-api-version 3.9 or above)'
),
)
parser.add_argument(
'--description',
metavar='<description>',
help=_(
'New backup description '
'(supported by --os-volume-api-version 3.9 or above)'
),
)
parser.add_argument(
'--state',
metavar='<state>',
@ -478,27 +404,6 @@ class SetVolumeBackup(command.Command):
'exercise caution when using)'
),
)
parser.add_argument(
'--no-property',
action='store_true',
help=_(
'Remove all properties from this backup '
'(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,
dest='properties',
default={},
help=_(
'Set a property on this backup '
'(repeat option to set multiple values) '
'(supported by --os-volume-api-version 3.43 or above)'
),
)
return parser
def take_action(self, parsed_args):
@ -513,122 +418,11 @@ class SetVolumeBackup(command.Command):
LOG.error(_("Failed to set backup state: %s"), e)
result += 1
kwargs = {}
if parsed_args.name:
if volume_client.api_version < api_versions.APIVersion('3.9'):
msg = _(
'--os-volume-api-version 3.9 or greater is required to '
'support the --name option'
)
raise exceptions.CommandError(msg)
kwargs['name'] = parsed_args.name
if parsed_args.description:
if volume_client.api_version < api_versions.APIVersion('3.9'):
msg = _(
'--os-volume-api-version 3.9 or greater is required to '
'support the --description option'
)
raise exceptions.CommandError(msg)
kwargs['description'] = parsed_args.description
if parsed_args.no_property:
if volume_client.api_version < api_versions.APIVersion('3.43'):
msg = _(
'--os-volume-api-version 3.43 or greater is required to '
'support the --no-property option'
)
raise exceptions.CommandError(msg)
if parsed_args.properties:
if volume_client.api_version < api_versions.APIVersion('3.43'):
msg = _(
'--os-volume-api-version 3.43 or greater is required to '
'support the --property option'
)
raise exceptions.CommandError(msg)
if volume_client.api_version >= api_versions.APIVersion('3.43'):
metadata = copy.deepcopy(backup.metadata)
if parsed_args.no_property:
metadata = {}
metadata.update(parsed_args.properties)
kwargs['metadata'] = metadata
if kwargs:
try:
volume_client.backups.update(backup.id, **kwargs)
except Exception as e:
LOG.error("Failed to update backup: %s", e)
result += 1
if result > 0:
msg = _("One or more of the set operations failed")
raise exceptions.CommandError(msg)
class UnsetVolumeBackup(command.Command):
"""Unset volume backup properties.
This command requires ``--os-volume-api-version`` 3.43 or greater.
"""
def get_parser(self, prog_name):
parser = super().get_parser(prog_name)
parser.add_argument(
'backup',
metavar='<backup>',
help=_('Backup to modify (name or ID)'),
)
parser.add_argument(
'--property',
metavar='<key>',
action='append',
dest='properties',
help=_(
'Property to remove from this backup '
'(repeat option to unset multiple values) '
),
)
return parser
def take_action(self, parsed_args):
volume_client = self.app.client_manager.volume
if volume_client.api_version < api_versions.APIVersion('3.43'):
msg = _(
'--os-volume-api-version 3.43 or greater is required to '
'support the --property option'
)
raise exceptions.CommandError(msg)
backup = utils.find_resource(volume_client.backups, parsed_args.backup)
metadata = copy.deepcopy(backup.metadata)
for key in parsed_args.properties:
if key not in metadata:
# ignore invalid properties but continue
LOG.warning(
"'%s' is not a valid property for backup '%s'",
key,
parsed_args.backup,
)
continue
del metadata[key]
kwargs = {
'metadata': metadata,
}
volume_client.backups.update(backup.id, **kwargs)
class ShowVolumeBackup(command.ShowOne):
_description = _("Display volume backup details")
@ -650,20 +444,16 @@ class ShowVolumeBackup(command.ShowOne):
"created_at",
"data_timestamp",
"description",
"encryption_key_id",
"fail_reason",
"has_dependent_backups",
"id",
"is_incremental",
"metadata",
"name",
"object_count",
"project_id",
"size",
"snapshot_id",
"status",
"updated_at",
"user_id",
"volume_id",
)
data = utils.get_dict_properties(backup, columns)

670
openstackclient/volume/v3/volume_backup.py Normal file
View File

@ -0,0 +1,670 @@
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
#
"""Volume v3 Backup action implementations"""
import copy
import functools
import logging
from cinderclient import api_versions
from cliff import columns as cliff_columns
from openstack import utils as sdk_utils
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 _
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 CreateVolumeBackup(command.ShowOne):
_description = _("Create new volume backup")
def get_parser(self, prog_name):
parser = super().get_parser(prog_name)
parser.add_argument(
"volume",
metavar="<volume>",
help=_("Volume to backup (name or ID)"),
)
parser.add_argument(
"--name", metavar="<name>", help=_("Name of the backup")
)
parser.add_argument(
"--description",
metavar="<description>",
help=_("Description of the backup"),
)
parser.add_argument(
"--container",
metavar="<container>",
help=_("Optional backup container name"),
)
parser.add_argument(
"--snapshot",
metavar="<snapshot>",
help=_("Snapshot to backup (name or ID)"),
)
parser.add_argument(
'--force',
action='store_true',
default=False,
help=_("Allow to back up an in-use volume"),
)
parser.add_argument(
'--incremental',
action='store_true',
default=False,
help=_("Perform an incremental backup"),
)
parser.add_argument(
'--no-incremental',
action='store_false',
help=_("Do not perform an incremental backup"),
)
parser.add_argument(
'--property',
metavar='<key=value>',
action=parseractions.KeyValueAction,
dest='properties',
help=_(
'Set a property on this backup '
'(repeat option to remove multiple values) '
'(supported by --os-volume-api-version 3.43 or above)'
),
)
parser.add_argument(
'--availability-zone',
metavar='<zone-name>',
help=_(
'AZ where the backup should be stored; by default it will be '
'the same as the source '
'(supported by --os-volume-api-version 3.51 or above)'
),
)
return parser
def take_action(self, parsed_args):
volume_client = self.app.client_manager.sdk_connection.volume
volume_id = volume_client.find_volume(
parsed_args.volume,
ignore_missing=False,
).id
kwargs = {}
if parsed_args.snapshot:
kwargs['snapshot_id'] = volume_client.find_snapshot(
parsed_args.snapshot,
ignore_missing=False,
).id
if parsed_args.properties:
if not sdk_utils.supports_microversion(volume_client, '3.43'):
msg = _(
'--os-volume-api-version 3.43 or greater is required to '
'support the --property option'
)
raise exceptions.CommandError(msg)
kwargs['metadata'] = parsed_args.properties
if parsed_args.availability_zone:
if not sdk_utils.supports_microversion(volume_client, '3.51'):
msg = _(
'--os-volume-api-version 3.51 or greater is required to '
'support the --availability-zone option'
)
raise exceptions.CommandError(msg)
kwargs['availability_zone'] = parsed_args.availability_zone
columns = (
"id",
"name",
"volume_id",
)
backup = volume_client.create_backup(
volume_id=volume_id,
container=parsed_args.container,
name=parsed_args.name,
description=parsed_args.description,
force=parsed_args.force,
is_incremental=parsed_args.incremental,
**kwargs,
)
data = utils.get_dict_properties(backup, columns)
return (columns, data)
class DeleteVolumeBackup(command.Command):
_description = _("Delete volume backup(s)")
def get_parser(self, prog_name):
parser = super().get_parser(prog_name)
parser.add_argument(
"backups",
metavar="<backup>",
nargs="+",
help=_("Backup(s) to delete (name or ID)"),
)
parser.add_argument(
'--force',
action='store_true',
default=False,
help=_("Allow delete in state other than error or available"),
)
return parser
def take_action(self, parsed_args):
volume_client = self.app.client_manager.sdk_connection.volume
result = 0
for backup in parsed_args.backups:
try:
backup_id = volume_client.find_backup(
backup, ignore_missing=False
).id
volume_client.delete_backup(
backup_id,
ignore_missing=False,
force=parsed_args.force,
)
except Exception as e:
result += 1
LOG.error(
_(
"Failed to delete backup with "
"name or ID '%(backup)s': %(e)s"
)
% {'backup': backup, 'e': e}
)
if result > 0:
total = len(parsed_args.backups)
msg = _("%(result)s of %(total)s backups failed to delete.") % {
'result': result,
'total': total,
}
raise exceptions.CommandError(msg)
class ListVolumeBackup(command.Lister):
_description = _("List volume backups")
def get_parser(self, prog_name):
parser = super().get_parser(prog_name)
parser.add_argument(
"--long",
action="store_true",
default=False,
help=_("List additional fields in output"),
)
parser.add_argument(
"--name",
metavar="<name>",
help=_("Filters results by the backup name"),
)
parser.add_argument(
"--status",
metavar="<status>",
choices=[
'creating',
'available',
'deleting',
'error',
'restoring',
'error_restoring',
],
help=_(
"Filters results by the backup status, one of: "
"creating, available, deleting, error, restoring or "
"error_restoring"
),
)
parser.add_argument(
"--volume",
metavar="<volume>",
help=_(
"Filters results by the volume which they backup (name or ID)"
),
)
pagination.add_marker_pagination_option_to_parser(parser)
parser.add_argument(
'--all-projects',
action='store_true',
default=False,
help=_('Include all projects (admin only)'),
)
# TODO(stephenfin): Add once we have an equivalent command for
# 'cinder list-filters'
# parser.add_argument(
# '--filter',
# metavar='<key=value>',
# action=parseractions.KeyValueAction,
# dest='filters',
# help=_(
# "Filter key and value pairs. Use 'foo' to "
# "check enabled filters from server. Use 'key~=value' for "
# "inexact filtering if the key supports "
# "(supported by --os-volume-api-version 3.33 or above)"
# ),
# )
return parser
def take_action(self, parsed_args):
volume_client = self.app.client_manager.sdk_connection.volume
columns = (
'id',
'name',
'description',
'status',
'size',
'is_incremental',
)
column_headers = (
'ID',
'Name',
'Description',
'Status',
'Size',
'Incremental',
)
if parsed_args.long:
columns += ('availability_zone', 'volume_id', 'container')
column_headers += ('Availability Zone', 'Volume', 'Container')
# Cache the volume list
volume_cache = {}
try:
for s in volume_client.volumes():
volume_cache[s.id] = s
except Exception:
# Just forget it if there's any trouble
pass # nosec: B110
_VolumeIdColumn = functools.partial(
VolumeIdColumn, volume_cache=volume_cache
)
filter_volume_id = None
if parsed_args.volume:
try:
filter_volume_id = volume_client.find_volume(
parsed_args.volume,
ignore_missing=False,
).id
except exceptions.CommandError:
# Volume with that ID does not exist, but search for backups
# for that volume nevertheless
LOG.debug(
"No volume with ID %s existing, continuing to "
"search for backups for that volume ID",
parsed_args.volume,
)
filter_volume_id = parsed_args.volume
marker_backup_id = None
if parsed_args.marker:
marker_backup_id = volume_client.find_backup(
parsed_args.marker,
ignore_missing=False,
).id
data = volume_client.backups(
name=parsed_args.name,
status=parsed_args.status,
volume_id=filter_volume_id,
all_tenants=parsed_args.all_projects,
marker=marker_backup_id,
limit=parsed_args.limit,
)
return (
column_headers,
(
utils.get_item_properties(
s,
columns,
formatters={'volume_id': _VolumeIdColumn},
)
for s in data
),
)
class RestoreVolumeBackup(command.ShowOne):
_description = _("Restore volume backup")
def get_parser(self, prog_name):
parser = super().get_parser(prog_name)
parser.add_argument(
"backup",
metavar="<backup>",
help=_("Backup to restore (name or ID)"),
)
parser.add_argument(
"volume",
metavar="<volume>",
nargs="?",
help=_(
"Volume to restore to "
"(name or ID for existing volume, name only for new volume) "
"(default to None)"
),
)
parser.add_argument(
"--force",
action="store_true",
help=_(
"Restore the backup to an existing volume "
"(default to False)"
),
)
return parser
def take_action(self, parsed_args):
volume_client = self.app.client_manager.sdk_connection.volume
backup = volume_client.find_backup(
parsed_args.backup,
ignore_missing=False,
)
volume_name = None
volume_id = None
try:
volume_id = volume_client.find_volume(
parsed_args.volume,
ignore_missing=False,
).id
except Exception:
volume_name = parsed_args.volume
else:
# If we didn't fail, the volume must already exist. We only allow
# this to work if the user forced things
if not parsed_args.force:
msg = _(
"Volume '%s' already exists; if you want to restore the "
"backup to it you need to specify the '--force' option"
)
raise exceptions.CommandError(msg % parsed_args.volume)
return volume_client.restore_backup(
backup.id,
volume_id=volume_id,
name=volume_name,
)
class SetVolumeBackup(command.Command):
_description = _("Set volume backup properties")
def get_parser(self, prog_name):
parser = super().get_parser(prog_name)
parser.add_argument(
"backup",
metavar="<backup>",
help=_("Backup to modify (name or ID)"),
)
parser.add_argument(
'--name',
metavar='<name>',
help=_(
'New backup name'
'(supported by --os-volume-api-version 3.9 or above)'
),
)
parser.add_argument(
'--description',
metavar='<description>',
help=_(
'New backup description '
'(supported by --os-volume-api-version 3.9 or above)'
),
)
parser.add_argument(
'--state',
metavar='<state>',
choices=['available', 'error'],
help=_(
'New backup state ("available" or "error") (admin only) '
'(This option simply changes the state of the backup '
'in the database with no regard to actual status; '
'exercise caution when using)'
),
)
parser.add_argument(
'--no-property',
action='store_true',
help=_(
'Remove all properties from this backup '
'(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,
dest='properties',
default={},
help=_(
'Set a property on this backup '
'(repeat option to set multiple values) '
'(supported by --os-volume-api-version 3.43 or above)'
),
)
return parser
def take_action(self, parsed_args):
volume_client = self.app.client_manager.volume
backup = utils.find_resource(volume_client.backups, parsed_args.backup)
result = 0
if parsed_args.state:
try:
volume_client.backups.reset_state(backup.id, parsed_args.state)
except Exception as e:
LOG.error(_("Failed to set backup state: %s"), e)
result += 1
kwargs = {}
if parsed_args.name:
if volume_client.api_version < api_versions.APIVersion('3.9'):
msg = _(
'--os-volume-api-version 3.9 or greater is required to '
'support the --name option'
)
raise exceptions.CommandError(msg)
kwargs['name'] = parsed_args.name
if parsed_args.description:
if volume_client.api_version < api_versions.APIVersion('3.9'):
msg = _(
'--os-volume-api-version 3.9 or greater is required to '
'support the --description option'
)
raise exceptions.CommandError(msg)
kwargs['description'] = parsed_args.description
if parsed_args.no_property:
if volume_client.api_version < api_versions.APIVersion('3.43'):
msg = _(
'--os-volume-api-version 3.43 or greater is required to '
'support the --no-property option'
)
raise exceptions.CommandError(msg)
if parsed_args.properties:
if volume_client.api_version < api_versions.APIVersion('3.43'):
msg = _(
'--os-volume-api-version 3.43 or greater is required to '
'support the --property option'
)
raise exceptions.CommandError(msg)
if volume_client.api_version >= api_versions.APIVersion('3.43'):
metadata = copy.deepcopy(backup.metadata)
if parsed_args.no_property:
metadata = {}
metadata.update(parsed_args.properties)
kwargs['metadata'] = metadata
if kwargs:
try:
volume_client.backups.update(backup.id, **kwargs)
except Exception as e:
LOG.error("Failed to update backup: %s", e)
result += 1
if result > 0:
msg = _("One or more of the set operations failed")
raise exceptions.CommandError(msg)
class UnsetVolumeBackup(command.Command):
"""Unset volume backup properties.
This command requires ``--os-volume-api-version`` 3.43 or greater.
"""
def get_parser(self, prog_name):
parser = super().get_parser(prog_name)
parser.add_argument(
'backup',
metavar='<backup>',
help=_('Backup to modify (name or ID)'),
)
parser.add_argument(
'--property',
metavar='<key>',
action='append',
dest='properties',
help=_(
'Property to remove from this backup '
'(repeat option to unset multiple values) '
),
)
return parser
def take_action(self, parsed_args):
volume_client = self.app.client_manager.volume
if volume_client.api_version < api_versions.APIVersion('3.43'):
msg = _(
'--os-volume-api-version 3.43 or greater is required to '
'support the --property option'
)
raise exceptions.CommandError(msg)
backup = utils.find_resource(volume_client.backups, parsed_args.backup)
metadata = copy.deepcopy(backup.metadata)
for key in parsed_args.properties:
if key not in metadata:
# ignore invalid properties but continue
LOG.warning(
"'%s' is not a valid property for backup '%s'",
key,
parsed_args.backup,
)
continue
del metadata[key]
kwargs = {
'metadata': metadata,
}
volume_client.backups.update(backup.id, **kwargs)
class ShowVolumeBackup(command.ShowOne):
_description = _("Display volume backup details")
def get_parser(self, prog_name):
parser = super().get_parser(prog_name)
parser.add_argument(
"backup",
metavar="<backup>",
help=_("Backup to display (name or ID)"),
)
return parser
def take_action(self, parsed_args):
volume_client = self.app.client_manager.sdk_connection.volume
backup = volume_client.get_backup(parsed_args.backup)
columns = (
"availability_zone",
"container",
"created_at",
"data_timestamp",
"description",
"encryption_key_id",
"fail_reason",
"has_dependent_backups",
"id",
"is_incremental",
"metadata",
"name",
"object_count",
"project_id",
"size",
"snapshot_id",
"status",
"updated_at",
"user_id",
"volume_id",
)
data = utils.get_dict_properties(backup, columns)
return (columns, data)

14
setup.cfg
View File

@ -780,13 +780,13 @@ openstack.volume.v3 =
volume_attachment_set = openstackclient.volume.v3.volume_attachment:SetVolumeAttachment
volume_attachment_show = openstackclient.volume.v3.volume_attachment:ShowVolumeAttachment
volume_backup_create = openstackclient.volume.v2.volume_backup:CreateVolumeBackup
volume_backup_delete = openstackclient.volume.v2.volume_backup:DeleteVolumeBackup
volume_backup_list = openstackclient.volume.v2.volume_backup:ListVolumeBackup
volume_backup_restore = openstackclient.volume.v2.volume_backup:RestoreVolumeBackup
volume_backup_set = openstackclient.volume.v2.volume_backup:SetVolumeBackup
volume_backup_unset = openstackclient.volume.v2.volume_backup:UnsetVolumeBackup
volume_backup_show = openstackclient.volume.v2.volume_backup:ShowVolumeBackup
volume_backup_create = openstackclient.volume.v3.volume_backup:CreateVolumeBackup
volume_backup_delete = openstackclient.volume.v3.volume_backup:DeleteVolumeBackup
volume_backup_list = openstackclient.volume.v3.volume_backup:ListVolumeBackup
volume_backup_restore = openstackclient.volume.v3.volume_backup:RestoreVolumeBackup
volume_backup_set = openstackclient.volume.v3.volume_backup:SetVolumeBackup
volume_backup_unset = openstackclient.volume.v3.volume_backup:UnsetVolumeBackup
volume_backup_show = openstackclient.volume.v3.volume_backup:ShowVolumeBackup
volume_backend_capability_show = openstackclient.volume.v2.volume_backend:ShowCapability
volume_backend_pool_list = openstackclient.volume.v2.volume_backend:ListPool