diff options
Diffstat (limited to 'openstackclient/tests/volume')
| -rw-r--r-- | openstackclient/tests/volume/__init__.py | 0 | ||||
| -rw-r--r-- | openstackclient/tests/volume/test_find_resource.py | 83 | ||||
| -rw-r--r-- | openstackclient/tests/volume/v1/__init__.py | 0 | ||||
| -rw-r--r-- | openstackclient/tests/volume/v1/fakes.py | 280 | ||||
| -rw-r--r-- | openstackclient/tests/volume/v1/test_qos_specs.py | 472 | ||||
| -rw-r--r-- | openstackclient/tests/volume/v1/test_transfer_request.py | 114 | ||||
| -rw-r--r-- | openstackclient/tests/volume/v1/test_volume.py | 716 | ||||
| -rw-r--r-- | openstackclient/tests/volume/v2/__init__.py | 0 | ||||
| -rw-r--r-- | openstackclient/tests/volume/v2/fakes.py | 696 | ||||
| -rw-r--r-- | openstackclient/tests/volume/v2/test_backup.py | 388 | ||||
| -rw-r--r-- | openstackclient/tests/volume/v2/test_qos_specs.py | 453 | ||||
| -rw-r--r-- | openstackclient/tests/volume/v2/test_service.py | 286 | ||||
| -rw-r--r-- | openstackclient/tests/volume/v2/test_snapshot.py | 458 | ||||
| -rw-r--r-- | openstackclient/tests/volume/v2/test_transfer_request.py | 114 | ||||
| -rw-r--r-- | openstackclient/tests/volume/v2/test_type.py | 575 | ||||
| -rw-r--r-- | openstackclient/tests/volume/v2/test_volume.py | 966 |
16 files changed, 0 insertions, 5601 deletions
diff --git a/openstackclient/tests/volume/__init__.py b/openstackclient/tests/volume/__init__.py deleted file mode 100644 index e69de29b..00000000 --- a/openstackclient/tests/volume/__init__.py +++ /dev/null diff --git a/openstackclient/tests/volume/test_find_resource.py b/openstackclient/tests/volume/test_find_resource.py deleted file mode 100644 index 982b02f0..00000000 --- a/openstackclient/tests/volume/test_find_resource.py +++ /dev/null @@ -1,83 +0,0 @@ -# Copyright 2013 Nebula Inc. -# -# Licensed under the Apache License, Version 2.0 (the "License"); you may -# not use this file except in compliance with the License. You may obtain -# a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT -# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the -# License for the specific language governing permissions and limitations -# under the License. -# - -import mock - -from cinderclient.v1 import volume_snapshots -from cinderclient.v1 import volumes -from osc_lib import exceptions -from osc_lib import utils - -from openstackclient.tests import utils as test_utils -from openstackclient.volume import client # noqa - - -# Monkey patch for v1 cinderclient -# NOTE(dtroyer): Do here because openstackclient.volume.client -# doesn't do it until the client object is created now. -volumes.Volume.NAME_ATTR = 'display_name' -volume_snapshots.Snapshot.NAME_ATTR = 'display_name' - - -ID = '1after909' -NAME = 'PhilSpector' - - -class TestFindResourceVolumes(test_utils.TestCase): - - def setUp(self): - super(TestFindResourceVolumes, self).setUp() - api = mock.Mock() - api.client = mock.Mock() - api.client.get = mock.Mock() - resp = mock.Mock() - body = {"volumes": [{"id": ID, 'display_name': NAME}]} - api.client.get.side_effect = [Exception("Not found"), - Exception("Not found"), - (resp, body)] - self.manager = volumes.VolumeManager(api) - - def test_find(self): - result = utils.find_resource(self.manager, NAME) - self.assertEqual(ID, result.id) - self.assertEqual(NAME, result.display_name) - - def test_not_find(self): - self.assertRaises(exceptions.CommandError, utils.find_resource, - self.manager, 'GeorgeMartin') - - -class TestFindResourceVolumeSnapshots(test_utils.TestCase): - - def setUp(self): - super(TestFindResourceVolumeSnapshots, self).setUp() - api = mock.Mock() - api.client = mock.Mock() - api.client.get = mock.Mock() - resp = mock.Mock() - body = {"snapshots": [{"id": ID, 'display_name': NAME}]} - api.client.get.side_effect = [Exception("Not found"), - Exception("Not found"), - (resp, body)] - self.manager = volume_snapshots.SnapshotManager(api) - - def test_find(self): - result = utils.find_resource(self.manager, NAME) - self.assertEqual(ID, result.id) - self.assertEqual(NAME, result.display_name) - - def test_not_find(self): - self.assertRaises(exceptions.CommandError, utils.find_resource, - self.manager, 'GeorgeMartin') diff --git a/openstackclient/tests/volume/v1/__init__.py b/openstackclient/tests/volume/v1/__init__.py deleted file mode 100644 index e69de29b..00000000 --- a/openstackclient/tests/volume/v1/__init__.py +++ /dev/null diff --git a/openstackclient/tests/volume/v1/fakes.py b/openstackclient/tests/volume/v1/fakes.py deleted file mode 100644 index b96f925d..00000000 --- a/openstackclient/tests/volume/v1/fakes.py +++ /dev/null @@ -1,280 +0,0 @@ -# Copyright 2013 Nebula Inc. -# -# Licensed under the Apache License, Version 2.0 (the "License"); you may -# not use this file except in compliance with the License. You may obtain -# a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT -# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the -# License for the specific language governing permissions and limitations -# under the License. -# - -import mock - -from openstackclient.tests import fakes -from openstackclient.tests.identity.v2_0 import fakes as identity_fakes -from openstackclient.tests import utils - - -volume_id = 'vvvvvvvv-vvvv-vvvv-vvvvvvvv' -volume_name = 'nigel' -volume_description = 'Nigel Tufnel' -volume_status = 'available' -volume_size = 120 -volume_type = 'to-eleven' -volume_zone = 'stonehenge' -volume_metadata = { - 'Alpha': 'a', - 'Beta': 'b', - 'Gamma': 'g', -} -volume_metadata_str = "Alpha='a', Beta='b', Gamma='g'" - -VOLUME = { - 'id': volume_id, - 'display_name': volume_name, - 'display_description': volume_description, - 'size': volume_size, - 'status': volume_status, - 'attach_status': 'detached', - 'availability_zone': volume_zone, - 'volume_type': volume_type, - 'metadata': volume_metadata, -} - -extension_name = 'SchedulerHints' -extension_namespace = 'http://docs.openstack.org/'\ - 'block-service/ext/scheduler-hints/api/v2' -extension_description = 'Pass arbitrary key/value'\ - 'pairs to the scheduler.' -extension_updated = '2014-02-07T12:00:0-00:00' -extension_alias = 'OS-SCH-HNT' -extension_links = '[{"href":'\ - '"https://github.com/openstack/block-api", "type":'\ - ' "text/html", "rel": "describedby"}]' - -EXTENSION = { - 'name': extension_name, - 'namespace': extension_namespace, - 'description': extension_description, - 'updated': extension_updated, - 'alias': extension_alias, - 'links': extension_links, -} - -# NOTE(dtroyer): duplicating here the minimum image info needed to test -# volume create --image until circular references can be -# avoided by refactoring the test fakes. - -image_id = 'im1' -image_name = 'graven' - - -IMAGE = { - 'id': image_id, - 'name': image_name, -} - -type_id = "5520dc9e-6f9b-4378-a719-729911c0f407" -type_name = "fake-lvmdriver-1" - -TYPE = { - 'id': type_id, - 'name': type_name -} - -qos_id = '6f2be1de-997b-4230-b76c-a3633b59e8fb' -qos_consumer = 'front-end' -qos_default_consumer = 'both' -qos_name = "fake-qos-specs" -qos_specs = { - 'foo': 'bar', - 'iops': '9001' -} -qos_association = { - 'association_type': 'volume_type', - 'name': type_name, - 'id': type_id -} - -QOS = { - 'id': qos_id, - 'consumer': qos_consumer, - 'name': qos_name -} - -QOS_DEFAULT_CONSUMER = { - 'id': qos_id, - 'consumer': qos_default_consumer, - 'name': qos_name -} - -QOS_WITH_SPECS = { - 'id': qos_id, - 'consumer': qos_consumer, - 'name': qos_name, - 'specs': qos_specs -} - -QOS_WITH_ASSOCIATIONS = { - 'id': qos_id, - 'consumer': qos_consumer, - 'name': qos_name, - 'specs': qos_specs, - 'associations': [qos_association] -} - - -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': 'ce26708d-a7f8-4b4b-9861-4a80256615a7', - 'name': 'fake_transfer_name', - 'id': '731a7f53-aa92-4fbd-9de3-6f7d729c926b' - } - - # 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 - - -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.MagicMock(side_effect=services) - - -class FakeImagev1Client(object): - - def __init__(self, **kwargs): - self.images = mock.Mock() - - -class FakeVolumev1Client(object): - - def __init__(self, **kwargs): - self.volumes = mock.Mock() - self.volumes.resource_class = fakes.FakeResource(None, {}) - self.services = mock.Mock() - self.services.resource_class = fakes.FakeResource(None, {}) - self.extensions = mock.Mock() - self.extensions.resource_class = fakes.FakeResource(None, {}) - self.qos_specs = mock.Mock() - self.qos_specs.resource_class = fakes.FakeResource(None, {}) - self.volume_types = mock.Mock() - self.volume_types.resource_class = fakes.FakeResource(None, {}) - self.transfers = mock.Mock() - self.transfers.resource_class = fakes.FakeResource(None, {}) - self.auth_token = kwargs['token'] - self.management_url = kwargs['endpoint'] - - -class TestVolumev1(utils.TestCommand): - - def setUp(self): - super(TestVolumev1, self).setUp() - - self.app.client_manager.volume = FakeVolumev1Client( - endpoint=fakes.AUTH_URL, - token=fakes.AUTH_TOKEN, - ) - - self.app.client_manager.identity = identity_fakes.FakeIdentityv2Client( - endpoint=fakes.AUTH_URL, - token=fakes.AUTH_TOKEN, - ) - - self.app.client_manager.image = FakeImagev1Client( - endpoint=fakes.AUTH_URL, - token=fakes.AUTH_TOKEN, - ) diff --git a/openstackclient/tests/volume/v1/test_qos_specs.py b/openstackclient/tests/volume/v1/test_qos_specs.py deleted file mode 100644 index 4e1733fd..00000000 --- a/openstackclient/tests/volume/v1/test_qos_specs.py +++ /dev/null @@ -1,472 +0,0 @@ -# Copyright 2015 iWeb Technologies Inc. -# -# Licensed under the Apache License, Version 2.0 (the "License"); you may -# not use this file except in compliance with the License. You may obtain -# a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT -# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the -# License for the specific language governing permissions and limitations -# under the License. -# - -import copy - -from osc_lib import utils - -from openstackclient.tests import fakes -from openstackclient.tests.volume.v1 import fakes as volume_fakes -from openstackclient.volume.v1 import qos_specs - - -class TestQos(volume_fakes.TestVolumev1): - - def setUp(self): - super(TestQos, self).setUp() - - self.qos_mock = self.app.client_manager.volume.qos_specs - self.qos_mock.reset_mock() - - self.types_mock = self.app.client_manager.volume.volume_types - self.types_mock.reset_mock() - - -class TestQosAssociate(TestQos): - - def setUp(self): - super(TestQosAssociate, self).setUp() - - # Get the command object to test - self.cmd = qos_specs.AssociateQos(self.app, None) - - def test_qos_associate(self): - self.qos_mock.get.return_value = fakes.FakeResource( - None, - copy.deepcopy(volume_fakes.QOS), - loaded=True - ) - self.types_mock.get.return_value = fakes.FakeResource( - None, - copy.deepcopy(volume_fakes.TYPE), - loaded=True - ) - arglist = [ - volume_fakes.qos_id, - volume_fakes.type_id - ] - verifylist = [ - ('qos_spec', volume_fakes.qos_id), - ('volume_type', volume_fakes.type_id) - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - result = self.cmd.take_action(parsed_args) - - self.qos_mock.associate.assert_called_with( - volume_fakes.qos_id, - volume_fakes.type_id - ) - self.assertIsNone(result) - - -class TestQosCreate(TestQos): - - columns = ( - 'consumer', - 'id', - 'name', - ) - datalist = ( - volume_fakes.qos_consumer, - volume_fakes.qos_id, - volume_fakes.qos_name - ) - - def setUp(self): - super(TestQosCreate, self).setUp() - - # Get the command object to test - self.cmd = qos_specs.CreateQos(self.app, None) - - def test_qos_create_without_properties(self): - self.qos_mock.create.return_value = fakes.FakeResource( - None, - copy.deepcopy(volume_fakes.QOS_DEFAULT_CONSUMER), - loaded=True - ) - - arglist = [ - volume_fakes.qos_name, - ] - verifylist = [ - ('name', volume_fakes.qos_name), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - columns, data = self.cmd.take_action(parsed_args) - - self.qos_mock.create.assert_called_with( - volume_fakes.qos_name, - {'consumer': volume_fakes.qos_default_consumer} - ) - - self.assertEqual(self.columns, columns) - datalist = ( - volume_fakes.qos_default_consumer, - volume_fakes.qos_id, - volume_fakes.qos_name - ) - self.assertEqual(datalist, data) - - def test_qos_create_with_consumer(self): - self.qos_mock.create.return_value = fakes.FakeResource( - None, - copy.deepcopy(volume_fakes.QOS), - loaded=True - ) - - arglist = [ - volume_fakes.qos_name, - '--consumer', volume_fakes.qos_consumer - ] - verifylist = [ - ('name', volume_fakes.qos_name), - ('consumer', volume_fakes.qos_consumer) - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - columns, data = self.cmd.take_action(parsed_args) - - self.qos_mock.create.assert_called_with( - volume_fakes.qos_name, - {'consumer': volume_fakes.qos_consumer} - ) - self.assertEqual(self.columns, columns) - self.assertEqual(self.datalist, data) - - def test_qos_create_with_properties(self): - self.qos_mock.create.return_value = fakes.FakeResource( - None, - copy.deepcopy(volume_fakes.QOS_WITH_SPECS), - loaded=True - ) - - arglist = [ - volume_fakes.qos_name, - '--consumer', volume_fakes.qos_consumer, - '--property', 'foo=bar', - '--property', 'iops=9001' - ] - verifylist = [ - ('name', volume_fakes.qos_name), - ('consumer', volume_fakes.qos_consumer), - ('property', volume_fakes.qos_specs) - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - columns, data = self.cmd.take_action(parsed_args) - - specs = volume_fakes.qos_specs.copy() - specs.update({'consumer': volume_fakes.qos_consumer}) - self.qos_mock.create.assert_called_with( - volume_fakes.qos_name, - specs - ) - - columns = self.columns + ( - 'specs', - ) - self.assertEqual(columns, columns) - datalist = self.datalist + ( - volume_fakes.qos_specs, - ) - self.assertEqual(datalist, data) - - -class TestQosDelete(TestQos): - - def setUp(self): - super(TestQosDelete, self).setUp() - - self.qos_mock.get.return_value = fakes.FakeResource( - None, - copy.deepcopy(volume_fakes.QOS), - loaded=True, - ) - - # Get the command object to test - self.cmd = qos_specs.DeleteQos(self.app, None) - - def test_qos_delete_with_id(self): - arglist = [ - volume_fakes.qos_id - ] - verifylist = [ - ('qos_specs', [volume_fakes.qos_id]) - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - result = self.cmd.take_action(parsed_args) - - self.qos_mock.delete.assert_called_with(volume_fakes.qos_id, False) - self.assertIsNone(result) - - def test_qos_delete_with_name(self): - arglist = [ - volume_fakes.qos_name - ] - verifylist = [ - ('qos_specs', [volume_fakes.qos_name]) - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - result = self.cmd.take_action(parsed_args) - - self.qos_mock.delete.assert_called_with(volume_fakes.qos_id, False) - self.assertIsNone(result) - - def test_qos_delete_with_force(self): - arglist = [ - '--force', - volume_fakes.qos_id - ] - verifylist = [ - ('force', True), - ('qos_specs', [volume_fakes.qos_id]) - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - result = self.cmd.take_action(parsed_args) - - self.qos_mock.delete.assert_called_with(volume_fakes.qos_id, True) - self.assertIsNone(result) - - -class TestQosDisassociate(TestQos): - - def setUp(self): - super(TestQosDisassociate, self).setUp() - - # Get the command object to test - self.cmd = qos_specs.DisassociateQos(self.app, None) - - def test_qos_disassociate_with_volume_type(self): - self.qos_mock.get.return_value = fakes.FakeResource( - None, - copy.deepcopy(volume_fakes.QOS), - loaded=True - ) - self.types_mock.get.return_value = fakes.FakeResource( - None, - copy.deepcopy(volume_fakes.TYPE), - loaded=True - ) - arglist = [ - volume_fakes.qos_id, - '--volume-type', volume_fakes.type_id - ] - verifylist = [ - ('qos_spec', volume_fakes.qos_id), - ('volume_type', volume_fakes.type_id) - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - result = self.cmd.take_action(parsed_args) - - self.qos_mock.disassociate.assert_called_with( - volume_fakes.qos_id, - volume_fakes.type_id - ) - self.assertIsNone(result) - - def test_qos_disassociate_with_all_volume_types(self): - self.qos_mock.get.return_value = fakes.FakeResource( - None, - copy.deepcopy(volume_fakes.QOS), - loaded=True - ) - - arglist = [ - volume_fakes.qos_id, - '--all' - ] - verifylist = [ - ('qos_spec', volume_fakes.qos_id) - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - result = self.cmd.take_action(parsed_args) - - self.qos_mock.disassociate_all.assert_called_with(volume_fakes.qos_id) - self.assertIsNone(result) - - -class TestQosList(TestQos): - - def setUp(self): - super(TestQosList, self).setUp() - - self.qos_mock.get.return_value = fakes.FakeResource( - None, - copy.deepcopy(volume_fakes.QOS_WITH_ASSOCIATIONS), - loaded=True, - ) - self.qos_mock.list.return_value = [self.qos_mock.get.return_value] - self.qos_mock.get_associations.return_value = [fakes.FakeResource( - None, - copy.deepcopy(volume_fakes.qos_association), - loaded=True, - )] - - # Get the command object to test - self.cmd = qos_specs.ListQos(self.app, None) - - def test_qos_list(self): - arglist = [] - verifylist = [] - - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - columns, data = self.cmd.take_action(parsed_args) - self.qos_mock.list.assert_called_with() - - collist = ( - 'ID', - 'Name', - 'Consumer', - 'Associations', - 'Specs', - ) - self.assertEqual(collist, columns) - datalist = (( - volume_fakes.qos_id, - volume_fakes.qos_name, - volume_fakes.qos_consumer, - volume_fakes.type_name, - utils.format_dict(volume_fakes.qos_specs), - ), ) - self.assertEqual(datalist, tuple(data)) - - -class TestQosSet(TestQos): - - def setUp(self): - super(TestQosSet, self).setUp() - - # Get the command object to test - self.cmd = qos_specs.SetQos(self.app, None) - - def test_qos_set_with_properties_with_id(self): - self.qos_mock.get.return_value = fakes.FakeResource( - None, - copy.deepcopy(volume_fakes.QOS_WITH_SPECS), - loaded=True - ) - arglist = [ - volume_fakes.qos_id, - '--property', 'foo=bar', - '--property', 'iops=9001' - ] - verifylist = [ - ('qos_spec', volume_fakes.qos_id), - ('property', volume_fakes.qos_specs) - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - result = self.cmd.take_action(parsed_args) - - self.qos_mock.set_keys.assert_called_with( - volume_fakes.qos_id, - volume_fakes.qos_specs - ) - self.assertIsNone(result) - - -class TestQosShow(TestQos): - - def setUp(self): - super(TestQosShow, self).setUp() - - self.qos_mock.get.return_value = fakes.FakeResource( - None, - copy.deepcopy(volume_fakes.QOS_WITH_ASSOCIATIONS), - loaded=True, - ) - self.qos_mock.get_associations.return_value = [fakes.FakeResource( - None, - copy.deepcopy(volume_fakes.qos_association), - loaded=True, - )] - - # Get the command object to test - self.cmd = qos_specs.ShowQos(self.app, None) - - def test_qos_show(self): - arglist = [ - volume_fakes.qos_id - ] - verifylist = [ - ('qos_spec', volume_fakes.qos_id) - ] - - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - columns, data = self.cmd.take_action(parsed_args) - self.qos_mock.get.assert_called_with( - volume_fakes.qos_id - ) - - collist = ( - 'associations', - 'consumer', - 'id', - 'name', - 'specs' - ) - self.assertEqual(collist, columns) - datalist = ( - volume_fakes.type_name, - volume_fakes.qos_consumer, - volume_fakes.qos_id, - volume_fakes.qos_name, - utils.format_dict(volume_fakes.qos_specs), - ) - self.assertEqual(datalist, tuple(data)) - - -class TestQosUnset(TestQos): - - def setUp(self): - super(TestQosUnset, self).setUp() - - # Get the command object to test - self.cmd = qos_specs.UnsetQos(self.app, None) - - def test_qos_unset_with_properties(self): - self.qos_mock.get.return_value = fakes.FakeResource( - None, - copy.deepcopy(volume_fakes.QOS), - loaded=True - ) - arglist = [ - volume_fakes.qos_id, - '--property', 'iops', - '--property', 'foo' - ] - - verifylist = [ - ('qos_spec', volume_fakes.qos_id), - ('property', ['iops', 'foo']) - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - result = self.cmd.take_action(parsed_args) - - self.qos_mock.unset_keys.assert_called_with( - volume_fakes.qos_id, - ['iops', 'foo'] - ) - self.assertIsNone(result) diff --git a/openstackclient/tests/volume/v1/test_transfer_request.py b/openstackclient/tests/volume/v1/test_transfer_request.py deleted file mode 100644 index 91f5f4b2..00000000 --- a/openstackclient/tests/volume/v1/test_transfer_request.py +++ /dev/null @@ -1,114 +0,0 @@ -# -# Licensed under the Apache License, Version 2.0 (the "License"); you may -# not use this file except in compliance with the License. You may obtain -# a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT -# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the -# License for the specific language governing permissions and limitations -# under the License. -# - - -from openstackclient.tests.volume.v1 import fakes as transfer_fakes -from openstackclient.volume.v1 import volume_transfer_request - - -class TestTransfer(transfer_fakes.TestVolumev1): - - def setUp(self): - super(TestTransfer, self).setUp() - - # Get a shortcut to the TransferManager Mock - self.transfer_mock = self.app.client_manager.volume.transfers - self.transfer_mock.reset_mock() - - -class TestTransferList(TestTransfer): - - # The Transfers to be listed - volume_transfers = transfer_fakes.FakeTransfer.create_one_transfer() - - def setUp(self): - super(TestTransferList, self).setUp() - - self.transfer_mock.list.return_value = [self.volume_transfers] - - # Get the command object to test - self.cmd = volume_transfer_request.ListTransferRequests(self.app, None) - - def test_transfer_list_without_argument(self): - arglist = [] - verifylist = [] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - # In base command class Lister in cliff, abstract method take_action() - # returns a tuple containing the column names and an iterable - # containing the data to be listed. - columns, data = self.cmd.take_action(parsed_args) - - expected_columns = [ - 'ID', - 'Volume', - 'Name', - ] - - # confirming if all expected columns are present in the result. - self.assertEqual(expected_columns, columns) - - datalist = (( - self.volume_transfers.id, - self.volume_transfers.volume_id, - self.volume_transfers.name, - ), ) - - # confirming if all expected values are present in the result. - self.assertEqual(datalist, tuple(data)) - - # checking if proper call was made to list volume_transfers - self.transfer_mock.list.assert_called_with( - detailed=True, - search_opts={'all_tenants': 0} - ) - - def test_transfer_list_with_argument(self): - arglist = [ - "--all-projects" - ] - verifylist = [ - ("all_projects", True) - ] - - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - # In base command class Lister in cliff, abstract method take_action() - # returns a tuple containing the column names and an iterable - # containing the data to be listed. - columns, data = self.cmd.take_action(parsed_args) - - expected_columns = [ - 'ID', - 'Volume', - 'Name', - ] - - # confirming if all expected columns are present in the result. - self.assertEqual(expected_columns, columns) - - datalist = (( - self.volume_transfers.id, - self.volume_transfers.volume_id, - self.volume_transfers.name, - ), ) - - # confirming if all expected values are present in the result. - self.assertEqual(datalist, tuple(data)) - - # checking if proper call was made to list volume_transfers - self.transfer_mock.list.assert_called_with( - detailed=True, - search_opts={'all_tenants': 1} - ) diff --git a/openstackclient/tests/volume/v1/test_volume.py b/openstackclient/tests/volume/v1/test_volume.py deleted file mode 100644 index e41c2a72..00000000 --- a/openstackclient/tests/volume/v1/test_volume.py +++ /dev/null @@ -1,716 +0,0 @@ -# Copyright 2013 Nebula Inc. -# -# Licensed under the Apache License, Version 2.0 (the "License"); you may -# not use this file except in compliance with the License. You may obtain -# a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT -# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the -# License for the specific language governing permissions and limitations -# under the License. -# - -import copy -import mock - -from openstackclient.tests import fakes -from openstackclient.tests.identity.v2_0 import fakes as identity_fakes -from openstackclient.tests.volume.v1 import fakes as volume_fakes -from openstackclient.volume.v1 import volume - - -class TestVolume(volume_fakes.TestVolumev1): - - def setUp(self): - super(TestVolume, self).setUp() - - # Get a shortcut to the VolumeManager Mock - self.volumes_mock = self.app.client_manager.volume.volumes - self.volumes_mock.reset_mock() - - # Get a shortcut to the TenantManager Mock - self.projects_mock = self.app.client_manager.identity.tenants - self.projects_mock.reset_mock() - - # Get a shortcut to the UserManager Mock - self.users_mock = self.app.client_manager.identity.users - self.users_mock.reset_mock() - - # Get a shortcut to the ImageManager Mock - self.images_mock = self.app.client_manager.image.images - self.images_mock.reset_mock() - - -# TODO(dtroyer): The volume create tests are incomplete, only the minimal -# options and the options that require additional processing -# are implemented at this time. - -class TestVolumeCreate(TestVolume): - - project = identity_fakes.FakeProject.create_one_project() - user = identity_fakes.FakeUser.create_one_user() - - columns = ( - 'attach_status', - 'availability_zone', - 'display_description', - 'display_name', - 'id', - 'properties', - 'size', - 'status', - 'type', - ) - datalist = ( - 'detached', - volume_fakes.volume_zone, - volume_fakes.volume_description, - volume_fakes.volume_name, - volume_fakes.volume_id, - volume_fakes.volume_metadata_str, - volume_fakes.volume_size, - volume_fakes.volume_status, - volume_fakes.volume_type, - ) - - def setUp(self): - super(TestVolumeCreate, self).setUp() - - self.volumes_mock.create.return_value = fakes.FakeResource( - None, - copy.deepcopy(volume_fakes.VOLUME), - loaded=True, - ) - - # Get the command object to test - self.cmd = volume.CreateVolume(self.app, None) - - def test_volume_create_min_options(self): - arglist = [ - '--size', str(volume_fakes.volume_size), - volume_fakes.volume_name, - ] - verifylist = [ - ('size', volume_fakes.volume_size), - ('name', volume_fakes.volume_name), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - # In base command class ShowOne in cliff, abstract method take_action() - # returns a two-part tuple with a tuple of column names and a tuple of - # data to be shown. - columns, data = self.cmd.take_action(parsed_args) - - # VolumeManager.create(size, snapshot_id=, source_volid=, - # display_name=, display_description=, - # volume_type=, user_id=, - # project_id=, availability_zone=, - # metadata=, imageRef=) - self.volumes_mock.create.assert_called_with( - volume_fakes.volume_size, - None, - None, - volume_fakes.volume_name, - None, - None, - None, - None, - None, - None, - None, - ) - self.assertEqual(self.columns, columns) - self.assertEqual(self.datalist, data) - - def test_volume_create_options(self): - arglist = [ - '--size', str(volume_fakes.volume_size), - '--description', volume_fakes.volume_description, - '--type', volume_fakes.volume_type, - '--availability-zone', volume_fakes.volume_zone, - volume_fakes.volume_name, - ] - verifylist = [ - ('size', volume_fakes.volume_size), - ('description', volume_fakes.volume_description), - ('type', volume_fakes.volume_type), - ('availability_zone', volume_fakes.volume_zone), - ('name', volume_fakes.volume_name), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - # In base command class ShowOne in cliff, abstract method take_action() - # returns a two-part tuple with a tuple of column names and a tuple of - # data to be shown. - columns, data = self.cmd.take_action(parsed_args) - - # VolumeManager.create(size, snapshot_id=, source_volid=, - # display_name=, display_description=, - # volume_type=, user_id=, - # project_id=, availability_zone=, - # metadata=, imageRef=) - self.volumes_mock.create.assert_called_with( - volume_fakes.volume_size, - None, - None, - volume_fakes.volume_name, - volume_fakes.volume_description, - volume_fakes.volume_type, - None, - None, - volume_fakes.volume_zone, - None, - None, - ) - - self.assertEqual(self.columns, columns) - self.assertEqual(self.datalist, data) - - def test_volume_create_user_project_id(self): - # Return a project - self.projects_mock.get.return_value = self.project - # Return a user - self.users_mock.get.return_value = self.user - - arglist = [ - '--size', str(volume_fakes.volume_size), - '--project', self.project.id, - '--user', self.user.id, - volume_fakes.volume_name, - ] - verifylist = [ - ('size', volume_fakes.volume_size), - ('project', self.project.id), - ('user', self.user.id), - ('name', volume_fakes.volume_name), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - # In base command class ShowOne in cliff, abstract method take_action() - # returns a two-part tuple with a tuple of column names and a tuple of - # data to be shown. - columns, data = self.cmd.take_action(parsed_args) - - # VolumeManager.create(size, snapshot_id=, source_volid=, - # display_name=, display_description=, - # volume_type=, user_id=, - # project_id=, availability_zone=, - # metadata=, imageRef=) - self.volumes_mock.create.assert_called_with( - volume_fakes.volume_size, - None, - None, - volume_fakes.volume_name, - None, - None, - self.user.id, - self.project.id, - None, - None, - None, - ) - - self.assertEqual(self.columns, columns) - self.assertEqual(self.datalist, data) - - def test_volume_create_user_project_name(self): - # Return a project - self.projects_mock.get.return_value = self.project - # Return a user - self.users_mock.get.return_value = self.user - - arglist = [ - '--size', str(volume_fakes.volume_size), - '--project', self.project.name, - '--user', self.user.name, - volume_fakes.volume_name, - ] - verifylist = [ - ('size', volume_fakes.volume_size), - ('project', self.project.name), - ('user', self.user.name), - ('name', volume_fakes.volume_name), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - # In base command class ShowOne in cliff, abstract method take_action() - # returns a two-part tuple with a tuple of column names and a tuple of - # data to be shown. - columns, data = self.cmd.take_action(parsed_args) - - # VolumeManager.create(size, snapshot_id=, source_volid=, - # display_name=, display_description=, - # volume_type=, user_id=, - # project_id=, availability_zone=, - # metadata=, imageRef=) - self.volumes_mock.create.assert_called_with( - volume_fakes.volume_size, - None, - None, - volume_fakes.volume_name, - None, - None, - self.user.id, - self.project.id, - None, - None, - None, - ) - - self.assertEqual(self.columns, columns) - self.assertEqual(self.datalist, data) - - def test_volume_create_properties(self): - arglist = [ - '--property', 'Alpha=a', - '--property', 'Beta=b', - '--size', str(volume_fakes.volume_size), - volume_fakes.volume_name, - ] - verifylist = [ - ('property', {'Alpha': 'a', 'Beta': 'b'}), - ('size', volume_fakes.volume_size), - ('name', volume_fakes.volume_name), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - # In base command class ShowOne in cliff, abstract method take_action() - # returns a two-part tuple with a tuple of column names and a tuple of - # data to be shown. - columns, data = self.cmd.take_action(parsed_args) - - # VolumeManager.create(size, snapshot_id=, source_volid=, - # display_name=, display_description=, - # volume_type=, user_id=, - # project_id=, availability_zone=, - # metadata=, imageRef=) - self.volumes_mock.create.assert_called_with( - volume_fakes.volume_size, - None, - None, - volume_fakes.volume_name, - None, - None, - None, - None, - None, - {'Alpha': 'a', 'Beta': 'b'}, - None, - ) - - self.assertEqual(self.columns, columns) - self.assertEqual(self.datalist, data) - - def test_volume_create_image_id(self): - self.images_mock.get.return_value = fakes.FakeResource( - None, - copy.deepcopy(volume_fakes.IMAGE), - loaded=True, - ) - - arglist = [ - '--image', volume_fakes.image_id, - '--size', str(volume_fakes.volume_size), - volume_fakes.volume_name, - ] - verifylist = [ - ('image', volume_fakes.image_id), - ('size', volume_fakes.volume_size), - ('name', volume_fakes.volume_name), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - # In base command class ShowOne in cliff, abstract method take_action() - # returns a two-part tuple with a tuple of column names and a tuple of - # data to be shown. - columns, data = self.cmd.take_action(parsed_args) - - # VolumeManager.create(size, snapshot_id=, source_volid=, - # display_name=, display_description=, - # volume_type=, user_id=, - # project_id=, availability_zone=, - # metadata=, imageRef=) - self.volumes_mock.create.assert_called_with( - volume_fakes.volume_size, - None, - None, - volume_fakes.volume_name, - None, - None, - None, - None, - None, - None, - volume_fakes.image_id, - ) - - self.assertEqual(self.columns, columns) - self.assertEqual(self.datalist, data) - - def test_volume_create_image_name(self): - self.images_mock.get.return_value = fakes.FakeResource( - None, - copy.deepcopy(volume_fakes.IMAGE), - loaded=True, - ) - - arglist = [ - '--image', volume_fakes.image_name, - '--size', str(volume_fakes.volume_size), - volume_fakes.volume_name, - ] - verifylist = [ - ('image', volume_fakes.image_name), - ('size', volume_fakes.volume_size), - ('name', volume_fakes.volume_name), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - # In base command class ShowOne in cliff, abstract method take_action() - # returns a two-part tuple with a tuple of column names and a tuple of - # data to be shown. - columns, data = self.cmd.take_action(parsed_args) - - # VolumeManager.create(size, snapshot_id=, source_volid=, - # display_name=, display_description=, - # volume_type=, user_id=, - # project_id=, availability_zone=, - # metadata=, imageRef=) - self.volumes_mock.create.assert_called_with( - volume_fakes.volume_size, - None, - None, - volume_fakes.volume_name, - None, - None, - None, - None, - None, - None, - volume_fakes.image_id, - ) - - self.assertEqual(self.columns, columns) - self.assertEqual(self.datalist, data) - - -class TestVolumeList(TestVolume): - - columns = ( - 'ID', - 'Display Name', - 'Status', - 'Size', - 'Attached to', - ) - datalist = ( - ( - volume_fakes.volume_id, - volume_fakes.volume_name, - volume_fakes.volume_status, - volume_fakes.volume_size, - '', - ), - ) - - def setUp(self): - super(TestVolumeList, self).setUp() - - self.volumes_mock.list.return_value = [ - fakes.FakeResource( - None, - copy.deepcopy(volume_fakes.VOLUME), - loaded=True, - ), - ] - - # Get the command object to test - self.cmd = volume.ListVolume(self.app, None) - - def test_volume_list_no_options(self): - arglist = [] - verifylist = [ - ('long', False), - ('all_projects', False), - ('name', None), - ('status', None), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - columns, data = self.cmd.take_action(parsed_args) - - self.assertEqual(self.columns, columns) - self.assertEqual(self.datalist, tuple(data)) - - def test_volume_list_name(self): - arglist = [ - '--name', volume_fakes.volume_name, - ] - verifylist = [ - ('long', False), - ('all_projects', False), - ('name', volume_fakes.volume_name), - ('status', None), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - columns, data = self.cmd.take_action(parsed_args) - self.assertEqual(self.columns, tuple(columns)) - self.assertEqual(self.datalist, tuple(data)) - - def test_volume_list_status(self): - arglist = [ - '--status', volume_fakes.volume_status, - ] - verifylist = [ - ('long', False), - ('all_projects', False), - ('name', None), - ('status', volume_fakes.volume_status), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - columns, data = self.cmd.take_action(parsed_args) - self.assertEqual(self.columns, tuple(columns)) - self.assertEqual(self.datalist, tuple(data)) - - def test_volume_list_all_projects(self): - arglist = [ - '--all-projects', - ] - verifylist = [ - ('long', False), - ('all_projects', True), - ('name', None), - ('status', None), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - columns, data = self.cmd.take_action(parsed_args) - self.assertEqual(self.columns, tuple(columns)) - self.assertEqual(self.datalist, tuple(data)) - - def test_volume_list_long(self): - arglist = [ - '--long', - ] - verifylist = [ - ('long', True), - ('all_projects', False), - ('name', None), - ('status', None), - ] - - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - columns, data = self.cmd.take_action(parsed_args) - - collist = ( - 'ID', - 'Display Name', - 'Status', - 'Size', - 'Type', - 'Bootable', - 'Attached to', - 'Properties', - ) - self.assertEqual(collist, columns) - - datalist = (( - volume_fakes.volume_id, - volume_fakes.volume_name, - volume_fakes.volume_status, - volume_fakes.volume_size, - volume_fakes.volume_type, - '', - '', - "Alpha='a', Beta='b', Gamma='g'", - ), ) - self.assertEqual(datalist, tuple(data)) - - -class TestVolumeSet(TestVolume): - - def setUp(self): - super(TestVolumeSet, self).setUp() - - self.volumes_mock.get.return_value = fakes.FakeResource( - None, - copy.deepcopy(volume_fakes.VOLUME), - loaded=True, - ) - - self.volumes_mock.update.return_value = fakes.FakeResource( - None, - copy.deepcopy(volume_fakes.VOLUME), - loaded=True, - ) - # Get the command object to test - self.cmd = volume.SetVolume(self.app, None) - - def test_volume_set_no_options(self): - arglist = [ - volume_fakes.volume_name, - ] - verifylist = [ - ('name', None), - ('description', None), - ('size', None), - ('property', None), - ('volume', volume_fakes.volume_name), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - result = self.cmd.take_action(parsed_args) - self.assertIsNone(result) - - def test_volume_set_name(self): - arglist = [ - '--name', 'qwerty', - volume_fakes.volume_name, - ] - verifylist = [ - ('name', 'qwerty'), - ('description', None), - ('size', None), - ('property', None), - ('volume', volume_fakes.volume_name), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - result = self.cmd.take_action(parsed_args) - - # Set expected values - kwargs = { - 'display_name': 'qwerty', - } - self.volumes_mock.update.assert_called_with( - volume_fakes.volume_id, - **kwargs - ) - self.assertIsNone(result) - - def test_volume_set_description(self): - arglist = [ - '--description', 'new desc', - volume_fakes.volume_name, - ] - verifylist = [ - ('name', None), - ('description', 'new desc'), - ('size', None), - ('property', None), - ('volume', volume_fakes.volume_name), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - result = self.cmd.take_action(parsed_args) - - # Set expected values - kwargs = { - 'display_description': 'new desc', - } - self.volumes_mock.update.assert_called_with( - volume_fakes.volume_id, - **kwargs - ) - self.assertIsNone(result) - - def test_volume_set_size(self): - arglist = [ - '--size', '130', - volume_fakes.volume_name, - ] - verifylist = [ - ('name', None), - ('description', None), - ('size', 130), - ('property', None), - ('volume', volume_fakes.volume_name), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - result = self.cmd.take_action(parsed_args) - - # Set expected values - size = 130 - self.volumes_mock.extend.assert_called_with( - volume_fakes.volume_id, - size - ) - self.assertIsNone(result) - - @mock.patch.object(volume.LOG, 'error') - def test_volume_set_size_smaller(self, mock_log_error): - arglist = [ - '--size', '100', - volume_fakes.volume_name, - ] - verifylist = [ - ('name', None), - ('description', None), - ('size', 100), - ('property', None), - ('volume', volume_fakes.volume_name), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - result = self.cmd.take_action(parsed_args) - - mock_log_error.assert_called_with("New size must be greater " - "than %s GB", - volume_fakes.volume_size) - self.assertIsNone(result) - - @mock.patch.object(volume.LOG, 'error') - def test_volume_set_size_not_available(self, mock_log_error): - self.volumes_mock.get.return_value.status = 'error' - arglist = [ - '--size', '130', - volume_fakes.volume_name, - ] - verifylist = [ - ('name', None), - ('description', None), - ('size', 130), - ('property', None), - ('volume', volume_fakes.volume_name), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - result = self.cmd.take_action(parsed_args) - - mock_log_error.assert_called_with("Volume is in %s state, it must be " - "available before size can be " - "extended", 'error') - self.assertIsNone(result) - - def test_volume_set_property(self): - arglist = [ - '--property', 'myprop=myvalue', - volume_fakes.volume_name, - ] - verifylist = [ - ('name', None), - ('description', None), - ('size', None), - ('property', {'myprop': 'myvalue'}), - ('volume', volume_fakes.volume_name), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - result = self.cmd.take_action(parsed_args) - - # Set expected values - metadata = { - 'myprop': 'myvalue' - } - self.volumes_mock.set_metadata.assert_called_with( - volume_fakes.volume_id, - metadata - ) - self.assertIsNone(result) diff --git a/openstackclient/tests/volume/v2/__init__.py b/openstackclient/tests/volume/v2/__init__.py deleted file mode 100644 index e69de29b..00000000 --- a/openstackclient/tests/volume/v2/__init__.py +++ /dev/null diff --git a/openstackclient/tests/volume/v2/fakes.py b/openstackclient/tests/volume/v2/fakes.py deleted file mode 100644 index 49384bd8..00000000 --- a/openstackclient/tests/volume/v2/fakes.py +++ /dev/null @@ -1,696 +0,0 @@ -# -# Licensed under the Apache License, Version 2.0 (the "License"); you may -# not use this file except in compliance with the License. You may obtain -# a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT -# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the -# License for the specific language governing permissions and limitations -# under the License. -# - -import copy -import mock -import random -import uuid - -from osc_lib import utils as common_utils - -from openstackclient.tests import fakes -from openstackclient.tests.identity.v3 import fakes as identity_fakes -from openstackclient.tests.image.v2 import fakes as image_fakes -from openstackclient.tests import utils - - -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': 'ce26708d-a7f8-4b4b-9861-4a80256615a7', - 'name': 'fake_transfer_name', - 'id': '731a7f53-aa92-4fbd-9de3-6f7d729c926b' - } - - # 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 - - -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 Dictionary 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 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 - - -class FakeVolumeClient(object): - - def __init__(self, **kwargs): - self.volumes = mock.Mock() - self.volumes.resource_class = fakes.FakeResource(None, {}) - self.extensions = mock.Mock() - self.extensions.resource_class = fakes.FakeResource(None, {}) - self.volume_snapshots = mock.Mock() - self.volume_snapshots.resource_class = fakes.FakeResource(None, {}) - self.backups = mock.Mock() - self.backups.resource_class = fakes.FakeResource(None, {}) - self.volume_types = mock.Mock() - self.volume_types.resource_class = fakes.FakeResource(None, {}) - self.volume_type_access = mock.Mock() - self.volume_type_access.resource_class = fakes.FakeResource(None, {}) - self.restores = mock.Mock() - self.restores.resource_class = fakes.FakeResource(None, {}) - self.qos_specs = mock.Mock() - self.qos_specs.resource_class = fakes.FakeResource(None, {}) - self.availability_zones = mock.Mock() - self.availability_zones.resource_class = fakes.FakeResource(None, {}) - self.transfers = mock.Mock() - self.transfers.resource_class = fakes.FakeResource(None, {}) - self.services = mock.Mock() - self.services.resource_class = fakes.FakeResource(None, {}) - self.auth_token = kwargs['token'] - self.management_url = kwargs['endpoint'] - - -class TestVolume(utils.TestCommand): - - def setUp(self): - super(TestVolume, self).setUp() - - self.app.client_manager.volume = FakeVolumeClient( - endpoint=fakes.AUTH_URL, - token=fakes.AUTH_TOKEN - ) - self.app.client_manager.identity = identity_fakes.FakeIdentityv3Client( - endpoint=fakes.AUTH_URL, - token=fakes.AUTH_TOKEN - ) - self.app.client_manager.image = image_fakes.FakeImagev2Client( - endpoint=fakes.AUTH_URL, - token=fakes.AUTH_TOKEN - ) - - -class FakeVolume(object): - """Fake one or more volumes. - - TODO(xiexs): Currently, only volume API v2 is supported by this class. - """ - - @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, - '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 - - @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.MagicMock(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( - common_utils.format_list(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 Dictionary 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 - - @staticmethod - def create_availability_zones(attrs=None, count=2): - """Create multiple fake AZs. - - :param Dictionary 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 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, - } - - # 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.MagicMock(side_effect=backups) - - -class FakeExtension(object): - """Fake one or more extension.""" - - @staticmethod - def create_one_extension(attrs=None): - """Create a fake extension. - - :param Dictionary 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 - - -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.MagicMock(side_effect=qoses) - - -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, - "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 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.MagicMock(side_effect=snapshots) - - -class FakeType(object): - """Fake one or more type.""" - - @staticmethod - def create_one_type(attrs=None, methods=None): - """Create a fake 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. - 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. - type_info.update(attrs) - - volume_type = fakes.FakeResource( - info=copy.deepcopy(type_info), - methods=methods, - loaded=True) - return volume_type - - @staticmethod - def create_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 = FakeType.create_one_type(attrs) - volume_types.append(volume_type) - - return volume_types diff --git a/openstackclient/tests/volume/v2/test_backup.py b/openstackclient/tests/volume/v2/test_backup.py deleted file mode 100644 index 67064352..00000000 --- a/openstackclient/tests/volume/v2/test_backup.py +++ /dev/null @@ -1,388 +0,0 @@ -# -# Licensed under the Apache License, Version 2.0 (the "License"); you may -# not use this file except in compliance with the License. You may obtain -# a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT -# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the -# License for the specific language governing permissions and limitations -# under the License. -# - -import mock -from mock import call - -from osc_lib import exceptions -from osc_lib import utils - -from openstackclient.tests.volume.v2 import fakes as volume_fakes -from openstackclient.volume.v2 import backup - - -class TestBackup(volume_fakes.TestVolume): - - def setUp(self): - super(TestBackup, self).setUp() - - self.backups_mock = self.app.client_manager.volume.backups - self.backups_mock.reset_mock() - self.volumes_mock = self.app.client_manager.volume.volumes - self.volumes_mock.reset_mock() - self.snapshots_mock = self.app.client_manager.volume.volume_snapshots - self.snapshots_mock.reset_mock() - self.restores_mock = self.app.client_manager.volume.restores - self.restores_mock.reset_mock() - - -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( - attrs={'volume_id': volume.id, 'snapshot_id': snapshot.id}) - - columns = ( - 'availability_zone', - 'container', - 'description', - 'id', - 'name', - 'object_count', - 'size', - 'snapshot_id', - 'status', - 'volume_id', - ) - data = ( - new_backup.availability_zone, - new_backup.container, - new_backup.description, - new_backup.id, - new_backup.name, - new_backup.object_count, - new_backup.size, - new_backup.snapshot_id, - new_backup.status, - new_backup.volume_id, - ) - - def setUp(self): - super(TestBackupCreate, self).setUp() - - self.volumes_mock.get.return_value = self.volume - self.snapshots_mock.get.return_value = self.snapshot - self.backups_mock.create.return_value = self.new_backup - - # Get the command object to test - self.cmd = backup.CreateVolumeBackup(self.app, None) - - def test_backup_create(self): - arglist = [ - "--name", self.new_backup.name, - "--description", self.new_backup.description, - "--container", self.new_backup.container, - "--force", - "--incremental", - "--snapshot", self.new_backup.snapshot_id, - self.new_backup.volume_id, - ] - verifylist = [ - ("name", self.new_backup.name), - ("description", self.new_backup.description), - ("container", self.new_backup.container), - ("force", True), - ("incremental", True), - ("snapshot", self.new_backup.snapshot_id), - ("volume", self.new_backup.volume_id), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - columns, data = self.cmd.take_action(parsed_args) - - self.backups_mock.create.assert_called_with( - self.new_backup.volume_id, - container=self.new_backup.container, - name=self.new_backup.name, - description=self.new_backup.description, - force=True, - incremental=True, - snapshot_id=self.new_backup.snapshot_id, - ) - self.assertEqual(self.columns, columns) - self.assertEqual(self.data, data) - - def test_backup_create_without_name(self): - arglist = [ - "--description", self.new_backup.description, - "--container", self.new_backup.container, - self.new_backup.volume_id, - ] - verifylist = [ - ("description", self.new_backup.description), - ("container", self.new_backup.container), - ("volume", self.new_backup.volume_id), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - columns, data = self.cmd.take_action(parsed_args) - - self.backups_mock.create.assert_called_with( - self.new_backup.volume_id, - container=self.new_backup.container, - name=None, - description=self.new_backup.description, - force=False, - incremental=False, - snapshot_id=None, - ) - self.assertEqual(self.columns, columns) - self.assertEqual(self.data, data) - - -class TestBackupDelete(TestBackup): - - backups = volume_fakes.FakeBackup.create_backups(count=2) - - def setUp(self): - super(TestBackupDelete, self).setUp() - - self.backups_mock.get = ( - volume_fakes.FakeBackup.get_backups(self.backups)) - self.backups_mock.delete.return_value = None - - # Get the command object to mock - self.cmd = backup.DeleteVolumeBackup(self.app, None) - - def test_backup_delete(self): - arglist = [ - self.backups[0].id - ] - verifylist = [ - ("backups", [self.backups[0].id]) - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - result = self.cmd.take_action(parsed_args) - - self.backups_mock.delete.assert_called_with( - self.backups[0].id, False) - self.assertIsNone(result) - - def test_backup_delete_with_force(self): - arglist = [ - '--force', - self.backups[0].id, - ] - verifylist = [ - ('force', True), - ("backups", [self.backups[0].id]) - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - result = self.cmd.take_action(parsed_args) - - self.backups_mock.delete.assert_called_with(self.backups[0].id, True) - self.assertIsNone(result) - - def test_delete_multiple_backups(self): - arglist = [] - for b in self.backups: - arglist.append(b.id) - verifylist = [ - ('backups', arglist), - ] - - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - result = self.cmd.take_action(parsed_args) - - calls = [] - for b in self.backups: - calls.append(call(b.id, False)) - self.backups_mock.delete.assert_has_calls(calls) - self.assertIsNone(result) - - def test_delete_multiple_backups_with_exception(self): - arglist = [ - self.backups[0].id, - 'unexist_backup', - ] - verifylist = [ - ('backups', arglist), - ] - - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - find_mock_result = [self.backups[0], exceptions.CommandError] - with mock.patch.object(utils, 'find_resource', - side_effect=find_mock_result) as find_mock: - try: - self.cmd.take_action(parsed_args) - self.fail('CommandError should be raised.') - except exceptions.CommandError as e: - self.assertEqual('1 of 2 backups failed to delete.', - str(e)) - - find_mock.assert_any_call(self.backups_mock, self.backups[0].id) - find_mock.assert_any_call(self.backups_mock, 'unexist_backup') - - self.assertEqual(2, find_mock.call_count) - self.backups_mock.delete.assert_called_once_with( - self.backups[0].id, False - ) - - -class TestBackupList(TestBackup): - - volume = volume_fakes.FakeVolume.create_one_volume() - backups = volume_fakes.FakeBackup.create_backups( - attrs={'volume_id': volume.name}, count=3) - - columns = [ - 'ID', - 'Name', - 'Description', - 'Status', - 'Size', - ] - columns_long = columns + [ - 'Availability Zone', - 'Volume', - 'Container', - ] - - data = [] - for b in backups: - data.append(( - b.id, - b.name, - b.description, - b.status, - b.size, - )) - data_long = [] - for b in backups: - data_long.append(( - b.id, - b.name, - b.description, - b.status, - b.size, - b.availability_zone, - b.volume_id, - b.container, - )) - - def setUp(self): - super(TestBackupList, self).setUp() - - self.volumes_mock.list.return_value = [self.volume] - self.backups_mock.list.return_value = self.backups - # Get the command to test - self.cmd = backup.ListVolumeBackup(self.app, None) - - def test_backup_list_without_options(self): - arglist = [] - verifylist = [("long", False)] - - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - columns, data = self.cmd.take_action(parsed_args) - - self.assertEqual(self.columns, columns) - self.assertEqual(self.data, list(data)) - - def test_backup_list_with_options(self): - arglist = ["--long"] - verifylist = [("long", True)] - - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - columns, data = self.cmd.take_action(parsed_args) - - self.assertEqual(self.columns_long, columns) - self.assertEqual(self.data_long, list(data)) - - -class TestBackupRestore(TestBackup): - - volume = volume_fakes.FakeVolume.create_one_volume() - backup = volume_fakes.FakeBackup.create_one_backup( - attrs={'volume_id': volume.id}) - - def setUp(self): - super(TestBackupRestore, self).setUp() - - self.backups_mock.get.return_value = self.backup - self.volumes_mock.get.return_value = self.volume - self.restores_mock.restore.return_value = None - # Get the command object to mock - self.cmd = backup.RestoreVolumeBackup(self.app, None) - - def test_backup_restore(self): - arglist = [ - self.backup.id, - self.backup.volume_id - ] - verifylist = [ - ("backup", self.backup.id), - ("volume", self.backup.volume_id) - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - result = self.cmd.take_action(parsed_args) - self.restores_mock.restore.assert_called_with(self.backup.id, - self.backup.volume_id) - self.assertIsNone(result) - - -class TestBackupShow(TestBackup): - - backup = volume_fakes.FakeBackup.create_one_backup() - - columns = ( - 'availability_zone', - 'container', - 'description', - 'id', - 'name', - 'object_count', - 'size', - 'snapshot_id', - 'status', - 'volume_id', - ) - data = ( - backup.availability_zone, - backup.container, - backup.description, - backup.id, - backup.name, - backup.object_count, - backup.size, - backup.snapshot_id, - backup.status, - backup.volume_id, - ) - - def setUp(self): - super(TestBackupShow, self).setUp() - - self.backups_mock.get.return_value = self.backup - # Get the command object to test - self.cmd = backup.ShowVolumeBackup(self.app, None) - - def test_backup_show(self): - arglist = [ - self.backup.id - ] - verifylist = [ - ("backup", self.backup.id) - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - columns, data = self.cmd.take_action(parsed_args) - self.backups_mock.get.assert_called_with(self.backup.id) - - self.assertEqual(self.columns, columns) - self.assertEqual(self.data, data) diff --git a/openstackclient/tests/volume/v2/test_qos_specs.py b/openstackclient/tests/volume/v2/test_qos_specs.py deleted file mode 100644 index 56b8ae03..00000000 --- a/openstackclient/tests/volume/v2/test_qos_specs.py +++ /dev/null @@ -1,453 +0,0 @@ -# Copyright 2015 iWeb Technologies Inc. -# -# Licensed under the Apache License, Version 2.0 (the "License"); you may -# not use this file except in compliance with the License. You may obtain -# a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT -# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the -# License for the specific language governing permissions and limitations -# under the License. -# - -import mock -from mock import call - -from osc_lib import exceptions -from osc_lib import utils - -from openstackclient.tests.volume.v2 import fakes as volume_fakes -from openstackclient.volume.v2 import qos_specs - - -class TestQos(volume_fakes.TestVolume): - - def setUp(self): - super(TestQos, self).setUp() - - self.qos_mock = self.app.client_manager.volume.qos_specs - self.qos_mock.reset_mock() - - self.types_mock = self.app.client_manager.volume.volume_types - self.types_mock.reset_mock() - - -class TestQosAssociate(TestQos): - - volume_type = volume_fakes.FakeType.create_one_type() - qos_spec = volume_fakes.FakeQos.create_one_qos() - - def setUp(self): - super(TestQosAssociate, self).setUp() - - self.qos_mock.get.return_value = self.qos_spec - self.types_mock.get.return_value = self.volume_type - # Get the command object to test - self.cmd = qos_specs.AssociateQos(self.app, None) - - def test_qos_associate(self): - arglist = [ - self.qos_spec.id, - self.volume_type.id - ] - verifylist = [ - ('qos_spec', self.qos_spec.id), - ('volume_type', self.volume_type.id) - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - result = self.cmd.take_action(parsed_args) - - self.qos_mock.associate.assert_called_with( - self.qos_spec.id, - self.volume_type.id - ) - self.assertIsNone(result) - - -class TestQosCreate(TestQos): - - new_qos_spec = volume_fakes.FakeQos.create_one_qos() - columns = ( - 'consumer', - 'id', - 'name', - 'specs' - ) - data = ( - new_qos_spec.consumer, - new_qos_spec.id, - new_qos_spec.name, - new_qos_spec.specs - ) - - def setUp(self): - super(TestQosCreate, self).setUp() - - self.qos_mock.create.return_value = self.new_qos_spec - # Get the command object to test - self.cmd = qos_specs.CreateQos(self.app, None) - - def test_qos_create_without_properties(self): - arglist = [ - self.new_qos_spec.name, - ] - verifylist = [ - ('name', self.new_qos_spec.name), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - columns, data = self.cmd.take_action(parsed_args) - - self.qos_mock.create.assert_called_with( - self.new_qos_spec.name, - {'consumer': 'both'} - ) - - self.assertEqual(self.columns, columns) - self.assertEqual(self.data, data) - - def test_qos_create_with_consumer(self): - arglist = [ - '--consumer', self.new_qos_spec.consumer, - self.new_qos_spec.name, - ] - verifylist = [ - ('consumer', self.new_qos_spec.consumer), - ('name', self.new_qos_spec.name), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - columns, data = self.cmd.take_action(parsed_args) - - self.qos_mock.create.assert_called_with( - self.new_qos_spec.name, - {'consumer': self.new_qos_spec.consumer} - ) - - self.assertEqual(self.columns, columns) - self.assertEqual(self.data, data) - - def test_qos_create_with_properties(self): - arglist = [ - '--consumer', self.new_qos_spec.consumer, - '--property', 'foo=bar', - '--property', 'iops=9001', - self.new_qos_spec.name, - ] - verifylist = [ - ('consumer', self.new_qos_spec.consumer), - ('property', self.new_qos_spec.specs), - ('name', self.new_qos_spec.name), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - columns, data = self.cmd.take_action(parsed_args) - - self.new_qos_spec.specs.update( - {'consumer': self.new_qos_spec.consumer}) - self.qos_mock.create.assert_called_with( - self.new_qos_spec.name, - self.new_qos_spec.specs - ) - - self.assertEqual(self.columns, columns) - self.assertEqual(self.data, data) - - -class TestQosDelete(TestQos): - - qos_specs = volume_fakes.FakeQos.create_qoses(count=2) - - def setUp(self): - super(TestQosDelete, self).setUp() - - self.qos_mock.get = ( - volume_fakes.FakeQos.get_qoses(self.qos_specs)) - # Get the command object to test - self.cmd = qos_specs.DeleteQos(self.app, None) - - def test_qos_delete(self): - arglist = [ - self.qos_specs[0].id - ] - verifylist = [ - ('qos_specs', [self.qos_specs[0].id]) - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - result = self.cmd.take_action(parsed_args) - - self.qos_mock.delete.assert_called_with( - self.qos_specs[0].id, False) - self.assertIsNone(result) - - def test_qos_delete_with_force(self): - arglist = [ - '--force', - self.qos_specs[0].id - ] - verifylist = [ - ('force', True), - ('qos_specs', [self.qos_specs[0].id]) - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - result = self.cmd.take_action(parsed_args) - - self.qos_mock.delete.assert_called_with( - self.qos_specs[0].id, True) - self.assertIsNone(result) - - def test_delete_multiple_qoses(self): - arglist = [] - for q in self.qos_specs: - arglist.append(q.id) - verifylist = [ - ('qos_specs', arglist), - ] - - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - result = self.cmd.take_action(parsed_args) - - calls = [] - for q in self.qos_specs: - calls.append(call(q.id, False)) - self.qos_mock.delete.assert_has_calls(calls) - self.assertIsNone(result) - - def test_delete_multiple_qoses_with_exception(self): - arglist = [ - self.qos_specs[0].id, - 'unexist_qos', - ] - verifylist = [ - ('qos_specs', arglist), - ] - - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - find_mock_result = [self.qos_specs[0], exceptions.CommandError] - with mock.patch.object(utils, 'find_resource', - side_effect=find_mock_result) as find_mock: - try: - self.cmd.take_action(parsed_args) - self.fail('CommandError should be raised.') - except exceptions.CommandError as e: - self.assertEqual( - '1 of 2 QoS specifications failed to delete.', str(e)) - - find_mock.assert_any_call(self.qos_mock, self.qos_specs[0].id) - find_mock.assert_any_call(self.qos_mock, 'unexist_qos') - - self.assertEqual(2, find_mock.call_count) - self.qos_mock.delete.assert_called_once_with( - self.qos_specs[0].id, False - ) - - -class TestQosDisassociate(TestQos): - - volume_type = volume_fakes.FakeType.create_one_type() - qos_spec = volume_fakes.FakeQos.create_one_qos() - - def setUp(self): - super(TestQosDisassociate, self).setUp() - - self.qos_mock.get.return_value = self.qos_spec - self.types_mock.get.return_value = self.volume_type - # Get the command object to test - self.cmd = qos_specs.DisassociateQos(self.app, None) - - def test_qos_disassociate_with_volume_type(self): - arglist = [ - '--volume-type', self.volume_type.id, - self.qos_spec.id, - ] - verifylist = [ - ('volume_type', self.volume_type.id), - ('qos_spec', self.qos_spec.id), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - result = self.cmd.take_action(parsed_args) - - self.qos_mock.disassociate.assert_called_with( - self.qos_spec.id, - self.volume_type.id - ) - self.assertIsNone(result) - - def test_qos_disassociate_with_all_volume_types(self): - arglist = [ - '--all', - self.qos_spec.id, - ] - verifylist = [ - ('qos_spec', self.qos_spec.id) - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - result = self.cmd.take_action(parsed_args) - - self.qos_mock.disassociate_all.assert_called_with(self.qos_spec.id) - self.assertIsNone(result) - - -class TestQosList(TestQos): - - qos_specs = volume_fakes.FakeQos.create_qoses(count=2) - qos_association = volume_fakes.FakeQos.create_one_qos_association() - - columns = ( - 'ID', - 'Name', - 'Consumer', - 'Associations', - 'Specs', - ) - data = [] - for q in qos_specs: - data.append(( - q.id, - q.name, - q.consumer, - qos_association.name, - utils.format_dict(q.specs), - )) - - def setUp(self): - super(TestQosList, self).setUp() - - self.qos_mock.list.return_value = self.qos_specs - self.qos_mock.get_associations.return_value = [self.qos_association] - - # Get the command object to test - self.cmd = qos_specs.ListQos(self.app, None) - - def test_qos_list(self): - arglist = [] - verifylist = [] - - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - columns, data = self.cmd.take_action(parsed_args) - self.qos_mock.list.assert_called_with() - - self.assertEqual(self.columns, columns) - self.assertEqual(self.data, list(data)) - - -class TestQosSet(TestQos): - - qos_spec = volume_fakes.FakeQos.create_one_qos() - - def setUp(self): - super(TestQosSet, self).setUp() - - self.qos_mock.get.return_value = self.qos_spec - # Get the command object to test - self.cmd = qos_specs.SetQos(self.app, None) - - def test_qos_set_with_properties_with_id(self): - arglist = [ - '--property', 'foo=bar', - '--property', 'iops=9001', - self.qos_spec.id, - ] - verifylist = [ - ('property', self.qos_spec.specs), - ('qos_spec', self.qos_spec.id), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - result = self.cmd.take_action(parsed_args) - - self.qos_mock.set_keys.assert_called_with( - self.qos_spec.id, - self.qos_spec.specs - ) - self.assertIsNone(result) - - -class TestQosShow(TestQos): - - qos_spec = volume_fakes.FakeQos.create_one_qos() - qos_association = volume_fakes.FakeQos.create_one_qos_association() - - columns = ( - 'associations', - 'consumer', - 'id', - 'name', - 'specs' - ) - data = ( - qos_association.name, - qos_spec.consumer, - qos_spec.id, - qos_spec.name, - utils.format_dict(qos_spec.specs), - ) - - def setUp(self): - super(TestQosShow, self).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 - self.cmd = qos_specs.ShowQos(self.app, None) - - def test_qos_show(self): - arglist = [ - self.qos_spec.id - ] - verifylist = [ - ('qos_spec', self.qos_spec.id) - ] - - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - columns, data = self.cmd.take_action(parsed_args) - self.qos_mock.get.assert_called_with( - self.qos_spec.id - ) - - self.assertEqual(self.columns, columns) - self.assertEqual(self.data, tuple(data)) - - -class TestQosUnset(TestQos): - - qos_spec = volume_fakes.FakeQos.create_one_qos() - - def setUp(self): - super(TestQosUnset, self).setUp() - - self.qos_mock.get.return_value = self.qos_spec - # Get the command object to test - self.cmd = qos_specs.UnsetQos(self.app, None) - - def test_qos_unset_with_properties(self): - arglist = [ - '--property', 'iops', - '--property', 'foo', - self.qos_spec.id, - ] - verifylist = [ - ('property', ['iops', 'foo']), - ('qos_spec', self.qos_spec.id), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - result = self.cmd.take_action(parsed_args) - - self.qos_mock.unset_keys.assert_called_with( - self.qos_spec.id, - ['iops', 'foo'] - ) - self.assertIsNone(result) diff --git a/openstackclient/tests/volume/v2/test_service.py b/openstackclient/tests/volume/v2/test_service.py deleted file mode 100644 index 5959a5e3..00000000 --- a/openstackclient/tests/volume/v2/test_service.py +++ /dev/null @@ -1,286 +0,0 @@ -# -# Licensed under the Apache License, Version 2.0 (the "License"); you may -# not use this file except in compliance with the License. You may obtain -# a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT -# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the -# License for the specific language governing permissions and limitations -# under the License. -# - -from osc_lib import exceptions - -from openstackclient.tests.volume.v2 import fakes as service_fakes -from openstackclient.volume.v2 import service - - -class TestService(service_fakes.TestVolume): - - def setUp(self): - super(TestService, self).setUp() - - # Get a shortcut to the ServiceManager Mock - self.service_mock = self.app.client_manager.volume.services - self.service_mock.reset_mock() - - -class TestServiceList(TestService): - - # The service to be listed - services = service_fakes.FakeService.create_one_service() - - def setUp(self): - super(TestServiceList, self).setUp() - - self.service_mock.list.return_value = [self.services] - - # Get the command object to test - self.cmd = service.ListService(self.app, None) - - def test_service_list(self): - arglist = [ - '--host', self.services.host, - '--service', self.services.binary, - ] - verifylist = [ - ('host', self.services.host), - ('service', self.services.binary), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - # In base command class Lister in cliff, abstract method take_action() - # returns a tuple containing the column names and an iterable - # containing the data to be listed. - columns, data = self.cmd.take_action(parsed_args) - - expected_columns = [ - 'Binary', - 'Host', - 'Zone', - 'Status', - 'State', - 'Updated At', - ] - - # confirming if all expected columns are present in the result. - self.assertEqual(expected_columns, columns) - - datalist = (( - self.services.binary, - self.services.host, - self.services.zone, - self.services.status, - self.services.state, - self.services.updated_at, - ), ) - - # confirming if all expected values are present in the result. - self.assertEqual(datalist, tuple(data)) - - # checking if proper call was made to list services - self.service_mock.list.assert_called_with( - self.services.host, - self.services.binary, - ) - - # checking if prohibited columns are present in output - self.assertNotIn("Disabled Reason", columns) - self.assertNotIn(self.services.disabled_reason, - tuple(data)) - - def test_service_list_with_long_option(self): - arglist = [ - '--host', self.services.host, - '--service', self.services.binary, - '--long' - ] - verifylist = [ - ('host', self.services.host), - ('service', self.services.binary), - ('long', True) - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - # In base command class Lister in cliff, abstract method take_action() - # returns a tuple containing the column names and an iterable - # containing the data to be listed. - columns, data = self.cmd.take_action(parsed_args) - - expected_columns = [ - 'Binary', - 'Host', - 'Zone', - 'Status', - 'State', - 'Updated At', - 'Disabled Reason' - ] - - # confirming if all expected columns are present in the result. - self.assertEqual(expected_columns, columns) - - datalist = (( - self.services.binary, - self.services.host, - self.services.zone, - self.services.status, - self.services.state, - self.services.updated_at, - self.services.disabled_reason, - ), ) - - # confirming if all expected values are present in the result. - self.assertEqual(datalist, tuple(data)) - - self.service_mock.list.assert_called_with( - self.services.host, - self.services.binary, - ) - - -class TestServiceSet(TestService): - - service = service_fakes.FakeService.create_one_service() - - def setUp(self): - super(TestServiceSet, self).setUp() - - self.service_mock.enable.return_value = self.service - self.service_mock.disable.return_value = self.service - self.service_mock.disable_log_reason.return_value = self.service - - self.cmd = service.SetService(self.app, None) - - def test_service_set_nothing(self): - arglist = [ - self.service.host, - self.service.binary, - ] - verifylist = [ - ('host', self.service.host), - ('service', self.service.binary), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - result = self.cmd.take_action(parsed_args) - - self.service_mock.enable.assert_not_called() - self.service_mock.disable.assert_not_called() - self.service_mock.disable_log_reason.assert_not_called() - self.assertIsNone(result) - - def test_service_set_enable(self): - arglist = [ - '--enable', - self.service.host, - self.service.binary, - ] - verifylist = [ - ('enable', True), - ('host', self.service.host), - ('service', self.service.binary), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - result = self.cmd.take_action(parsed_args) - - self.service_mock.enable.assert_called_with( - self.service.host, - self.service.binary - ) - self.service_mock.disable.assert_not_called() - self.service_mock.disable_log_reason.assert_not_called() - self.assertIsNone(result) - - def test_service_set_disable(self): - arglist = [ - '--disable', - self.service.host, - self.service.binary, - ] - verifylist = [ - ('disable', True), - ('host', self.service.host), - ('service', self.service.binary), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - result = self.cmd.take_action(parsed_args) - - self.service_mock.disable.assert_called_with( - self.service.host, - self.service.binary - ) - self.service_mock.enable.assert_not_called() - self.service_mock.disable_log_reason.assert_not_called() - self.assertIsNone(result) - - def test_service_set_disable_with_reason(self): - reason = 'earthquake' - arglist = [ - '--disable', - '--disable-reason', reason, - self.service.host, - self.service.binary, - ] - verifylist = [ - ('disable', True), - ('disable_reason', reason), - ('host', self.service.host), - ('service', self.service.binary), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - result = self.cmd.take_action(parsed_args) - - self.service_mock.disable_log_reason.assert_called_with( - self.service.host, - self.service.binary, - reason - ) - self.assertIsNone(result) - - def test_service_set_only_with_disable_reason(self): - reason = 'earthquake' - arglist = [ - '--disable-reason', reason, - self.service.host, - self.service.binary, - ] - verifylist = [ - ('disable_reason', reason), - ('host', self.service.host), - ('service', self.service.binary), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - try: - self.cmd.take_action(parsed_args) - self.fail("CommandError should be raised.") - except exceptions.CommandError as e: - self.assertEqual("Cannot specify option --disable-reason without " - "--disable specified.", str(e)) - - def test_service_set_enable_with_disable_reason(self): - reason = 'earthquake' - arglist = [ - '--enable', - '--disable-reason', reason, - self.service.host, - self.service.binary, - ] - verifylist = [ - ('enable', True), - ('disable_reason', reason), - ('host', self.service.host), - ('service', self.service.binary), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - try: - self.cmd.take_action(parsed_args) - self.fail("CommandError should be raised.") - except exceptions.CommandError as e: - self.assertEqual("Cannot specify option --disable-reason without " - "--disable specified.", str(e)) diff --git a/openstackclient/tests/volume/v2/test_snapshot.py b/openstackclient/tests/volume/v2/test_snapshot.py deleted file mode 100644 index 3eb740ba..00000000 --- a/openstackclient/tests/volume/v2/test_snapshot.py +++ /dev/null @@ -1,458 +0,0 @@ -# -# Licensed under the Apache License, Version 2.0 (the "License"); you may -# not use this file except in compliance with the License. You may obtain -# a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT -# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the -# License for the specific language governing permissions and limitations -# under the License. -# - -import argparse -import mock -from mock import call - -from osc_lib import exceptions -from osc_lib import utils - -from openstackclient.tests.volume.v2 import fakes as volume_fakes -from openstackclient.volume.v2 import snapshot - - -class TestSnapshot(volume_fakes.TestVolume): - - def setUp(self): - super(TestSnapshot, self).setUp() - - self.snapshots_mock = self.app.client_manager.volume.volume_snapshots - self.snapshots_mock.reset_mock() - self.volumes_mock = self.app.client_manager.volume.volumes - self.volumes_mock.reset_mock() - - -class TestSnapshotCreate(TestSnapshot): - - columns = ( - 'created_at', - 'description', - 'id', - 'name', - 'properties', - 'size', - 'status', - 'volume_id', - ) - - def setUp(self): - super(TestSnapshotCreate, self).setUp() - - self.volume = volume_fakes.FakeVolume.create_one_volume() - self.new_snapshot = volume_fakes.FakeSnapshot.create_one_snapshot( - attrs={'volume_id': self.volume.id}) - - self.data = ( - self.new_snapshot.created_at, - self.new_snapshot.description, - self.new_snapshot.id, - self.new_snapshot.name, - utils.format_dict(self.new_snapshot.metadata), - self.new_snapshot.size, - self.new_snapshot.status, - self.new_snapshot.volume_id, - ) - - self.volumes_mock.get.return_value = self.volume - self.snapshots_mock.create.return_value = self.new_snapshot - # Get the command object to test - self.cmd = snapshot.CreateSnapshot(self.app, None) - - def test_snapshot_create(self): - arglist = [ - "--name", self.new_snapshot.name, - "--description", self.new_snapshot.description, - "--force", - '--property', 'Alpha=a', - '--property', 'Beta=b', - self.new_snapshot.volume_id, - ] - verifylist = [ - ("name", self.new_snapshot.name), - ("description", self.new_snapshot.description), - ("force", True), - ('property', {'Alpha': 'a', 'Beta': 'b'}), - ("volume", self.new_snapshot.volume_id), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - columns, data = self.cmd.take_action(parsed_args) - - self.snapshots_mock.create.assert_called_with( - self.new_snapshot.volume_id, - force=True, - name=self.new_snapshot.name, - description=self.new_snapshot.description, - metadata={'Alpha': 'a', 'Beta': 'b'}, - ) - self.assertEqual(self.columns, columns) - self.assertEqual(self.data, data) - - def test_snapshot_create_without_name(self): - arglist = [ - self.new_snapshot.volume_id, - "--description", self.new_snapshot.description, - "--force" - ] - verifylist = [ - ("volume", self.new_snapshot.volume_id), - ("description", self.new_snapshot.description), - ("force", True) - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - columns, data = self.cmd.take_action(parsed_args) - - self.snapshots_mock.create.assert_called_with( - self.new_snapshot.volume_id, - force=True, - name=None, - description=self.new_snapshot.description, - metadata=None, - ) - self.assertEqual(self.columns, columns) - self.assertEqual(self.data, data) - - -class TestSnapshotDelete(TestSnapshot): - - snapshots = volume_fakes.FakeSnapshot.create_snapshots(count=2) - - def setUp(self): - super(TestSnapshotDelete, self).setUp() - - self.snapshots_mock.get = ( - volume_fakes.FakeSnapshot.get_snapshots(self.snapshots)) - self.snapshots_mock.delete.return_value = None - - # Get the command object to mock - self.cmd = snapshot.DeleteSnapshot(self.app, None) - - def test_snapshot_delete(self): - arglist = [ - self.snapshots[0].id - ] - verifylist = [ - ("snapshots", [self.snapshots[0].id]) - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - result = self.cmd.take_action(parsed_args) - - self.snapshots_mock.delete.assert_called_with( - self.snapshots[0].id) - self.assertIsNone(result) - - def test_delete_multiple_snapshots(self): - arglist = [] - for s in self.snapshots: - arglist.append(s.id) - verifylist = [ - ('snapshots', arglist), - ] - - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - result = self.cmd.take_action(parsed_args) - - calls = [] - for s in self.snapshots: - calls.append(call(s.id)) - self.snapshots_mock.delete.assert_has_calls(calls) - self.assertIsNone(result) - - def test_delete_multiple_snapshots_with_exception(self): - arglist = [ - self.snapshots[0].id, - 'unexist_snapshot', - ] - verifylist = [ - ('snapshots', arglist), - ] - - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - find_mock_result = [self.snapshots[0], exceptions.CommandError] - with mock.patch.object(utils, 'find_resource', - side_effect=find_mock_result) as find_mock: - try: - self.cmd.take_action(parsed_args) - self.fail('CommandError should be raised.') - except exceptions.CommandError as e: - self.assertEqual('1 of 2 snapshots failed to delete.', - str(e)) - - find_mock.assert_any_call( - self.snapshots_mock, self.snapshots[0].id) - find_mock.assert_any_call(self.snapshots_mock, 'unexist_snapshot') - - self.assertEqual(2, find_mock.call_count) - self.snapshots_mock.delete.assert_called_once_with( - self.snapshots[0].id - ) - - -class TestSnapshotList(TestSnapshot): - - volume = volume_fakes.FakeVolume.create_one_volume() - snapshots = volume_fakes.FakeSnapshot.create_snapshots( - attrs={'volume_id': volume.name}, count=3) - - columns = [ - "ID", - "Name", - "Description", - "Status", - "Size" - ] - columns_long = columns + [ - "Created At", - "Volume", - "Properties" - ] - - data = [] - for s in snapshots: - data.append(( - s.id, - s.name, - s.description, - s.status, - s.size, - )) - data_long = [] - for s in snapshots: - data_long.append(( - s.id, - s.name, - s.description, - s.status, - s.size, - s.created_at, - s.volume_id, - utils.format_dict(s.metadata), - )) - - def setUp(self): - super(TestSnapshotList, self).setUp() - - self.volumes_mock.list.return_value = [self.volume] - self.snapshots_mock.list.return_value = self.snapshots - # Get the command to test - self.cmd = snapshot.ListSnapshot(self.app, None) - - def test_snapshot_list_without_options(self): - arglist = [] - verifylist = [ - ('all_projects', False), - ("long", False) - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - columns, data = self.cmd.take_action(parsed_args) - - self.snapshots_mock.list.assert_called_once_with( - limit=None, marker=None, search_opts={'all_tenants': False}) - self.assertEqual(self.columns, columns) - self.assertEqual(self.data, list(data)) - - def test_snapshot_list_with_options(self): - arglist = [ - "--long", - "--limit", "2", - "--marker", self.snapshots[0].id, - ] - verifylist = [ - ("long", True), - ("limit", 2), - ("marker", self.snapshots[0].id), - ('all_projects', False), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - columns, data = self.cmd.take_action(parsed_args) - - self.snapshots_mock.list.assert_called_once_with( - limit=2, - marker=self.snapshots[0].id, - search_opts={'all_tenants': False} - ) - self.assertEqual(self.columns_long, columns) - self.assertEqual(self.data_long, list(data)) - - def test_snapshot_list_all_projects(self): - arglist = [ - '--all-projects', - ] - verifylist = [ - ('long', False), - ('all_projects', True) - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - columns, data = self.cmd.take_action(parsed_args) - - self.snapshots_mock.list.assert_called_once_with( - limit=None, marker=None, search_opts={'all_tenants': True}) - self.assertEqual(self.columns, columns) - self.assertEqual(self.data, list(data)) - - def test_snapshot_list_negative_limit(self): - arglist = [ - "--limit", "-2", - ] - verifylist = [ - ("limit", -2), - ] - self.assertRaises(argparse.ArgumentTypeError, self.check_parser, - self.cmd, arglist, verifylist) - - -class TestSnapshotSet(TestSnapshot): - - snapshot = volume_fakes.FakeSnapshot.create_one_snapshot() - - def setUp(self): - super(TestSnapshotSet, self).setUp() - - self.snapshots_mock.get.return_value = self.snapshot - self.snapshots_mock.set_metadata.return_value = None - self.snapshots_mock.update.return_value = None - # Get the command object to mock - self.cmd = snapshot.SetSnapshot(self.app, None) - - def test_snapshot_set(self): - arglist = [ - "--name", "new_snapshot", - "--property", "x=y", - "--property", "foo=foo", - self.snapshot.id, - ] - new_property = {"x": "y", "foo": "foo"} - verifylist = [ - ("name", "new_snapshot"), - ("property", new_property), - ("snapshot", self.snapshot.id), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - result = self.cmd.take_action(parsed_args) - - kwargs = { - "name": "new_snapshot", - } - self.snapshots_mock.update.assert_called_with( - self.snapshot.id, **kwargs) - self.snapshots_mock.set_metadata.assert_called_with( - self.snapshot.id, new_property - ) - self.assertIsNone(result) - - def test_snapshot_set_state_to_error(self): - arglist = [ - "--state", "error", - self.snapshot.id - ] - verifylist = [ - ("state", "error"), - ("snapshot", self.snapshot.id) - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - result = self.cmd.take_action(parsed_args) - - self.snapshots_mock.reset_state.assert_called_with( - self.snapshot.id, "error") - self.assertIsNone(result) - - -class TestSnapshotShow(TestSnapshot): - - columns = ( - 'created_at', - 'description', - 'id', - 'name', - 'properties', - 'size', - 'status', - 'volume_id', - ) - - def setUp(self): - super(TestSnapshotShow, self).setUp() - - self.snapshot = volume_fakes.FakeSnapshot.create_one_snapshot() - - self.data = ( - self.snapshot.created_at, - self.snapshot.description, - self.snapshot.id, - self.snapshot.name, - utils.format_dict(self.snapshot.metadata), - self.snapshot.size, - self.snapshot.status, - self.snapshot.volume_id, - ) - - self.snapshots_mock.get.return_value = self.snapshot - # Get the command object to test - self.cmd = snapshot.ShowSnapshot(self.app, None) - - def test_snapshot_show(self): - arglist = [ - self.snapshot.id - ] - verifylist = [ - ("snapshot", self.snapshot.id) - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - columns, data = self.cmd.take_action(parsed_args) - self.snapshots_mock.get.assert_called_with(self.snapshot.id) - - self.assertEqual(self.columns, columns) - self.assertEqual(self.data, data) - - -class TestSnapshotUnset(TestSnapshot): - - snapshot = volume_fakes.FakeSnapshot.create_one_snapshot() - - def setUp(self): - super(TestSnapshotUnset, self).setUp() - - self.snapshots_mock.get.return_value = self.snapshot - self.snapshots_mock.delete_metadata.return_value = None - # Get the command object to mock - self.cmd = snapshot.UnsetSnapshot(self.app, None) - - def test_snapshot_unset(self): - arglist = [ - "--property", "foo", - self.snapshot.id, - ] - verifylist = [ - ("property", ["foo"]), - ("snapshot", self.snapshot.id), - ] - - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - result = self.cmd.take_action(parsed_args) - - self.snapshots_mock.delete_metadata.assert_called_with( - self.snapshot.id, ["foo"] - ) - self.assertIsNone(result) diff --git a/openstackclient/tests/volume/v2/test_transfer_request.py b/openstackclient/tests/volume/v2/test_transfer_request.py deleted file mode 100644 index ea39caa7..00000000 --- a/openstackclient/tests/volume/v2/test_transfer_request.py +++ /dev/null @@ -1,114 +0,0 @@ -# -# Licensed under the Apache License, Version 2.0 (the "License"); you may -# not use this file except in compliance with the License. You may obtain -# a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT -# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the -# License for the specific language governing permissions and limitations -# under the License. -# - - -from openstackclient.tests.volume.v2 import fakes as transfer_fakes -from openstackclient.volume.v2 import volume_transfer_request - - -class TestTransfer(transfer_fakes.TestVolume): - - def setUp(self): - super(TestTransfer, self).setUp() - - # Get a shortcut to the TransferManager Mock - self.transfer_mock = self.app.client_manager.volume.transfers - self.transfer_mock.reset_mock() - - -class TestTransferList(TestTransfer): - - # The Transfers to be listed - volume_transfers = transfer_fakes.FakeTransfer.create_one_transfer() - - def setUp(self): - super(TestTransferList, self).setUp() - - self.transfer_mock.list.return_value = [self.volume_transfers] - - # Get the command object to test - self.cmd = volume_transfer_request.ListTransferRequests(self.app, None) - - def test_transfer_list_without_argument(self): - arglist = [] - verifylist = [] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - # In base command class Lister in cliff, abstract method take_action() - # returns a tuple containing the column names and an iterable - # containing the data to be listed. - columns, data = self.cmd.take_action(parsed_args) - - expected_columns = [ - 'ID', - 'Volume', - 'Name', - ] - - # confirming if all expected columns are present in the result. - self.assertEqual(expected_columns, columns) - - datalist = (( - self.volume_transfers.id, - self.volume_transfers.volume_id, - self.volume_transfers.name, - ), ) - - # confirming if all expected values are present in the result. - self.assertEqual(datalist, tuple(data)) - - # checking if proper call was made to list volume_transfers - self.transfer_mock.list.assert_called_with( - detailed=True, - search_opts={'all_tenants': 0} - ) - - def test_transfer_list_with_argument(self): - arglist = [ - "--all-projects" - ] - verifylist = [ - ("all_projects", True) - ] - - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - # In base command class Lister in cliff, abstract method take_action() - # returns a tuple containing the column names and an iterable - # containing the data to be listed. - columns, data = self.cmd.take_action(parsed_args) - - expected_columns = [ - 'ID', - 'Volume', - 'Name', - ] - - # confirming if all expected columns are present in the result. - self.assertEqual(expected_columns, columns) - - datalist = (( - self.volume_transfers.id, - self.volume_transfers.volume_id, - self.volume_transfers.name, - ), ) - - # confirming if all expected values are present in the result. - self.assertEqual(datalist, tuple(data)) - - # checking if proper call was made to list volume_transfers - self.transfer_mock.list.assert_called_with( - detailed=True, - search_opts={'all_tenants': 1} - ) diff --git a/openstackclient/tests/volume/v2/test_type.py b/openstackclient/tests/volume/v2/test_type.py deleted file mode 100644 index b0316aef..00000000 --- a/openstackclient/tests/volume/v2/test_type.py +++ /dev/null @@ -1,575 +0,0 @@ -# -# Licensed under the Apache License, Version 2.0 (the "License"); you may -# not use this file except in compliance with the License. You may obtain -# a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT -# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the -# License for the specific language governing permissions and limitations -# under the License. -# - -import mock - -from osc_lib import exceptions -from osc_lib import utils - -from openstackclient.tests.identity.v3 import fakes as identity_fakes -from openstackclient.tests import utils as tests_utils -from openstackclient.tests.volume.v2 import fakes as volume_fakes -from openstackclient.volume.v2 import volume_type - - -class TestType(volume_fakes.TestVolume): - - def setUp(self): - super(TestType, self).setUp() - - self.types_mock = self.app.client_manager.volume.volume_types - self.types_mock.reset_mock() - - self.types_access_mock = ( - self.app.client_manager.volume.volume_type_access) - self.types_access_mock.reset_mock() - - self.projects_mock = self.app.client_manager.identity.projects - self.projects_mock.reset_mock() - - -class TestTypeCreate(TestType): - - project = identity_fakes.FakeProject.create_one_project() - columns = ( - 'description', - 'id', - 'is_public', - 'name', - ) - - def setUp(self): - super(TestTypeCreate, self).setUp() - - self.new_volume_type = volume_fakes.FakeType.create_one_type() - self.data = ( - self.new_volume_type.description, - self.new_volume_type.id, - True, - self.new_volume_type.name, - ) - - self.types_mock.create.return_value = self.new_volume_type - self.projects_mock.get.return_value = self.project - # Get the command object to test - self.cmd = volume_type.CreateVolumeType(self.app, None) - - def test_type_create_public(self): - arglist = [ - "--description", self.new_volume_type.description, - "--public", - self.new_volume_type.name, - ] - verifylist = [ - ("description", self.new_volume_type.description), - ("public", True), - ("private", False), - ("name", self.new_volume_type.name), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - columns, data = self.cmd.take_action(parsed_args) - self.types_mock.create.assert_called_with( - self.new_volume_type.name, - description=self.new_volume_type.description, - is_public=True, - ) - - self.assertEqual(self.columns, columns) - self.assertEqual(self.data, data) - - def test_type_create_private(self): - arglist = [ - "--description", self.new_volume_type.description, - "--private", - "--project", self.project.id, - self.new_volume_type.name, - ] - verifylist = [ - ("description", self.new_volume_type.description), - ("public", False), - ("private", True), - ("project", self.project.id), - ("name", self.new_volume_type.name), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - columns, data = self.cmd.take_action(parsed_args) - self.types_mock.create.assert_called_with( - self.new_volume_type.name, - description=self.new_volume_type.description, - is_public=False, - ) - - self.assertEqual(self.columns, columns) - self.assertEqual(self.data, data) - - def test_public_type_create_with_project(self): - arglist = [ - '--project', self.project.id, - self.new_volume_type.name, - ] - verifylist = [ - ('project', self.project.id), - ('name', self.new_volume_type.name), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - self.assertRaises(exceptions.CommandError, - self.cmd.take_action, - parsed_args) - - -class TestTypeDelete(TestType): - - volume_type = volume_fakes.FakeType.create_one_type() - - def setUp(self): - super(TestTypeDelete, self).setUp() - - self.types_mock.get.return_value = self.volume_type - self.types_mock.delete.return_value = None - - # Get the command object to mock - self.cmd = volume_type.DeleteVolumeType(self.app, None) - - def test_type_delete(self): - arglist = [ - self.volume_type.id - ] - verifylist = [ - ("volume_types", [self.volume_type.id]) - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - result = self.cmd.take_action(parsed_args) - - self.types_mock.delete.assert_called_with(self.volume_type) - self.assertIsNone(result) - - -class TestTypeList(TestType): - - volume_types = volume_fakes.FakeType.create_types() - - columns = [ - "ID", - "Name" - ] - columns_long = columns + [ - "Description", - "Properties" - ] - - data = [] - for t in volume_types: - data.append(( - t.id, - t.name, - )) - data_long = [] - for t in volume_types: - data_long.append(( - t.id, - t.name, - t.description, - utils.format_dict(t.extra_specs), - )) - - def setUp(self): - super(TestTypeList, self).setUp() - - self.types_mock.list.return_value = self.volume_types - # get the command to test - self.cmd = volume_type.ListVolumeType(self.app, None) - - def test_type_list_without_options(self): - arglist = [] - verifylist = [ - ("long", False), - ("private", False), - ("public", False), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - columns, data = self.cmd.take_action(parsed_args) - self.types_mock.list.assert_called_once_with(is_public=None) - self.assertEqual(self.columns, columns) - self.assertEqual(self.data, list(data)) - - def test_type_list_with_options(self): - arglist = [ - "--long", - "--public", - ] - verifylist = [ - ("long", True), - ("private", False), - ("public", True), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - columns, data = self.cmd.take_action(parsed_args) - self.types_mock.list.assert_called_once_with(is_public=True) - self.assertEqual(self.columns_long, columns) - self.assertEqual(self.data_long, list(data)) - - def test_type_list_with_private_option(self): - arglist = [ - "--private", - ] - verifylist = [ - ("long", False), - ("private", True), - ("public", False), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - columns, data = self.cmd.take_action(parsed_args) - self.types_mock.list.assert_called_once_with(is_public=False) - self.assertEqual(self.columns, columns) - self.assertEqual(self.data, list(data)) - - -class TestTypeSet(TestType): - - project = identity_fakes.FakeProject.create_one_project() - volume_type = volume_fakes.FakeType.create_one_type( - methods={'set_keys': None}) - - def setUp(self): - super(TestTypeSet, self).setUp() - - self.types_mock.get.return_value = self.volume_type - - # Return a project - self.projects_mock.get.return_value = self.project - # Get the command object to test - self.cmd = volume_type.SetVolumeType(self.app, None) - - def test_type_set_name(self): - new_name = 'new_name' - arglist = [ - '--name', new_name, - self.volume_type.id, - ] - verifylist = [ - ('name', new_name), - ('description', None), - ('property', None), - ('volume_type', self.volume_type.id), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - result = self.cmd.take_action(parsed_args) - - # Set expected values - kwargs = { - 'name': new_name, - } - self.types_mock.update.assert_called_with( - self.volume_type.id, - **kwargs - ) - self.assertIsNone(result) - - def test_type_set_description(self): - new_desc = 'new_desc' - arglist = [ - '--description', new_desc, - self.volume_type.id, - ] - verifylist = [ - ('name', None), - ('description', new_desc), - ('property', None), - ('volume_type', self.volume_type.id), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - result = self.cmd.take_action(parsed_args) - - # Set expected values - kwargs = { - 'description': new_desc, - } - self.types_mock.update.assert_called_with( - self.volume_type.id, - **kwargs - ) - self.assertIsNone(result) - - def test_type_set_property(self): - arglist = [ - '--property', 'myprop=myvalue', - self.volume_type.id, - ] - verifylist = [ - ('name', None), - ('description', None), - ('property', {'myprop': 'myvalue'}), - ('volume_type', self.volume_type.id), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - result = self.cmd.take_action(parsed_args) - self.volume_type.set_keys.assert_called_once_with( - {'myprop': 'myvalue'}) - self.assertIsNone(result) - - def test_type_set_not_called_without_project_argument(self): - arglist = [ - '--project', '', - self.volume_type.id, - ] - verifylist = [ - ('project', ''), - ('volume_type', self.volume_type.id), - ] - - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - result = self.cmd.take_action(parsed_args) - self.assertIsNone(result) - - self.assertFalse(self.types_access_mock.add_project_access.called) - - def test_type_set_failed_with_missing_volume_type_argument(self): - arglist = [ - '--project', 'identity_fakes.project_id', - ] - verifylist = [ - ('project', 'identity_fakes.project_id'), - ] - - self.assertRaises(tests_utils.ParserException, - self.check_parser, - self.cmd, - arglist, - verifylist) - - def test_type_set_project_access(self): - arglist = [ - '--project', self.project.id, - self.volume_type.id, - ] - verifylist = [ - ('project', self.project.id), - ('volume_type', self.volume_type.id), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - result = self.cmd.take_action(parsed_args) - self.assertIsNone(result) - - self.types_access_mock.add_project_access.assert_called_with( - self.volume_type.id, - self.project.id, - ) - - -class TestTypeShow(TestType): - - columns = ( - 'access_project_ids', - 'description', - 'id', - 'is_public', - 'name', - 'properties', - ) - - def setUp(self): - super(TestTypeShow, self).setUp() - - self.volume_type = volume_fakes.FakeType.create_one_type() - self.data = ( - None, - self.volume_type.description, - self.volume_type.id, - True, - self.volume_type.name, - utils.format_dict(self.volume_type.extra_specs) - ) - - self.types_mock.get.return_value = self.volume_type - - # Get the command object to test - self.cmd = volume_type.ShowVolumeType(self.app, None) - - def test_type_show(self): - arglist = [ - self.volume_type.id - ] - verifylist = [ - ("volume_type", self.volume_type.id) - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - columns, data = self.cmd.take_action(parsed_args) - self.types_mock.get.assert_called_with(self.volume_type.id) - - self.assertEqual(self.columns, columns) - self.assertEqual(self.data, data) - - def test_type_show_with_access(self): - arglist = [ - self.volume_type.id - ] - verifylist = [ - ("volume_type", self.volume_type.id) - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - private_type = volume_fakes.FakeType.create_one_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]): - columns, data = self.cmd.take_action(parsed_args) - self.types_mock.get.assert_called_once_with( - self.volume_type.id) - self.types_access_mock.list.assert_called_once_with( - private_type.id) - - self.assertEqual(self.columns, columns) - private_type_data = ( - utils.format_list([type_access_list.project_id]), - private_type.description, - private_type.id, - private_type.is_public, - private_type.name, - utils.format_dict(private_type.extra_specs) - ) - self.assertEqual(private_type_data, data) - - def test_type_show_with_list_access_exec(self): - arglist = [ - self.volume_type.id - ] - verifylist = [ - ("volume_type", self.volume_type.id) - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - private_type = volume_fakes.FakeType.create_one_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', - side_effect=Exception()): - columns, data = self.cmd.take_action(parsed_args) - self.types_mock.get.assert_called_once_with( - self.volume_type.id) - self.types_access_mock.list.assert_called_once_with( - private_type.id) - - self.assertEqual(self.columns, columns) - private_type_data = ( - None, - private_type.description, - private_type.id, - private_type.is_public, - private_type.name, - utils.format_dict(private_type.extra_specs) - ) - self.assertEqual(private_type_data, data) - - -class TestTypeUnset(TestType): - - project = identity_fakes.FakeProject.create_one_project() - volume_type = volume_fakes.FakeType.create_one_type( - methods={'unset_keys': None}) - - def setUp(self): - super(TestTypeUnset, self).setUp() - - self.types_mock.get.return_value = self.volume_type - - # Return a project - self.projects_mock.get.return_value = self.project - - # Get the command object to test - self.cmd = volume_type.UnsetVolumeType(self.app, None) - - def test_type_unset(self): - arglist = [ - '--property', 'property', - '--property', 'multi_property', - self.volume_type.id, - ] - verifylist = [ - ('property', ['property', 'multi_property']), - ('volume_type', self.volume_type.id), - ] - - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - result = self.cmd.take_action(parsed_args) - self.volume_type.unset_keys.assert_called_once_with( - ['property', 'multi_property']) - self.assertIsNone(result) - - def test_type_unset_project_access(self): - arglist = [ - '--project', self.project.id, - self.volume_type.id, - ] - verifylist = [ - ('project', self.project.id), - ('volume_type', self.volume_type.id), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - result = self.cmd.take_action(parsed_args) - self.assertIsNone(result) - - self.types_access_mock.remove_project_access.assert_called_with( - self.volume_type.id, - self.project.id, - ) - - def test_type_unset_not_called_without_project_argument(self): - arglist = [ - '--project', '', - self.volume_type.id, - ] - verifylist = [ - ('project', ''), - ('volume_type', self.volume_type.id), - ] - - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - result = self.cmd.take_action(parsed_args) - self.assertIsNone(result) - - self.assertFalse(self.types_access_mock.remove_project_access.called) - - def test_type_unset_failed_with_missing_volume_type_argument(self): - arglist = [ - '--project', 'identity_fakes.project_id', - ] - verifylist = [ - ('project', 'identity_fakes.project_id'), - ] - - self.assertRaises(tests_utils.ParserException, - self.check_parser, - self.cmd, - arglist, - verifylist) diff --git a/openstackclient/tests/volume/v2/test_volume.py b/openstackclient/tests/volume/v2/test_volume.py deleted file mode 100644 index 74181a25..00000000 --- a/openstackclient/tests/volume/v2/test_volume.py +++ /dev/null @@ -1,966 +0,0 @@ -# -# Licensed under the Apache License, Version 2.0 (the "License"); you may -# not use this file except in compliance with the License. You may obtain -# a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT -# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the -# License for the specific language governing permissions and limitations -# under the License. -# - -import mock -from mock import call - -from osc_lib import exceptions -from osc_lib import utils - -from openstackclient.tests.identity.v3 import fakes as identity_fakes -from openstackclient.tests.image.v2 import fakes as image_fakes -from openstackclient.tests.volume.v2 import fakes as volume_fakes -from openstackclient.volume.v2 import volume - - -class TestVolume(volume_fakes.TestVolume): - - def setUp(self): - super(TestVolume, self).setUp() - - self.volumes_mock = self.app.client_manager.volume.volumes - self.volumes_mock.reset_mock() - - self.projects_mock = self.app.client_manager.identity.projects - self.projects_mock.reset_mock() - - self.users_mock = self.app.client_manager.identity.users - self.users_mock.reset_mock() - - self.images_mock = self.app.client_manager.image.images - self.images_mock.reset_mock() - - self.snapshots_mock = self.app.client_manager.volume.volume_snapshots - self.snapshots_mock.reset_mock() - - def setup_volumes_mock(self, count): - volumes = volume_fakes.FakeVolume.create_volumes(count=count) - - self.volumes_mock.get = volume_fakes.FakeVolume.get_volumes( - volumes, - 0) - return volumes - - -class TestVolumeCreate(TestVolume): - - project = identity_fakes.FakeProject.create_one_project() - user = identity_fakes.FakeUser.create_one_user() - - columns = ( - 'attachments', - 'availability_zone', - 'bootable', - 'description', - 'id', - 'name', - 'properties', - 'size', - 'snapshot_id', - 'status', - 'type', - ) - - def setUp(self): - super(TestVolumeCreate, self).setUp() - - self.new_volume = volume_fakes.FakeVolume.create_one_volume() - self.volumes_mock.create.return_value = self.new_volume - - self.datalist = ( - self.new_volume.attachments, - self.new_volume.availability_zone, - self.new_volume.bootable, - self.new_volume.description, - self.new_volume.id, - self.new_volume.name, - utils.format_dict(self.new_volume.metadata), - self.new_volume.size, - self.new_volume.snapshot_id, - self.new_volume.status, - self.new_volume.volume_type, - ) - - # Get the command object to test - self.cmd = volume.CreateVolume(self.app, None) - - def test_volume_create_min_options(self): - arglist = [ - '--size', str(self.new_volume.size), - self.new_volume.name, - ] - verifylist = [ - ('size', self.new_volume.size), - ('name', self.new_volume.name), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - # In base command class ShowOne in cliff, abstract method take_action() - # returns a two-part tuple with a tuple of column names and a tuple of - # data to be shown. - columns, data = self.cmd.take_action(parsed_args) - - self.volumes_mock.create.assert_called_with( - size=self.new_volume.size, - snapshot_id=None, - name=self.new_volume.name, - description=None, - volume_type=None, - user_id=None, - project_id=None, - availability_zone=None, - metadata=None, - imageRef=None, - source_volid=None - ) - - self.assertEqual(self.columns, columns) - self.assertEqual(self.datalist, data) - - def test_volume_create_options(self): - arglist = [ - '--size', str(self.new_volume.size), - '--description', self.new_volume.description, - '--type', self.new_volume.volume_type, - '--availability-zone', self.new_volume.availability_zone, - self.new_volume.name, - ] - verifylist = [ - ('size', self.new_volume.size), - ('description', self.new_volume.description), - ('type', self.new_volume.volume_type), - ('availability_zone', self.new_volume.availability_zone), - ('name', self.new_volume.name), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - # In base command class ShowOne in cliff, abstract method take_action() - # returns a two-part tuple with a tuple of column names and a tuple of - # data to be shown. - columns, data = self.cmd.take_action(parsed_args) - - self.volumes_mock.create.assert_called_with( - size=self.new_volume.size, - snapshot_id=None, - name=self.new_volume.name, - description=self.new_volume.description, - volume_type=self.new_volume.volume_type, - user_id=None, - project_id=None, - availability_zone=self.new_volume.availability_zone, - metadata=None, - imageRef=None, - source_volid=None - ) - - self.assertEqual(self.columns, columns) - self.assertEqual(self.datalist, data) - - def test_volume_create_user_project_id(self): - # Return a project - self.projects_mock.get.return_value = self.project - # Return a user - self.users_mock.get.return_value = self.user - - arglist = [ - '--size', str(self.new_volume.size), - '--project', self.project.id, - '--user', self.user.id, - self.new_volume.name, - ] - verifylist = [ - ('size', self.new_volume.size), - ('project', self.project.id), - ('user', self.user.id), - ('name', self.new_volume.name), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - # In base command class ShowOne in cliff, abstract method take_action() - # returns a two-part tuple with a tuple of column names and a tuple of - # data to be shown. - columns, data = self.cmd.take_action(parsed_args) - - self.volumes_mock.create.assert_called_with( - size=self.new_volume.size, - snapshot_id=None, - name=self.new_volume.name, - description=None, - volume_type=None, - user_id=self.user.id, - project_id=self.project.id, - availability_zone=None, - metadata=None, - imageRef=None, - source_volid=None - ) - - self.assertEqual(self.columns, columns) - self.assertEqual(self.datalist, data) - - def test_volume_create_user_project_name(self): - # Return a project - self.projects_mock.get.return_value = self.project - # Return a user - self.users_mock.get.return_value = self.user - - arglist = [ - '--size', str(self.new_volume.size), - '--project', self.project.name, - '--user', self.user.name, - self.new_volume.name, - ] - verifylist = [ - ('size', self.new_volume.size), - ('project', self.project.name), - ('user', self.user.name), - ('name', self.new_volume.name), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - # In base command class ShowOne in cliff, abstract method take_action() - # returns a two-part tuple with a tuple of column names and a tuple of - # data to be shown. - columns, data = self.cmd.take_action(parsed_args) - - self.volumes_mock.create.assert_called_with( - size=self.new_volume.size, - snapshot_id=None, - name=self.new_volume.name, - description=None, - volume_type=None, - user_id=self.user.id, - project_id=self.project.id, - availability_zone=None, - metadata=None, - imageRef=None, - source_volid=None - ) - - self.assertEqual(self.columns, columns) - self.assertEqual(self.datalist, data) - - def test_volume_create_properties(self): - arglist = [ - '--property', 'Alpha=a', - '--property', 'Beta=b', - '--size', str(self.new_volume.size), - self.new_volume.name, - ] - verifylist = [ - ('property', {'Alpha': 'a', 'Beta': 'b'}), - ('size', self.new_volume.size), - ('name', self.new_volume.name), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - # In base command class ShowOne in cliff, abstract method take_action() - # returns a two-part tuple with a tuple of column names and a tuple of - # data to be shown. - columns, data = self.cmd.take_action(parsed_args) - - self.volumes_mock.create.assert_called_with( - size=self.new_volume.size, - snapshot_id=None, - name=self.new_volume.name, - description=None, - volume_type=None, - user_id=None, - project_id=None, - availability_zone=None, - metadata={'Alpha': 'a', 'Beta': 'b'}, - imageRef=None, - source_volid=None - ) - - self.assertEqual(self.columns, columns) - self.assertEqual(self.datalist, data) - - def test_volume_create_image_id(self): - image = image_fakes.FakeImage.create_one_image() - self.images_mock.get.return_value = image - - arglist = [ - '--image', image.id, - '--size', str(self.new_volume.size), - self.new_volume.name, - ] - verifylist = [ - ('image', image.id), - ('size', self.new_volume.size), - ('name', self.new_volume.name), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - # In base command class ShowOne in cliff, abstract method take_action() - # returns a two-part tuple with a tuple of column names and a tuple of - # data to be shown. - columns, data = self.cmd.take_action(parsed_args) - - self.volumes_mock.create.assert_called_with( - size=self.new_volume.size, - snapshot_id=None, - name=self.new_volume.name, - description=None, - volume_type=None, - user_id=None, - project_id=None, - availability_zone=None, - metadata=None, - imageRef=image.id, - source_volid=None, - ) - - self.assertEqual(self.columns, columns) - self.assertEqual(self.datalist, data) - - def test_volume_create_image_name(self): - image = image_fakes.FakeImage.create_one_image() - self.images_mock.get.return_value = image - - arglist = [ - '--image', image.name, - '--size', str(self.new_volume.size), - self.new_volume.name, - ] - verifylist = [ - ('image', image.name), - ('size', self.new_volume.size), - ('name', self.new_volume.name), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - # In base command class ShowOne in cliff, abstract method take_action() - # returns a two-part tuple with a tuple of column names and a tuple of - # data to be shown. - columns, data = self.cmd.take_action(parsed_args) - - self.volumes_mock.create.assert_called_with( - size=self.new_volume.size, - snapshot_id=None, - name=self.new_volume.name, - description=None, - volume_type=None, - user_id=None, - project_id=None, - availability_zone=None, - metadata=None, - imageRef=image.id, - source_volid=None - ) - - self.assertEqual(self.columns, columns) - self.assertEqual(self.datalist, data) - - def test_volume_create_with_snapshot(self): - snapshot = volume_fakes.FakeSnapshot.create_one_snapshot() - self.new_volume.snapshot_id = snapshot.id - arglist = [ - '--size', str(self.new_volume.size), - '--snapshot', self.new_volume.snapshot_id, - self.new_volume.name, - ] - verifylist = [ - ('size', self.new_volume.size), - ('snapshot', self.new_volume.snapshot_id), - ('name', self.new_volume.name), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - self.snapshots_mock.get.return_value = snapshot - - # In base command class ShowOne in cliff, abstract method take_action() - # returns a two-part tuple with a tuple of column names and a tuple of - # data to be shown. - columns, data = self.cmd.take_action(parsed_args) - - self.volumes_mock.create.assert_called_once_with( - size=self.new_volume.size, - snapshot_id=snapshot.id, - name=self.new_volume.name, - description=None, - volume_type=None, - user_id=None, - project_id=None, - availability_zone=None, - metadata=None, - imageRef=None, - source_volid=None - ) - - self.assertEqual(self.columns, columns) - self.assertEqual(self.datalist, data) - - -class TestVolumeDelete(TestVolume): - - def setUp(self): - super(TestVolumeDelete, self).setUp() - - self.volumes_mock.delete.return_value = None - - # Get the command object to mock - self.cmd = volume.DeleteVolume(self.app, None) - - def test_volume_delete_one_volume(self): - volumes = self.setup_volumes_mock(count=1) - - arglist = [ - volumes[0].id - ] - verifylist = [ - ("force", False), - ("purge", False), - ("volumes", [volumes[0].id]), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - result = self.cmd.take_action(parsed_args) - - self.volumes_mock.delete.assert_called_once_with( - volumes[0].id, cascade=False) - self.assertIsNone(result) - - def test_volume_delete_multi_volumes(self): - volumes = self.setup_volumes_mock(count=3) - - arglist = [v.id for v in volumes] - verifylist = [ - ('force', False), - ('purge', False), - ('volumes', arglist), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - result = self.cmd.take_action(parsed_args) - - calls = [call(v.id, cascade=False) for v in volumes] - self.volumes_mock.delete.assert_has_calls(calls) - self.assertIsNone(result) - - def test_volume_delete_multi_volumes_with_exception(self): - volumes = self.setup_volumes_mock(count=2) - - arglist = [ - volumes[0].id, - 'unexist_volume', - ] - verifylist = [ - ('force', False), - ('purge', False), - ('volumes', arglist), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - find_mock_result = [volumes[0], exceptions.CommandError] - with mock.patch.object(utils, 'find_resource', - side_effect=find_mock_result) as find_mock: - try: - self.cmd.take_action(parsed_args) - self.fail('CommandError should be raised.') - except exceptions.CommandError as e: - self.assertEqual('1 of 2 volumes failed to delete.', - str(e)) - - find_mock.assert_any_call(self.volumes_mock, volumes[0].id) - find_mock.assert_any_call(self.volumes_mock, 'unexist_volume') - - self.assertEqual(2, find_mock.call_count) - self.volumes_mock.delete.assert_called_once_with( - volumes[0].id, cascade=False) - - def test_volume_delete_with_purge(self): - volumes = self.setup_volumes_mock(count=1) - - arglist = [ - '--purge', - volumes[0].id, - ] - verifylist = [ - ('force', False), - ('purge', True), - ('volumes', [volumes[0].id]), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - result = self.cmd.take_action(parsed_args) - - self.volumes_mock.delete.assert_called_once_with( - volumes[0].id, cascade=True) - self.assertIsNone(result) - - def test_volume_delete_with_force(self): - volumes = self.setup_volumes_mock(count=1) - - arglist = [ - '--force', - volumes[0].id, - ] - verifylist = [ - ('force', True), - ('purge', False), - ('volumes', [volumes[0].id]), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - result = self.cmd.take_action(parsed_args) - - self.volumes_mock.force_delete.assert_called_once_with(volumes[0].id) - self.assertIsNone(result) - - -class TestVolumeList(TestVolume): - - project = identity_fakes.FakeProject.create_one_project() - user = identity_fakes.FakeUser.create_one_user() - - columns = [ - 'ID', - 'Display Name', - 'Status', - 'Size', - 'Attached to', - ] - - def setUp(self): - super(TestVolumeList, self).setUp() - - self.mock_volume = volume_fakes.FakeVolume.create_one_volume() - self.volumes_mock.list.return_value = [self.mock_volume] - - self.users_mock.get.return_value = self.user - - self.projects_mock.get.return_value = self.project - - # Get the command object to test - self.cmd = volume.ListVolume(self.app, None) - - def test_volume_list_no_options(self): - arglist = [] - verifylist = [ - ('long', False), - ('all_projects', False), - ('name', None), - ('status', None), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - columns, data = self.cmd.take_action(parsed_args) - - self.assertEqual(self.columns, columns) - - server = self.mock_volume.attachments[0]['server_id'] - device = self.mock_volume.attachments[0]['device'] - msg = 'Attached to %s on %s ' % (server, device) - datalist = (( - self.mock_volume.id, - self.mock_volume.name, - self.mock_volume.status, - self.mock_volume.size, - msg, - ), ) - self.assertEqual(datalist, tuple(data)) - - def test_volume_list_project(self): - arglist = [ - '--project', self.project.name, - ] - verifylist = [ - ('project', self.project.name), - ('long', False), - ('all_projects', False), - ('status', None), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - columns, data = self.cmd.take_action(parsed_args) - - self.assertEqual(self.columns, columns) - - server = self.mock_volume.attachments[0]['server_id'] - device = self.mock_volume.attachments[0]['device'] - msg = 'Attached to %s on %s ' % (server, device) - datalist = (( - self.mock_volume.id, - self.mock_volume.name, - self.mock_volume.status, - self.mock_volume.size, - msg, - ), ) - self.assertEqual(datalist, tuple(data)) - - def test_volume_list_project_domain(self): - arglist = [ - '--project', self.project.name, - '--project-domain', self.project.domain_id, - ] - verifylist = [ - ('project', self.project.name), - ('project_domain', self.project.domain_id), - ('long', False), - ('all_projects', False), - ('status', None), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - columns, data = self.cmd.take_action(parsed_args) - - self.assertEqual(self.columns, columns) - - server = self.mock_volume.attachments[0]['server_id'] - device = self.mock_volume.attachments[0]['device'] - msg = 'Attached to %s on %s ' % (server, device) - datalist = (( - self.mock_volume.id, - self.mock_volume.name, - self.mock_volume.status, - self.mock_volume.size, - msg, - ), ) - self.assertEqual(datalist, tuple(data)) - - def test_volume_list_user(self): - arglist = [ - '--user', self.user.name, - ] - verifylist = [ - ('user', self.user.name), - ('long', False), - ('all_projects', False), - ('status', None), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - columns, data = self.cmd.take_action(parsed_args) - - self.assertEqual(self.columns, columns) - server = self.mock_volume.attachments[0]['server_id'] - device = self.mock_volume.attachments[0]['device'] - msg = 'Attached to %s on %s ' % (server, device) - datalist = (( - self.mock_volume.id, - self.mock_volume.name, - self.mock_volume.status, - self.mock_volume.size, - msg, - ), ) - self.assertEqual(datalist, tuple(data)) - - def test_volume_list_user_domain(self): - arglist = [ - '--user', self.user.name, - '--user-domain', self.user.domain_id, - ] - verifylist = [ - ('user', self.user.name), - ('user_domain', self.user.domain_id), - ('long', False), - ('all_projects', False), - ('status', None), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - columns, data = self.cmd.take_action(parsed_args) - - self.assertEqual(self.columns, columns) - - server = self.mock_volume.attachments[0]['server_id'] - device = self.mock_volume.attachments[0]['device'] - msg = 'Attached to %s on %s ' % (server, device) - datalist = (( - self.mock_volume.id, - self.mock_volume.name, - self.mock_volume.status, - self.mock_volume.size, - msg, - ), ) - self.assertEqual(datalist, tuple(data)) - - def test_volume_list_name(self): - arglist = [ - '--name', self.mock_volume.name, - ] - verifylist = [ - ('long', False), - ('all_projects', False), - ('name', self.mock_volume.name), - ('status', None), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - columns, data = self.cmd.take_action(parsed_args) - - self.assertEqual(self.columns, columns) - - server = self.mock_volume.attachments[0]['server_id'] - device = self.mock_volume.attachments[0]['device'] - msg = 'Attached to %s on %s ' % (server, device) - datalist = (( - self.mock_volume.id, - self.mock_volume.name, - self.mock_volume.status, - self.mock_volume.size, - msg, - ), ) - self.assertEqual(datalist, tuple(data)) - - def test_volume_list_status(self): - arglist = [ - '--status', self.mock_volume.status, - ] - verifylist = [ - ('long', False), - ('all_projects', False), - ('name', None), - ('status', self.mock_volume.status), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - columns, data = self.cmd.take_action(parsed_args) - - self.assertEqual(self.columns, columns) - - server = self.mock_volume.attachments[0]['server_id'] - device = self.mock_volume.attachments[0]['device'] - msg = 'Attached to %s on %s ' % (server, device) - datalist = (( - self.mock_volume.id, - self.mock_volume.name, - self.mock_volume.status, - self.mock_volume.size, - msg, - ), ) - self.assertEqual(datalist, tuple(data)) - - def test_volume_list_all_projects(self): - arglist = [ - '--all-projects', - ] - verifylist = [ - ('long', False), - ('all_projects', True), - ('name', None), - ('status', None), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - columns, data = self.cmd.take_action(parsed_args) - - self.assertEqual(self.columns, columns) - - server = self.mock_volume.attachments[0]['server_id'] - device = self.mock_volume.attachments[0]['device'] - msg = 'Attached to %s on %s ' % (server, device) - datalist = (( - self.mock_volume.id, - self.mock_volume.name, - self.mock_volume.status, - self.mock_volume.size, - msg, - ), ) - self.assertEqual(datalist, tuple(data)) - - def test_volume_list_long(self): - arglist = [ - '--long', - ] - verifylist = [ - ('long', True), - ('all_projects', False), - ('name', None), - ('status', None), - ] - - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - columns, data = self.cmd.take_action(parsed_args) - - collist = [ - 'ID', - 'Display Name', - 'Status', - 'Size', - 'Type', - 'Bootable', - 'Attached to', - 'Properties', - ] - self.assertEqual(collist, columns) - - server = self.mock_volume.attachments[0]['server_id'] - device = self.mock_volume.attachments[0]['device'] - msg = 'Attached to %s on %s ' % (server, device) - datalist = (( - self.mock_volume.id, - self.mock_volume.name, - self.mock_volume.status, - self.mock_volume.size, - self.mock_volume.volume_type, - self.mock_volume.bootable, - msg, - utils.format_dict(self.mock_volume.metadata), - ), ) - self.assertEqual(datalist, tuple(data)) - - -class TestVolumeSet(TestVolume): - - def setUp(self): - super(TestVolumeSet, self).setUp() - - self.new_volume = volume_fakes.FakeVolume.create_one_volume() - self.volumes_mock.get.return_value = self.new_volume - - # Get the command object to test - self.cmd = volume.SetVolume(self.app, None) - - def test_volume_set_image_property(self): - arglist = [ - '--image-property', 'Alpha=a', - '--image-property', 'Beta=b', - self.new_volume.id, - ] - verifylist = [ - ('image_property', {'Alpha': 'a', 'Beta': 'b'}), - ('volume', self.new_volume.id), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - # In base command class ShowOne in cliff, abstract method take_action() - # returns nothing - self.cmd.take_action(parsed_args) - self.volumes_mock.set_image_metadata.assert_called_with( - self.new_volume.id, parsed_args.image_property) - - def test_volume_set_state(self): - arglist = [ - '--state', 'error', - self.new_volume.id - ] - verifylist = [ - ('state', 'error'), - ('volume', self.new_volume.id) - ] - - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - result = self.cmd.take_action(parsed_args) - self.volumes_mock.reset_state.assert_called_with( - self.new_volume.id, 'error') - self.assertIsNone(result) - - def test_volume_set_state_failed(self): - self.volumes_mock.reset_state.side_effect = exceptions.CommandError() - arglist = [ - '--state', 'error', - self.new_volume.id - ] - verifylist = [ - ('state', 'error'), - ('volume', self.new_volume.id) - ] - - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - try: - self.cmd.take_action(parsed_args) - self.fail('CommandError should be raised.') - except exceptions.CommandError as e: - self.assertEqual('One or more of the set operations failed', - str(e)) - self.volumes_mock.reset_state.assert_called_with( - self.new_volume.id, 'error') - - -class TestVolumeShow(TestVolume): - - def setUp(self): - super(TestVolumeShow, self).setUp() - - self._volume = volume_fakes.FakeVolume.create_one_volume() - self.volumes_mock.get.return_value = self._volume - # Get the command object to test - self.cmd = volume.ShowVolume(self.app, None) - - def test_volume_show(self): - arglist = [ - self._volume.id - ] - verifylist = [ - ("volume", self._volume.id) - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - columns, data = self.cmd.take_action(parsed_args) - self.volumes_mock.get.assert_called_with(self._volume.id) - - self.assertEqual( - volume_fakes.FakeVolume.get_volume_columns(self._volume), - columns) - - self.assertEqual( - volume_fakes.FakeVolume.get_volume_data(self._volume), - data) - - -class TestVolumeUnset(TestVolume): - - def setUp(self): - super(TestVolumeUnset, self).setUp() - - self.new_volume = volume_fakes.FakeVolume.create_one_volume() - self.volumes_mock.get.return_value = self.new_volume - - # Get the command object to set property - self.cmd_set = volume.SetVolume(self.app, None) - - # Get the command object to unset property - self.cmd_unset = volume.UnsetVolume(self.app, None) - - def test_volume_unset_image_property(self): - - # Arguments for setting image properties - arglist = [ - '--image-property', 'Alpha=a', - '--image-property', 'Beta=b', - self.new_volume.id, - ] - verifylist = [ - ('image_property', {'Alpha': 'a', 'Beta': 'b'}), - ('volume', self.new_volume.id), - ] - parsed_args = self.check_parser(self.cmd_set, arglist, verifylist) - - # In base command class ShowOne in cliff, abstract method take_action() - # returns nothing - self.cmd_set.take_action(parsed_args) - - # Arguments for unsetting image properties - arglist_unset = [ - '--image-property', 'Alpha', - self.new_volume.id, - ] - verifylist_unset = [ - ('image_property', ['Alpha']), - ('volume', self.new_volume.id), - ] - parsed_args_unset = self.check_parser(self.cmd_unset, - arglist_unset, - verifylist_unset) - - # In base command class ShowOne in cliff, abstract method take_action() - # returns nothing - self.cmd_unset.take_action(parsed_args_unset) - - self.volumes_mock.delete_image_metadata.assert_called_with( - self.new_volume.id, parsed_args_unset.image_property) |