summaryrefslogtreecommitdiff
path: root/openstackclient/tests
diff options
context:
space:
mode:
authorZuul <zuul@review.opendev.org>2022-11-05 09:10:14 +0000
committerGerrit Code Review <review@openstack.org>2022-11-05 09:10:14 +0000
commitae5f20c1ab58a49d951b3df696b98c4f10488650 (patch)
tree23afabe7d3cdf8838f34b252b90edcd47fe9e3be /openstackclient/tests
parent495e7fa90770daf932570068c129d35f02cbe42b (diff)
parent1aaaa6f1d14ebc2498513aabe788239c9f867639 (diff)
downloadpython-openstackclient-ae5f20c1ab58a49d951b3df696b98c4f10488650.tar.gz
Merge "tests: Remove unnecessary nesting of volume resources"
Diffstat (limited to 'openstackclient/tests')
-rw-r--r--openstackclient/tests/unit/common/test_availability_zone.py7
-rw-r--r--openstackclient/tests/unit/common/test_extension.py8
-rw-r--r--openstackclient/tests/unit/common/test_limits.py4
-rw-r--r--openstackclient/tests/unit/common/test_project_purge.py10
-rw-r--r--openstackclient/tests/unit/common/test_quota.py23
-rw-r--r--openstackclient/tests/unit/compute/v2/test_server.py6
-rw-r--r--openstackclient/tests/unit/volume/v1/fakes.py1112
-rw-r--r--openstackclient/tests/unit/volume/v1/test_qos_specs.py44
-rw-r--r--openstackclient/tests/unit/volume/v1/test_service.py14
-rw-r--r--openstackclient/tests/unit/volume/v1/test_transfer_request.py47
-rw-r--r--openstackclient/tests/unit/volume/v1/test_type.py69
-rw-r--r--openstackclient/tests/unit/volume/v1/test_volume.py38
-rw-r--r--openstackclient/tests/unit/volume/v1/test_volume_backup.py41
-rw-r--r--openstackclient/tests/unit/volume/v2/fakes.py2208
-rw-r--r--openstackclient/tests/unit/volume/v2/test_backup_record.py20
-rw-r--r--openstackclient/tests/unit/volume/v2/test_consistency_group.py70
-rw-r--r--openstackclient/tests/unit/volume/v2/test_consistency_group_snapshot.py26
-rw-r--r--openstackclient/tests/unit/volume/v2/test_qos_specs.py26
-rw-r--r--openstackclient/tests/unit/volume/v2/test_service.py14
-rw-r--r--openstackclient/tests/unit/volume/v2/test_type.py94
-rw-r--r--openstackclient/tests/unit/volume/v2/test_volume.py56
-rw-r--r--openstackclient/tests/unit/volume/v2/test_volume_backend.py8
-rw-r--r--openstackclient/tests/unit/volume/v2/test_volume_backup.py38
-rw-r--r--openstackclient/tests/unit/volume/v2/test_volume_host.py14
-rw-r--r--openstackclient/tests/unit/volume/v2/test_volume_snapshot.py18
-rw-r--r--openstackclient/tests/unit/volume/v2/test_volume_transfer_request.py47
-rw-r--r--openstackclient/tests/unit/volume/v3/fakes.py780
-rw-r--r--openstackclient/tests/unit/volume/v3/test_block_storage_cluster.py6
-rw-r--r--openstackclient/tests/unit/volume/v3/test_block_storage_resource_filter.py6
-rw-r--r--openstackclient/tests/unit/volume/v3/test_volume_attachment.py20
-rw-r--r--openstackclient/tests/unit/volume/v3/test_volume_group.py19
-rw-r--r--openstackclient/tests/unit/volume/v3/test_volume_group_snapshot.py9
-rw-r--r--openstackclient/tests/unit/volume/v3/test_volume_group_type.py29
-rw-r--r--openstackclient/tests/unit/volume/v3/test_volume_message.py14
34 files changed, 2378 insertions, 2567 deletions
diff --git a/openstackclient/tests/unit/common/test_availability_zone.py b/openstackclient/tests/unit/common/test_availability_zone.py
index e5348ec3..096038ca 100644
--- a/openstackclient/tests/unit/common/test_availability_zone.py
+++ b/openstackclient/tests/unit/common/test_availability_zone.py
@@ -78,7 +78,7 @@ def _build_network_az_datalist(network_az, long_datalist=False):
class TestAvailabilityZone(utils.TestCommand):
def setUp(self):
- super(TestAvailabilityZone, self).setUp()
+ super().setUp()
compute_client = compute_fakes.FakeComputev2Client(
endpoint=fakes.AUTH_URL,
@@ -113,8 +113,7 @@ class TestAvailabilityZoneList(TestAvailabilityZone):
compute_azs = \
compute_fakes.FakeAvailabilityZone.create_availability_zones()
- volume_azs = \
- volume_fakes.FakeAvailabilityZone.create_availability_zones(count=1)
+ volume_azs = volume_fakes.create_availability_zones(count=1)
network_azs = network_fakes.create_availability_zones()
short_columnslist = ('Zone Name', 'Zone Status')
@@ -128,7 +127,7 @@ class TestAvailabilityZoneList(TestAvailabilityZone):
)
def setUp(self):
- super(TestAvailabilityZoneList, self).setUp()
+ super().setUp()
self.compute_azs_mock.list.return_value = self.compute_azs
self.volume_azs_mock.list.return_value = self.volume_azs
diff --git a/openstackclient/tests/unit/common/test_extension.py b/openstackclient/tests/unit/common/test_extension.py
index 5093cbbb..bd90b32d 100644
--- a/openstackclient/tests/unit/common/test_extension.py
+++ b/openstackclient/tests/unit/common/test_extension.py
@@ -26,7 +26,7 @@ from openstackclient.tests.unit.volume.v2 import fakes as volume_fakes
class TestExtension(utils.TestCommand):
def setUp(self):
- super(TestExtension, self).setUp()
+ super().setUp()
identity_client = identity_fakes.FakeIdentityv2Client(
endpoint=fakes.AUTH_URL,
@@ -66,13 +66,13 @@ class TestExtensionList(TestExtension):
long_columns = ('Name', 'Alias', 'Description', 'Namespace', 'Updated',
'Links')
- volume_extension = volume_fakes.FakeExtension.create_one_extension()
+ volume_extension = volume_fakes.create_one_extension()
identity_extension = identity_fakes.FakeExtension.create_one_extension()
compute_extension = compute_fakes.FakeExtension.create_one_extension()
network_extension = network_fakes.FakeExtension.create_one_extension()
def setUp(self):
- super(TestExtensionList, self).setUp()
+ super().setUp()
self.identity_extensions_mock.list.return_value = [
self.identity_extension]
@@ -310,7 +310,7 @@ class TestExtensionShow(TestExtension):
)
def setUp(self):
- super(TestExtensionShow, self).setUp()
+ super().setUp()
self.cmd = extension.ShowExtension(self.app, None)
diff --git a/openstackclient/tests/unit/common/test_limits.py b/openstackclient/tests/unit/common/test_limits.py
index d73db2cb..e3cdcf45 100644
--- a/openstackclient/tests/unit/common/test_limits.py
+++ b/openstackclient/tests/unit/common/test_limits.py
@@ -33,7 +33,7 @@ class TestComputeLimits(compute_fakes.TestComputev2):
]
def setUp(self):
- super(TestComputeLimits, self).setUp()
+ super().setUp()
self.app.client_manager.volume_endpoint_enabled = False
self.compute = self.app.client_manager.compute
@@ -87,7 +87,7 @@ class TestVolumeLimits(volume_fakes.TestVolume):
]
def setUp(self):
- super(TestVolumeLimits, self).setUp()
+ super().setUp()
self.app.client_manager.compute_endpoint_enabled = False
self.volume = self.app.client_manager.volume
diff --git a/openstackclient/tests/unit/common/test_project_purge.py b/openstackclient/tests/unit/common/test_project_purge.py
index 5199093c..26333d70 100644
--- a/openstackclient/tests/unit/common/test_project_purge.py
+++ b/openstackclient/tests/unit/common/test_project_purge.py
@@ -26,7 +26,7 @@ from openstackclient.tests.unit.volume.v2 import fakes as volume_fakes
class TestProjectPurgeInit(tests_utils.TestCommand):
def setUp(self):
- super(TestProjectPurgeInit, self).setUp()
+ super().setUp()
compute_client = compute_fakes.FakeComputev2Client(
endpoint=fakes.AUTH_URL,
token=fakes.AUTH_TOKEN,
@@ -71,12 +71,12 @@ class TestProjectPurge(TestProjectPurgeInit):
project = identity_fakes.FakeProject.create_one_project()
server = compute_fakes.FakeServer.create_one_server()
image = image_fakes.create_one_image()
- volume = volume_fakes.FakeVolume.create_one_volume()
- backup = volume_fakes.FakeBackup.create_one_backup()
- snapshot = volume_fakes.FakeSnapshot.create_one_snapshot()
+ volume = volume_fakes.create_one_volume()
+ backup = volume_fakes.create_one_backup()
+ snapshot = volume_fakes.create_one_snapshot()
def setUp(self):
- super(TestProjectPurge, self).setUp()
+ super().setUp()
self.projects_mock.get.return_value = self.project
self.projects_mock.delete.return_value = None
self.images_mock.list.return_value = [self.image]
diff --git a/openstackclient/tests/unit/common/test_quota.py b/openstackclient/tests/unit/common/test_quota.py
index 0900200e..682799b3 100644
--- a/openstackclient/tests/unit/common/test_quota.py
+++ b/openstackclient/tests/unit/common/test_quota.py
@@ -176,12 +176,12 @@ class TestQuotaList(TestQuota):
)
self.volume_quotas = [
- volume_fakes.FakeQuota.create_one_vol_quota(),
- volume_fakes.FakeQuota.create_one_vol_quota(),
+ volume_fakes.create_one_vol_quota(),
+ volume_fakes.create_one_vol_quota(),
]
self.volume_default_quotas = [
- volume_fakes.FakeQuota.create_one_default_vol_quota(),
- volume_fakes.FakeQuota.create_one_default_vol_quota(),
+ volume_fakes.create_one_default_vol_quota(),
+ volume_fakes.create_one_default_vol_quota(),
]
self.volume = self.app.client_manager.volume
self.volume.quotas.defaults = mock.Mock(
@@ -280,8 +280,7 @@ class TestQuotaList(TestQuota):
sorted(detailed_reference_data), sorted(ret_quotas))
def test_quota_list_details_volume(self):
- detailed_quota = (
- volume_fakes.FakeQuota.create_one_detailed_quota())
+ detailed_quota = volume_fakes.create_one_detailed_quota()
detailed_column_header = (
'Resource',
@@ -567,7 +566,7 @@ class TestQuotaList(TestQuota):
self.volume.quotas.get = mock.Mock(
side_effect=[
self.volume_quotas[0],
- volume_fakes.FakeQuota.create_one_default_vol_quota(),
+ volume_fakes.create_one_default_vol_quota(),
],
)
@@ -591,7 +590,7 @@ class TestQuotaList(TestQuota):
self.volume.quotas.get = mock.Mock(
side_effect=[
self.volume_quotas[0],
- volume_fakes.FakeQuota.create_one_default_vol_quota(),
+ volume_fakes.create_one_default_vol_quota(),
],
)
@@ -1046,10 +1045,9 @@ class TestQuotaShow(TestQuota):
loaded=True,
)
- self.volume_quota = volume_fakes.FakeQuota.create_one_vol_quota()
+ self.volume_quota = volume_fakes.create_one_vol_quota()
self.volume_quotas_mock.get.return_value = self.volume_quota
- self.volume_default_quota = \
- volume_fakes.FakeQuota.create_one_default_vol_quota()
+ self.volume_default_quota = volume_fakes.create_one_default_vol_quota()
self.volume_quotas_mock.defaults.return_value = \
self.volume_default_quota
self.volume_quotas_class_mock.get.return_value = FakeQuotaResource(
@@ -1217,8 +1215,7 @@ class TestQuotaShow(TestQuota):
self.compute_quota = \
compute_fakes.FakeQuota.create_one_comp_detailed_quota()
self.compute_quotas_mock.get.return_value = self.compute_quota
- self.volume_quota = \
- volume_fakes.FakeQuota.create_one_detailed_quota()
+ self.volume_quota = volume_fakes.create_one_detailed_quota()
self.volume_quotas_mock.get.return_value = self.volume_quota
self.network.get_quota.return_value = \
network_fakes.FakeQuota.create_one_net_detailed_quota()
diff --git a/openstackclient/tests/unit/compute/v2/test_server.py b/openstackclient/tests/unit/compute/v2/test_server.py
index 2e64e071..81c8a8a5 100644
--- a/openstackclient/tests/unit/compute/v2/test_server.py
+++ b/openstackclient/tests/unit/compute/v2/test_server.py
@@ -153,7 +153,7 @@ class TestServer(compute_fakes.TestComputev2):
return servers
def setup_sdk_volumes_mock(self, count):
- volumes = volume_fakes.FakeVolume.create_sdk_volumes(count=count)
+ volumes = volume_fakes.create_sdk_volumes(count=count)
# This is the return value for volume_client.find_volume()
self.sdk_volume_client.find_volume.side_effect = volumes
@@ -1444,10 +1444,10 @@ class TestServerCreate(TestServer):
self.flavor = compute_fakes.FakeFlavor.create_one_flavor()
self.flavors_mock.get.return_value = self.flavor
- self.volume = volume_fakes.FakeVolume.create_one_volume()
+ self.volume = volume_fakes.create_one_volume()
self.volumes_mock.get.return_value = self.volume
- self.snapshot = volume_fakes.FakeSnapshot.create_one_snapshot()
+ self.snapshot = volume_fakes.create_one_snapshot()
self.snapshots_mock.get.return_value = self.snapshot
# Get the command object to test
diff --git a/openstackclient/tests/unit/volume/v1/fakes.py b/openstackclient/tests/unit/volume/v1/fakes.py
index 2ec40190..76b208b2 100644
--- a/openstackclient/tests/unit/volume/v1/fakes.py
+++ b/openstackclient/tests/unit/volume/v1/fakes.py
@@ -24,7 +24,6 @@ from openstackclient.tests.unit import utils
class FakeVolumev1Client:
-
def __init__(self, **kwargs):
self.volumes = mock.Mock()
self.volumes.resource_class = fakes.FakeResource(None, {})
@@ -37,8 +36,9 @@ class FakeVolumev1Client:
self.volume_types = mock.Mock()
self.volume_types.resource_class = fakes.FakeResource(None, {})
self.volume_encryption_types = mock.Mock()
- self.volume_encryption_types.resource_class = (
- fakes.FakeResource(None, {}))
+ self.volume_encryption_types.resource_class = fakes.FakeResource(
+ None, {}
+ )
self.transfers = mock.Mock()
self.transfers.resource_class = fakes.FakeResource(None, {})
self.volume_snapshots = mock.Mock()
@@ -52,7 +52,6 @@ class FakeVolumev1Client:
class TestVolumev1(utils.TestCommand):
-
def setUp(self):
super().setUp()
@@ -75,577 +74,538 @@ class TestVolumev1(utils.TestCommand):
)
-class FakeTransfer(object):
- """Fake one or more Transfer."""
-
- @staticmethod
- def create_one_transfer(attrs=None):
- """Create a fake transfer.
-
- :param Dictionary attrs:
- A dictionary with all attributes of Transfer Request
- :return:
- A FakeResource object with volume_id, name, id.
- """
- # Set default attribute
- transfer_info = {
- 'volume_id': 'volume-id-' + uuid.uuid4().hex,
- 'name': 'fake_transfer_name',
- 'id': 'id-' + uuid.uuid4().hex,
- 'links': 'links-' + uuid.uuid4().hex,
- }
-
- # Overwrite default attributes if there are some attributes set
- attrs = attrs or {}
-
- transfer_info.update(attrs)
-
- transfer = fakes.FakeResource(
- None,
- transfer_info,
- loaded=True)
-
- return transfer
-
- @staticmethod
- def create_transfers(attrs=None, count=2):
- """Create multiple fake transfers.
-
- :param Dictionary attrs:
- A dictionary with all attributes of transfer
- :param Integer count:
- The number of transfers to be faked
- :return:
- A list of FakeResource objects
- """
- transfers = []
- for n in range(0, count):
- transfers.append(FakeTransfer.create_one_transfer(attrs))
-
- return transfers
-
- @staticmethod
- def get_transfers(transfers=None, count=2):
- """Get an iterable MagicMock object with a list of faked transfers.
-
- If transfers list is provided, then initialize the Mock object with the
- list. Otherwise create one.
-
- :param List transfers:
- A list of FakeResource objects faking transfers
- :param Integer count:
- The number of transfers to be faked
- :return
- An iterable Mock object with side_effect set to a list of faked
- transfers
- """
- if transfers is None:
- transfers = FakeTransfer.create_transfers(count)
-
- return mock.Mock(side_effect=transfers)
-
-
-class FakeService(object):
- """Fake one or more Services."""
-
- @staticmethod
- def create_one_service(attrs=None):
- """Create a fake service.
-
- :param Dictionary attrs:
- A dictionary with all attributes of service
- :return:
- A FakeResource object with host, status, etc.
- """
- # Set default attribute
- service_info = {
- 'host': 'host_test',
- 'binary': 'cinder_test',
- 'status': 'enabled',
- 'disabled_reason': 'LongHoliday-GoldenWeek',
- 'zone': 'fake_zone',
- 'updated_at': 'fake_date',
- 'state': 'fake_state',
- }
-
- # Overwrite default attributes if there are some attributes set
- attrs = attrs or {}
-
- service_info.update(attrs)
-
- service = fakes.FakeResource(
- None,
- service_info,
- loaded=True)
-
- return service
-
- @staticmethod
- def create_services(attrs=None, count=2):
- """Create multiple fake services.
-
- :param Dictionary attrs:
- A dictionary with all attributes of service
- :param Integer count:
- The number of services to be faked
- :return:
- A list of FakeResource objects
- """
- services = []
- for n in range(0, count):
- services.append(FakeService.create_one_service(attrs))
-
- return services
-
- @staticmethod
- def get_services(services=None, count=2):
- """Get an iterable MagicMock object with a list of faked services.
-
- If services list is provided, then initialize the Mock object with the
- list. Otherwise create one.
-
- :param List services:
- A list of FakeResource objects faking services
- :param Integer count:
- The number of services to be faked
- :return
- An iterable Mock object with side_effect set to a list of faked
- services
- """
- if services is None:
- services = FakeService.create_services(count)
-
- return mock.Mock(side_effect=services)
-
-
-class FakeQos(object):
- """Fake one or more Qos specification."""
-
- @staticmethod
- def create_one_qos(attrs=None):
- """Create a fake Qos specification.
-
- :param Dictionary attrs:
- A dictionary with all attributes
- :return:
- A FakeResource object with id, name, consumer, etc.
- """
- attrs = attrs or {}
-
- # Set default attributes.
- qos_info = {
- "id": 'qos-id-' + uuid.uuid4().hex,
- "name": 'qos-name-' + uuid.uuid4().hex,
- "consumer": 'front-end',
- "specs": {"foo": "bar", "iops": "9001"},
- }
-
- # Overwrite default attributes.
- qos_info.update(attrs)
-
- qos = fakes.FakeResource(
- info=copy.deepcopy(qos_info),
- loaded=True)
- return qos
-
- @staticmethod
- def create_one_qos_association(attrs=None):
- """Create a fake Qos specification association.
-
- :param Dictionary attrs:
- A dictionary with all attributes
- :return:
- A FakeResource object with id, name, association_type, etc.
- """
- attrs = attrs or {}
-
- # Set default attributes.
- qos_association_info = {
- "id": 'type-id-' + uuid.uuid4().hex,
- "name": 'type-name-' + uuid.uuid4().hex,
- "association_type": 'volume_type',
- }
-
- # Overwrite default attributes.
- qos_association_info.update(attrs)
-
- qos_association = fakes.FakeResource(
- info=copy.deepcopy(qos_association_info),
- loaded=True)
- return qos_association
-
- @staticmethod
- def create_qoses(attrs=None, count=2):
- """Create multiple fake Qos specifications.
-
- :param Dictionary attrs:
- A dictionary with all attributes
- :param int count:
- The number of Qos specifications to fake
- :return:
- A list of FakeResource objects faking the Qos specifications
- """
- qoses = []
- for i in range(0, count):
- qos = FakeQos.create_one_qos(attrs)
- qoses.append(qos)
-
- return qoses
-
- @staticmethod
- def get_qoses(qoses=None, count=2):
- """Get an iterable MagicMock object with a list of faked qoses.
-
- If qoses list is provided, then initialize the Mock object with the
- list. Otherwise create one.
-
- :param List volumes:
- A list of FakeResource objects faking qoses
- :param Integer count:
- The number of qoses to be faked
- :return
- An iterable Mock object with side_effect set to a list of faked
- qoses
- """
- if qoses is None:
- qoses = FakeQos.create_qoses(count)
-
- return mock.Mock(side_effect=qoses)
-
-
-class FakeVolume(object):
- """Fake one or more volumes."""
-
- @staticmethod
- def create_one_volume(attrs=None):
- """Create a fake volume.
-
- :param Dictionary attrs:
- A dictionary with all attributes of volume
- :return:
- A FakeResource object with id, name, status, etc.
- """
- attrs = attrs or {}
-
- # Set default attribute
- volume_info = {
- 'id': 'volume-id' + uuid.uuid4().hex,
- 'display_name': 'volume-name' + uuid.uuid4().hex,
- 'display_description': 'description' + uuid.uuid4().hex,
- 'status': 'available',
- 'size': 10,
- 'volume_type':
- random.choice(['fake_lvmdriver-1', 'fake_lvmdriver-2']),
- 'bootable': 'true',
- 'metadata': {
- 'key' + uuid.uuid4().hex: 'val' + uuid.uuid4().hex,
- 'key' + uuid.uuid4().hex: 'val' + uuid.uuid4().hex,
- 'key' + uuid.uuid4().hex: 'val' + uuid.uuid4().hex},
- 'snapshot_id': 'snapshot-id-' + uuid.uuid4().hex,
- 'availability_zone': 'zone' + uuid.uuid4().hex,
- 'attachments': [{
+def create_one_transfer(attrs=None):
+ """Create a fake transfer.
+
+ :param Dictionary attrs:
+ A dictionary with all attributes of Transfer Request
+ :return:
+ A FakeResource object with volume_id, name, id.
+ """
+ # Set default attribute
+ transfer_info = {
+ 'volume_id': 'volume-id-' + uuid.uuid4().hex,
+ 'name': 'fake_transfer_name',
+ 'id': 'id-' + uuid.uuid4().hex,
+ 'links': 'links-' + uuid.uuid4().hex,
+ }
+
+ # Overwrite default attributes if there are some attributes set
+ attrs = attrs or {}
+
+ transfer_info.update(attrs)
+
+ transfer = fakes.FakeResource(None, transfer_info, loaded=True)
+
+ return transfer
+
+
+def create_transfers(attrs=None, count=2):
+ """Create multiple fake transfers.
+
+ :param Dictionary attrs:
+ A dictionary with all attributes of transfer
+ :param Integer count:
+ The number of transfers to be faked
+ :return:
+ A list of FakeResource objects
+ """
+ transfers = []
+ for n in range(0, count):
+ transfers.append(create_one_transfer(attrs))
+
+ return transfers
+
+
+def get_transfers(transfers=None, count=2):
+ """Get an iterable MagicMock object with a list of faked transfers.
+
+ If transfers list is provided, then initialize the Mock object with the
+ list. Otherwise create one.
+
+ :param List transfers:
+ A list of FakeResource objects faking transfers
+ :param Integer count:
+ The number of transfers to be faked
+ :return
+ An iterable Mock object with side_effect set to a list of faked
+ transfers
+ """
+ if transfers is None:
+ transfers = create_transfers(count)
+
+ return mock.Mock(side_effect=transfers)
+
+
+def create_one_service(attrs=None):
+ """Create a fake service.
+
+ :param Dictionary attrs:
+ A dictionary with all attributes of service
+ :return:
+ A FakeResource object with host, status, etc.
+ """
+ # Set default attribute
+ service_info = {
+ 'host': 'host_test',
+ 'binary': 'cinder_test',
+ 'status': 'enabled',
+ 'disabled_reason': 'LongHoliday-GoldenWeek',
+ 'zone': 'fake_zone',
+ 'updated_at': 'fake_date',
+ 'state': 'fake_state',
+ }
+
+ # Overwrite default attributes if there are some attributes set
+ attrs = attrs or {}
+
+ service_info.update(attrs)
+
+ service = fakes.FakeResource(None, service_info, loaded=True)
+
+ return service
+
+
+def create_services(attrs=None, count=2):
+ """Create multiple fake services.
+
+ :param Dictionary attrs:
+ A dictionary with all attributes of service
+ :param Integer count:
+ The number of services to be faked
+ :return:
+ A list of FakeResource objects
+ """
+ services = []
+ for n in range(0, count):
+ services.append(create_one_service(attrs))
+
+ return services
+
+
+def get_services(services=None, count=2):
+ """Get an iterable MagicMock object with a list of faked services.
+
+ If services list is provided, then initialize the Mock object with the
+ list. Otherwise create one.
+
+ :param List services:
+ A list of FakeResource objects faking services
+ :param Integer count:
+ The number of services to be faked
+ :return
+ An iterable Mock object with side_effect set to a list of faked
+ services
+ """
+ if services is None:
+ services = create_services(count)
+
+ return mock.Mock(side_effect=services)
+
+
+def create_one_qos(attrs=None):
+ """Create a fake Qos specification.
+
+ :param Dictionary attrs:
+ A dictionary with all attributes
+ :return:
+ A FakeResource object with id, name, consumer, etc.
+ """
+ attrs = attrs or {}
+
+ # Set default attributes.
+ qos_info = {
+ "id": 'qos-id-' + uuid.uuid4().hex,
+ "name": 'qos-name-' + uuid.uuid4().hex,
+ "consumer": 'front-end',
+ "specs": {"foo": "bar", "iops": "9001"},
+ }
+
+ # Overwrite default attributes.
+ qos_info.update(attrs)
+
+ qos = fakes.FakeResource(info=copy.deepcopy(qos_info), loaded=True)
+ return qos
+
+
+def create_one_qos_association(attrs=None):
+ """Create a fake Qos specification association.
+
+ :param Dictionary attrs:
+ A dictionary with all attributes
+ :return:
+ A FakeResource object with id, name, association_type, etc.
+ """
+ attrs = attrs or {}
+
+ # Set default attributes.
+ qos_association_info = {
+ "id": 'type-id-' + uuid.uuid4().hex,
+ "name": 'type-name-' + uuid.uuid4().hex,
+ "association_type": 'volume_type',
+ }
+
+ # Overwrite default attributes.
+ qos_association_info.update(attrs)
+
+ qos_association = fakes.FakeResource(
+ info=copy.deepcopy(qos_association_info), loaded=True
+ )
+ return qos_association
+
+
+def create_qoses(attrs=None, count=2):
+ """Create multiple fake Qos specifications.
+
+ :param Dictionary attrs:
+ A dictionary with all attributes
+ :param int count:
+ The number of Qos specifications to fake
+ :return:
+ A list of FakeResource objects faking the Qos specifications
+ """
+ qoses = []
+ for i in range(0, count):
+ qos = create_one_qos(attrs)
+ qoses.append(qos)
+
+ return qoses
+
+
+def get_qoses(qoses=None, count=2):
+ """Get an iterable MagicMock object with a list of faked qoses.
+
+ If qoses list is provided, then initialize the Mock object with the
+ list. Otherwise create one.
+
+ :param List volumes:
+ A list of FakeResource objects faking qoses
+ :param Integer count:
+ The number of qoses to be faked
+ :return
+ An iterable Mock object with side_effect set to a list of faked
+ qoses
+ """
+ if qoses is None:
+ qoses = create_qoses(count)
+
+ return mock.Mock(side_effect=qoses)
+
+
+def create_one_volume(attrs=None):
+ """Create a fake volume.
+
+ :param Dictionary attrs:
+ A dictionary with all attributes of volume
+ :return:
+ A FakeResource object with id, name, status, etc.
+ """
+ attrs = attrs or {}
+
+ # Set default attribute
+ volume_info = {
+ 'id': 'volume-id' + uuid.uuid4().hex,
+ 'display_name': 'volume-name' + uuid.uuid4().hex,
+ 'display_description': 'description' + uuid.uuid4().hex,
+ 'status': 'available',
+ 'size': 10,
+ 'volume_type': random.choice(['fake_lvmdriver-1', 'fake_lvmdriver-2']),
+ 'bootable': 'true',
+ 'metadata': {
+ 'key' + uuid.uuid4().hex: 'val' + uuid.uuid4().hex,
+ 'key' + uuid.uuid4().hex: 'val' + uuid.uuid4().hex,
+ 'key' + uuid.uuid4().hex: 'val' + uuid.uuid4().hex,
+ },
+ 'snapshot_id': 'snapshot-id-' + uuid.uuid4().hex,
+ 'availability_zone': 'zone' + uuid.uuid4().hex,
+ 'attachments': [
+ {
'device': '/dev/' + uuid.uuid4().hex,
'server_id': uuid.uuid4().hex,
- }, ],
- 'created_at': 'time-' + uuid.uuid4().hex,
- }
-
- # Overwrite default attributes if there are some attributes set
- volume_info.update(attrs)
-
- volume = fakes.FakeResource(
- None,
- volume_info,
- loaded=True)
- return volume
-
- @staticmethod
- def create_volumes(attrs=None, count=2):
- """Create multiple fake volumes.
-
- :param Dictionary attrs:
- A dictionary with all attributes of volume
- :param Integer count:
- The number of volumes to be faked
- :return:
- A list of FakeResource objects
- """
- volumes = []
- for n in range(0, count):
- volumes.append(FakeVolume.create_one_volume(attrs))
-
- return volumes
-
- @staticmethod
- def get_volumes(volumes=None, count=2):
- """Get an iterable MagicMock object with a list of faked volumes.
-
- If volumes list is provided, then initialize the Mock object with the
- list. Otherwise create one.
-
- :param List volumes:
- A list of FakeResource objects faking volumes
- :param Integer count:
- The number of volumes to be faked
- :return
- An iterable Mock object with side_effect set to a list of faked
- volumes
- """
- if volumes is None:
- volumes = FakeVolume.create_volumes(count)
-
- return mock.Mock(side_effect=volumes)
-
-
-class FakeVolumeType(object):
- """Fake one or more type."""
-
- @staticmethod
- def create_one_volume_type(attrs=None, methods=None):
- """Create a fake volume type.
-
- :param Dictionary attrs:
- A dictionary with all attributes
- :param Dictionary methods:
- A dictionary with all methods
- :return:
- A FakeResource object with id, name, description, etc.
- """
- attrs = attrs or {}
- methods = methods or {}
-
- # Set default attributes.
- volume_type_info = {
- "id": 'type-id-' + uuid.uuid4().hex,
- "name": 'type-name-' + uuid.uuid4().hex,
- "description": 'type-description-' + uuid.uuid4().hex,
- "extra_specs": {"foo": "bar"},
- "is_public": True,
- }
-
- # Overwrite default attributes.
- volume_type_info.update(attrs)
-
- volume_type = fakes.FakeResource(
- info=copy.deepcopy(volume_type_info),
- methods=methods,
- loaded=True)
- return volume_type
-
- @staticmethod
- def create_volume_types(attrs=None, count=2):
- """Create multiple fake types.
-
- :param Dictionary attrs:
- A dictionary with all attributes
- :param int count:
- The number of types to fake
- :return:
- A list of FakeResource objects faking the types
- """
- volume_types = []
- for i in range(0, count):
- volume_type = FakeVolumeType.create_one_volume_type(attrs)
- volume_types.append(volume_type)
-
- return volume_types
-
- @staticmethod
- def get_volume_types(volume_types=None, count=2):
- """Get an iterable MagicMock object with a list of faked types.
-
- If types list is provided, then initialize the Mock object with the
- list. Otherwise create one.
-
- :param List volume_types:
- A list of FakeResource objects faking types
- :param Integer count:
- The number of types to be faked
- :return
- An iterable Mock object with side_effect set to a list of faked
- types
- """
- if volume_types is None:
- volume_types = FakeVolumeType.create_volume_types(count)
-
- return mock.Mock(side_effect=volume_types)
-
- @staticmethod
- def create_one_encryption_volume_type(attrs=None):
- """Create a fake encryption volume type.
-
- :param Dictionary attrs:
- A dictionary with all attributes
- :return:
- A FakeResource object with volume_type_id etc.
- """
- attrs = attrs or {}
-
- # Set default attributes.
- encryption_info = {
- "volume_type_id": 'type-id-' + uuid.uuid4().hex,
- 'provider': 'LuksEncryptor',
- 'cipher': None,
- 'key_size': None,
- 'control_location': 'front-end',
- }
-
- # Overwrite default attributes.
- encryption_info.update(attrs)
-
- encryption_type = fakes.FakeResource(
- info=copy.deepcopy(encryption_info),
- loaded=True)
- return encryption_type
-
-
-class FakeSnapshot(object):
- """Fake one or more snapshot."""
-
- @staticmethod
- def create_one_snapshot(attrs=None):
- """Create a fake snapshot.
-
- :param Dictionary attrs:
- A dictionary with all attributes
- :return:
- A FakeResource object with id, name, description, etc.
- """
- attrs = attrs or {}
-
- # Set default attributes.
- snapshot_info = {
- "id": 'snapshot-id-' + uuid.uuid4().hex,
- "display_name": 'snapshot-name-' + uuid.uuid4().hex,
- "display_description": 'snapshot-description-' + uuid.uuid4().hex,
- "size": 10,
- "status": "available",
- "metadata": {"foo": "bar"},
- "created_at": "2015-06-03T18:49:19.000000",
- "volume_id": 'vloume-id-' + uuid.uuid4().hex,
- }
-
- # Overwrite default attributes.
- snapshot_info.update(attrs)
-
- snapshot_method = {'update': None}
-
- snapshot = fakes.FakeResource(
- info=copy.deepcopy(snapshot_info),
- methods=copy.deepcopy(snapshot_method),
- loaded=True)
- return snapshot
-
- @staticmethod
- def create_snapshots(attrs=None, count=2):
- """Create multiple fake snapshots.
-
- :param Dictionary attrs:
- A dictionary with all attributes
- :param int count:
- The number of snapshots to fake
- :return:
- A list of FakeResource objects faking the snapshots
- """
- snapshots = []
- for i in range(0, count):
- snapshot = FakeSnapshot.create_one_snapshot(attrs)
- snapshots.append(snapshot)
-
- return snapshots
-
- @staticmethod
- def get_snapshots(snapshots=None, count=2):
- """Get an iterable MagicMock object with a list of faked snapshots.
-
- If snapshots list is provided, then initialize the Mock object with the
- list. Otherwise create one.
-
- :param List volumes:
- A list of FakeResource objects faking snapshots
- :param Integer count:
- The number of snapshots to be faked
- :return
- An iterable Mock object with side_effect set to a list of faked
- snapshots
- """
- if snapshots is None:
- snapshots = FakeSnapshot.create_snapshots(count)
-
- return mock.Mock(side_effect=snapshots)
-
-
-class FakeBackup(object):
- """Fake one or more backup."""
-
- @staticmethod
- def create_one_backup(attrs=None):
- """Create a fake backup.
-
- :param Dictionary attrs:
- A dictionary with all attributes
- :return:
- A FakeResource object with id, name, volume_id, etc.
- """
- attrs = attrs or {}
-
- # Set default attributes.
- backup_info = {
- "id": 'backup-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),
- "status": "error",
- "availability_zone": 'zone' + uuid.uuid4().hex,
- "links": 'links-' + uuid.uuid4().hex,
- }
-
- # Overwrite default attributes.
- backup_info.update(attrs)
-
- backup = fakes.FakeResource(
- info=copy.deepcopy(backup_info),
- loaded=True)
- return backup
-
- @staticmethod
- def create_backups(attrs=None, count=2):
- """Create multiple fake backups.
-
- :param Dictionary attrs:
- A dictionary with all attributes
- :param int count:
- The number of backups to fake
- :return:
- A list of FakeResource objects faking the backups
- """
- backups = []
- for i in range(0, count):
- backup = FakeBackup.create_one_backup(attrs)
- backups.append(backup)
-
- return backups
-
- @staticmethod
- 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 volumes:
- 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 = FakeBackup.create_backups(count)
-
- return mock.Mock(side_effect=backups)
+ },
+ ],
+ 'created_at': 'time-' + uuid.uuid4().hex,
+ }
+
+ # Overwrite default attributes if there are some attributes set
+ volume_info.update(attrs)
+
+ volume = fakes.FakeResource(None, volume_info, loaded=True)
+ return volume
+
+
+def create_volumes(attrs=None, count=2):
+ """Create multiple fake volumes.
+
+ :param Dictionary attrs:
+ A dictionary with all attributes of volume
+ :param Integer count:
+ The number of volumes to be faked
+ :return:
+ A list of FakeResource objects
+ """
+ volumes = []
+ for n in range(0, count):
+ volumes.append(create_one_volume(attrs))
+
+ return volumes
+
+
+def get_volumes(volumes=None, count=2):
+ """Get an iterable MagicMock object with a list of faked volumes.
+
+ If volumes list is provided, then initialize the Mock object with the
+ list. Otherwise create one.
+
+ :param List volumes:
+ A list of FakeResource objects faking volumes
+ :param Integer count:
+ The number of volumes to be faked
+ :return
+ An iterable Mock object with side_effect set to a list of faked
+ volumes
+ """
+ if volumes is None:
+ volumes = create_volumes(count)
+
+ return mock.Mock(side_effect=volumes)
+
+
+def create_one_volume_type(attrs=None, methods=None):
+ """Create a fake volume type.
+
+ :param Dictionary attrs:
+ A dictionary with all attributes
+ :param Dictionary methods:
+ A dictionary with all methods
+ :return:
+ A FakeResource object with id, name, description, etc.
+ """
+ attrs = attrs or {}
+ methods = methods or {}
+
+ # Set default attributes.
+ volume_type_info = {
+ "id": 'type-id-' + uuid.uuid4().hex,
+ "name": 'type-name-' + uuid.uuid4().hex,
+ "description": 'type-description-' + uuid.uuid4().hex,
+ "extra_specs": {"foo": "bar"},
+ "is_public": True,
+ }
+
+ # Overwrite default attributes.
+ volume_type_info.update(attrs)
+
+ volume_type = fakes.FakeResource(
+ info=copy.deepcopy(volume_type_info), methods=methods, loaded=True
+ )
+ return volume_type
+
+
+def create_volume_types(attrs=None, count=2):
+ """Create multiple fake types.
+
+ :param Dictionary attrs:
+ A dictionary with all attributes
+ :param int count:
+ The number of types to fake
+ :return:
+ A list of FakeResource objects faking the types
+ """
+ volume_types = []
+ for i in range(0, count):
+ volume_type = create_one_volume_type(attrs)
+ volume_types.append(volume_type)
+
+ return volume_types
+
+
+def get_volume_types(volume_types=None, count=2):
+ """Get an iterable MagicMock object with a list of faked types.
+
+ If types list is provided, then initialize the Mock object with the
+ list. Otherwise create one.
+
+ :param List volume_types:
+ A list of FakeResource objects faking types
+ :param Integer count:
+ The number of types to be faked
+ :return
+ An iterable Mock object with side_effect set to a list of faked
+ types
+ """
+ if volume_types is None:
+ volume_types = create_volume_types(count)
+
+ return mock.Mock(side_effect=volume_types)
+
+
+def create_one_encryption_volume_type(attrs=None):
+ """Create a fake encryption volume type.
+
+ :param Dictionary attrs:
+ A dictionary with all attributes
+ :return:
+ A FakeResource object with volume_type_id etc.
+ """
+ attrs = attrs or {}
+
+ # Set default attributes.
+ encryption_info = {
+ "volume_type_id": 'type-id-' + uuid.uuid4().hex,
+ 'provider': 'LuksEncryptor',
+ 'cipher': None,
+ 'key_size': None,
+ 'control_location': 'front-end',
+ }
+
+ # Overwrite default attributes.
+ encryption_info.update(attrs)
+
+ encryption_type = fakes.FakeResource(
+ info=copy.deepcopy(encryption_info), loaded=True
+ )
+ return encryption_type
+
+
+def create_one_snapshot(attrs=None):
+ """Create a fake snapshot.
+
+ :param Dictionary attrs:
+ A dictionary with all attributes
+ :return:
+ A FakeResource object with id, name, description, etc.
+ """
+ attrs = attrs or {}
+
+ # Set default attributes.
+ snapshot_info = {
+ "id": 'snapshot-id-' + uuid.uuid4().hex,
+ "display_name": 'snapshot-name-' + uuid.uuid4().hex,
+ "display_description": 'snapshot-description-' + uuid.uuid4().hex,
+ "size": 10,
+ "status": "available",
+ "metadata": {"foo": "bar"},
+ "created_at": "2015-06-03T18:49:19.000000",
+ "volume_id": 'vloume-id-' + uuid.uuid4().hex,
+ }
+
+ # Overwrite default attributes.
+ snapshot_info.update(attrs)
+
+ snapshot_method = {'update': None}
+
+ snapshot = fakes.FakeResource(
+ info=copy.deepcopy(snapshot_info),
+ methods=copy.deepcopy(snapshot_method),
+ loaded=True,
+ )
+ return snapshot
+
+
+def create_snapshots(attrs=None, count=2):
+ """Create multiple fake snapshots.
+
+ :param Dictionary attrs:
+ A dictionary with all attributes
+ :param int count:
+ The number of snapshots to fake
+ :return:
+ A list of FakeResource objects faking the snapshots
+ """
+ snapshots = []
+ for i in range(0, count):
+ snapshot = create_one_snapshot(attrs)
+ snapshots.append(snapshot)
+
+ return snapshots
+
+
+def get_snapshots(snapshots=None, count=2):
+ """Get an iterable MagicMock object with a list of faked snapshots.
+
+ If snapshots list is provided, then initialize the Mock object with the
+ list. Otherwise create one.
+
+ :param List volumes:
+ A list of FakeResource objects faking snapshots
+ :param Integer count:
+ The number of snapshots to be faked
+ :return
+ An iterable Mock object with side_effect set to a list of faked
+ snapshots
+ """
+ if snapshots is None:
+ snapshots = create_snapshots(count)
+
+ return mock.Mock(side_effect=snapshots)
+
+
+def create_one_backup(attrs=None):
+ """Create a fake backup.
+
+ :param Dictionary attrs:
+ A dictionary with all attributes
+ :return:
+ A FakeResource object with id, name, volume_id, etc.
+ """
+ attrs = attrs or {}
+
+ # Set default attributes.
+ backup_info = {
+ "id": 'backup-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),
+ "status": "error",
+ "availability_zone": 'zone' + uuid.uuid4().hex,
+ "links": 'links-' + uuid.uuid4().hex,
+ }
+
+ # Overwrite default attributes.
+ backup_info.update(attrs)
+
+ backup = fakes.FakeResource(info=copy.deepcopy(backup_info), loaded=True)
+ return backup
+
+
+def create_backups(attrs=None, count=2):
+ """Create multiple fake backups.
+
+ :param Dictionary attrs:
+ A dictionary with all attributes
+ :param int count:
+ The number of backups to fake
+ :return:
+ A list of FakeResource objects faking the backups
+ """
+ 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 volumes:
+ 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)
diff --git a/openstackclient/tests/unit/volume/v1/test_qos_specs.py b/openstackclient/tests/unit/volume/v1/test_qos_specs.py
index d66a8558..f5b35143 100644
--- a/openstackclient/tests/unit/volume/v1/test_qos_specs.py
+++ b/openstackclient/tests/unit/volume/v1/test_qos_specs.py
@@ -28,7 +28,7 @@ from openstackclient.volume.v1 import qos_specs
class TestQos(volume_fakes.TestVolumev1):
def setUp(self):
- super(TestQos, self).setUp()
+ super().setUp()
self.qos_mock = self.app.client_manager.volume.qos_specs
self.qos_mock.reset_mock()
@@ -39,11 +39,11 @@ class TestQos(volume_fakes.TestVolumev1):
class TestQosAssociate(TestQos):
- volume_type = volume_fakes.FakeVolumeType.create_one_volume_type()
- qos_spec = volume_fakes.FakeQos.create_one_qos()
+ volume_type = volume_fakes.create_one_volume_type()
+ qos_spec = volume_fakes.create_one_qos()
def setUp(self):
- super(TestQosAssociate, self).setUp()
+ super().setUp()
self.qos_mock.get.return_value = self.qos_spec
self.types_mock.get.return_value = self.volume_type
@@ -80,8 +80,8 @@ class TestQosCreate(TestQos):
)
def setUp(self):
- super(TestQosCreate, self).setUp()
- self.new_qos_spec = volume_fakes.FakeQos.create_one_qos()
+ super().setUp()
+ self.new_qos_spec = volume_fakes.create_one_qos()
self.datalist = (
self.new_qos_spec.consumer,
self.new_qos_spec.id,
@@ -160,13 +160,13 @@ class TestQosCreate(TestQos):
class TestQosDelete(TestQos):
- qos_specs = volume_fakes.FakeQos.create_qoses(count=2)
+ qos_specs = volume_fakes.create_qoses(count=2)
def setUp(self):
- super(TestQosDelete, self).setUp()
+ super().setUp()
self.qos_mock.get = (
- volume_fakes.FakeQos.get_qoses(self.qos_specs))
+ volume_fakes.get_qoses(self.qos_specs))
# Get the command object to test
self.cmd = qos_specs.DeleteQos(self.app, None)
@@ -263,11 +263,11 @@ class TestQosDelete(TestQos):
class TestQosDisassociate(TestQos):
- volume_type = volume_fakes.FakeVolumeType.create_one_volume_type()
- qos_spec = volume_fakes.FakeQos.create_one_qos()
+ volume_type = volume_fakes.create_one_volume_type()
+ qos_spec = volume_fakes.create_one_qos()
def setUp(self):
- super(TestQosDisassociate, self).setUp()
+ super().setUp()
self.qos_mock.get.return_value = self.qos_spec
self.types_mock.get.return_value = self.volume_type
@@ -311,8 +311,8 @@ class TestQosDisassociate(TestQos):
class TestQosList(TestQos):
- qos_specs = volume_fakes.FakeQos.create_qoses(count=2)
- qos_association = volume_fakes.FakeQos.create_one_qos_association()
+ qos_specs = volume_fakes.create_qoses(count=2)
+ qos_association = volume_fakes.create_one_qos_association()
columns = (
'ID',
@@ -332,7 +332,7 @@ class TestQosList(TestQos):
))
def setUp(self):
- super(TestQosList, self).setUp()
+ super().setUp()
self.qos_mock.list.return_value = self.qos_specs
self.qos_mock.get_associations.return_value = [self.qos_association]
@@ -382,10 +382,10 @@ class TestQosList(TestQos):
class TestQosSet(TestQos):
- qos_spec = volume_fakes.FakeQos.create_one_qos()
+ qos_spec = volume_fakes.create_one_qos()
def setUp(self):
- super(TestQosSet, self).setUp()
+ super().setUp()
self.qos_mock.get.return_value = self.qos_spec
# Get the command object to test
@@ -414,11 +414,11 @@ class TestQosSet(TestQos):
class TestQosShow(TestQos):
- qos_spec = volume_fakes.FakeQos.create_one_qos()
- qos_association = volume_fakes.FakeQos.create_one_qos_association()
+ qos_spec = volume_fakes.create_one_qos()
+ qos_association = volume_fakes.create_one_qos_association()
def setUp(self):
- super(TestQosShow, self).setUp()
+ super().setUp()
self.qos_mock.get.return_value = self.qos_spec
self.qos_mock.get_associations.return_value = [self.qos_association]
# Get the command object to test
@@ -459,10 +459,10 @@ class TestQosShow(TestQos):
class TestQosUnset(TestQos):
- qos_spec = volume_fakes.FakeQos.create_one_qos()
+ qos_spec = volume_fakes.create_one_qos()
def setUp(self):
- super(TestQosUnset, self).setUp()
+ super().setUp()
self.qos_mock.get.return_value = self.qos_spec
# Get the command object to test
diff --git a/openstackclient/tests/unit/volume/v1/test_service.py b/openstackclient/tests/unit/volume/v1/test_service.py
index 82d21bfc..a199c913 100644
--- a/openstackclient/tests/unit/volume/v1/test_service.py
+++ b/openstackclient/tests/unit/volume/v1/test_service.py
@@ -14,14 +14,14 @@
from osc_lib import exceptions
-from openstackclient.tests.unit.volume.v1 import fakes as service_fakes
+from openstackclient.tests.unit.volume.v1 import fakes as volume_fakes
from openstackclient.volume.v1 import service
-class TestService(service_fakes.TestVolumev1):
+class TestService(volume_fakes.TestVolumev1):
def setUp(self):
- super(TestService, self).setUp()
+ super().setUp()
# Get a shortcut to the ServiceManager Mock
self.service_mock = self.app.client_manager.volume.services
@@ -31,10 +31,10 @@ class TestService(service_fakes.TestVolumev1):
class TestServiceList(TestService):
# The service to be listed
- services = service_fakes.FakeService.create_one_service()
+ services = volume_fakes.create_one_service()
def setUp(self):
- super(TestServiceList, self).setUp()
+ super().setUp()
self.service_mock.list.return_value = [self.services]
@@ -144,10 +144,10 @@ class TestServiceList(TestService):
class TestServiceSet(TestService):
- service = service_fakes.FakeService.create_one_service()
+ service = volume_fakes.create_one_service()
def setUp(self):
- super(TestServiceSet, self).setUp()
+ super().setUp()
self.service_mock.enable.return_value = self.service
self.service_mock.disable.return_value = self.service
diff --git a/openstackclient/tests/unit/volume/v1/test_transfer_request.py b/openstackclient/tests/unit/volume/v1/test_transfer_request.py
index 333bf526..97700fbb 100644
--- a/openstackclient/tests/unit/volume/v1/test_transfer_request.py
+++ b/openstackclient/tests/unit/volume/v1/test_transfer_request.py
@@ -18,14 +18,14 @@ from unittest.mock import call
from osc_lib import exceptions
from osc_lib import utils
-from openstackclient.tests.unit.volume.v1 import fakes as transfer_fakes
+from openstackclient.tests.unit.volume.v1 import fakes as volume_fakes
from openstackclient.volume.v1 import volume_transfer_request
-class TestTransfer(transfer_fakes.TestVolumev1):
+class TestTransfer(volume_fakes.TestVolumev1):
def setUp(self):
- super(TestTransfer, self).setUp()
+ super().setUp()
# Get a shortcut to the TransferManager Mock
self.transfer_mock = self.app.client_manager.volume.transfers
@@ -45,10 +45,9 @@ class TestTransferAccept(TestTransfer):
)
def setUp(self):
- super(TestTransferAccept, self).setUp()
+ super().setUp()
- self.volume_transfer = (
- transfer_fakes.FakeTransfer.create_one_transfer())
+ self.volume_transfer = volume_fakes.create_one_transfer()
self.data = (
self.volume_transfer.id,
self.volume_transfer.name,
@@ -103,7 +102,7 @@ class TestTransferAccept(TestTransfer):
class TestTransferCreate(TestTransfer):
- volume = transfer_fakes.FakeVolume.create_one_volume()
+ volume = volume_fakes.create_one_volume()
columns = (
'auth_key',
@@ -114,12 +113,14 @@ class TestTransferCreate(TestTransfer):
)
def setUp(self):
- super(TestTransferCreate, self).setUp()
-
- self.volume_transfer = transfer_fakes.FakeTransfer.create_one_transfer(
- attrs={'volume_id': self.volume.id,
- 'auth_key': 'key',
- 'created_at': 'time'}
+ super().setUp()
+
+ self.volume_transfer = volume_fakes.create_one_transfer(
+ attrs={
+ 'volume_id': self.volume.id,
+ 'auth_key': 'key',
+ 'created_at': 'time',
+ },
)
self.data = (
self.volume_transfer.auth_key,
@@ -173,13 +174,14 @@ class TestTransferCreate(TestTransfer):
class TestTransferDelete(TestTransfer):
- volume_transfers = transfer_fakes.FakeTransfer.create_transfers(count=2)
+ volume_transfers = volume_fakes.create_transfers(count=2)
def setUp(self):
- super(TestTransferDelete, self).setUp()
+ super().setUp()
- self.transfer_mock.get = (
- transfer_fakes.FakeTransfer.get_transfers(self.volume_transfers))
+ self.transfer_mock.get = volume_fakes.get_transfers(
+ self.volume_transfers,
+ )
self.transfer_mock.delete.return_value = None
# Get the command object to mock
@@ -252,10 +254,10 @@ class TestTransferDelete(TestTransfer):
class TestTransferList(TestTransfer):
# The Transfers to be listed
- volume_transfers = transfer_fakes.FakeTransfer.create_one_transfer()
+ volume_transfers = volume_fakes.create_one_transfer()
def setUp(self):
- super(TestTransferList, self).setUp()
+ super().setUp()
self.transfer_mock.list.return_value = [self.volume_transfers]
@@ -346,11 +348,10 @@ class TestTransferShow(TestTransfer):
)
def setUp(self):
- super(TestTransferShow, self).setUp()
+ super().setUp()
- self.volume_transfer = (
- transfer_fakes.FakeTransfer.create_one_transfer(
- attrs={'created_at': 'time'})
+ self.volume_transfer = volume_fakes.create_one_transfer(
+ attrs={'created_at': 'time'}
)
self.data = (
self.volume_transfer.created_at,
diff --git a/openstackclient/tests/unit/volume/v1/test_type.py b/openstackclient/tests/unit/volume/v1/test_type.py
index ca74c3e6..c8788249 100644
--- a/openstackclient/tests/unit/volume/v1/test_type.py
+++ b/openstackclient/tests/unit/volume/v1/test_type.py
@@ -27,7 +27,7 @@ from openstackclient.volume.v1 import volume_type
class TestType(volume_fakes.TestVolumev1):
def setUp(self):
- super(TestType, self).setUp()
+ super().setUp()
self.types_mock = self.app.client_manager.volume.volume_types
self.types_mock.reset_mock()
@@ -47,11 +47,11 @@ class TestTypeCreate(TestType):
)
def setUp(self):
- super(TestTypeCreate, self).setUp()
+ super().setUp()
- self.new_volume_type = \
- volume_fakes.FakeVolumeType.create_one_volume_type(
- methods={'set_keys': {'myprop': 'myvalue'}})
+ self.new_volume_type = volume_fakes.create_one_volume_type(
+ methods={'set_keys': {'myprop': 'myvalue'}},
+ )
self.data = (
self.new_volume_type.description,
self.new_volume_type.id,
@@ -87,12 +87,12 @@ class TestTypeCreate(TestType):
'key_size': '128',
'control_location': 'front-end',
}
- encryption_type = \
- volume_fakes.FakeVolumeType.create_one_encryption_volume_type(
- attrs=encryption_info)
- self.new_volume_type = \
- volume_fakes.FakeVolumeType.create_one_volume_type(
- attrs={'encryption': encryption_info})
+ encryption_type = volume_fakes.create_one_encryption_volume_type(
+ attrs=encryption_info,
+ )
+ self.new_volume_type = volume_fakes.create_one_volume_type(
+ attrs={'encryption': encryption_info},
+ )
self.types_mock.create.return_value = self.new_volume_type
self.encryption_types_mock.create.return_value = encryption_type
encryption_columns = (
@@ -145,13 +145,12 @@ class TestTypeCreate(TestType):
class TestTypeDelete(TestType):
- volume_types = volume_fakes.FakeVolumeType.create_volume_types(count=2)
+ volume_types = volume_fakes.create_volume_types(count=2)
def setUp(self):
- super(TestTypeDelete, self).setUp()
+ super().setUp()
- self.types_mock.get = volume_fakes.FakeVolumeType.get_volume_types(
- self.volume_types)
+ self.types_mock.get = volume_fakes.get_volume_types(self.volume_types)
self.types_mock.delete.return_value = None
# Get the command object to mock
@@ -221,7 +220,7 @@ class TestTypeDelete(TestType):
class TestTypeList(TestType):
- volume_types = volume_fakes.FakeVolumeType.create_volume_types()
+ volume_types = volume_fakes.create_volume_types()
columns = [
"ID",
@@ -252,7 +251,7 @@ class TestTypeList(TestType):
))
def setUp(self):
- super(TestTypeList, self).setUp()
+ super().setUp()
self.types_mock.list.return_value = self.volume_types
self.encryption_types_mock.create.return_value = None
@@ -288,9 +287,9 @@ class TestTypeList(TestType):
self.assertCountEqual(self.data_long, list(data))
def test_type_list_with_encryption(self):
- encryption_type = \
- volume_fakes.FakeVolumeType.create_one_encryption_volume_type(
- attrs={'volume_type_id': self.volume_types[0].id})
+ encryption_type = volume_fakes.create_one_encryption_volume_type(
+ attrs={'volume_type_id': self.volume_types[0].id},
+ )
encryption_info = {
'provider': 'LuksEncryptor',
'cipher': None,
@@ -335,11 +334,12 @@ class TestTypeList(TestType):
class TestTypeSet(TestType):
- volume_type = volume_fakes.FakeVolumeType.create_one_volume_type(
- methods={'set_keys': None})
+ volume_type = volume_fakes.create_one_volume_type(
+ methods={'set_keys': None},
+ )
def setUp(self):
- super(TestTypeSet, self).setUp()
+ super().setUp()
self.types_mock.get.return_value = self.volume_type
@@ -441,9 +441,9 @@ class TestTypeShow(TestType):
)
def setUp(self):
- super(TestTypeShow, self).setUp()
+ super().setUp()
- self.volume_type = volume_fakes.FakeVolumeType.create_one_volume_type()
+ self.volume_type = volume_fakes.create_one_volume_type()
self.data = (
self.volume_type.description,
self.volume_type.id,
@@ -474,16 +474,16 @@ class TestTypeShow(TestType):
self.assertCountEqual(self.data, data)
def test_type_show_with_encryption(self):
- encryption_type = \
- volume_fakes.FakeVolumeType.create_one_encryption_volume_type()
+ encryption_type = volume_fakes.create_one_encryption_volume_type()
encryption_info = {
'provider': 'LuksEncryptor',
'cipher': None,
'key_size': None,
'control_location': 'front-end',
}
- self.volume_type = volume_fakes.FakeVolumeType.create_one_volume_type(
- attrs={'encryption': encryption_info})
+ self.volume_type = volume_fakes.create_one_volume_type(
+ attrs={'encryption': encryption_info},
+ )
self.types_mock.get.return_value = self.volume_type
self.encryption_types_mock.get.return_value = encryption_type
encryption_columns = (
@@ -521,11 +521,12 @@ class TestTypeShow(TestType):
class TestTypeUnset(TestType):
- volume_type = volume_fakes.FakeVolumeType.create_one_volume_type(
- methods={'unset_keys': None})
+ volume_type = volume_fakes.create_one_volume_type(
+ methods={'unset_keys': None},
+ )
def setUp(self):
- super(TestTypeUnset, self).setUp()
+ super().setUp()
self.types_mock.get.return_value = self.volume_type
@@ -599,7 +600,7 @@ class TestTypeUnset(TestType):
class TestColumns(TestType):
def test_encryption_info_column_with_info(self):
- fake_volume_type = volume_fakes.FakeVolumeType.create_one_volume_type()
+ fake_volume_type = volume_fakes.create_one_volume_type()
type_id = fake_volume_type.id
encryption_info = {
@@ -615,7 +616,7 @@ class TestColumns(TestType):
self.assertEqual(encryption_info, col.machine_readable())
def test_encryption_info_column_without_info(self):
- fake_volume_type = volume_fakes.FakeVolumeType.create_one_volume_type()
+ fake_volume_type = volume_fakes.create_one_volume_type()
type_id = fake_volume_type.id
col = volume_type.EncryptionInfoColumn(type_id, {})
diff --git a/openstackclient/tests/unit/volume/v1/test_volume.py b/openstackclient/tests/unit/volume/v1/test_volume.py
index 584eca2a..9f16b398 100644
--- a/openstackclient/tests/unit/volume/v1/test_volume.py
+++ b/openstackclient/tests/unit/volume/v1/test_volume.py
@@ -31,7 +31,7 @@ from openstackclient.volume.v1 import volume
class TestVolume(volume_fakes.TestVolumev1):
def setUp(self):
- super(TestVolume, self).setUp()
+ super().setUp()
# Get a shortcut to the VolumeManager Mock
self.volumes_mock = self.app.client_manager.volume.volumes
@@ -50,11 +50,9 @@ class TestVolume(volume_fakes.TestVolumev1):
self.images_mock.reset_mock()
def setup_volumes_mock(self, count):
- volumes = volume_fakes.FakeVolume.create_volumes(count=count)
+ volumes = volume_fakes.create_volumes(count=count)
- self.volumes_mock.get = volume_fakes.FakeVolume.get_volumes(
- volumes,
- 0)
+ self.volumes_mock.get = volume_fakes.get_volumes(volumes, 0)
return volumes
@@ -79,8 +77,8 @@ class TestVolumeCreate(TestVolume):
)
def setUp(self):
- super(TestVolumeCreate, self).setUp()
- self.new_volume = volume_fakes.FakeVolume.create_one_volume()
+ super().setUp()
+ self.new_volume = volume_fakes.create_one_volume()
self.datalist = (
self.new_volume.attachments,
self.new_volume.availability_zone,
@@ -635,7 +633,7 @@ class TestVolumeCreate(TestVolume):
class TestVolumeDelete(TestVolume):
def setUp(self):
- super(TestVolumeDelete, self).setUp()
+ super().setUp()
self.volumes_mock.delete.return_value = None
@@ -725,7 +723,7 @@ class TestVolumeDelete(TestVolume):
class TestVolumeList(TestVolume):
- _volume = volume_fakes.FakeVolume.create_one_volume()
+ _volume = volume_fakes.create_one_volume()
columns = (
'ID',
'Name',
@@ -744,7 +742,7 @@ class TestVolumeList(TestVolume):
)
def setUp(self):
- super(TestVolumeList, self).setUp()
+ super().setUp()
self.volumes_mock.list.return_value = [self._volume]
@@ -921,10 +919,10 @@ class TestVolumeList(TestVolume):
class TestVolumeMigrate(TestVolume):
- _volume = volume_fakes.FakeVolume.create_one_volume()
+ _volume = volume_fakes.create_one_volume()
def setUp(self):
- super(TestVolumeMigrate, self).setUp()
+ super().setUp()
self.volumes_mock.get.return_value = self._volume
self.volumes_mock.migrate_volume.return_value = None
@@ -983,10 +981,10 @@ class TestVolumeMigrate(TestVolume):
class TestVolumeSet(TestVolume):
- _volume = volume_fakes.FakeVolume.create_one_volume()
+ _volume = volume_fakes.create_one_volume()
def setUp(self):
- super(TestVolumeSet, self).setUp()
+ super().setUp()
self.volumes_mock.get.return_value = self._volume
@@ -1243,8 +1241,8 @@ class TestVolumeShow(TestVolume):
)
def setUp(self):
- super(TestVolumeShow, self).setUp()
- self._volume = volume_fakes.FakeVolume.create_one_volume()
+ super().setUp()
+ self._volume = volume_fakes.create_one_volume()
self.datalist = (
self._volume.attachments,
self._volume.availability_zone,
@@ -1300,10 +1298,10 @@ class TestVolumeShow(TestVolume):
class TestVolumeUnset(TestVolume):
- _volume = volume_fakes.FakeVolume.create_one_volume()
+ _volume = volume_fakes.create_one_volume()
def setUp(self):
- super(TestVolumeUnset, self).setUp()
+ super().setUp()
self.volumes_mock.get.return_value = self._volume
@@ -1346,7 +1344,7 @@ class TestVolumeUnset(TestVolume):
class TestColumns(TestVolume):
def test_attachments_column_without_server_cache(self):
- _volume = volume_fakes.FakeVolume.create_one_volume()
+ _volume = volume_fakes.create_one_volume()
server_id = _volume.attachments[0]['server_id']
device = _volume.attachments[0]['device']
@@ -1356,7 +1354,7 @@ class TestColumns(TestVolume):
self.assertEqual(_volume.attachments, col.machine_readable())
def test_attachments_column_with_server_cache(self):
- _volume = volume_fakes.FakeVolume.create_one_volume()
+ _volume = volume_fakes.create_one_volume()
server_id = _volume.attachments[0]['server_id']
device = _volume.attachments[0]['device']
diff --git a/openstackclient/tests/unit/volume/v1/test_volume_backup.py b/openstackclient/tests/unit/volume/v1/test_volume_backup.py
index 2f568929..b705b4b9 100644
--- a/openstackclient/tests/unit/volume/v1/test_volume_backup.py
+++ b/openstackclient/tests/unit/volume/v1/test_volume_backup.py
@@ -25,7 +25,7 @@ from openstackclient.volume.v1 import volume_backup
class TestBackup(volume_fakes.TestVolumev1):
def setUp(self):
- super(TestBackup, self).setUp()
+ super().setUp()
self.backups_mock = self.app.client_manager.volume.backups
self.backups_mock.reset_mock()
@@ -39,7 +39,7 @@ class TestBackup(volume_fakes.TestVolumev1):
class TestBackupCreate(TestBackup):
- volume = volume_fakes.FakeVolume.create_one_volume()
+ volume = volume_fakes.create_one_volume()
columns = (
'availability_zone',
@@ -55,9 +55,10 @@ class TestBackupCreate(TestBackup):
)
def setUp(self):
- super(TestBackupCreate, self).setUp()
- self.new_backup = volume_fakes.FakeBackup.create_one_backup(
- attrs={'volume_id': self.volume.id})
+ super().setUp()
+ self.new_backup = volume_fakes.create_one_backup(
+ attrs={'volume_id': self.volume.id},
+ )
self.data = (
self.new_backup.availability_zone,
self.new_backup.container,
@@ -129,13 +130,12 @@ class TestBackupCreate(TestBackup):
class TestBackupDelete(TestBackup):
- backups = volume_fakes.FakeBackup.create_backups(count=2)
+ backups = volume_fakes.create_backups(count=2)
def setUp(self):
- super(TestBackupDelete, self).setUp()
+ super().setUp()
- self.backups_mock.get = (
- volume_fakes.FakeBackup.get_backups(self.backups))
+ self.backups_mock.get = volume_fakes.get_backups(self.backups)
self.backups_mock.delete.return_value = None
# Get the command object to mock
@@ -205,9 +205,11 @@ class TestBackupDelete(TestBackup):
class TestBackupList(TestBackup):
- volume = volume_fakes.FakeVolume.create_one_volume()
- backups = volume_fakes.FakeBackup.create_backups(
- attrs={'volume_id': volume.display_name}, count=3)
+ volume = volume_fakes.create_one_volume()
+ backups = volume_fakes.create_backups(
+ attrs={'volume_id': volume.display_name},
+ count=3,
+ )
columns = [
'ID',
@@ -245,7 +247,7 @@ class TestBackupList(TestBackup):
))
def setUp(self):
- super(TestBackupList, self).setUp()
+ super().setUp()
self.volumes_mock.list.return_value = [self.volume]
self.backups_mock.list.return_value = self.backups
@@ -314,9 +316,10 @@ class TestBackupList(TestBackup):
class TestBackupRestore(TestBackup):
- volume = volume_fakes.FakeVolume.create_one_volume()
- backup = volume_fakes.FakeBackup.create_one_backup(
- attrs={'volume_id': volume.id})
+ volume = volume_fakes.create_one_volume()
+ backup = volume_fakes.create_one_backup(
+ attrs={'volume_id': volume.id},
+ )
def setUp(self):
super().setUp()
@@ -324,7 +327,7 @@ class TestBackupRestore(TestBackup):
self.backups_mock.get.return_value = self.backup
self.volumes_mock.get.return_value = self.volume
self.restores_mock.restore.return_value = (
- volume_fakes.FakeVolume.create_one_volume(
+ volume_fakes.create_one_volume(
{'id': self.volume['id']},
)
)
@@ -400,8 +403,8 @@ class TestBackupShow(TestBackup):
)
def setUp(self):
- super(TestBackupShow, self).setUp()
- self.backup = volume_fakes.FakeBackup.create_one_backup()
+ super().setUp()
+ self.backup = volume_fakes.create_one_backup()
self.data = (
self.backup.availability_zone,
self.backup.container,
diff --git a/openstackclient/tests/unit/volume/v2/fakes.py b/openstackclient/tests/unit/volume/v2/fakes.py
index 33588e04..a3ef142f 100644
--- a/openstackclient/tests/unit/volume/v2/fakes.py
+++ b/openstackclient/tests/unit/volume/v2/fakes.py
@@ -41,7 +41,6 @@ QUOTA = {
class FakeVolumeClient:
-
def __init__(self, **kwargs):
self.auth_token = kwargs['token']
self.management_url = kwargs['endpoint']
@@ -76,8 +75,9 @@ class FakeVolumeClient:
self.transfers = mock.Mock()
self.transfers.resource_class = fakes.FakeResource(None, {})
self.volume_encryption_types = mock.Mock()
- self.volume_encryption_types.resource_class = (
- fakes.FakeResource(None, {}))
+ self.volume_encryption_types.resource_class = fakes.FakeResource(
+ None, {}
+ )
self.volume_snapshots = mock.Mock()
self.volume_snapshots.resource_class = fakes.FakeResource(None, {})
self.volume_type_access = mock.Mock()
@@ -89,1138 +89,1047 @@ class FakeVolumeClient:
class TestVolume(utils.TestCommand):
-
def setUp(self):
super().setUp()
self.app.client_manager.volume = FakeVolumeClient(
- endpoint=fakes.AUTH_URL,
- token=fakes.AUTH_TOKEN
+ endpoint=fakes.AUTH_URL, token=fakes.AUTH_TOKEN
)
self.app.client_manager.identity = identity_fakes.FakeIdentityv3Client(
- endpoint=fakes.AUTH_URL,
- token=fakes.AUTH_TOKEN
+ endpoint=fakes.AUTH_URL, token=fakes.AUTH_TOKEN
)
self.app.client_manager.image = image_fakes.FakeImagev2Client(
- endpoint=fakes.AUTH_URL,
- token=fakes.AUTH_TOKEN
+ endpoint=fakes.AUTH_URL, token=fakes.AUTH_TOKEN
)
-class FakeTransfer(object):
- """Fake one or more Transfer."""
-
- @staticmethod
- def create_one_transfer(attrs=None):
- """Create a fake transfer.
-
- :param dict attrs:
- A dictionary with all attributes of Transfer Request
- :return:
- A FakeResource object with volume_id, name, id.
- """
- # Set default attribute
- transfer_info = {
- 'volume_id': 'volume-id-' + uuid.uuid4().hex,
- 'name': 'fake_transfer_name',
- 'id': 'id-' + uuid.uuid4().hex,
- 'links': 'links-' + uuid.uuid4().hex,
- }
-
- # Overwrite default attributes if there are some attributes set
- attrs = attrs or {}
-
- transfer_info.update(attrs)
-
- transfer = fakes.FakeResource(
- None,
- transfer_info,
- loaded=True)
-
- return transfer
-
- @staticmethod
- def create_transfers(attrs=None, count=2):
- """Create multiple fake transfers.
-
- :param dict attrs:
- A dictionary with all attributes of transfer
- :param Integer count:
- The number of transfers to be faked
- :return:
- A list of FakeResource objects
- """
- transfers = []
- for n in range(0, count):
- transfers.append(FakeTransfer.create_one_transfer(attrs))
-
- return transfers
-
- @staticmethod
- def get_transfers(transfers=None, count=2):
- """Get an iterable MagicMock object with a list of faked transfers.
-
- If transfers list is provided, then initialize the Mock object with the
- list. Otherwise create one.
-
- :param List transfers:
- A list of FakeResource objects faking transfers
- :param Integer count:
- The number of transfers to be faked
- :return
- An iterable Mock object with side_effect set to a list of faked
- transfers
- """
- if transfers is None:
- transfers = FakeTransfer.create_transfers(count)
-
- return mock.Mock(side_effect=transfers)
-
-
-class FakeTypeAccess(object):
- """Fake one or more volume type access."""
-
- @staticmethod
- def create_one_type_access(attrs=None):
- """Create a fake volume type access for project.
-
- :param dict attrs:
- A dictionary with all attributes
- :return:
- A FakeResource object, with Volume_type_ID and Project_ID.
- """
- if attrs is None:
- attrs = {}
-
- # Set default attributes.
- type_access_attrs = {
- 'volume_type_id': 'volume-type-id-' + uuid.uuid4().hex,
- 'project_id': 'project-id-' + uuid.uuid4().hex,
- }
-
- # Overwrite default attributes.
- type_access_attrs.update(attrs)
-
- type_access = fakes.FakeResource(
- None,
- type_access_attrs,
- loaded=True)
-
- return type_access
-
-
-class FakeService(object):
- """Fake one or more Services."""
-
- @staticmethod
- def create_one_service(attrs=None):
- """Create a fake service.
-
- :param dict attrs:
- A dictionary with all attributes of service
- :return:
- A FakeResource object with host, status, etc.
- """
- # Set default attribute
- service_info = {
- 'host': 'host_test',
- 'binary': 'cinder_test',
- 'status': 'enabled',
- 'disabled_reason': 'LongHoliday-GoldenWeek',
- 'zone': 'fake_zone',
- 'updated_at': 'fake_date',
- 'state': 'fake_state',
- }
-
- # Overwrite default attributes if there are some attributes set
- attrs = attrs or {}
-
- service_info.update(attrs)
-
- service = fakes.FakeResource(
- None,
- service_info,
- loaded=True)
-
- return service
-
- @staticmethod
- def create_services(attrs=None, count=2):
- """Create multiple fake services.
-
- :param dict attrs:
- A dictionary with all attributes of service
- :param Integer count:
- The number of services to be faked
- :return:
- A list of FakeResource objects
- """
- services = []
- for n in range(0, count):
- services.append(FakeService.create_one_service(attrs))
-
- return services
-
-
-class FakeCapability(object):
- """Fake capability."""
-
- @staticmethod
- def create_one_capability(attrs=None):
- """Create a fake volume backend capability.
-
- :param dict attrs:
- A dictionary with all attributes of the Capabilities.
- :return:
- A FakeResource object with capability name and attrs.
- """
- # Set default attribute
- capability_info = {
- "namespace": "OS::Storage::Capabilities::fake",
- "vendor_name": "OpenStack",
- "volume_backend_name": "lvmdriver-1",
- "pool_name": "pool",
- "driver_version": "2.0.0",
- "storage_protocol": "iSCSI",
- "display_name": "Capabilities of Cinder LVM driver",
- "description": "Blah, blah.",
- "visibility": "public",
- "replication_targets": [],
- "properties": {
- "compression": {
- "title": "Compression",
- "description": "Enables compression.",
- "type": "boolean"
- },
- "qos": {
- "title": "QoS",
- "description": "Enables QoS.",
- "type": "boolean"
- },
- "replication": {
- "title": "Replication",
- "description": "Enables replication.",
- "type": "boolean"
- },
- "thin_provisioning": {
- "title": "Thin Provisioning",
- "description": "Sets thin provisioning.",
- "type": "boolean"
- }
- }
- }
+def create_one_transfer(attrs=None):
+ """Create a fake transfer.
- # Overwrite default attributes if there are some attributes set
- capability_info.update(attrs or {})
-
- capability = fakes.FakeResource(
- None,
- capability_info,
- loaded=True)
-
- return capability
-
-
-class FakePool(object):
- """Fake Pools."""
-
- @staticmethod
- def create_one_pool(attrs=None):
- """Create a fake pool.
-
- :param dict attrs:
- A dictionary with all attributes of the pool
- :return:
- A FakeResource object with pool name and attrs.
- """
- # Set default attribute
- pool_info = {
- 'name': 'host@lvmdriver-1#lvmdriver-1',
- 'storage_protocol': 'iSCSI',
- 'thick_provisioning_support': False,
- 'thin_provisioning_support': True,
- 'total_volumes': 99,
- 'total_capacity_gb': 1000.00,
- 'allocated_capacity_gb': 100,
- 'max_over_subscription_ratio': 200.0,
- }
-
- # Overwrite default attributes if there are some attributes set
- pool_info.update(attrs or {})
-
- pool = fakes.FakeResource(
- None,
- pool_info,
- loaded=True)
-
- return pool
-
-
-class FakeVolume(object):
- """Fake one or more volumes."""
-
- @staticmethod
- def create_one_volume(attrs=None):
- """Create a fake volume.
-
- :param dict attrs:
- A dictionary with all attributes of volume
- :return:
- A FakeResource object with id, name, status, etc.
- """
- attrs = attrs or {}
-
- # Set default attribute
- volume_info = {
- 'id': 'volume-id' + uuid.uuid4().hex,
- 'name': 'volume-name' + uuid.uuid4().hex,
- 'description': 'description' + uuid.uuid4().hex,
- 'status': random.choice(['available', 'in_use']),
- 'size': random.randint(1, 20),
- 'volume_type':
- random.choice(['fake_lvmdriver-1', 'fake_lvmdriver-2']),
- 'bootable':
- random.randint(0, 1),
- 'metadata': {
- 'key' + uuid.uuid4().hex: 'val' + uuid.uuid4().hex,
- 'key' + uuid.uuid4().hex: 'val' + uuid.uuid4().hex,
- 'key' + uuid.uuid4().hex: 'val' + uuid.uuid4().hex},
- 'snapshot_id': random.randint(1, 5),
- 'availability_zone': 'zone' + uuid.uuid4().hex,
- 'attachments': [{
- 'device': '/dev/' + uuid.uuid4().hex,
- 'server_id': uuid.uuid4().hex,
- }, ],
- }
+ :param dict attrs:
+ A dictionary with all attributes of Transfer Request
+ :return:
+ A FakeResource object with volume_id, name, id.
+ """
+ # Set default attribute
+ transfer_info = {
+ 'volume_id': 'volume-id-' + uuid.uuid4().hex,
+ 'name': 'fake_transfer_name',
+ 'id': 'id-' + uuid.uuid4().hex,
+ 'links': 'links-' + uuid.uuid4().hex,
+ }
- # Overwrite default attributes if there are some attributes set
- volume_info.update(attrs)
-
- volume = fakes.FakeResource(
- None,
- volume_info,
- loaded=True)
- return volume
-
- @staticmethod
- def create_volumes(attrs=None, count=2):
- """Create multiple fake volumes.
-
- :param dict attrs:
- A dictionary with all attributes of volume
- :param Integer count:
- The number of volumes to be faked
- :return:
- A list of FakeResource objects
- """
- volumes = []
- for n in range(0, count):
- volumes.append(FakeVolume.create_one_volume(attrs))
-
- return volumes
-
- @staticmethod
- def create_one_sdk_volume(attrs=None):
- """Create a fake volume.
-
- :param dict attrs:
- A dictionary with all attributes of volume
- :return:
- A FakeResource object with id, name, status, etc.
- """
- attrs = attrs or {}
-
- # Set default attribute
- volume_info = {
- 'id': 'volume-id' + uuid.uuid4().hex,
- 'name': 'volume-name' + uuid.uuid4().hex,
- 'description': 'description' + uuid.uuid4().hex,
- 'status': random.choice(['available', 'in_use']),
- 'size': random.randint(1, 20),
- 'volume_type':
- random.choice(['fake_lvmdriver-1', 'fake_lvmdriver-2']),
- 'bootable':
- random.choice(['true', 'false']),
- 'metadata': {
- 'key' + uuid.uuid4().hex: 'val' + uuid.uuid4().hex,
- 'key' + uuid.uuid4().hex: 'val' + uuid.uuid4().hex,
- 'key' + uuid.uuid4().hex: 'val' + uuid.uuid4().hex},
- 'snapshot_id': random.randint(1, 5),
- 'availability_zone': 'zone' + uuid.uuid4().hex,
- 'attachments': [{
- 'device': '/dev/' + uuid.uuid4().hex,
- 'server_id': uuid.uuid4().hex,
- }, ],
- }
+ # Overwrite default attributes if there are some attributes set
+ attrs = attrs or {}
- # Overwrite default attributes if there are some attributes set
- volume_info.update(attrs)
- return volume.Volume(**volume_info)
-
- @staticmethod
- def create_sdk_volumes(attrs=None, count=2):
- """Create multiple fake volumes.
-
- :param dict attrs:
- A dictionary with all attributes of volume
- :param Integer count:
- The number of volumes to be faked
- :return:
- A list of FakeResource objects
- """
- volumes = []
- for n in range(0, count):
- volumes.append(FakeVolume.create_one_sdk_volume(attrs))
-
- return volumes
-
- @staticmethod
- def get_volumes(volumes=None, count=2):
- """Get an iterable MagicMock object with a list of faked volumes.
-
- If volumes list is provided, then initialize the Mock object with the
- list. Otherwise create one.
-
- :param List volumes:
- A list of FakeResource objects faking volumes
- :param Integer count:
- The number of volumes to be faked
- :return
- An iterable Mock object with side_effect set to a list of faked
- volumes
- """
- if volumes is None:
- volumes = FakeVolume.create_volumes(count)
-
- return mock.Mock(side_effect=volumes)
-
- @staticmethod
- def get_volume_columns(volume=None):
- """Get the volume columns from a faked volume object.
-
- :param volume:
- A FakeResource objects faking volume
- :return
- A tuple which may include the following keys:
- ('id', 'name', 'description', 'status', 'size', 'volume_type',
- 'metadata', 'snapshot', 'availability_zone', 'attachments')
- """
- if volume is not None:
- return tuple(k for k in sorted(volume.keys()))
- return tuple([])
-
- @staticmethod
- def get_volume_data(volume=None):
- """Get the volume data from a faked volume object.
-
- :param volume:
- A FakeResource objects faking volume
- :return
- A tuple which may include the following values:
- ('ce26708d', 'fake_volume', 'fake description', 'available',
- 20, 'fake_lvmdriver-1', "Alpha='a', Beta='b', Gamma='g'",
- 1, 'nova', [{'device': '/dev/ice', 'server_id': '1233'}])
- """
- data_list = []
- if volume is not None:
- for x in sorted(volume.keys()):
- if x == 'tags':
- # The 'tags' should be format_list
- data_list.append(
- format_columns.ListColumn(volume.info.get(x)))
- else:
- data_list.append(volume.info.get(x))
- return tuple(data_list)
-
-
-class FakeAvailabilityZone(object):
- """Fake one or more volume availability zones (AZs)."""
-
- @staticmethod
- def create_one_availability_zone(attrs=None):
- """Create a fake AZ.
-
- :param dict attrs:
- A dictionary with all attributes
- :return:
- A FakeResource object with zoneName, zoneState, etc.
- """
- attrs = attrs or {}
-
- # Set default attributes.
- availability_zone = {
- 'zoneName': uuid.uuid4().hex,
- 'zoneState': {'available': True},
- }
+ transfer_info.update(attrs)
- # Overwrite default attributes.
- availability_zone.update(attrs)
-
- availability_zone = fakes.FakeResource(
- info=copy.deepcopy(availability_zone),
- loaded=True)
- return availability_zone
-
- @staticmethod
- def create_availability_zones(attrs=None, count=2):
- """Create multiple fake AZs.
-
- :param dict attrs:
- A dictionary with all attributes
- :param int count:
- The number of AZs to fake
- :return:
- A list of FakeResource objects faking the AZs
- """
- availability_zones = []
- for i in range(0, count):
- availability_zone = \
- FakeAvailabilityZone.create_one_availability_zone(attrs)
- availability_zones.append(availability_zone)
-
- return availability_zones
-
-
-class FakeBackup(object):
- """Fake one or more backup."""
-
- @staticmethod
- def create_one_backup(attrs=None):
- """Create a fake backup.
-
- :param dict attrs:
- A dictionary with all attributes
- :return:
- A FakeResource object with id, name, volume_id, etc.
- """
- attrs = attrs or {}
-
- # Set default attributes.
- backup_info = {
- "id": 'backup-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),
- "status": "error",
- "availability_zone": 'zone' + uuid.uuid4().hex,
- }
+ transfer = fakes.FakeResource(None, transfer_info, loaded=True)
- # Overwrite default attributes.
- backup_info.update(attrs)
-
- backup = fakes.FakeResource(
- info=copy.deepcopy(backup_info),
- loaded=True)
- return backup
-
- @staticmethod
- 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 FakeResource objects faking the backups
- """
- backups = []
- for i in range(0, count):
- backup = FakeBackup.create_one_backup(attrs)
- backups.append(backup)
-
- return backups
-
- @staticmethod
- 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 = FakeBackup.create_backups(count)
-
- return mock.Mock(side_effect=backups)
-
- @staticmethod
- def create_backup_record():
- """Gets a fake backup record for a given backup.
-
- :return: An "exported" backup record.
- """
-
- return {
- 'backup_service': 'cinder.backup.drivers.swift.SwiftBackupDriver',
- 'backup_url': 'eyJzdGF0dXMiOiAiYXZh',
- }
+ return transfer
- @staticmethod
- def import_backup_record():
- """Creates a fake backup record import response from a backup.
-
- :return: The fake backup object that was encoded.
- """
- return {
- 'backup': {
- 'id': 'backup.id',
- 'name': 'backup.name',
- 'links': [
- {'href': 'link1', 'rel': 'self'},
- {'href': 'link2', 'rel': 'bookmark'},
- ],
- },
- }
+def create_transfers(attrs=None, count=2):
+ """Create multiple fake transfers.
-class FakeConsistencyGroup(object):
- """Fake one or more consistency group."""
-
- @staticmethod
- def create_one_consistency_group(attrs=None):
- """Create a fake consistency group.
-
- :param dict attrs:
- A dictionary with all attributes
- :return:
- A FakeResource object with id, name, description, etc.
- """
- attrs = attrs or {}
-
- # Set default attributes.
- consistency_group_info = {
- "id": 'backup-id-' + uuid.uuid4().hex,
- "name": 'backup-name-' + uuid.uuid4().hex,
- "description": 'description-' + uuid.uuid4().hex,
- "status": "error",
- "availability_zone": 'zone' + uuid.uuid4().hex,
- "created_at": 'time-' + uuid.uuid4().hex,
- "volume_types": ['volume-type1'],
- }
+ :param dict attrs:
+ A dictionary with all attributes of transfer
+ :param Integer count:
+ The number of transfers to be faked
+ :return:
+ A list of FakeResource objects
+ """
+ transfers = []
+ for n in range(0, count):
+ transfers.append(create_one_transfer(attrs))
- # Overwrite default attributes.
- consistency_group_info.update(attrs)
-
- consistency_group = fakes.FakeResource(
- info=copy.deepcopy(consistency_group_info),
- loaded=True)
- return consistency_group
-
- @staticmethod
- def create_consistency_groups(attrs=None, count=2):
- """Create multiple fake consistency groups.
-
- :param dict attrs:
- A dictionary with all attributes
- :param int count:
- The number of consistency groups to fake
- :return:
- A list of FakeResource objects faking the consistency groups
- """
- consistency_groups = []
- for i in range(0, count):
- consistency_group = (
- FakeConsistencyGroup.create_one_consistency_group(attrs))
- consistency_groups.append(consistency_group)
-
- return consistency_groups
-
- @staticmethod
- def get_consistency_groups(consistency_groups=None, count=2):
- """Note:
-
- Get an iterable MagicMock object with a list of faked
- consistency_groups.
-
- If consistency_groups list is provided, then initialize
- the Mock object with the list. Otherwise create one.
-
- :param List consistency_groups:
- A list of FakeResource objects faking consistency_groups
- :param Integer count:
- The number of consistency_groups to be faked
- :return
- An iterable Mock object with side_effect set to a list of faked
- consistency_groups
- """
- if consistency_groups is None:
- consistency_groups = (FakeConsistencyGroup.
- create_consistency_groups(count))
-
- return mock.Mock(side_effect=consistency_groups)
-
-
-class FakeConsistencyGroupSnapshot(object):
- """Fake one or more consistency group snapshot."""
-
- @staticmethod
- def create_one_consistency_group_snapshot(attrs=None):
- """Create a fake consistency group snapshot.
-
- :param dict attrs:
- A dictionary with all attributes
- :return:
- A FakeResource object with id, name, description, etc.
- """
- attrs = attrs or {}
-
- # Set default attributes.
- consistency_group_snapshot_info = {
- "id": 'id-' + uuid.uuid4().hex,
- "name": 'backup-name-' + uuid.uuid4().hex,
- "description": 'description-' + uuid.uuid4().hex,
- "status": "error",
- "consistencygroup_id": 'consistency-group-id' + uuid.uuid4().hex,
- "created_at": 'time-' + uuid.uuid4().hex,
- }
+ return transfers
- # Overwrite default attributes.
- consistency_group_snapshot_info.update(attrs)
-
- consistency_group_snapshot = fakes.FakeResource(
- info=copy.deepcopy(consistency_group_snapshot_info),
- loaded=True)
- return consistency_group_snapshot
-
- @staticmethod
- def create_consistency_group_snapshots(attrs=None, count=2):
- """Create multiple fake consistency group snapshots.
-
- :param dict attrs:
- A dictionary with all attributes
- :param int count:
- The number of consistency group snapshots to fake
- :return:
- A list of FakeResource objects faking the
- consistency group snapshots
- """
- consistency_group_snapshots = []
- for i in range(0, count):
- consistency_group_snapshot = (
- FakeConsistencyGroupSnapshot.
- create_one_consistency_group_snapshot(attrs)
- )
- consistency_group_snapshots.append(consistency_group_snapshot)
-
- return consistency_group_snapshots
-
- @staticmethod
- def get_consistency_group_snapshots(snapshots=None, count=2):
- """Get an iterable MagicMock object with a list of faked cgsnapshots.
-
- If consistenct group snapshots list is provided, then initialize
- the Mock object with the list. Otherwise create one.
-
- :param List snapshots:
- A list of FakeResource objects faking consistency group snapshots
- :param Integer count:
- The number of consistency group snapshots to be faked
- :return
- An iterable Mock object with side_effect set to a list of faked
- consistency groups
- """
- if snapshots is None:
- snapshots = (FakeConsistencyGroupSnapshot.
- create_consistency_group_snapshots(count))
-
- return mock.Mock(side_effect=snapshots)
-
-
-class FakeExtension(object):
- """Fake one or more extension."""
-
- @staticmethod
- def create_one_extension(attrs=None):
- """Create a fake extension.
-
- :param dict attrs:
- A dictionary with all attributes
- :return:
- A FakeResource object with name, namespace, etc.
- """
- attrs = attrs or {}
-
- # Set default attributes.
- extension_info = {
- 'name': 'name-' + uuid.uuid4().hex,
- 'namespace': ('http://docs.openstack.org/'
- 'block-service/ext/scheduler-hints/api/v2'),
- 'description': 'description-' + uuid.uuid4().hex,
- 'updated': '2013-04-18T00:00:00+00:00',
- 'alias': 'OS-SCH-HNT',
- 'links': ('[{"href":'
- '"https://github.com/openstack/block-api", "type":'
- ' "text/html", "rel": "describedby"}]'),
- }
- # Overwrite default attributes.
- extension_info.update(attrs)
+def get_transfers(transfers=None, count=2):
+ """Get an iterable MagicMock object with a list of faked transfers.
- extension = fakes.FakeResource(
- info=copy.deepcopy(extension_info),
- loaded=True)
- return extension
+ If transfers list is provided, then initialize the Mock object with the
+ list. Otherwise create one.
+ :param List transfers:
+ A list of FakeResource objects faking transfers
+ :param Integer count:
+ The number of transfers to be faked
+ :return
+ An iterable Mock object with side_effect set to a list of faked
+ transfers
+ """
+ if transfers is None:
+ transfers = create_transfers(count)
+
+ return mock.Mock(side_effect=transfers)
+
+
+def create_one_type_access(attrs=None):
+ """Create a fake volume type access for project.
+
+ :param dict attrs:
+ A dictionary with all attributes
+ :return:
+ A FakeResource object, with Volume_type_ID and Project_ID.
+ """
+ if attrs is None:
+ attrs = {}
+
+ # Set default attributes.
+ type_access_attrs = {
+ 'volume_type_id': 'volume-type-id-' + uuid.uuid4().hex,
+ 'project_id': 'project-id-' + uuid.uuid4().hex,
+ }
+
+ # Overwrite default attributes.
+ type_access_attrs.update(attrs)
+
+ type_access = fakes.FakeResource(None, type_access_attrs, loaded=True)
+
+ return type_access
+
+
+def create_one_service(attrs=None):
+ """Create a fake service.
+
+ :param dict attrs:
+ A dictionary with all attributes of service
+ :return:
+ A FakeResource object with host, status, etc.
+ """
+ # Set default attribute
+ service_info = {
+ 'host': 'host_test',
+ 'binary': 'cinder_test',
+ 'status': 'enabled',
+ 'disabled_reason': 'LongHoliday-GoldenWeek',
+ 'zone': 'fake_zone',
+ 'updated_at': 'fake_date',
+ 'state': 'fake_state',
+ }
+
+ # Overwrite default attributes if there are some attributes set
+ attrs = attrs or {}
+
+ service_info.update(attrs)
+
+ service = fakes.FakeResource(None, service_info, loaded=True)
+
+ return service
+
+
+def create_services(attrs=None, count=2):
+ """Create multiple fake services.
+
+ :param dict attrs:
+ A dictionary with all attributes of service
+ :param Integer count:
+ The number of services to be faked
+ :return:
+ A list of FakeResource objects
+ """
+ services = []
+ for n in range(0, count):
+ services.append(create_one_service(attrs))
+
+ return services
+
+
+def create_one_capability(attrs=None):
+ """Create a fake volume backend capability.
+
+ :param dict attrs:
+ A dictionary with all attributes of the Capabilities.
+ :return:
+ A FakeResource object with capability name and attrs.
+ """
+ # Set default attribute
+ capability_info = {
+ "namespace": "OS::Storage::Capabilities::fake",
+ "vendor_name": "OpenStack",
+ "volume_backend_name": "lvmdriver-1",
+ "pool_name": "pool",
+ "driver_version": "2.0.0",
+ "storage_protocol": "iSCSI",
+ "display_name": "Capabilities of Cinder LVM driver",
+ "description": "Blah, blah.",
+ "visibility": "public",
+ "replication_targets": [],
+ "properties": {
+ "compression": {
+ "title": "Compression",
+ "description": "Enables compression.",
+ "type": "boolean",
+ },
+ "qos": {
+ "title": "QoS",
+ "description": "Enables QoS.",
+ "type": "boolean",
+ },
+ "replication": {
+ "title": "Replication",
+ "description": "Enables replication.",
+ "type": "boolean",
+ },
+ "thin_provisioning": {
+ "title": "Thin Provisioning",
+ "description": "Sets thin provisioning.",
+ "type": "boolean",
+ },
+ },
+ }
+
+ # Overwrite default attributes if there are some attributes set
+ capability_info.update(attrs or {})
+
+ capability = fakes.FakeResource(None, capability_info, loaded=True)
+
+ return capability
+
+
+def create_one_pool(attrs=None):
+ """Create a fake pool.
+
+ :param dict attrs:
+ A dictionary with all attributes of the pool
+ :return:
+ A FakeResource object with pool name and attrs.
+ """
+ # Set default attribute
+ pool_info = {
+ 'name': 'host@lvmdriver-1#lvmdriver-1',
+ 'storage_protocol': 'iSCSI',
+ 'thick_provisioning_support': False,
+ 'thin_provisioning_support': True,
+ 'total_volumes': 99,
+ 'total_capacity_gb': 1000.00,
+ 'allocated_capacity_gb': 100,
+ 'max_over_subscription_ratio': 200.0,
+ }
+
+ # Overwrite default attributes if there are some attributes set
+ pool_info.update(attrs or {})
+
+ pool = fakes.FakeResource(None, pool_info, loaded=True)
+
+ return pool
+
+
+def create_one_volume(attrs=None):
+ """Create a fake volume.
+
+ :param dict attrs:
+ A dictionary with all attributes of volume
+ :return:
+ A FakeResource object with id, name, status, etc.
+ """
+ attrs = attrs or {}
+
+ # Set default attribute
+ volume_info = {
+ 'id': 'volume-id' + uuid.uuid4().hex,
+ 'name': 'volume-name' + uuid.uuid4().hex,
+ 'description': 'description' + uuid.uuid4().hex,
+ 'status': random.choice(['available', 'in_use']),
+ 'size': random.randint(1, 20),
+ 'volume_type': random.choice(['fake_lvmdriver-1', 'fake_lvmdriver-2']),
+ 'bootable': random.randint(0, 1),
+ 'metadata': {
+ 'key' + uuid.uuid4().hex: 'val' + uuid.uuid4().hex,
+ 'key' + uuid.uuid4().hex: 'val' + uuid.uuid4().hex,
+ 'key' + uuid.uuid4().hex: 'val' + uuid.uuid4().hex,
+ },
+ 'snapshot_id': random.randint(1, 5),
+ 'availability_zone': 'zone' + uuid.uuid4().hex,
+ 'attachments': [
+ {
+ 'device': '/dev/' + uuid.uuid4().hex,
+ 'server_id': uuid.uuid4().hex,
+ },
+ ],
+ }
+
+ # Overwrite default attributes if there are some attributes set
+ volume_info.update(attrs)
+
+ volume = fakes.FakeResource(None, volume_info, loaded=True)
+ return volume
+
+
+def create_volumes(attrs=None, count=2):
+ """Create multiple fake volumes.
+
+ :param dict attrs:
+ A dictionary with all attributes of volume
+ :param Integer count:
+ The number of volumes to be faked
+ :return:
+ A list of FakeResource objects
+ """
+ volumes = []
+ for n in range(0, count):
+ volumes.append(create_one_volume(attrs))
+
+ return volumes
+
+
+def create_one_sdk_volume(attrs=None):
+ """Create a fake volume.
+
+ :param dict attrs:
+ A dictionary with all attributes of volume
+ :return:
+ A FakeResource object with id, name, status, etc.
+ """
+ attrs = attrs or {}
+
+ # Set default attribute
+ volume_info = {
+ 'id': 'volume-id' + uuid.uuid4().hex,
+ 'name': 'volume-name' + uuid.uuid4().hex,
+ 'description': 'description' + uuid.uuid4().hex,
+ 'status': random.choice(['available', 'in_use']),
+ 'size': random.randint(1, 20),
+ 'volume_type': random.choice(['fake_lvmdriver-1', 'fake_lvmdriver-2']),
+ 'bootable': random.choice(['true', 'false']),
+ 'metadata': {
+ 'key' + uuid.uuid4().hex: 'val' + uuid.uuid4().hex,
+ 'key' + uuid.uuid4().hex: 'val' + uuid.uuid4().hex,
+ 'key' + uuid.uuid4().hex: 'val' + uuid.uuid4().hex,
+ },
+ 'snapshot_id': random.randint(1, 5),
+ 'availability_zone': 'zone' + uuid.uuid4().hex,
+ 'attachments': [
+ {
+ 'device': '/dev/' + uuid.uuid4().hex,
+ 'server_id': uuid.uuid4().hex,
+ },
+ ],
+ }
+
+ # Overwrite default attributes if there are some attributes set
+ volume_info.update(attrs)
+ return volume.Volume(**volume_info)
+
+
+def create_sdk_volumes(attrs=None, count=2):
+ """Create multiple fake volumes.
+
+ :param dict attrs:
+ A dictionary with all attributes of volume
+ :param Integer count:
+ The number of volumes to be faked
+ :return:
+ A list of FakeResource objects
+ """
+ volumes = []
+ for n in range(0, count):
+ volumes.append(create_one_sdk_volume(attrs))
+
+ return volumes
+
+
+def get_volumes(volumes=None, count=2):
+ """Get an iterable MagicMock object with a list of faked volumes.
+
+ If volumes list is provided, then initialize the Mock object with the
+ list. Otherwise create one.
+
+ :param List volumes:
+ A list of FakeResource objects faking volumes
+ :param Integer count:
+ The number of volumes to be faked
+ :return
+ An iterable Mock object with side_effect set to a list of faked
+ volumes
+ """
+ if volumes is None:
+ volumes = create_volumes(count)
+
+ return mock.Mock(side_effect=volumes)
+
+
+def get_volume_columns(volume=None):
+ """Get the volume columns from a faked volume object.
+
+ :param volume:
+ A FakeResource objects faking volume
+ :return
+ A tuple which may include the following keys:
+ ('id', 'name', 'description', 'status', 'size', 'volume_type',
+ 'metadata', 'snapshot', 'availability_zone', 'attachments')
+ """
+ if volume is not None:
+ return tuple(k for k in sorted(volume.keys()))
+ return tuple([])
+
+
+def get_volume_data(volume=None):
+ """Get the volume data from a faked volume object.
+
+ :param volume:
+ A FakeResource objects faking volume
+ :return
+ A tuple which may include the following values:
+ ('ce26708d', 'fake_volume', 'fake description', 'available',
+ 20, 'fake_lvmdriver-1', "Alpha='a', Beta='b', Gamma='g'",
+ 1, 'nova', [{'device': '/dev/ice', 'server_id': '1233'}])
+ """
+ data_list = []
+ if volume is not None:
+ for x in sorted(volume.keys()):
+ if x == 'tags':
+ # The 'tags' should be format_list
+ data_list.append(format_columns.ListColumn(volume.info.get(x)))
+ else:
+ data_list.append(volume.info.get(x))
+ return tuple(data_list)
+
+
+def create_one_availability_zone(attrs=None):
+ """Create a fake AZ.
+
+ :param dict attrs:
+ A dictionary with all attributes
+ :return:
+ A FakeResource object with zoneName, zoneState, etc.
+ """
+ attrs = attrs or {}
+
+ # Set default attributes.
+ availability_zone = {
+ 'zoneName': uuid.uuid4().hex,
+ 'zoneState': {'available': True},
+ }
+
+ # Overwrite default attributes.
+ availability_zone.update(attrs)
+
+ availability_zone = fakes.FakeResource(
+ info=copy.deepcopy(availability_zone), loaded=True
+ )
+ return availability_zone
+
+
+def create_availability_zones(attrs=None, count=2):
+ """Create multiple fake AZs.
+
+ :param dict attrs:
+ A dictionary with all attributes
+ :param int count:
+ The number of AZs to fake
+ :return:
+ A list of FakeResource objects faking the AZs
+ """
+ availability_zones = []
+ for i in range(0, count):
+ availability_zone = create_one_availability_zone(attrs)
+ availability_zones.append(availability_zone)
+
+ return availability_zones
+
+
+def create_one_backup(attrs=None):
+ """Create a fake backup.
+
+ :param dict attrs:
+ A dictionary with all attributes
+ :return:
+ A FakeResource object with id, name, volume_id, etc.
+ """
+ attrs = attrs or {}
+
+ # Set default attributes.
+ backup_info = {
+ "id": 'backup-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),
+ "status": "error",
+ "availability_zone": 'zone' + uuid.uuid4().hex,
+ }
+
+ # Overwrite default attributes.
+ backup_info.update(attrs)
+
+ backup = fakes.FakeResource(info=copy.deepcopy(backup_info), loaded=True)
+ 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 FakeResource objects faking the backups
+ """
+ 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_backup_record():
+ """Gets a fake backup record for a given backup.
+
+ :return: An "exported" backup record.
+ """
+
+ return {
+ 'backup_service': 'cinder.backup.drivers.swift.SwiftBackupDriver',
+ 'backup_url': 'eyJzdGF0dXMiOiAiYXZh',
+ }
+
+
+def import_backup_record():
+ """Creates a fake backup record import response from a backup.
+
+ :return: The fake backup object that was encoded.
+ """
+ return {
+ 'backup': {
+ 'id': 'backup.id',
+ 'name': 'backup.name',
+ 'links': [
+ {'href': 'link1', 'rel': 'self'},
+ {'href': 'link2', 'rel': 'bookmark'},
+ ],
+ },
+ }
+
+
+def create_one_consistency_group(attrs=None):
+ """Create a fake consistency group.
+
+ :param dict attrs:
+ A dictionary with all attributes
+ :return:
+ A FakeResource object with id, name, description, etc.
+ """
+ attrs = attrs or {}
+
+ # Set default attributes.
+ consistency_group_info = {
+ "id": 'backup-id-' + uuid.uuid4().hex,
+ "name": 'backup-name-' + uuid.uuid4().hex,
+ "description": 'description-' + uuid.uuid4().hex,
+ "status": "error",
+ "availability_zone": 'zone' + uuid.uuid4().hex,
+ "created_at": 'time-' + uuid.uuid4().hex,
+ "volume_types": ['volume-type1'],
+ }
+
+ # Overwrite default attributes.
+ consistency_group_info.update(attrs)
+
+ consistency_group = fakes.FakeResource(
+ info=copy.deepcopy(consistency_group_info), loaded=True
+ )
+ return consistency_group
+
+
+def create_consistency_groups(attrs=None, count=2):
+ """Create multiple fake consistency groups.
+
+ :param dict attrs:
+ A dictionary with all attributes
+ :param int count:
+ The number of consistency groups to fake
+ :return:
+ A list of FakeResource objects faking the consistency groups
+ """
+ consistency_groups = []
+ for i in range(0, count):
+ consistency_group = create_one_consistency_group(attrs)
+ consistency_groups.append(consistency_group)
+
+ return consistency_groups
+
+
+def get_consistency_groups(consistency_groups=None, count=2):
+ """Note:
+
+ Get an iterable MagicMock object with a list of faked
+ consistency_groups.
+
+ If consistency_groups list is provided, then initialize
+ the Mock object with the list. Otherwise create one.
+
+ :param List consistency_groups:
+ A list of FakeResource objects faking consistency_groups
+ :param Integer count:
+ The number of consistency_groups to be faked
+ :return
+ An iterable Mock object with side_effect set to a list of faked
+ consistency_groups
+ """
+ if consistency_groups is None:
+ consistency_groups = create_consistency_groups(count)
+
+ return mock.Mock(side_effect=consistency_groups)
+
+
+def create_one_consistency_group_snapshot(attrs=None):
+ """Create a fake consistency group snapshot.
+
+ :param dict attrs:
+ A dictionary with all attributes
+ :return:
+ A FakeResource object with id, name, description, etc.
+ """
+ attrs = attrs or {}
+
+ # Set default attributes.
+ consistency_group_snapshot_info = {
+ "id": 'id-' + uuid.uuid4().hex,
+ "name": 'backup-name-' + uuid.uuid4().hex,
+ "description": 'description-' + uuid.uuid4().hex,
+ "status": "error",
+ "consistencygroup_id": 'consistency-group-id' + uuid.uuid4().hex,
+ "created_at": 'time-' + uuid.uuid4().hex,
+ }
+
+ # Overwrite default attributes.
+ consistency_group_snapshot_info.update(attrs)
+
+ consistency_group_snapshot = fakes.FakeResource(
+ info=copy.deepcopy(consistency_group_snapshot_info), loaded=True
+ )
+ return consistency_group_snapshot
+
+
+def create_consistency_group_snapshots(attrs=None, count=2):
+ """Create multiple fake consistency group snapshots.
+
+ :param dict attrs:
+ A dictionary with all attributes
+ :param int count:
+ The number of consistency group snapshots to fake
+ :return:
+ A list of FakeResource objects faking the
+ consistency group snapshots
+ """
+ consistency_group_snapshots = []
+ for i in range(0, count):
+ consistency_group_snapshot = create_one_consistency_group_snapshot(
+ attrs,
+ )
+ consistency_group_snapshots.append(consistency_group_snapshot)
+
+ return consistency_group_snapshots
+
+
+def get_consistency_group_snapshots(snapshots=None, count=2):
+ """Get an iterable MagicMock object with a list of faked cgsnapshots.
+
+ If consistenct group snapshots list is provided, then initialize
+ the Mock object with the list. Otherwise create one.
+
+ :param List snapshots:
+ A list of FakeResource objects faking consistency group snapshots
+ :param Integer count:
+ The number of consistency group snapshots to be faked
+ :return
+ An iterable Mock object with side_effect set to a list of faked
+ consistency groups
+ """
+ if snapshots is None:
+ snapshots = create_consistency_group_snapshots(count)
+
+ return mock.Mock(side_effect=snapshots)
+
+
+def create_one_extension(attrs=None):
+ """Create a fake extension.
+
+ :param dict attrs:
+ A dictionary with all attributes
+ :return:
+ A FakeResource object with name, namespace, etc.
+ """
+ attrs = attrs or {}
+
+ # Set default attributes.
+ extension_info = {
+ 'name': 'name-' + uuid.uuid4().hex,
+ 'namespace': (
+ 'http://docs.openstack.org/'
+ 'block-service/ext/scheduler-hints/api/v2'
+ ),
+ 'description': 'description-' + uuid.uuid4().hex,
+ 'updated': '2013-04-18T00:00:00+00:00',
+ 'alias': 'OS-SCH-HNT',
+ 'links': (
+ '[{"href":'
+ '"https://github.com/openstack/block-api", "type":'
+ ' "text/html", "rel": "describedby"}]'
+ ),
+ }
+
+ # Overwrite default attributes.
+ extension_info.update(attrs)
+
+ extension = fakes.FakeResource(
+ info=copy.deepcopy(extension_info), loaded=True
+ )
+ return extension
+
+
+def create_one_qos(attrs=None):
+ """Create a fake Qos specification.
+
+ :param dict attrs:
+ A dictionary with all attributes
+ :return:
+ A FakeResource object with id, name, consumer, etc.
+ """
+ attrs = attrs or {}
+
+ # Set default attributes.
+ qos_info = {
+ "id": 'qos-id-' + uuid.uuid4().hex,
+ "name": 'qos-name-' + uuid.uuid4().hex,
+ "consumer": 'front-end',
+ "specs": {"foo": "bar", "iops": "9001"},
+ }
+
+ # Overwrite default attributes.
+ qos_info.update(attrs)
+
+ qos = fakes.FakeResource(info=copy.deepcopy(qos_info), loaded=True)
+ return qos
+
+
+def create_one_qos_association(attrs=None):
+ """Create a fake Qos specification association.
+
+ :param dict attrs:
+ A dictionary with all attributes
+ :return:
+ A FakeResource object with id, name, association_type, etc.
+ """
+ attrs = attrs or {}
+
+ # Set default attributes.
+ qos_association_info = {
+ "id": 'type-id-' + uuid.uuid4().hex,
+ "name": 'type-name-' + uuid.uuid4().hex,
+ "association_type": 'volume_type',
+ }
+
+ # Overwrite default attributes.
+ qos_association_info.update(attrs)
+
+ qos_association = fakes.FakeResource(
+ info=copy.deepcopy(qos_association_info), loaded=True
+ )
+ return qos_association
+
+
+def create_qoses(attrs=None, count=2):
+ """Create multiple fake Qos specifications.
+
+ :param dict attrs:
+ A dictionary with all attributes
+ :param int count:
+ The number of Qos specifications to fake
+ :return:
+ A list of FakeResource objects faking the Qos specifications
+ """
+ qoses = []
+ for i in range(0, count):
+ qos = create_one_qos(attrs)
+ qoses.append(qos)
+
+ return qoses
+
+
+def get_qoses(qoses=None, count=2):
+ """Get an iterable MagicMock object with a list of faked qoses.
+
+ If qoses list is provided, then initialize the Mock object with the
+ list. Otherwise create one.
+
+ :param List qoses:
+ A list of FakeResource objects faking qoses
+ :param Integer count:
+ The number of qoses to be faked
+ :return
+ An iterable Mock object with side_effect set to a list of faked
+ qoses
+ """
+ if qoses is None:
+ qoses = create_qoses(count)
+
+ return mock.Mock(side_effect=qoses)
+
+
+def create_one_snapshot(attrs=None):
+ """Create a fake snapshot.
+
+ :param dict attrs:
+ A dictionary with all attributes
+ :return:
+ A FakeResource object with id, name, description, etc.
+ """
+ attrs = attrs or {}
+
+ # Set default attributes.
+ snapshot_info = {
+ "id": 'snapshot-id-' + uuid.uuid4().hex,
+ "name": 'snapshot-name-' + uuid.uuid4().hex,
+ "description": 'snapshot-description-' + uuid.uuid4().hex,
+ "size": 10,
+ "status": "available",
+ "metadata": {"foo": "bar"},
+ "created_at": "2015-06-03T18:49:19.000000",
+ "volume_id": 'vloume-id-' + uuid.uuid4().hex,
+ }
+
+ # Overwrite default attributes.
+ snapshot_info.update(attrs)
+
+ snapshot = fakes.FakeResource(
+ info=copy.deepcopy(snapshot_info), loaded=True
+ )
+ return snapshot
+
+
+def create_snapshots(attrs=None, count=2):
+ """Create multiple fake snapshots.
+
+ :param dict attrs:
+ A dictionary with all attributes
+ :param int count:
+ The number of snapshots to fake
+ :return:
+ A list of FakeResource objects faking the snapshots
+ """
+ snapshots = []
+ for i in range(0, count):
+ snapshot = create_one_snapshot(attrs)
+ snapshots.append(snapshot)
+
+ return snapshots
+
+
+def get_snapshots(snapshots=None, count=2):
+ """Get an iterable MagicMock object with a list of faked snapshots.
+
+ If snapshots list is provided, then initialize the Mock object with the
+ list. Otherwise create one.
+
+ :param List snapshots:
+ A list of FakeResource objects faking snapshots
+ :param Integer count:
+ The number of snapshots to be faked
+ :return
+ An iterable Mock object with side_effect set to a list of faked
+ snapshots
+ """
+ if snapshots is None:
+ snapshots = create_snapshots(count)
+
+ return mock.Mock(side_effect=snapshots)
+
+
+def create_one_volume_type(attrs=None, methods=None):
+ """Create a fake volume type.
+
+ :param dict attrs:
+ A dictionary with all attributes
+ :param dict methods:
+ A dictionary with all methods
+ :return:
+ A FakeResource object with id, name, description, etc.
+ """
+ attrs = attrs or {}
+ methods = methods or {}
+
+ # Set default attributes.
+ volume_type_info = {
+ "id": 'type-id-' + uuid.uuid4().hex,
+ "name": 'type-name-' + uuid.uuid4().hex,
+ "description": 'type-description-' + uuid.uuid4().hex,
+ "extra_specs": {"foo": "bar"},
+ "is_public": True,
+ }
+
+ # Overwrite default attributes.
+ volume_type_info.update(attrs)
+
+ volume_type = fakes.FakeResource(
+ info=copy.deepcopy(volume_type_info), methods=methods, loaded=True
+ )
+ return volume_type
+
+
+def create_volume_types(attrs=None, count=2):
+ """Create multiple fake volume_types.
+
+ :param dict attrs:
+ A dictionary with all attributes
+ :param int count:
+ The number of types to fake
+ :return:
+ A list of FakeResource objects faking the types
+ """
+ volume_types = []
+ for i in range(0, count):
+ volume_type = create_one_volume_type(attrs)
+ volume_types.append(volume_type)
+
+ return volume_types
+
+
+def get_volume_types(volume_types=None, count=2):
+ """Get an iterable MagicMock object with a list of faked volume types.
+
+ If volume_types list is provided, then initialize the Mock object with
+ the list. Otherwise create one.
+
+ :param List volume_types:
+ A list of FakeResource objects faking volume types
+ :param Integer count:
+ The number of volume types to be faked
+ :return
+ An iterable Mock object with side_effect set to a list of faked
+ volume types
+ """
+ if volume_types is None:
+ volume_types = create_volume_types(count)
+
+ return mock.Mock(side_effect=volume_types)
+
+
+def create_one_encryption_volume_type(attrs=None):
+ """Create a fake encryption volume type.
+
+ :param dict attrs:
+ A dictionary with all attributes
+ :return:
+ A FakeResource object with volume_type_id etc.
+ """
+ attrs = attrs or {}
+
+ # Set default attributes.
+ encryption_info = {
+ "volume_type_id": 'type-id-' + uuid.uuid4().hex,
+ 'provider': 'LuksEncryptor',
+ 'cipher': None,
+ 'key_size': None,
+ 'control_location': 'front-end',
+ }
+
+ # Overwrite default attributes.
+ encryption_info.update(attrs)
+
+ encryption_type = fakes.FakeResource(
+ info=copy.deepcopy(encryption_info), loaded=True
+ )
+ return encryption_type
+
+
+def create_one_vol_quota(attrs=None):
+ """Create one quota"""
+ attrs = attrs or {}
+
+ quota_attrs = {
+ 'id': 'project-id-' + uuid.uuid4().hex,
+ 'backups': 100,
+ 'backup_gigabytes': 100,
+ 'gigabytes': 10,
+ 'per_volume_gigabytes': 10,
+ 'snapshots': 0,
+ 'volumes': 10,
+ }
+
+ quota_attrs.update(attrs)
+
+ quota = fakes.FakeResource(info=copy.deepcopy(quota_attrs), loaded=True)
+ quota.project_id = quota_attrs['id']
+
+ return quota
+
+
+def create_one_default_vol_quota(attrs=None):
+ """Create one quota"""
+ attrs = attrs or {}
+
+ quota_attrs = {
+ 'id': 'project-id-' + uuid.uuid4().hex,
+ 'backups': 100,
+ 'backup_gigabytes': 100,
+ 'gigabytes': 100,
+ 'per_volume_gigabytes': 100,
+ 'snapshots': 100,
+ 'volumes': 100,
+ }
+
+ quota_attrs.update(attrs)
+
+ quota = fakes.FakeResource(info=copy.deepcopy(quota_attrs), loaded=True)
+ quota.project_id = quota_attrs['id']
+
+ return quota
+
+
+def create_one_detailed_quota(attrs=None):
+ """Create one quota"""
+ attrs = attrs or {}
-class FakeQos(object):
- """Fake one or more Qos specification."""
-
- @staticmethod
- def create_one_qos(attrs=None):
- """Create a fake Qos specification.
-
- :param dict attrs:
- A dictionary with all attributes
- :return:
- A FakeResource object with id, name, consumer, etc.
- """
- attrs = attrs or {}
-
- # Set default attributes.
- qos_info = {
- "id": 'qos-id-' + uuid.uuid4().hex,
- "name": 'qos-name-' + uuid.uuid4().hex,
- "consumer": 'front-end',
- "specs": {"foo": "bar", "iops": "9001"},
- }
-
- # Overwrite default attributes.
- qos_info.update(attrs)
-
- qos = fakes.FakeResource(
- info=copy.deepcopy(qos_info),
- loaded=True)
- return qos
-
- @staticmethod
- def create_one_qos_association(attrs=None):
- """Create a fake Qos specification association.
-
- :param dict attrs:
- A dictionary with all attributes
- :return:
- A FakeResource object with id, name, association_type, etc.
- """
- attrs = attrs or {}
-
- # Set default attributes.
- qos_association_info = {
- "id": 'type-id-' + uuid.uuid4().hex,
- "name": 'type-name-' + uuid.uuid4().hex,
- "association_type": 'volume_type',
- }
-
- # Overwrite default attributes.
- qos_association_info.update(attrs)
-
- qos_association = fakes.FakeResource(
- info=copy.deepcopy(qos_association_info),
- loaded=True)
- return qos_association
-
- @staticmethod
- def create_qoses(attrs=None, count=2):
- """Create multiple fake Qos specifications.
-
- :param dict attrs:
- A dictionary with all attributes
- :param int count:
- The number of Qos specifications to fake
- :return:
- A list of FakeResource objects faking the Qos specifications
- """
- qoses = []
- for i in range(0, count):
- qos = FakeQos.create_one_qos(attrs)
- qoses.append(qos)
-
- return qoses
-
- @staticmethod
- def get_qoses(qoses=None, count=2):
- """Get an iterable MagicMock object with a list of faked qoses.
-
- If qoses list is provided, then initialize the Mock object with the
- list. Otherwise create one.
-
- :param List qoses:
- A list of FakeResource objects faking qoses
- :param Integer count:
- The number of qoses to be faked
- :return
- An iterable Mock object with side_effect set to a list of faked
- qoses
- """
- if qoses is None:
- qoses = FakeQos.create_qoses(count)
-
- return mock.Mock(side_effect=qoses)
-
-
-class FakeSnapshot(object):
- """Fake one or more snapshot."""
-
- @staticmethod
- def create_one_snapshot(attrs=None):
- """Create a fake snapshot.
-
- :param dict attrs:
- A dictionary with all attributes
- :return:
- A FakeResource object with id, name, description, etc.
- """
- attrs = attrs or {}
-
- # Set default attributes.
- snapshot_info = {
- "id": 'snapshot-id-' + uuid.uuid4().hex,
- "name": 'snapshot-name-' + uuid.uuid4().hex,
- "description": 'snapshot-description-' + uuid.uuid4().hex,
- "size": 10,
- "status": "available",
- "metadata": {"foo": "bar"},
- "created_at": "2015-06-03T18:49:19.000000",
- "volume_id": 'vloume-id-' + uuid.uuid4().hex,
- }
-
- # Overwrite default attributes.
- snapshot_info.update(attrs)
-
- snapshot = fakes.FakeResource(
- info=copy.deepcopy(snapshot_info),
- loaded=True)
- return snapshot
-
- @staticmethod
- def create_snapshots(attrs=None, count=2):
- """Create multiple fake snapshots.
-
- :param dict attrs:
- A dictionary with all attributes
- :param int count:
- The number of snapshots to fake
- :return:
- A list of FakeResource objects faking the snapshots
- """
- snapshots = []
- for i in range(0, count):
- snapshot = FakeSnapshot.create_one_snapshot(attrs)
- snapshots.append(snapshot)
-
- return snapshots
-
- @staticmethod
- def get_snapshots(snapshots=None, count=2):
- """Get an iterable MagicMock object with a list of faked snapshots.
-
- If snapshots list is provided, then initialize the Mock object with the
- list. Otherwise create one.
-
- :param List snapshots:
- A list of FakeResource objects faking snapshots
- :param Integer count:
- The number of snapshots to be faked
- :return
- An iterable Mock object with side_effect set to a list of faked
- snapshots
- """
- if snapshots is None:
- snapshots = FakeSnapshot.create_snapshots(count)
-
- return mock.Mock(side_effect=snapshots)
-
-
-class FakeVolumeType(object):
- """Fake one or more volume type."""
-
- @staticmethod
- def create_one_volume_type(attrs=None, methods=None):
- """Create a fake volume type.
-
- :param dict attrs:
- A dictionary with all attributes
- :param dict methods:
- A dictionary with all methods
- :return:
- A FakeResource object with id, name, description, etc.
- """
- attrs = attrs or {}
- methods = methods or {}
-
- # Set default attributes.
- volume_type_info = {
- "id": 'type-id-' + uuid.uuid4().hex,
- "name": 'type-name-' + uuid.uuid4().hex,
- "description": 'type-description-' + uuid.uuid4().hex,
- "extra_specs": {"foo": "bar"},
- "is_public": True,
- }
-
- # Overwrite default attributes.
- volume_type_info.update(attrs)
-
- volume_type = fakes.FakeResource(
- info=copy.deepcopy(volume_type_info),
- methods=methods,
- loaded=True)
- return volume_type
-
- @staticmethod
- def create_volume_types(attrs=None, count=2):
- """Create multiple fake volume_types.
-
- :param dict attrs:
- A dictionary with all attributes
- :param int count:
- The number of types to fake
- :return:
- A list of FakeResource objects faking the types
- """
- volume_types = []
- for i in range(0, count):
- volume_type = FakeVolumeType.create_one_volume_type(attrs)
- volume_types.append(volume_type)
-
- return volume_types
-
- @staticmethod
- def get_volume_types(volume_types=None, count=2):
- """Get an iterable MagicMock object with a list of faked volume types.
-
- If volume_types list is provided, then initialize the Mock object with
- the list. Otherwise create one.
-
- :param List volume_types:
- A list of FakeResource objects faking volume types
- :param Integer count:
- The number of volume types to be faked
- :return
- An iterable Mock object with side_effect set to a list of faked
- volume types
- """
- if volume_types is None:
- volume_types = FakeVolumeType.create_volume_types(count)
-
- return mock.Mock(side_effect=volume_types)
-
- @staticmethod
- def create_one_encryption_volume_type(attrs=None):
- """Create a fake encryption volume type.
-
- :param dict attrs:
- A dictionary with all attributes
- :return:
- A FakeResource object with volume_type_id etc.
- """
- attrs = attrs or {}
-
- # Set default attributes.
- encryption_info = {
- "volume_type_id": 'type-id-' + uuid.uuid4().hex,
- 'provider': 'LuksEncryptor',
- 'cipher': None,
- 'key_size': None,
- 'control_location': 'front-end',
- }
-
- # Overwrite default attributes.
- encryption_info.update(attrs)
-
- encryption_type = fakes.FakeResource(
- info=copy.deepcopy(encryption_info),
- loaded=True)
- return encryption_type
-
-
-class FakeQuota(object):
- """Fake quota"""
-
- @staticmethod
- def create_one_vol_quota(attrs=None):
- """Create one quota"""
- attrs = attrs or {}
-
- quota_attrs = {
- 'id': 'project-id-' + uuid.uuid4().hex,
- 'backups': 100,
- 'backup_gigabytes': 100,
- 'gigabytes': 10,
- 'per_volume_gigabytes': 10,
- 'snapshots': 0,
- 'volumes': 10}
-
- quota_attrs.update(attrs)
-
- quota = fakes.FakeResource(
- info=copy.deepcopy(quota_attrs),
- loaded=True)
- quota.project_id = quota_attrs['id']
-
- return quota
-
- @staticmethod
- def create_one_default_vol_quota(attrs=None):
- """Create one quota"""
- attrs = attrs or {}
-
- quota_attrs = {
- 'id': 'project-id-' + uuid.uuid4().hex,
- 'backups': 100,
- 'backup_gigabytes': 100,
- 'gigabytes': 100,
- 'per_volume_gigabytes': 100,
- 'snapshots': 100,
- 'volumes': 100}
-
- quota_attrs.update(attrs)
-
- quota = fakes.FakeResource(
- info=copy.deepcopy(quota_attrs),
- loaded=True)
- quota.project_id = quota_attrs['id']
-
- return quota
-
- @staticmethod
- def create_one_detailed_quota(attrs=None):
- """Create one quota"""
- attrs = attrs or {}
-
- quota_attrs = {
- 'volumes': {'limit': 3, 'in_use': 1, 'reserved': 0},
- 'per_volume_gigabytes': {'limit': -1, 'in_use': 0, 'reserved': 0},
- 'snapshots': {'limit': 10, 'in_use': 0, 'reserved': 0},
- 'gigabytes': {'limit': 1000, 'in_use': 5, 'reserved': 0},
- 'backups': {'limit': 10, 'in_use': 0, 'reserved': 0},
- 'backup_gigabytes': {'limit': 1000, 'in_use': 0, 'reserved': 0},
- 'volumes_lvmdriver-1': {'limit': -1, 'in_use': 1, 'reserved': 0},
- 'gigabytes_lvmdriver-1': {'limit': -1, 'in_use': 5, 'reserved': 0},
- 'snapshots_lvmdriver-1': {'limit': -1, 'in_use': 0, 'reserved': 0},
- 'volumes___DEFAULT__': {'limit': -1, 'in_use': 0, 'reserved': 0},
- 'gigabytes___DEFAULT__': {'limit': -1, 'in_use': 0, 'reserved': 0},
- 'snapshots___DEFAULT__': {'limit': -1, 'in_use': 0, 'reserved': 0},
- 'groups': {'limit': 10, 'in_use': 0, 'reserved': 0},
- 'id': uuid.uuid4().hex,
- }
- quota_attrs.update(attrs)
+ quota_attrs = {
+ 'volumes': {'limit': 3, 'in_use': 1, 'reserved': 0},
+ 'per_volume_gigabytes': {'limit': -1, 'in_use': 0, 'reserved': 0},
+ 'snapshots': {'limit': 10, 'in_use': 0, 'reserved': 0},
+ 'gigabytes': {'limit': 1000, 'in_use': 5, 'reserved': 0},
+ 'backups': {'limit': 10, 'in_use': 0, 'reserved': 0},
+ 'backup_gigabytes': {'limit': 1000, 'in_use': 0, 'reserved': 0},
+ 'volumes_lvmdriver-1': {'limit': -1, 'in_use': 1, 'reserved': 0},
+ 'gigabytes_lvmdriver-1': {'limit': -1, 'in_use': 5, 'reserved': 0},
+ 'snapshots_lvmdriver-1': {'limit': -1, 'in_use': 0, 'reserved': 0},
+ 'volumes___DEFAULT__': {'limit': -1, 'in_use': 0, 'reserved': 0},
+ 'gigabytes___DEFAULT__': {'limit': -1, 'in_use': 0, 'reserved': 0},
+ 'snapshots___DEFAULT__': {'limit': -1, 'in_use': 0, 'reserved': 0},
+ 'groups': {'limit': 10, 'in_use': 0, 'reserved': 0},
+ 'id': uuid.uuid4().hex,
+ }
+ quota_attrs.update(attrs)
- quota = fakes.FakeResource(
- info=copy.deepcopy(quota_attrs),
- loaded=True)
+ quota = fakes.FakeResource(info=copy.deepcopy(quota_attrs), loaded=True)
- return quota
+ return quota
class FakeLimits(object):
@@ -1237,37 +1146,39 @@ class FakeLimits(object):
'maxTotalVolumes': 10,
'totalVolumesUsed': 4,
'totalBackupsUsed': 0,
- 'totalGigabytesUsed': 35
+ 'totalGigabytesUsed': 35,
}
absolute_attrs = absolute_attrs or {}
self.absolute_limits_attrs.update(absolute_attrs)
- self.rate_limits_attrs = [{
- "uri": "*",
- "limit": [
- {
- "value": 10,
- "verb": "POST",
- "remaining": 2,
- "unit": "MINUTE",
- "next-available": "2011-12-15T22:42:45Z"
- },
- {
- "value": 10,
- "verb": "PUT",
- "remaining": 2,
- "unit": "MINUTE",
- "next-available": "2011-12-15T22:42:45Z"
- },
- {
- "value": 100,
- "verb": "DELETE",
- "remaining": 100,
- "unit": "MINUTE",
- "next-available": "2011-12-15T22:42:45Z"
- }
- ]
- }]
+ self.rate_limits_attrs = [
+ {
+ "uri": "*",
+ "limit": [
+ {
+ "value": 10,
+ "verb": "POST",
+ "remaining": 2,
+ "unit": "MINUTE",
+ "next-available": "2011-12-15T22:42:45Z",
+ },
+ {
+ "value": 10,
+ "verb": "PUT",
+ "remaining": 2,
+ "unit": "MINUTE",
+ "next-available": "2011-12-15T22:42:45Z",
+ },
+ {
+ "value": 100,
+ "verb": "DELETE",
+ "remaining": 100,
+ "unit": "MINUTE",
+ "next-available": "2011-12-15T22:42:45Z",
+ },
+ ],
+ }
+ ]
@property
def absolute(self):
@@ -1285,18 +1196,30 @@ class FakeLimits(object):
for group in self.rate_limits_attrs:
uri = group['uri']
for rate in group['limit']:
- yield FakeRateLimit(rate['verb'], uri, rate['value'],
- rate['remaining'], rate['unit'],
- rate['next-available'])
+ yield FakeRateLimit(
+ rate['verb'],
+ uri,
+ rate['value'],
+ rate['remaining'],
+ rate['unit'],
+ rate['next-available'],
+ )
def rate_limits(self):
reference_data = []
for group in self.rate_limits_attrs:
uri = group['uri']
for rate in group['limit']:
- reference_data.append((rate['verb'], uri, rate['value'],
- rate['remaining'], rate['unit'],
- rate['next-available']))
+ reference_data.append(
+ (
+ rate['verb'],
+ uri,
+ rate['value'],
+ rate['remaining'],
+ rate['unit'],
+ rate['next-available'],
+ )
+ )
return reference_data
@@ -1311,8 +1234,7 @@ class FakeAbsoluteLimit(object):
class FakeRateLimit(object):
"""Data model that represents a flattened view of a single rate limit."""
- def __init__(self, verb, uri, value, remain,
- unit, next_available):
+ def __init__(self, verb, uri, value, remain, unit, next_available):
self.verb = verb
self.uri = uri
self.value = value
diff --git a/openstackclient/tests/unit/volume/v2/test_backup_record.py b/openstackclient/tests/unit/volume/v2/test_backup_record.py
index 0e24174c..aa9a25a2 100644
--- a/openstackclient/tests/unit/volume/v2/test_backup_record.py
+++ b/openstackclient/tests/unit/volume/v2/test_backup_record.py
@@ -19,7 +19,7 @@ from openstackclient.volume.v2 import backup_record
class TestBackupRecord(volume_fakes.TestVolume):
def setUp(self):
- super(TestBackupRecord, self).setUp()
+ super().setUp()
self.backups_mock = self.app.client_manager.volume.backups
self.backups_mock.reset_mock()
@@ -27,12 +27,13 @@ class TestBackupRecord(volume_fakes.TestVolume):
class TestBackupRecordExport(TestBackupRecord):
- new_backup = volume_fakes.FakeBackup.create_one_backup(
- attrs={'volume_id': 'a54708a2-0388-4476-a909-09579f885c25'})
- new_record = volume_fakes.FakeBackup.create_backup_record()
+ new_backup = volume_fakes.create_one_backup(
+ attrs={'volume_id': 'a54708a2-0388-4476-a909-09579f885c25'},
+ )
+ new_record = volume_fakes.create_backup_record()
def setUp(self):
- super(TestBackupRecordExport, self).setUp()
+ super().setUp()
self.backups_mock.export_record.return_value = self.new_record
self.backups_mock.get.return_value = self.new_backup
@@ -81,12 +82,13 @@ class TestBackupRecordExport(TestBackupRecord):
class TestBackupRecordImport(TestBackupRecord):
- new_backup = volume_fakes.FakeBackup.create_one_backup(
- attrs={'volume_id': 'a54708a2-0388-4476-a909-09579f885c25'})
- new_import = volume_fakes.FakeBackup.import_backup_record()
+ new_backup = volume_fakes.create_one_backup(
+ attrs={'volume_id': 'a54708a2-0388-4476-a909-09579f885c25'},
+ )
+ new_import = volume_fakes.import_backup_record()
def setUp(self):
- super(TestBackupRecordImport, self).setUp()
+ super().setUp()
self.backups_mock.import_record.return_value = self.new_import
diff --git a/openstackclient/tests/unit/volume/v2/test_consistency_group.py b/openstackclient/tests/unit/volume/v2/test_consistency_group.py
index 434d0cf9..7ef4a08e 100644
--- a/openstackclient/tests/unit/volume/v2/test_consistency_group.py
+++ b/openstackclient/tests/unit/volume/v2/test_consistency_group.py
@@ -26,7 +26,7 @@ from openstackclient.volume.v2 import consistency_group
class TestConsistencyGroup(volume_fakes.TestVolume):
def setUp(self):
- super(TestConsistencyGroup, self).setUp()
+ super().setUp()
# Get a shortcut to the TransferManager Mock
self.consistencygroups_mock = (
@@ -47,11 +47,10 @@ class TestConsistencyGroup(volume_fakes.TestVolume):
class TestConsistencyGroupAddVolume(TestConsistencyGroup):
- _consistency_group = (
- volume_fakes.FakeConsistencyGroup.create_one_consistency_group())
+ _consistency_group = volume_fakes.create_one_consistency_group()
def setUp(self):
- super(TestConsistencyGroupAddVolume, self).setUp()
+ super().setUp()
self.consistencygroups_mock.get.return_value = (
self._consistency_group)
@@ -60,7 +59,7 @@ class TestConsistencyGroupAddVolume(TestConsistencyGroup):
consistency_group.AddVolumeToConsistencyGroup(self.app, None)
def test_add_one_volume_to_consistency_group(self):
- volume = volume_fakes.FakeVolume.create_one_volume()
+ volume = volume_fakes.create_one_volume()
self.volumes_mock.get.return_value = volume
arglist = [
self._consistency_group.id,
@@ -85,8 +84,8 @@ class TestConsistencyGroupAddVolume(TestConsistencyGroup):
self.assertIsNone(result)
def test_add_multiple_volumes_to_consistency_group(self):
- volumes = volume_fakes.FakeVolume.create_volumes(count=2)
- self.volumes_mock.get = volume_fakes.FakeVolume.get_volumes(volumes)
+ volumes = volume_fakes.create_volumes(count=2)
+ self.volumes_mock.get = volume_fakes.get_volumes(volumes)
arglist = [
self._consistency_group.id,
volumes[0].id,
@@ -112,8 +111,9 @@ class TestConsistencyGroupAddVolume(TestConsistencyGroup):
@mock.patch.object(consistency_group.LOG, 'error')
def test_add_multiple_volumes_to_consistency_group_with_exception(
- self, mock_error):
- volume = volume_fakes.FakeVolume.create_one_volume()
+ self, mock_error,
+ ):
+ volume = volume_fakes.create_one_volume()
arglist = [
self._consistency_group.id,
volume.id,
@@ -148,13 +148,10 @@ class TestConsistencyGroupAddVolume(TestConsistencyGroup):
class TestConsistencyGroupCreate(TestConsistencyGroup):
- volume_type = volume_fakes.FakeVolumeType.create_one_volume_type()
- new_consistency_group = (
- volume_fakes.FakeConsistencyGroup.create_one_consistency_group())
+ volume_type = volume_fakes.create_one_volume_type()
+ new_consistency_group = volume_fakes.create_one_consistency_group()
consistency_group_snapshot = (
- volume_fakes.
- FakeConsistencyGroupSnapshot.
- create_one_consistency_group_snapshot()
+ volume_fakes.create_one_consistency_group_snapshot()
)
columns = (
@@ -177,7 +174,7 @@ class TestConsistencyGroupCreate(TestConsistencyGroup):
)
def setUp(self):
- super(TestConsistencyGroupCreate, self).setUp()
+ super().setUp()
self.consistencygroups_mock.create.return_value = (
self.new_consistency_group)
self.consistencygroups_mock.create_from_src.return_value = (
@@ -313,13 +310,14 @@ class TestConsistencyGroupCreate(TestConsistencyGroup):
class TestConsistencyGroupDelete(TestConsistencyGroup):
consistency_groups =\
- volume_fakes.FakeConsistencyGroup.create_consistency_groups(count=2)
+ volume_fakes.create_consistency_groups(count=2)
def setUp(self):
- super(TestConsistencyGroupDelete, self).setUp()
+ super().setUp()
- self.consistencygroups_mock.get = volume_fakes.FakeConsistencyGroup.\
- get_consistency_groups(self.consistency_groups)
+ self.consistencygroups_mock.get = volume_fakes.get_consistency_groups(
+ self.consistency_groups,
+ )
self.consistencygroups_mock.delete.return_value = None
# Get the command object to mock
@@ -409,8 +407,7 @@ class TestConsistencyGroupDelete(TestConsistencyGroup):
class TestConsistencyGroupList(TestConsistencyGroup):
- consistency_groups = (
- volume_fakes.FakeConsistencyGroup.create_consistency_groups(count=2))
+ consistency_groups = volume_fakes.create_consistency_groups(count=2)
columns = [
'ID',
@@ -444,7 +441,7 @@ class TestConsistencyGroupList(TestConsistencyGroup):
))
def setUp(self):
- super(TestConsistencyGroupList, self).setUp()
+ super().setUp()
self.consistencygroups_mock.list.return_value = self.consistency_groups
# Get the command to test
@@ -502,11 +499,10 @@ class TestConsistencyGroupList(TestConsistencyGroup):
class TestConsistencyGroupRemoveVolume(TestConsistencyGroup):
- _consistency_group = (
- volume_fakes.FakeConsistencyGroup.create_one_consistency_group())
+ _consistency_group = volume_fakes.create_one_consistency_group()
def setUp(self):
- super(TestConsistencyGroupRemoveVolume, self).setUp()
+ super().setUp()
self.consistencygroups_mock.get.return_value = (
self._consistency_group)
@@ -515,7 +511,7 @@ class TestConsistencyGroupRemoveVolume(TestConsistencyGroup):
consistency_group.RemoveVolumeFromConsistencyGroup(self.app, None)
def test_remove_one_volume_from_consistency_group(self):
- volume = volume_fakes.FakeVolume.create_one_volume()
+ volume = volume_fakes.create_one_volume()
self.volumes_mock.get.return_value = volume
arglist = [
self._consistency_group.id,
@@ -540,8 +536,8 @@ class TestConsistencyGroupRemoveVolume(TestConsistencyGroup):
self.assertIsNone(result)
def test_remove_multi_volumes_from_consistency_group(self):
- volumes = volume_fakes.FakeVolume.create_volumes(count=2)
- self.volumes_mock.get = volume_fakes.FakeVolume.get_volumes(volumes)
+ volumes = volume_fakes.create_volumes(count=2)
+ self.volumes_mock.get = volume_fakes.get_volumes(volumes)
arglist = [
self._consistency_group.id,
volumes[0].id,
@@ -567,8 +563,10 @@ class TestConsistencyGroupRemoveVolume(TestConsistencyGroup):
@mock.patch.object(consistency_group.LOG, 'error')
def test_remove_multiple_volumes_from_consistency_group_with_exception(
- self, mock_error):
- volume = volume_fakes.FakeVolume.create_one_volume()
+ self,
+ mock_error,
+ ):
+ volume = volume_fakes.create_one_volume()
arglist = [
self._consistency_group.id,
volume.id,
@@ -603,11 +601,10 @@ class TestConsistencyGroupRemoveVolume(TestConsistencyGroup):
class TestConsistencyGroupSet(TestConsistencyGroup):
- consistency_group = (
- volume_fakes.FakeConsistencyGroup.create_one_consistency_group())
+ consistency_group = volume_fakes.create_one_consistency_group()
def setUp(self):
- super(TestConsistencyGroupSet, self).setUp()
+ super().setUp()
self.consistencygroups_mock.get.return_value = (
self.consistency_group)
@@ -677,10 +674,9 @@ class TestConsistencyGroupShow(TestConsistencyGroup):
)
def setUp(self):
- super(TestConsistencyGroupShow, self).setUp()
+ super().setUp()
- self.consistency_group = (
- volume_fakes.FakeConsistencyGroup.create_one_consistency_group())
+ self.consistency_group = volume_fakes.create_one_consistency_group()
self.data = (
self.consistency_group.availability_zone,
self.consistency_group.created_at,
diff --git a/openstackclient/tests/unit/volume/v2/test_consistency_group_snapshot.py b/openstackclient/tests/unit/volume/v2/test_consistency_group_snapshot.py
index 2202b85b..e3c738c8 100644
--- a/openstackclient/tests/unit/volume/v2/test_consistency_group_snapshot.py
+++ b/openstackclient/tests/unit/volume/v2/test_consistency_group_snapshot.py
@@ -35,12 +35,9 @@ class TestConsistencyGroupSnapshot(volume_fakes.TestVolume):
class TestConsistencyGroupSnapshotCreate(TestConsistencyGroupSnapshot):
_consistency_group_snapshot = (
- volume_fakes.
- FakeConsistencyGroupSnapshot.
- create_one_consistency_group_snapshot()
+ volume_fakes.create_one_consistency_group_snapshot()
)
- consistency_group = (
- volume_fakes.FakeConsistencyGroup.create_one_consistency_group())
+ consistency_group = volume_fakes.create_one_consistency_group()
columns = (
'consistencygroup_id',
@@ -124,16 +121,16 @@ class TestConsistencyGroupSnapshotCreate(TestConsistencyGroupSnapshot):
class TestConsistencyGroupSnapshotDelete(TestConsistencyGroupSnapshot):
consistency_group_snapshots = (
- volume_fakes.FakeConsistencyGroupSnapshot.
- create_consistency_group_snapshots(count=2)
+ volume_fakes.create_consistency_group_snapshots(count=2)
)
def setUp(self):
super(TestConsistencyGroupSnapshotDelete, self).setUp()
self.cgsnapshots_mock.get = (
- volume_fakes.FakeConsistencyGroupSnapshot.
- get_consistency_group_snapshots(self.consistency_group_snapshots)
+ volume_fakes.get_consistency_group_snapshots(
+ self.consistency_group_snapshots
+ )
)
self.cgsnapshots_mock.delete.return_value = None
@@ -178,12 +175,9 @@ class TestConsistencyGroupSnapshotDelete(TestConsistencyGroupSnapshot):
class TestConsistencyGroupSnapshotList(TestConsistencyGroupSnapshot):
consistency_group_snapshots = (
- volume_fakes.FakeConsistencyGroupSnapshot.
- create_consistency_group_snapshots(count=2)
- )
- consistency_group = (
- volume_fakes.FakeConsistencyGroup.create_one_consistency_group()
+ volume_fakes.create_consistency_group_snapshots(count=2)
)
+ consistency_group = volume_fakes.create_one_consistency_group()
columns = [
'ID',
@@ -306,9 +300,7 @@ class TestConsistencyGroupSnapshotList(TestConsistencyGroupSnapshot):
class TestConsistencyGroupSnapshotShow(TestConsistencyGroupSnapshot):
_consistency_group_snapshot = (
- volume_fakes.
- FakeConsistencyGroupSnapshot.
- create_one_consistency_group_snapshot()
+ volume_fakes.create_one_consistency_group_snapshot()
)
columns = (
diff --git a/openstackclient/tests/unit/volume/v2/test_qos_specs.py b/openstackclient/tests/unit/volume/v2/test_qos_specs.py
index 41c18014..6f258dd5 100644
--- a/openstackclient/tests/unit/volume/v2/test_qos_specs.py
+++ b/openstackclient/tests/unit/volume/v2/test_qos_specs.py
@@ -39,8 +39,8 @@ class TestQos(volume_fakes.TestVolume):
class TestQosAssociate(TestQos):
- volume_type = volume_fakes.FakeVolumeType.create_one_volume_type()
- qos_spec = volume_fakes.FakeQos.create_one_qos()
+ volume_type = volume_fakes.create_one_volume_type()
+ qos_spec = volume_fakes.create_one_qos()
def setUp(self):
super(TestQosAssociate, self).setUp()
@@ -82,7 +82,7 @@ class TestQosCreate(TestQos):
def setUp(self):
super(TestQosCreate, self).setUp()
- self.new_qos_spec = volume_fakes.FakeQos.create_one_qos()
+ self.new_qos_spec = volume_fakes.create_one_qos()
self.qos_mock.create.return_value = self.new_qos_spec
self.data = (
@@ -164,13 +164,13 @@ class TestQosCreate(TestQos):
class TestQosDelete(TestQos):
- qos_specs = volume_fakes.FakeQos.create_qoses(count=2)
+ qos_specs = volume_fakes.create_qoses(count=2)
def setUp(self):
super(TestQosDelete, self).setUp()
self.qos_mock.get = (
- volume_fakes.FakeQos.get_qoses(self.qos_specs))
+ volume_fakes.get_qoses(self.qos_specs))
# Get the command object to test
self.cmd = qos_specs.DeleteQos(self.app, None)
@@ -255,8 +255,8 @@ class TestQosDelete(TestQos):
class TestQosDisassociate(TestQos):
- volume_type = volume_fakes.FakeVolumeType.create_one_volume_type()
- qos_spec = volume_fakes.FakeQos.create_one_qos()
+ volume_type = volume_fakes.create_one_volume_type()
+ qos_spec = volume_fakes.create_one_qos()
def setUp(self):
super(TestQosDisassociate, self).setUp()
@@ -303,8 +303,8 @@ class TestQosDisassociate(TestQos):
class TestQosList(TestQos):
- qos_specs = volume_fakes.FakeQos.create_qoses(count=2)
- qos_association = volume_fakes.FakeQos.create_one_qos_association()
+ qos_specs = volume_fakes.create_qoses(count=2)
+ qos_association = volume_fakes.create_one_qos_association()
columns = (
'ID',
@@ -374,7 +374,7 @@ class TestQosList(TestQos):
class TestQosSet(TestQos):
- qos_spec = volume_fakes.FakeQos.create_one_qos()
+ qos_spec = volume_fakes.create_one_qos()
def setUp(self):
super(TestQosSet, self).setUp()
@@ -406,8 +406,8 @@ class TestQosSet(TestQos):
class TestQosShow(TestQos):
- qos_spec = volume_fakes.FakeQos.create_one_qos()
- qos_association = volume_fakes.FakeQos.create_one_qos_association()
+ qos_spec = volume_fakes.create_one_qos()
+ qos_association = volume_fakes.create_one_qos_association()
columns = (
'associations',
@@ -454,7 +454,7 @@ class TestQosShow(TestQos):
class TestQosUnset(TestQos):
- qos_spec = volume_fakes.FakeQos.create_one_qos()
+ qos_spec = volume_fakes.create_one_qos()
def setUp(self):
super(TestQosUnset, self).setUp()
diff --git a/openstackclient/tests/unit/volume/v2/test_service.py b/openstackclient/tests/unit/volume/v2/test_service.py
index 3e9b2df9..e9e39f41 100644
--- a/openstackclient/tests/unit/volume/v2/test_service.py
+++ b/openstackclient/tests/unit/volume/v2/test_service.py
@@ -14,14 +14,14 @@
from osc_lib import exceptions
-from openstackclient.tests.unit.volume.v2 import fakes as service_fakes
+from openstackclient.tests.unit.volume.v2 import fakes as volume_fakes
from openstackclient.volume.v2 import service
-class TestService(service_fakes.TestVolume):
+class TestService(volume_fakes.TestVolume):
def setUp(self):
- super(TestService, self).setUp()
+ super().setUp()
# Get a shortcut to the ServiceManager Mock
self.service_mock = self.app.client_manager.volume.services
@@ -31,10 +31,10 @@ class TestService(service_fakes.TestVolume):
class TestServiceList(TestService):
# The service to be listed
- services = service_fakes.FakeService.create_one_service()
+ services = volume_fakes.create_one_service()
def setUp(self):
- super(TestServiceList, self).setUp()
+ super().setUp()
self.service_mock.list.return_value = [self.services]
@@ -144,10 +144,10 @@ class TestServiceList(TestService):
class TestServiceSet(TestService):
- service = service_fakes.FakeService.create_one_service()
+ service = volume_fakes.create_one_service()
def setUp(self):
- super(TestServiceSet, self).setUp()
+ super().setUp()
self.service_mock.enable.return_value = self.service
self.service_mock.disable.return_value = self.service
diff --git a/openstackclient/tests/unit/volume/v2/test_type.py b/openstackclient/tests/unit/volume/v2/test_type.py
index d94dc1e0..1cb46c45 100644
--- a/openstackclient/tests/unit/volume/v2/test_type.py
+++ b/openstackclient/tests/unit/volume/v2/test_type.py
@@ -28,7 +28,7 @@ from openstackclient.volume.v2 import volume_type
class TestType(volume_fakes.TestVolume):
def setUp(self):
- super(TestType, self).setUp()
+ super().setUp()
self.types_mock = self.app.client_manager.volume.volume_types
self.types_mock.reset_mock()
@@ -56,10 +56,9 @@ class TestTypeCreate(TestType):
)
def setUp(self):
- super(TestTypeCreate, self).setUp()
+ super().setUp()
- self.new_volume_type = \
- volume_fakes.FakeVolumeType.create_one_volume_type()
+ self.new_volume_type = volume_fakes.create_one_volume_type()
self.data = (
self.new_volume_type.description,
self.new_volume_type.id,
@@ -144,12 +143,12 @@ class TestTypeCreate(TestType):
'key_size': '128',
'control_location': 'front-end',
}
- encryption_type = \
- volume_fakes.FakeVolumeType.create_one_encryption_volume_type(
- attrs=encryption_info)
- self.new_volume_type = \
- volume_fakes.FakeVolumeType.create_one_volume_type(
- attrs={'encryption': encryption_info})
+ encryption_type = volume_fakes.create_one_encryption_volume_type(
+ attrs=encryption_info,
+ )
+ self.new_volume_type = volume_fakes.create_one_volume_type(
+ attrs={'encryption': encryption_info},
+ )
self.types_mock.create.return_value = self.new_volume_type
self.encryption_types_mock.create.return_value = encryption_type
encryption_columns = (
@@ -203,13 +202,14 @@ class TestTypeCreate(TestType):
class TestTypeDelete(TestType):
- volume_types = volume_fakes.FakeVolumeType.create_volume_types(count=2)
+ volume_types = volume_fakes.create_volume_types(count=2)
def setUp(self):
- super(TestTypeDelete, self).setUp()
+ super().setUp()
- self.types_mock.get = volume_fakes.FakeVolumeType.get_volume_types(
- self.volume_types)
+ self.types_mock.get = volume_fakes.get_volume_types(
+ self.volume_types,
+ )
self.types_mock.delete.return_value = None
# Get the command object to mock
@@ -278,7 +278,7 @@ class TestTypeDelete(TestType):
class TestTypeList(TestType):
- volume_types = volume_fakes.FakeVolumeType.create_volume_types()
+ volume_types = volume_fakes.create_volume_types()
columns = [
"ID",
@@ -312,7 +312,7 @@ class TestTypeList(TestType):
))
def setUp(self):
- super(TestTypeList, self).setUp()
+ super().setUp()
self.types_mock.list.return_value = self.volume_types
self.types_mock.default.return_value = self.volume_types[0]
@@ -388,9 +388,9 @@ class TestTypeList(TestType):
self.assertCountEqual(self.data_with_default_type, list(data))
def test_type_list_with_encryption(self):
- encryption_type = \
- volume_fakes.FakeVolumeType.create_one_encryption_volume_type(
- attrs={'volume_type_id': self.volume_types[0].id})
+ encryption_type = volume_fakes.create_one_encryption_volume_type(
+ attrs={'volume_type_id': self.volume_types[0].id},
+ )
encryption_info = {
'provider': 'LuksEncryptor',
'cipher': None,
@@ -436,11 +436,12 @@ class TestTypeList(TestType):
class TestTypeSet(TestType):
project = identity_fakes.FakeProject.create_one_project()
- volume_type = volume_fakes.FakeVolumeType.create_one_volume_type(
- methods={'set_keys': None})
+ volume_type = volume_fakes.create_one_volume_type(
+ methods={'set_keys': None},
+ )
def setUp(self):
- super(TestTypeSet, self).setUp()
+ super().setUp()
self.types_mock.get.return_value = self.volume_type
@@ -685,9 +686,9 @@ class TestTypeShow(TestType):
)
def setUp(self):
- super(TestTypeShow, self).setUp()
+ super().setUp()
- self.volume_type = volume_fakes.FakeVolumeType.create_one_volume_type()
+ self.volume_type = volume_fakes.create_one_volume_type()
self.data = (
None,
self.volume_type.description,
@@ -727,13 +728,20 @@ class TestTypeShow(TestType):
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
- private_type = volume_fakes.FakeVolumeType.create_one_volume_type(
- attrs={'is_public': False})
- type_access_list = volume_fakes.FakeTypeAccess.create_one_type_access()
- with mock.patch.object(self.types_mock, 'get',
- return_value=private_type):
- with mock.patch.object(self.types_access_mock, 'list',
- return_value=[type_access_list]):
+ private_type = volume_fakes.create_one_volume_type(
+ attrs={'is_public': False},
+ )
+ type_access_list = volume_fakes.create_one_type_access()
+ with mock.patch.object(
+ self.types_mock,
+ 'get',
+ return_value=private_type,
+ ):
+ with mock.patch.object(
+ self.types_access_mock,
+ 'list',
+ return_value=[type_access_list],
+ ):
columns, data = self.cmd.take_action(parsed_args)
self.types_mock.get.assert_called_once_with(
self.volume_type.id)
@@ -760,8 +768,9 @@ class TestTypeShow(TestType):
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
- private_type = volume_fakes.FakeVolumeType.create_one_volume_type(
- attrs={'is_public': False})
+ private_type = volume_fakes.create_one_volume_type(
+ attrs={'is_public': False},
+ )
with mock.patch.object(self.types_mock, 'get',
return_value=private_type):
with mock.patch.object(self.types_access_mock, 'list',
@@ -784,16 +793,16 @@ class TestTypeShow(TestType):
self.assertCountEqual(private_type_data, data)
def test_type_show_with_encryption(self):
- encryption_type = \
- volume_fakes.FakeVolumeType.create_one_encryption_volume_type()
+ encryption_type = volume_fakes.create_one_encryption_volume_type()
encryption_info = {
'provider': 'LuksEncryptor',
'cipher': None,
'key_size': None,
'control_location': 'front-end',
}
- self.volume_type = volume_fakes.FakeVolumeType.create_one_volume_type(
- attrs={'encryption': encryption_info})
+ self.volume_type = volume_fakes.create_one_volume_type(
+ attrs={'encryption': encryption_info},
+ )
self.types_mock.get.return_value = self.volume_type
self.encryption_types_mock.get.return_value = encryption_type
encryption_columns = (
@@ -834,11 +843,12 @@ class TestTypeShow(TestType):
class TestTypeUnset(TestType):
project = identity_fakes.FakeProject.create_one_project()
- volume_type = volume_fakes.FakeVolumeType.create_one_volume_type(
- methods={'unset_keys': None})
+ volume_type = volume_fakes.create_one_volume_type(
+ methods={'unset_keys': None},
+ )
def setUp(self):
- super(TestTypeUnset, self).setUp()
+ super().setUp()
self.types_mock.get.return_value = self.volume_type
@@ -936,7 +946,7 @@ class TestTypeUnset(TestType):
class TestColumns(TestType):
def test_encryption_info_column_with_info(self):
- fake_volume_type = volume_fakes.FakeVolumeType.create_one_volume_type()
+ fake_volume_type = volume_fakes.create_one_volume_type()
type_id = fake_volume_type.id
encryption_info = {
@@ -952,7 +962,7 @@ class TestColumns(TestType):
self.assertEqual(encryption_info, col.machine_readable())
def test_encryption_info_column_without_info(self):
- fake_volume_type = volume_fakes.FakeVolumeType.create_one_volume_type()
+ fake_volume_type = volume_fakes.create_one_volume_type()
type_id = fake_volume_type.id
col = volume_type.EncryptionInfoColumn(type_id, {})
diff --git a/openstackclient/tests/unit/volume/v2/test_volume.py b/openstackclient/tests/unit/volume/v2/test_volume.py
index b6e9f487..f802f637 100644
--- a/openstackclient/tests/unit/volume/v2/test_volume.py
+++ b/openstackclient/tests/unit/volume/v2/test_volume.py
@@ -30,7 +30,7 @@ from openstackclient.volume.v2 import volume
class TestVolume(volume_fakes.TestVolume):
def setUp(self):
- super(TestVolume, self).setUp()
+ super().setUp()
self.volumes_mock = self.app.client_manager.volume.volumes
self.volumes_mock.reset_mock()
@@ -55,11 +55,9 @@ class TestVolume(volume_fakes.TestVolume):
self.consistencygroups_mock.reset_mock()
def setup_volumes_mock(self, count):
- volumes = volume_fakes.FakeVolume.create_volumes(count=count)
+ volumes = volume_fakes.create_volumes(count=count)
- self.volumes_mock.get = volume_fakes.FakeVolume.get_volumes(
- volumes,
- 0)
+ self.volumes_mock.get = volume_fakes.get_volumes(volumes, 0)
return volumes
@@ -83,9 +81,9 @@ class TestVolumeCreate(TestVolume):
)
def setUp(self):
- super(TestVolumeCreate, self).setUp()
+ super().setUp()
- self.new_volume = volume_fakes.FakeVolume.create_one_volume()
+ self.new_volume = volume_fakes.create_one_volume()
self.volumes_mock.create.return_value = self.new_volume
self.datalist = (
@@ -137,8 +135,7 @@ class TestVolumeCreate(TestVolume):
self.assertCountEqual(self.datalist, data)
def test_volume_create_options(self):
- consistency_group = (
- volume_fakes.FakeConsistencyGroup.create_one_consistency_group())
+ consistency_group = volume_fakes.create_one_consistency_group()
self.consistencygroups_mock.get.return_value = consistency_group
arglist = [
'--size', str(self.new_volume.size),
@@ -295,7 +292,7 @@ class TestVolumeCreate(TestVolume):
self.assertCountEqual(self.datalist, data)
def test_volume_create_with_snapshot(self):
- snapshot = volume_fakes.FakeSnapshot.create_one_snapshot()
+ snapshot = volume_fakes.create_one_snapshot()
self.new_volume.snapshot_id = snapshot.id
arglist = [
'--snapshot', self.new_volume.snapshot_id,
@@ -502,7 +499,7 @@ class TestVolumeCreate(TestVolume):
class TestVolumeDelete(TestVolume):
def setUp(self):
- super(TestVolumeDelete, self).setUp()
+ super().setUp()
self.volumes_mock.delete.return_value = None
@@ -630,9 +627,9 @@ class TestVolumeList(TestVolume):
]
def setUp(self):
- super(TestVolumeList, self).setUp()
+ super().setUp()
- self.mock_volume = volume_fakes.FakeVolume.create_one_volume()
+ self.mock_volume = volume_fakes.create_one_volume()
self.volumes_mock.list.return_value = [self.mock_volume]
self.users_mock.get.return_value = self.user
@@ -1105,10 +1102,10 @@ class TestVolumeList(TestVolume):
class TestVolumeMigrate(TestVolume):
- _volume = volume_fakes.FakeVolume.create_one_volume()
+ _volume = volume_fakes.create_one_volume()
def setUp(self):
- super(TestVolumeMigrate, self).setUp()
+ super().setUp()
self.volumes_mock.get.return_value = self._volume
self.volumes_mock.migrate_volume.return_value = None
@@ -1171,12 +1168,12 @@ class TestVolumeMigrate(TestVolume):
class TestVolumeSet(TestVolume):
- volume_type = volume_fakes.FakeVolumeType.create_one_volume_type()
+ volume_type = volume_fakes.create_one_volume_type()
def setUp(self):
- super(TestVolumeSet, self).setUp()
+ super().setUp()
- self.new_volume = volume_fakes.FakeVolume.create_one_volume()
+ self.new_volume = volume_fakes.create_one_volume()
self.volumes_mock.get.return_value = self.new_volume
self.types_mock.get.return_value = self.volume_type
@@ -1425,9 +1422,9 @@ class TestVolumeSet(TestVolume):
class TestVolumeShow(TestVolume):
def setUp(self):
- super(TestVolumeShow, self).setUp()
+ super().setUp()
- self._volume = volume_fakes.FakeVolume.create_one_volume()
+ self._volume = volume_fakes.create_one_volume()
self.volumes_mock.get.return_value = self._volume
# Get the command object to test
self.cmd = volume.ShowVolume(self.app, None)
@@ -1445,20 +1442,21 @@ class TestVolumeShow(TestVolume):
self.volumes_mock.get.assert_called_with(self._volume.id)
self.assertEqual(
- volume_fakes.FakeVolume.get_volume_columns(self._volume),
- columns)
-
+ volume_fakes.get_volume_columns(self._volume),
+ columns,
+ )
self.assertCountEqual(
- volume_fakes.FakeVolume.get_volume_data(self._volume),
- data)
+ volume_fakes.get_volume_data(self._volume),
+ data,
+ )
class TestVolumeUnset(TestVolume):
def setUp(self):
- super(TestVolumeUnset, self).setUp()
+ super().setUp()
- self.new_volume = volume_fakes.FakeVolume.create_one_volume()
+ self.new_volume = volume_fakes.create_one_volume()
self.volumes_mock.get.return_value = self.new_volume
# Get the command object to set property
@@ -1536,7 +1534,7 @@ class TestVolumeUnset(TestVolume):
class TestColumns(TestVolume):
def test_attachments_column_without_server_cache(self):
- _volume = volume_fakes.FakeVolume.create_one_volume()
+ _volume = volume_fakes.create_one_volume()
server_id = _volume.attachments[0]['server_id']
device = _volume.attachments[0]['device']
@@ -1546,7 +1544,7 @@ class TestColumns(TestVolume):
self.assertEqual(_volume.attachments, col.machine_readable())
def test_attachments_column_with_server_cache(self):
- _volume = volume_fakes.FakeVolume.create_one_volume()
+ _volume = volume_fakes.create_one_volume()
server_id = _volume.attachments[0]['server_id']
device = _volume.attachments[0]['device']
diff --git a/openstackclient/tests/unit/volume/v2/test_volume_backend.py b/openstackclient/tests/unit/volume/v2/test_volume_backend.py
index d9ac2c96..6c64f645 100644
--- a/openstackclient/tests/unit/volume/v2/test_volume_backend.py
+++ b/openstackclient/tests/unit/volume/v2/test_volume_backend.py
@@ -20,10 +20,10 @@ class TestShowVolumeCapability(volume_fakes.TestVolume):
"""Test backend capability functionality."""
# The capability to be listed
- capability = volume_fakes.FakeCapability.create_one_capability()
+ capability = volume_fakes.create_one_capability()
def setUp(self):
- super(TestShowVolumeCapability, self).setUp()
+ super().setUp()
# Get a shortcut to the capability Mock
self.capability_mock = self.app.client_manager.volume.capabilities
@@ -77,10 +77,10 @@ class TestListVolumePool(volume_fakes.TestVolume):
"""Tests for volume backend pool listing."""
# The pool to be listed
- pools = volume_fakes.FakePool.create_one_pool()
+ pools = volume_fakes.create_one_pool()
def setUp(self):
- super(TestListVolumePool, self).setUp()
+ super().setUp()
self.pool_mock = self.app.client_manager.volume.pools
self.pool_mock.list.return_value = [self.pools]
diff --git a/openstackclient/tests/unit/volume/v2/test_volume_backup.py b/openstackclient/tests/unit/volume/v2/test_volume_backup.py
index ffd84901..7d00b8bf 100644
--- a/openstackclient/tests/unit/volume/v2/test_volume_backup.py
+++ b/openstackclient/tests/unit/volume/v2/test_volume_backup.py
@@ -26,7 +26,7 @@ from openstackclient.volume.v2 import volume_backup
class TestBackup(volume_fakes.TestVolume):
def setUp(self):
- super(TestBackup, self).setUp()
+ super().setUp()
self.backups_mock = self.app.client_manager.volume.backups
self.backups_mock.reset_mock()
@@ -40,9 +40,9 @@ class TestBackup(volume_fakes.TestVolume):
class TestBackupCreate(TestBackup):
- volume = volume_fakes.FakeVolume.create_one_volume()
- snapshot = volume_fakes.FakeSnapshot.create_one_snapshot()
- new_backup = volume_fakes.FakeBackup.create_one_backup(
+ 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 = (
@@ -71,7 +71,7 @@ class TestBackupCreate(TestBackup):
)
def setUp(self):
- super(TestBackupCreate, self).setUp()
+ super().setUp()
self.volumes_mock.get.return_value = self.volume
self.snapshots_mock.get.return_value = self.snapshot
@@ -242,13 +242,13 @@ class TestBackupCreate(TestBackup):
class TestBackupDelete(TestBackup):
- backups = volume_fakes.FakeBackup.create_backups(count=2)
+ backups = volume_fakes.create_backups(count=2)
def setUp(self):
- super(TestBackupDelete, self).setUp()
+ super().setUp()
self.backups_mock.get = (
- volume_fakes.FakeBackup.get_backups(self.backups))
+ volume_fakes.get_backups(self.backups))
self.backups_mock.delete.return_value = None
# Get the command object to mock
@@ -334,8 +334,8 @@ class TestBackupDelete(TestBackup):
class TestBackupList(TestBackup):
- volume = volume_fakes.FakeVolume.create_one_volume()
- backups = volume_fakes.FakeBackup.create_backups(
+ volume = volume_fakes.create_one_volume()
+ backups = volume_fakes.create_backups(
attrs={'volume_id': volume.name}, count=3)
columns = (
@@ -374,7 +374,7 @@ class TestBackupList(TestBackup):
))
def setUp(self):
- super(TestBackupList, self).setUp()
+ super().setUp()
self.volumes_mock.list.return_value = [self.volume]
self.backups_mock.list.return_value = self.backups
@@ -456,8 +456,8 @@ class TestBackupList(TestBackup):
class TestBackupRestore(TestBackup):
- volume = volume_fakes.FakeVolume.create_one_volume()
- backup = volume_fakes.FakeBackup.create_one_backup(
+ volume = volume_fakes.create_one_volume()
+ backup = volume_fakes.create_one_backup(
attrs={'volume_id': volume.id},
)
@@ -467,7 +467,7 @@ class TestBackupRestore(TestBackup):
self.backups_mock.get.return_value = self.backup
self.volumes_mock.get.return_value = self.volume
self.restores_mock.restore.return_value = (
- volume_fakes.FakeVolume.create_one_volume(
+ volume_fakes.create_one_volume(
{'id': self.volume['id']},
)
)
@@ -550,12 +550,12 @@ class TestBackupRestore(TestBackup):
class TestBackupSet(TestBackup):
- backup = volume_fakes.FakeBackup.create_one_backup(
+ backup = volume_fakes.create_one_backup(
attrs={'metadata': {'wow': 'cool'}},
)
def setUp(self):
- super(TestBackupSet, self).setUp()
+ super().setUp()
self.backups_mock.get.return_value = self.backup
@@ -784,7 +784,7 @@ class TestBackupSet(TestBackup):
class TestBackupUnset(TestBackup):
- backup = volume_fakes.FakeBackup.create_one_backup(
+ backup = volume_fakes.create_one_backup(
attrs={'metadata': {'foo': 'bar'}},
)
@@ -845,7 +845,7 @@ class TestBackupUnset(TestBackup):
class TestBackupShow(TestBackup):
- backup = volume_fakes.FakeBackup.create_one_backup()
+ backup = volume_fakes.create_one_backup()
columns = (
'availability_zone',
@@ -873,7 +873,7 @@ class TestBackupShow(TestBackup):
)
def setUp(self):
- super(TestBackupShow, self).setUp()
+ super().setUp()
self.backups_mock.get.return_value = self.backup
# Get the command object to test
diff --git a/openstackclient/tests/unit/volume/v2/test_volume_host.py b/openstackclient/tests/unit/volume/v2/test_volume_host.py
index b024329a..730085a3 100644
--- a/openstackclient/tests/unit/volume/v2/test_volume_host.py
+++ b/openstackclient/tests/unit/volume/v2/test_volume_host.py
@@ -12,14 +12,14 @@
# under the License.
#
-from openstackclient.tests.unit.volume.v2 import fakes as host_fakes
+from openstackclient.tests.unit.volume.v2 import fakes as volume_fakes
from openstackclient.volume.v2 import volume_host
-class TestVolumeHost(host_fakes.TestVolume):
+class TestVolumeHost(volume_fakes.TestVolume):
def setUp(self):
- super(TestVolumeHost, self).setUp()
+ super().setUp()
self.host_mock = self.app.client_manager.volume.services
self.host_mock.reset_mock()
@@ -27,10 +27,10 @@ class TestVolumeHost(host_fakes.TestVolume):
class TestVolumeHostSet(TestVolumeHost):
- service = host_fakes.FakeService.create_one_service()
+ service = volume_fakes.create_one_service()
def setUp(self):
- super(TestVolumeHostSet, self).setUp()
+ super().setUp()
self.host_mock.freeze_host.return_value = None
self.host_mock.thaw_host.return_value = None
@@ -89,10 +89,10 @@ class TestVolumeHostSet(TestVolumeHost):
class TestVolumeHostFailover(TestVolumeHost):
- service = host_fakes.FakeService.create_one_service()
+ service = volume_fakes.create_one_service()
def setUp(self):
- super(TestVolumeHostFailover, self).setUp()
+ super().setUp()
self.host_mock.failover_host.return_value = None
diff --git a/openstackclient/tests/unit/volume/v2/test_volume_snapshot.py b/openstackclient/tests/unit/volume/v2/test_volume_snapshot.py
index 33a5a98a..6cffcaac 100644
--- a/openstackclient/tests/unit/volume/v2/test_volume_snapshot.py
+++ b/openstackclient/tests/unit/volume/v2/test_volume_snapshot.py
@@ -54,8 +54,8 @@ class TestVolumeSnapshotCreate(TestVolumeSnapshot):
def setUp(self):
super().setUp()
- self.volume = volume_fakes.FakeVolume.create_one_volume()
- self.new_snapshot = volume_fakes.FakeSnapshot.create_one_snapshot(
+ self.volume = volume_fakes.create_one_volume()
+ self.new_snapshot = volume_fakes.create_one_snapshot(
attrs={'volume_id': self.volume.id})
self.data = (
@@ -179,13 +179,13 @@ class TestVolumeSnapshotCreate(TestVolumeSnapshot):
class TestVolumeSnapshotDelete(TestVolumeSnapshot):
- snapshots = volume_fakes.FakeSnapshot.create_snapshots(count=2)
+ snapshots = volume_fakes.create_snapshots(count=2)
def setUp(self):
super().setUp()
self.snapshots_mock.get = (
- volume_fakes.FakeSnapshot.get_snapshots(self.snapshots))
+ volume_fakes.get_snapshots(self.snapshots))
self.snapshots_mock.delete.return_value = None
# Get the command object to mock
@@ -273,9 +273,9 @@ class TestVolumeSnapshotDelete(TestVolumeSnapshot):
class TestVolumeSnapshotList(TestVolumeSnapshot):
- volume = volume_fakes.FakeVolume.create_one_volume()
+ volume = volume_fakes.create_one_volume()
project = project_fakes.FakeProject.create_one_project()
- snapshots = volume_fakes.FakeSnapshot.create_snapshots(
+ snapshots = volume_fakes.create_snapshots(
attrs={'volume_id': volume.name}, count=3)
columns = [
@@ -495,7 +495,7 @@ class TestVolumeSnapshotList(TestVolumeSnapshot):
class TestVolumeSnapshotSet(TestVolumeSnapshot):
- snapshot = volume_fakes.FakeSnapshot.create_one_snapshot()
+ snapshot = volume_fakes.create_one_snapshot()
def setUp(self):
super().setUp()
@@ -677,7 +677,7 @@ class TestVolumeSnapshotShow(TestVolumeSnapshot):
def setUp(self):
super().setUp()
- self.snapshot = volume_fakes.FakeSnapshot.create_one_snapshot()
+ self.snapshot = volume_fakes.create_one_snapshot()
self.data = (
self.snapshot.created_at,
@@ -712,7 +712,7 @@ class TestVolumeSnapshotShow(TestVolumeSnapshot):
class TestVolumeSnapshotUnset(TestVolumeSnapshot):
- snapshot = volume_fakes.FakeSnapshot.create_one_snapshot()
+ snapshot = volume_fakes.create_one_snapshot()
def setUp(self):
super().setUp()
diff --git a/openstackclient/tests/unit/volume/v2/test_volume_transfer_request.py b/openstackclient/tests/unit/volume/v2/test_volume_transfer_request.py
index 1a1f220f..c8c6fac9 100644
--- a/openstackclient/tests/unit/volume/v2/test_volume_transfer_request.py
+++ b/openstackclient/tests/unit/volume/v2/test_volume_transfer_request.py
@@ -20,14 +20,14 @@ from osc_lib import exceptions
from osc_lib import utils
from openstackclient.tests.unit import utils as test_utils
-from openstackclient.tests.unit.volume.v2 import fakes as transfer_fakes
+from openstackclient.tests.unit.volume.v2 import fakes as volume_fakes
from openstackclient.volume.v2 import volume_transfer_request
-class TestTransfer(transfer_fakes.TestVolume):
+class TestTransfer(volume_fakes.TestVolume):
def setUp(self):
- super(TestTransfer, self).setUp()
+ super().setUp()
# Get a shortcut to the TransferManager Mock
self.transfer_mock = self.app.client_manager.volume.transfers
@@ -47,10 +47,9 @@ class TestTransferAccept(TestTransfer):
)
def setUp(self):
- super(TestTransferAccept, self).setUp()
+ super().setUp()
- self.volume_transfer = (
- transfer_fakes.FakeTransfer.create_one_transfer())
+ self.volume_transfer = volume_fakes.create_one_transfer()
self.data = (
self.volume_transfer.id,
self.volume_transfer.name,
@@ -106,7 +105,7 @@ class TestTransferAccept(TestTransfer):
class TestTransferCreate(TestTransfer):
- volume = transfer_fakes.FakeVolume.create_one_volume()
+ volume = volume_fakes.create_one_volume()
columns = (
'auth_key',
@@ -117,12 +116,14 @@ class TestTransferCreate(TestTransfer):
)
def setUp(self):
- super(TestTransferCreate, self).setUp()
-
- self.volume_transfer = transfer_fakes.FakeTransfer.create_one_transfer(
- attrs={'volume_id': self.volume.id,
- 'auth_key': 'key',
- 'created_at': 'time'}
+ super().setUp()
+
+ self.volume_transfer = volume_fakes.create_one_transfer(
+ attrs={
+ 'volume_id': self.volume.id,
+ 'auth_key': 'key',
+ 'created_at': 'time',
+ },
)
self.data = (
self.volume_transfer.auth_key,
@@ -221,13 +222,14 @@ class TestTransferCreate(TestTransfer):
class TestTransferDelete(TestTransfer):
- volume_transfers = transfer_fakes.FakeTransfer.create_transfers(count=2)
+ volume_transfers = volume_fakes.create_transfers(count=2)
def setUp(self):
- super(TestTransferDelete, self).setUp()
+ super().setUp()
- self.transfer_mock.get = (
- transfer_fakes.FakeTransfer.get_transfers(self.volume_transfers))
+ self.transfer_mock.get = volume_fakes.get_transfers(
+ self.volume_transfers,
+ )
self.transfer_mock.delete.return_value = None
# Get the command object to mock
@@ -300,10 +302,10 @@ class TestTransferDelete(TestTransfer):
class TestTransferList(TestTransfer):
# The Transfers to be listed
- volume_transfers = transfer_fakes.FakeTransfer.create_one_transfer()
+ volume_transfers = volume_fakes.create_one_transfer()
def setUp(self):
- super(TestTransferList, self).setUp()
+ super().setUp()
self.transfer_mock.list.return_value = [self.volume_transfers]
@@ -394,11 +396,10 @@ class TestTransferShow(TestTransfer):
)
def setUp(self):
- super(TestTransferShow, self).setUp()
+ super().setUp()
- self.volume_transfer = (
- transfer_fakes.FakeTransfer.create_one_transfer(
- attrs={'created_at': 'time'})
+ self.volume_transfer = volume_fakes.create_one_transfer(
+ attrs={'created_at': 'time'},
)
self.data = (
self.volume_transfer.created_at,
diff --git a/openstackclient/tests/unit/volume/v3/fakes.py b/openstackclient/tests/unit/volume/v3/fakes.py
index caf7d269..3e3a05fa 100644
--- a/openstackclient/tests/unit/volume/v3/fakes.py
+++ b/openstackclient/tests/unit/volume/v3/fakes.py
@@ -24,7 +24,6 @@ from openstackclient.tests.unit.volume.v2 import fakes as volume_v2_fakes
class FakeVolumeClient:
-
def __init__(self, **kwargs):
self.auth_token = kwargs['token']
self.management_url = kwargs['endpoint']
@@ -51,17 +50,14 @@ class FakeVolumeClient:
class TestVolume(utils.TestCommand):
-
def setUp(self):
super().setUp()
self.app.client_manager.volume = FakeVolumeClient(
- endpoint=fakes.AUTH_URL,
- token=fakes.AUTH_TOKEN
+ endpoint=fakes.AUTH_URL, token=fakes.AUTH_TOKEN
)
self.app.client_manager.identity = identity_fakes.FakeIdentityv3Client(
- endpoint=fakes.AUTH_URL,
- token=fakes.AUTH_TOKEN
+ endpoint=fakes.AUTH_URL, token=fakes.AUTH_TOKEN
)
self.app.client_manager.compute = compute_fakes.FakeComputev2Client(
endpoint=fakes.AUTH_URL,
@@ -70,353 +66,309 @@ class TestVolume(utils.TestCommand):
# TODO(stephenfin): Check if the responses are actually the same
-FakeVolume = volume_v2_fakes.FakeVolume
-FakeVolumeType = volume_v2_fakes.FakeVolumeType
-
-
-class FakeCluster:
- """Fake one or more clusters."""
-
- @staticmethod
- def create_one_cluster(attrs=None):
- """Create a fake service cluster.
-
- :param attrs: A dictionary with all attributes of service cluster
- :return: A FakeResource object with id, name, status, etc.
- """
- attrs = attrs or {}
-
- # Set default attribute
- cluster_info = {
- 'name': f'cluster-{uuid.uuid4().hex}',
- 'binary': f'binary-{uuid.uuid4().hex}',
- 'state': random.choice(['up', 'down']),
- 'status': random.choice(['enabled', 'disabled']),
- 'disabled_reason': None,
- 'num_hosts': random.randint(1, 64),
- 'num_down_hosts': random.randint(1, 64),
- 'last_heartbeat': '2015-09-16T09:28:52.000000',
- 'created_at': '2015-09-16T09:28:52.000000',
- 'updated_at': '2015-09-16T09:28:52.000000',
- 'replication_status': None,
- 'frozen': False,
- 'active_backend_id': None,
- }
-
- # Overwrite default attributes if there are some attributes set
- cluster_info.update(attrs)
-
- return fakes.FakeResource(
- None,
- cluster_info,
- loaded=True)
-
- @staticmethod
- def create_clusters(attrs=None, count=2):
- """Create multiple fake service clusters.
-
- :param attrs: A dictionary with all attributes of service cluster
- :param count: The number of service clusters to be faked
- :return: A list of FakeResource objects
- """
- clusters = []
- for n in range(0, count):
- clusters.append(FakeCluster.create_one_cluster(attrs))
-
- return clusters
-
-
-class FakeResourceFilter:
- """Fake one or more resource filters."""
-
- @staticmethod
- def create_one_resource_filter(attrs=None):
- """Create a fake resource filter.
-
- :param attrs: A dictionary with all attributes of resource filter
- :return: A FakeResource object with id, name, status, etc.
- """
- attrs = attrs or {}
-
- # Set default attribute
-
- resource_filter_info = {
- 'filters': [
- 'name',
- 'status',
- 'image_metadata',
- 'bootable',
- 'migration_status',
- ],
- 'resource': 'volume',
- }
-
- # Overwrite default attributes if there are some attributes set
- resource_filter_info.update(attrs)
-
- return fakes.FakeResource(None, resource_filter_info, loaded=True)
-
- @staticmethod
- def create_resource_filters(attrs=None, count=2):
- """Create multiple fake resource filters.
-
- :param attrs: A dictionary with all attributes of resource filter
- :param count: The number of resource filters to be faked
- :return: A list of FakeResource objects
- """
- resource_filters = []
- for n in range(0, count):
- resource_filters.append(
- FakeResourceFilter.create_one_resource_filter(attrs)
- )
-
- return resource_filters
-
-
-class FakeVolumeGroup:
- """Fake one or more volume groups."""
-
- @staticmethod
- def create_one_volume_group(attrs=None):
- """Create a fake group.
-
- :param attrs: A dictionary with all attributes of group
- :return: A FakeResource object with id, name, status, etc.
- """
- attrs = attrs or {}
-
- group_type = attrs.pop('group_type', None) or uuid.uuid4().hex
- volume_types = attrs.pop('volume_types', None) or [uuid.uuid4().hex]
-
- # Set default attribute
- group_info = {
- 'id': uuid.uuid4().hex,
- 'status': random.choice([
+create_one_volume = volume_v2_fakes.create_one_volume
+create_one_volume_type = volume_v2_fakes.create_one_volume_type
+
+
+def create_one_cluster(attrs=None):
+ """Create a fake service cluster.
+
+ :param attrs: A dictionary with all attributes of service cluster
+ :return: A FakeResource object with id, name, status, etc.
+ """
+ attrs = attrs or {}
+
+ # Set default attribute
+ cluster_info = {
+ 'name': f'cluster-{uuid.uuid4().hex}',
+ 'binary': f'binary-{uuid.uuid4().hex}',
+ 'state': random.choice(['up', 'down']),
+ 'status': random.choice(['enabled', 'disabled']),
+ 'disabled_reason': None,
+ 'num_hosts': random.randint(1, 64),
+ 'num_down_hosts': random.randint(1, 64),
+ 'last_heartbeat': '2015-09-16T09:28:52.000000',
+ 'created_at': '2015-09-16T09:28:52.000000',
+ 'updated_at': '2015-09-16T09:28:52.000000',
+ 'replication_status': None,
+ 'frozen': False,
+ 'active_backend_id': None,
+ }
+
+ # Overwrite default attributes if there are some attributes set
+ cluster_info.update(attrs)
+
+ return fakes.FakeResource(None, cluster_info, loaded=True)
+
+
+def create_clusters(attrs=None, count=2):
+ """Create multiple fake service clusters.
+
+ :param attrs: A dictionary with all attributes of service cluster
+ :param count: The number of service clusters to be faked
+ :return: A list of FakeResource objects
+ """
+ clusters = []
+ for n in range(0, count):
+ clusters.append(create_one_cluster(attrs))
+
+ return clusters
+
+
+def create_one_resource_filter(attrs=None):
+ """Create a fake resource filter.
+
+ :param attrs: A dictionary with all attributes of resource filter
+ :return: A FakeResource object with id, name, status, etc.
+ """
+ attrs = attrs or {}
+
+ # Set default attribute
+
+ resource_filter_info = {
+ 'filters': [
+ 'name',
+ 'status',
+ 'image_metadata',
+ 'bootable',
+ 'migration_status',
+ ],
+ 'resource': 'volume',
+ }
+
+ # Overwrite default attributes if there are some attributes set
+ resource_filter_info.update(attrs)
+
+ return fakes.FakeResource(None, resource_filter_info, loaded=True)
+
+
+def create_resource_filters(attrs=None, count=2):
+ """Create multiple fake resource filters.
+
+ :param attrs: A dictionary with all attributes of resource filter
+ :param count: The number of resource filters to be faked
+ :return: A list of FakeResource objects
+ """
+ resource_filters = []
+ for n in range(0, count):
+ resource_filters.append(create_one_resource_filter(attrs))
+
+ return resource_filters
+
+
+def create_one_volume_group(attrs=None):
+ """Create a fake group.
+
+ :param attrs: A dictionary with all attributes of group
+ :return: A FakeResource object with id, name, status, etc.
+ """
+ attrs = attrs or {}
+
+ group_type = attrs.pop('group_type', None) or uuid.uuid4().hex
+ volume_types = attrs.pop('volume_types', None) or [uuid.uuid4().hex]
+
+ # Set default attribute
+ group_info = {
+ 'id': uuid.uuid4().hex,
+ 'status': random.choice(
+ [
'available',
- ]),
- 'availability_zone': f'az-{uuid.uuid4().hex}',
- 'created_at': '2015-09-16T09:28:52.000000',
- 'name': 'first_group',
- 'description': f'description-{uuid.uuid4().hex}',
- 'group_type': group_type,
- 'volume_types': volume_types,
- 'volumes': [f'volume-{uuid.uuid4().hex}'],
- 'group_snapshot_id': None,
- 'source_group_id': None,
- 'project_id': f'project-{uuid.uuid4().hex}',
- }
-
- # Overwrite default attributes if there are some attributes set
- group_info.update(attrs)
-
- group = fakes.FakeResource(
- None,
- group_info,
- loaded=True)
- return group
-
- @staticmethod
- def create_volume_groups(attrs=None, count=2):
- """Create multiple fake groups.
-
- :param attrs: A dictionary with all attributes of group
- :param count: The number of groups to be faked
- :return: A list of FakeResource objects
- """
- groups = []
- for n in range(0, count):
- groups.append(FakeVolumeGroup.create_one_volume_group(attrs))
-
- return groups
-
-
-class FakeVolumeGroupSnapshot:
- """Fake one or more volume group snapshots."""
-
- @staticmethod
- def create_one_volume_group_snapshot(attrs=None, methods=None):
- """Create a fake group snapshot.
-
- :param attrs: A dictionary with all attributes
- :param methods: A dictionary with all methods
- :return: A FakeResource object with id, name, description, etc.
- """
- attrs = attrs or {}
-
- # Set default attribute
- group_snapshot_info = {
- 'id': uuid.uuid4().hex,
- 'name': f'group-snapshot-{uuid.uuid4().hex}',
- 'description': f'description-{uuid.uuid4().hex}',
- 'status': random.choice(['available']),
- 'group_id': uuid.uuid4().hex,
- 'group_type_id': uuid.uuid4().hex,
- 'project_id': uuid.uuid4().hex,
- }
-
- # Overwrite default attributes if there are some attributes set
- group_snapshot_info.update(attrs)
-
- group_snapshot = fakes.FakeResource(
- None,
- group_snapshot_info,
- methods=methods,
- loaded=True)
- return group_snapshot
-
- @staticmethod
- def create_volume_group_snapshots(attrs=None, count=2):
- """Create multiple fake group snapshots.
-
- :param attrs: A dictionary with all attributes of group snapshot
- :param count: The number of group snapshots to be faked
- :return: A list of FakeResource objects
- """
- group_snapshots = []
- for n in range(0, count):
- group_snapshots.append(
- FakeVolumeGroupSnapshot.create_one_volume_group_snapshot(attrs)
- )
-
- return group_snapshots
-
-
-class FakeVolumeGroupType:
- """Fake one or more volume group types."""
-
- @staticmethod
- def create_one_volume_group_type(attrs=None, methods=None):
- """Create a fake group type.
-
- :param attrs: A dictionary with all attributes of group type
- :param methods: A dictionary with all methods
- :return: A FakeResource object with id, name, description, etc.
- """
- attrs = attrs or {}
-
- # Set default attribute
- group_type_info = {
- 'id': uuid.uuid4().hex,
- 'name': f'group-type-{uuid.uuid4().hex}',
- 'description': f'description-{uuid.uuid4().hex}',
- 'is_public': random.choice([True, False]),
- 'group_specs': {},
- }
-
- # Overwrite default attributes if there are some attributes set
- group_type_info.update(attrs)
-
- group_type = fakes.FakeResource(
- None,
- group_type_info,
- methods=methods,
- loaded=True)
- return group_type
-
- @staticmethod
- def create_volume_group_types(attrs=None, count=2):
- """Create multiple fake group types.
-
- :param attrs: A dictionary with all attributes of group type
- :param count: The number of group types to be faked
- :return: A list of FakeResource objects
- """
- group_types = []
- for n in range(0, count):
- group_types.append(
- FakeVolumeGroupType.create_one_volume_group_type(attrs)
- )
-
- return group_types
-
-
-class FakeVolumeMessage:
- """Fake one or more volume messages."""
-
- @staticmethod
- def create_one_volume_message(attrs=None):
- """Create a fake message.
-
- :param attrs: A dictionary with all attributes of message
- :return: A FakeResource object with id, name, status, etc.
- """
- attrs = attrs or {}
-
- # Set default attribute
- message_info = {
- 'created_at': '2016-02-11T11:17:37.000000',
- 'event_id': f'VOLUME_{random.randint(1, 999999):06d}',
- 'guaranteed_until': '2016-02-11T11:17:37.000000',
- 'id': uuid.uuid4().hex,
- 'message_level': 'ERROR',
- 'request_id': f'req-{uuid.uuid4().hex}',
- 'resource_type': 'VOLUME',
- 'resource_uuid': uuid.uuid4().hex,
- 'user_message': f'message-{uuid.uuid4().hex}',
- }
-
- # Overwrite default attributes if there are some attributes set
- message_info.update(attrs)
-
- return fakes.FakeResource(
- None,
- message_info,
- loaded=True)
-
- @staticmethod
- def create_volume_messages(attrs=None, count=2):
- """Create multiple fake messages.
-
- :param attrs: A dictionary with all attributes of message
- :param count: The number of messages to be faked
- :return: A list of FakeResource objects
- """
- messages = []
- for n in range(0, count):
- messages.append(FakeVolumeMessage.create_one_volume_message(attrs))
-
- return messages
-
- @staticmethod
- def get_volume_messages(messages=None, count=2):
- """Get an iterable MagicMock object with a list of faked messages.
-
- If messages list is provided, then initialize the Mock object with the
- list. Otherwise create one.
-
- :param messages: A list of FakeResource objects faking messages
- :param count: The number of messages to be faked
- :return An iterable Mock object with side_effect set to a list of faked
- messages
- """
- if messages is None:
- messages = FakeVolumeMessage.create_messages(count)
-
- return mock.Mock(side_effect=messages)
-
-
-class FakeVolumeAttachment:
- """Fake one or more volume attachments."""
-
- @staticmethod
- def create_one_volume_attachment(attrs=None):
- """Create a fake volume attachment.
-
- :param attrs: A dictionary with all attributes of volume attachment
- :return: A FakeResource object with id, status, etc.
- """
- attrs = attrs or {}
-
- attachment_id = uuid.uuid4().hex
- volume_id = attrs.pop('volume_id', None) or uuid.uuid4().hex
- server_id = attrs.pop('instance', None) or uuid.uuid4().hex
-
- # Set default attribute
- attachment_info = {
- 'id': attachment_id,
- 'volume_id': volume_id,
- 'instance': server_id,
- 'status': random.choice([
+ ]
+ ),
+ 'availability_zone': f'az-{uuid.uuid4().hex}',
+ 'created_at': '2015-09-16T09:28:52.000000',
+ 'name': 'first_group',
+ 'description': f'description-{uuid.uuid4().hex}',
+ 'group_type': group_type,
+ 'volume_types': volume_types,
+ 'volumes': [f'volume-{uuid.uuid4().hex}'],
+ 'group_snapshot_id': None,
+ 'source_group_id': None,
+ 'project_id': f'project-{uuid.uuid4().hex}',
+ }
+
+ # Overwrite default attributes if there are some attributes set
+ group_info.update(attrs)
+
+ group = fakes.FakeResource(None, group_info, loaded=True)
+ return group
+
+
+def create_volume_groups(attrs=None, count=2):
+ """Create multiple fake groups.
+
+ :param attrs: A dictionary with all attributes of group
+ :param count: The number of groups to be faked
+ :return: A list of FakeResource objects
+ """
+ groups = []
+ for n in range(0, count):
+ groups.append(create_one_volume_group(attrs))
+
+ return groups
+
+
+def create_one_volume_group_snapshot(attrs=None, methods=None):
+ """Create a fake group snapshot.
+
+ :param attrs: A dictionary with all attributes
+ :param methods: A dictionary with all methods
+ :return: A FakeResource object with id, name, description, etc.
+ """
+ attrs = attrs or {}
+
+ # Set default attribute
+ group_snapshot_info = {
+ 'id': uuid.uuid4().hex,
+ 'name': f'group-snapshot-{uuid.uuid4().hex}',
+ 'description': f'description-{uuid.uuid4().hex}',
+ 'status': random.choice(['available']),
+ 'group_id': uuid.uuid4().hex,
+ 'group_type_id': uuid.uuid4().hex,
+ 'project_id': uuid.uuid4().hex,
+ }
+
+ # Overwrite default attributes if there are some attributes set
+ group_snapshot_info.update(attrs)
+
+ group_snapshot = fakes.FakeResource(
+ None, group_snapshot_info, methods=methods, loaded=True
+ )
+ return group_snapshot
+
+
+def create_volume_group_snapshots(attrs=None, count=2):
+ """Create multiple fake group snapshots.
+
+ :param attrs: A dictionary with all attributes of group snapshot
+ :param count: The number of group snapshots to be faked
+ :return: A list of FakeResource objects
+ """
+ group_snapshots = []
+ for n in range(0, count):
+ group_snapshots.append(create_one_volume_group_snapshot(attrs))
+
+ return group_snapshots
+
+
+def create_one_volume_group_type(attrs=None, methods=None):
+ """Create a fake group type.
+
+ :param attrs: A dictionary with all attributes of group type
+ :param methods: A dictionary with all methods
+ :return: A FakeResource object with id, name, description, etc.
+ """
+ attrs = attrs or {}
+
+ # Set default attribute
+ group_type_info = {
+ 'id': uuid.uuid4().hex,
+ 'name': f'group-type-{uuid.uuid4().hex}',
+ 'description': f'description-{uuid.uuid4().hex}',
+ 'is_public': random.choice([True, False]),
+ 'group_specs': {},
+ }
+
+ # Overwrite default attributes if there are some attributes set
+ group_type_info.update(attrs)
+
+ group_type = fakes.FakeResource(
+ None, group_type_info, methods=methods, loaded=True
+ )
+ return group_type
+
+
+def create_volume_group_types(attrs=None, count=2):
+ """Create multiple fake group types.
+
+ :param attrs: A dictionary with all attributes of group type
+ :param count: The number of group types to be faked
+ :return: A list of FakeResource objects
+ """
+ group_types = []
+ for n in range(0, count):
+ group_types.append(create_one_volume_group_type(attrs))
+
+ return group_types
+
+
+def create_one_volume_message(attrs=None):
+ """Create a fake message.
+
+ :param attrs: A dictionary with all attributes of message
+ :return: A FakeResource object with id, name, status, etc.
+ """
+ attrs = attrs or {}
+
+ # Set default attribute
+ message_info = {
+ 'created_at': '2016-02-11T11:17:37.000000',
+ 'event_id': f'VOLUME_{random.randint(1, 999999):06d}',
+ 'guaranteed_until': '2016-02-11T11:17:37.000000',
+ 'id': uuid.uuid4().hex,
+ 'message_level': 'ERROR',
+ 'request_id': f'req-{uuid.uuid4().hex}',
+ 'resource_type': 'VOLUME',
+ 'resource_uuid': uuid.uuid4().hex,
+ 'user_message': f'message-{uuid.uuid4().hex}',
+ }
+
+ # Overwrite default attributes if there are some attributes set
+ message_info.update(attrs)
+
+ return fakes.FakeResource(None, message_info, loaded=True)
+
+
+def create_volume_messages(attrs=None, count=2):
+ """Create multiple fake messages.
+
+ :param attrs: A dictionary with all attributes of message
+ :param count: The number of messages to be faked
+ :return: A list of FakeResource objects
+ """
+ messages = []
+ for n in range(0, count):
+ messages.append(create_one_volume_message(attrs))
+
+ return messages
+
+
+def get_volume_messages(messages=None, count=2):
+ """Get an iterable MagicMock object with a list of faked messages.
+
+ If messages list is provided, then initialize the Mock object with the
+ list. Otherwise create one.
+
+ :param messages: A list of FakeResource objects faking messages
+ :param count: The number of messages to be faked
+ :return An iterable Mock object with side_effect set to a list of faked
+ messages
+ """
+ if messages is None:
+ messages = create_volume_messages(count)
+
+ return mock.Mock(side_effect=messages)
+
+
+def create_one_volume_attachment(attrs=None):
+ """Create a fake volume attachment.
+
+ :param attrs: A dictionary with all attributes of volume attachment
+ :return: A FakeResource object with id, status, etc.
+ """
+ attrs = attrs or {}
+
+ attachment_id = uuid.uuid4().hex
+ volume_id = attrs.pop('volume_id', None) or uuid.uuid4().hex
+ server_id = attrs.pop('instance', None) or uuid.uuid4().hex
+
+ # Set default attribute
+ attachment_info = {
+ 'id': attachment_id,
+ 'volume_id': volume_id,
+ 'instance': server_id,
+ 'status': random.choice(
+ [
'attached',
'attaching',
'detached',
@@ -424,67 +376,63 @@ class FakeVolumeAttachment:
'error_attaching',
'error_detaching',
'deleted',
- ]),
- 'attach_mode': random.choice(['ro', 'rw']),
- 'attached_at': '2015-09-16T09:28:52.000000',
- 'detached_at': None,
- 'connection_info': {
- 'access_mode': 'rw',
- 'attachment_id': attachment_id,
- 'auth_method': 'CHAP',
- 'auth_password': 'AcUZ8PpxLHwzypMC',
- 'auth_username': '7j3EZQWT3rbE6pcSGKvK',
- 'cacheable': False,
- 'driver_volume_type': 'iscsi',
- 'encrypted': False,
- 'qos_specs': None,
- 'target_discovered': False,
- 'target_iqn':
- f'iqn.2010-10.org.openstack:volume-{attachment_id}',
- 'target_lun': '1',
- 'target_portal': '192.168.122.170:3260',
- 'volume_id': volume_id,
- },
- }
-
- # Overwrite default attributes if there are some attributes set
- attachment_info.update(attrs)
-
- return fakes.FakeResource(
- None,
- attachment_info,
- loaded=True)
-
- @staticmethod
- def create_volume_attachments(attrs=None, count=2):
- """Create multiple fake volume attachments.
-
- :param attrs: A dictionary with all attributes of volume attachment
- :param count: The number of volume attachments to be faked
- :return: A list of FakeResource objects
- """
- attachments = []
-
- for n in range(0, count):
- attachments.append(
- FakeVolumeAttachment.create_one_volume_attachment(attrs))
-
- return attachments
-
- @staticmethod
- def get_volume_attachments(attachments=None, count=2):
- """Get an iterable MagicMock object with a list of faked volumes.
-
- If attachments list is provided, then initialize the Mock object with
- the list. Otherwise create one.
-
- :param attachments: A list of FakeResource objects faking volume
- attachments
- :param count: The number of volume attachments to be faked
- :return An iterable Mock object with side_effect set to a list of faked
- volume attachments
- """
- if attachments is None:
- attachments = FakeVolumeAttachment.create_volume_attachments(count)
-
- return mock.Mock(side_effect=attachments)
+ ]
+ ),
+ 'attach_mode': random.choice(['ro', 'rw']),
+ 'attached_at': '2015-09-16T09:28:52.000000',
+ 'detached_at': None,
+ 'connection_info': {
+ 'access_mode': 'rw',
+ 'attachment_id': attachment_id,
+ 'auth_method': 'CHAP',
+ 'auth_password': 'AcUZ8PpxLHwzypMC',
+ 'auth_username': '7j3EZQWT3rbE6pcSGKvK',
+ 'cacheable': False,
+ 'driver_volume_type': 'iscsi',
+ 'encrypted': False,
+ 'qos_specs': None,
+ 'target_discovered': False,
+ 'target_iqn': f'iqn.2010-10.org.openstack:volume-{attachment_id}',
+ 'target_lun': '1',
+ 'target_portal': '192.168.122.170:3260',
+ 'volume_id': volume_id,
+ },
+ }
+
+ # Overwrite default attributes if there are some attributes set
+ attachment_info.update(attrs)
+
+ return fakes.FakeResource(None, attachment_info, loaded=True)
+
+
+def create_volume_attachments(attrs=None, count=2):
+ """Create multiple fake volume attachments.
+
+ :param attrs: A dictionary with all attributes of volume attachment
+ :param count: The number of volume attachments to be faked
+ :return: A list of FakeResource objects
+ """
+ attachments = []
+
+ for n in range(0, count):
+ attachments.append(create_one_volume_attachment(attrs))
+
+ return attachments
+
+
+def get_volume_attachments(attachments=None, count=2):
+ """Get an iterable MagicMock object with a list of faked volumes.
+
+ If attachments list is provided, then initialize the Mock object with
+ the list. Otherwise create one.
+
+ :param attachments: A list of FakeResource objects faking volume
+ attachments
+ :param count: The number of volume attachments to be faked
+ :return An iterable Mock object with side_effect set to a list of faked
+ volume attachments
+ """
+ if attachments is None:
+ attachments = create_volume_attachments(count)
+
+ return mock.Mock(side_effect=attachments)
diff --git a/openstackclient/tests/unit/volume/v3/test_block_storage_cluster.py b/openstackclient/tests/unit/volume/v3/test_block_storage_cluster.py
index d87a946b..fdfd1100 100644
--- a/openstackclient/tests/unit/volume/v3/test_block_storage_cluster.py
+++ b/openstackclient/tests/unit/volume/v3/test_block_storage_cluster.py
@@ -30,7 +30,7 @@ class TestBlockStorageCluster(volume_fakes.TestVolume):
class TestBlockStorageClusterList(TestBlockStorageCluster):
# The cluster to be listed
- fake_clusters = volume_fakes.FakeCluster.create_clusters()
+ fake_clusters = volume_fakes.create_clusters()
def setUp(self):
super().setUp()
@@ -176,7 +176,7 @@ class TestBlockStorageClusterList(TestBlockStorageCluster):
class TestBlockStorageClusterSet(TestBlockStorageCluster):
- cluster = volume_fakes.FakeCluster.create_one_cluster()
+ cluster = volume_fakes.create_one_cluster()
columns = (
'Name',
'Binary',
@@ -347,7 +347,7 @@ class TestBlockStorageClusterSet(TestBlockStorageCluster):
class TestBlockStorageClusterShow(TestBlockStorageCluster):
- cluster = volume_fakes.FakeCluster.create_one_cluster()
+ cluster = volume_fakes.create_one_cluster()
columns = (
'Name',
'Binary',
diff --git a/openstackclient/tests/unit/volume/v3/test_block_storage_resource_filter.py b/openstackclient/tests/unit/volume/v3/test_block_storage_resource_filter.py
index b886726d..086339ff 100644
--- a/openstackclient/tests/unit/volume/v3/test_block_storage_resource_filter.py
+++ b/openstackclient/tests/unit/volume/v3/test_block_storage_resource_filter.py
@@ -31,8 +31,7 @@ class TestBlockStorageResourceFilter(volume_fakes.TestVolume):
class TestBlockStorageResourceFilterList(TestBlockStorageResourceFilter):
# The resource filters to be listed
- fake_resource_filters = \
- volume_fakes.FakeResourceFilter.create_resource_filters()
+ fake_resource_filters = volume_fakes.create_resource_filters()
def setUp(self):
super().setUp()
@@ -86,8 +85,7 @@ class TestBlockStorageResourceFilterList(TestBlockStorageResourceFilter):
class TestBlockStorageResourceFilterShow(TestBlockStorageResourceFilter):
# The resource filters to be listed
- fake_resource_filter = \
- volume_fakes.FakeResourceFilter.create_one_resource_filter()
+ fake_resource_filter = volume_fakes.create_one_resource_filter()
def setUp(self):
super().setUp()
diff --git a/openstackclient/tests/unit/volume/v3/test_volume_attachment.py b/openstackclient/tests/unit/volume/v3/test_volume_attachment.py
index 44fac6c5..c0bf5ae7 100644
--- a/openstackclient/tests/unit/volume/v3/test_volume_attachment.py
+++ b/openstackclient/tests/unit/volume/v3/test_volume_attachment.py
@@ -41,11 +41,11 @@ class TestVolumeAttachment(volume_fakes.TestVolume):
class TestVolumeAttachmentCreate(TestVolumeAttachment):
- volume = volume_fakes.FakeVolume.create_one_volume()
+ volume = volume_fakes.create_one_volume()
server = compute_fakes.FakeServer.create_one_server()
- volume_attachment = \
- volume_fakes.FakeVolumeAttachment.create_one_volume_attachment(
- attrs={'instance': server.id, 'volume_id': volume.id})
+ volume_attachment = volume_fakes.create_one_volume_attachment(
+ attrs={'instance': server.id, 'volume_id': volume.id},
+ )
columns = (
'ID',
@@ -238,8 +238,7 @@ class TestVolumeAttachmentCreate(TestVolumeAttachment):
class TestVolumeAttachmentDelete(TestVolumeAttachment):
- volume_attachment = \
- volume_fakes.FakeVolumeAttachment.create_one_volume_attachment()
+ volume_attachment = volume_fakes.create_one_volume_attachment()
def setUp(self):
super().setUp()
@@ -290,8 +289,7 @@ class TestVolumeAttachmentDelete(TestVolumeAttachment):
class TestVolumeAttachmentSet(TestVolumeAttachment):
- volume_attachment = \
- volume_fakes.FakeVolumeAttachment.create_one_volume_attachment()
+ volume_attachment = volume_fakes.create_one_volume_attachment()
columns = (
'ID',
@@ -391,8 +389,7 @@ class TestVolumeAttachmentSet(TestVolumeAttachment):
class TestVolumeAttachmentComplete(TestVolumeAttachment):
- volume_attachment = \
- volume_fakes.FakeVolumeAttachment.create_one_volume_attachment()
+ volume_attachment = volume_fakes.create_one_volume_attachment()
def setUp(self):
super().setUp()
@@ -444,8 +441,7 @@ class TestVolumeAttachmentComplete(TestVolumeAttachment):
class TestVolumeAttachmentList(TestVolumeAttachment):
project = identity_fakes.FakeProject.create_one_project()
- volume_attachments = \
- volume_fakes.FakeVolumeAttachment.create_volume_attachments()
+ volume_attachments = volume_fakes.create_volume_attachments()
columns = (
'ID',
diff --git a/openstackclient/tests/unit/volume/v3/test_volume_group.py b/openstackclient/tests/unit/volume/v3/test_volume_group.py
index 13ef38d2..96079a08 100644
--- a/openstackclient/tests/unit/volume/v3/test_volume_group.py
+++ b/openstackclient/tests/unit/volume/v3/test_volume_group.py
@@ -35,10 +35,9 @@ class TestVolumeGroup(volume_fakes.TestVolume):
class TestVolumeGroupCreate(TestVolumeGroup):
- fake_volume_type = volume_fakes.FakeVolumeType.create_one_volume_type()
- fake_volume_group_type = \
- volume_fakes.FakeVolumeGroupType.create_one_volume_group_type()
- fake_volume_group = volume_fakes.FakeVolumeGroup.create_one_volume_group(
+ fake_volume_type = volume_fakes.create_one_volume_type()
+ fake_volume_group_type = volume_fakes.create_one_volume_group_type()
+ fake_volume_group = volume_fakes.create_one_volume_group(
attrs={
'group_type': fake_volume_group_type.id,
'volume_types': [fake_volume_type.id],
@@ -180,8 +179,7 @@ class TestVolumeGroupCreate(TestVolumeGroup):
class TestVolumeGroupDelete(TestVolumeGroup):
- fake_volume_group = \
- volume_fakes.FakeVolumeGroup.create_one_volume_group()
+ fake_volume_group = volume_fakes.create_one_volume_group()
def setUp(self):
super().setUp()
@@ -236,8 +234,7 @@ class TestVolumeGroupDelete(TestVolumeGroup):
class TestVolumeGroupSet(TestVolumeGroup):
- fake_volume_group = \
- volume_fakes.FakeVolumeGroup.create_one_volume_group()
+ fake_volume_group = volume_fakes.create_one_volume_group()
columns = (
'ID',
@@ -368,8 +365,7 @@ class TestVolumeGroupSet(TestVolumeGroup):
class TestVolumeGroupList(TestVolumeGroup):
- fake_volume_groups = \
- volume_fakes.FakeVolumeGroup.create_volume_groups()
+ fake_volume_groups = volume_fakes.create_volume_groups()
columns = (
'ID',
@@ -436,8 +432,7 @@ class TestVolumeGroupList(TestVolumeGroup):
class TestVolumeGroupFailover(TestVolumeGroup):
- fake_volume_group = \
- volume_fakes.FakeVolumeGroup.create_one_volume_group()
+ fake_volume_group = volume_fakes.create_one_volume_group()
def setUp(self):
super().setUp()
diff --git a/openstackclient/tests/unit/volume/v3/test_volume_group_snapshot.py b/openstackclient/tests/unit/volume/v3/test_volume_group_snapshot.py
index 509d9f08..2a5a30f0 100644
--- a/openstackclient/tests/unit/volume/v3/test_volume_group_snapshot.py
+++ b/openstackclient/tests/unit/volume/v3/test_volume_group_snapshot.py
@@ -32,9 +32,9 @@ class TestVolumeGroupSnapshot(volume_fakes.TestVolume):
class TestVolumeGroupSnapshotCreate(TestVolumeGroupSnapshot):
- fake_volume_group = volume_fakes.FakeVolumeGroup.create_one_volume_group()
+ fake_volume_group = volume_fakes.create_one_volume_group()
fake_volume_group_snapshot = \
- volume_fakes.FakeVolumeGroupSnapshot.create_one_volume_group_snapshot()
+ volume_fakes.create_one_volume_group_snapshot()
columns = (
'ID',
@@ -141,7 +141,7 @@ class TestVolumeGroupSnapshotCreate(TestVolumeGroupSnapshot):
class TestVolumeGroupSnapshotDelete(TestVolumeGroupSnapshot):
fake_volume_group_snapshot = \
- volume_fakes.FakeVolumeGroupSnapshot.create_one_volume_group_snapshot()
+ volume_fakes.create_one_volume_group_snapshot()
def setUp(self):
super().setUp()
@@ -195,8 +195,7 @@ class TestVolumeGroupSnapshotDelete(TestVolumeGroupSnapshot):
class TestVolumeGroupSnapshotList(TestVolumeGroupSnapshot):
- fake_volume_group_snapshots = \
- volume_fakes.FakeVolumeGroupSnapshot.create_volume_group_snapshots()
+ fake_volume_group_snapshots = volume_fakes.create_volume_group_snapshots()
columns = (
'ID',
diff --git a/openstackclient/tests/unit/volume/v3/test_volume_group_type.py b/openstackclient/tests/unit/volume/v3/test_volume_group_type.py
index 7e758a2c..34b4e501 100644
--- a/openstackclient/tests/unit/volume/v3/test_volume_group_type.py
+++ b/openstackclient/tests/unit/volume/v3/test_volume_group_type.py
@@ -34,8 +34,7 @@ class TestVolumeGroupTypeCreate(TestVolumeGroupType):
maxDiff = 2000
- fake_volume_group_type = \
- volume_fakes.FakeVolumeGroupType.create_one_volume_group_type()
+ fake_volume_group_type = volume_fakes.create_one_volume_group_type()
columns = (
'ID',
@@ -133,8 +132,7 @@ class TestVolumeGroupTypeCreate(TestVolumeGroupType):
class TestVolumeGroupTypeDelete(TestVolumeGroupType):
- fake_volume_group_type = \
- volume_fakes.FakeVolumeGroupType.create_one_volume_group_type()
+ fake_volume_group_type = volume_fakes.create_one_volume_group_type()
def setUp(self):
super().setUp()
@@ -187,13 +185,13 @@ class TestVolumeGroupTypeDelete(TestVolumeGroupType):
class TestVolumeGroupTypeSet(TestVolumeGroupType):
- fake_volume_group_type = \
- volume_fakes.FakeVolumeGroupType.create_one_volume_group_type(
- methods={
- 'get_keys': {'foo': 'bar'},
- 'set_keys': None,
- 'unset_keys': None,
- })
+ fake_volume_group_type = volume_fakes.create_one_volume_group_type(
+ methods={
+ 'get_keys': {'foo': 'bar'},
+ 'set_keys': None,
+ 'unset_keys': None,
+ },
+ )
columns = (
'ID',
@@ -316,9 +314,9 @@ class TestVolumeGroupTypeSet(TestVolumeGroupType):
class TestVolumeGroupTypeUnset(TestVolumeGroupType):
- fake_volume_group_type = \
- volume_fakes.FakeVolumeGroupType.create_one_volume_group_type(
- methods={'unset_keys': None})
+ fake_volume_group_type = volume_fakes.create_one_volume_group_type(
+ methods={'unset_keys': None},
+ )
columns = (
'ID',
@@ -393,8 +391,7 @@ class TestVolumeGroupTypeUnset(TestVolumeGroupType):
class TestVolumeGroupTypeList(TestVolumeGroupType):
- fake_volume_group_types = \
- volume_fakes.FakeVolumeGroupType.create_volume_group_types()
+ fake_volume_group_types = volume_fakes.create_volume_group_types()
columns = (
'ID',
diff --git a/openstackclient/tests/unit/volume/v3/test_volume_message.py b/openstackclient/tests/unit/volume/v3/test_volume_message.py
index 8cabc0c3..45b0747e 100644
--- a/openstackclient/tests/unit/volume/v3/test_volume_message.py
+++ b/openstackclient/tests/unit/volume/v3/test_volume_message.py
@@ -34,15 +34,14 @@ class TestVolumeMessage(volume_fakes.TestVolume):
class TestVolumeMessageDelete(TestVolumeMessage):
- fake_messages = volume_fakes.FakeVolumeMessage.create_volume_messages(
- count=2)
+ fake_messages = volume_fakes.create_volume_messages(count=2)
def setUp(self):
super().setUp()
- self.volume_messages_mock.get = \
- volume_fakes.FakeVolumeMessage.get_volume_messages(
- self.fake_messages)
+ self.volume_messages_mock.get = volume_fakes.get_volume_messages(
+ self.fake_messages,
+ )
self.volume_messages_mock.delete.return_value = None
# Get the command object to mock
@@ -139,8 +138,7 @@ class TestVolumeMessageDelete(TestVolumeMessage):
class TestVolumeMessageList(TestVolumeMessage):
fake_project = identity_fakes.FakeProject.create_one_project()
- fake_messages = volume_fakes.FakeVolumeMessage.create_volume_messages(
- count=3)
+ fake_messages = volume_fakes.create_volume_messages(count=3)
columns = (
'ID',
@@ -253,7 +251,7 @@ class TestVolumeMessageList(TestVolumeMessage):
class TestVolumeMessageShow(TestVolumeMessage):
- fake_message = volume_fakes.FakeVolumeMessage.create_one_volume_message()
+ fake_message = volume_fakes.create_one_volume_message()
columns = (
'created_at',