diff options
Diffstat (limited to 'openstackclient/tests/unit/volume')
17 files changed, 5887 insertions, 0 deletions
diff --git a/openstackclient/tests/unit/volume/__init__.py b/openstackclient/tests/unit/volume/__init__.py new file mode 100644 index 00000000..e69de29b --- /dev/null +++ b/openstackclient/tests/unit/volume/__init__.py diff --git a/openstackclient/tests/unit/volume/test_find_resource.py b/openstackclient/tests/unit/volume/test_find_resource.py new file mode 100644 index 00000000..d2509315 --- /dev/null +++ b/openstackclient/tests/unit/volume/test_find_resource.py @@ -0,0 +1,83 @@ +# 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.unit 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/unit/volume/v1/__init__.py b/openstackclient/tests/unit/volume/v1/__init__.py new file mode 100644 index 00000000..e69de29b --- /dev/null +++ b/openstackclient/tests/unit/volume/v1/__init__.py diff --git a/openstackclient/tests/unit/volume/v1/fakes.py b/openstackclient/tests/unit/volume/v1/fakes.py new file mode 100644 index 00000000..c6fee7d1 --- /dev/null +++ b/openstackclient/tests/unit/volume/v1/fakes.py @@ -0,0 +1,280 @@ +# 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.unit import fakes +from openstackclient.tests.unit.identity.v2_0 import fakes as identity_fakes +from openstackclient.tests.unit 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/unit/volume/v1/test_qos_specs.py b/openstackclient/tests/unit/volume/v1/test_qos_specs.py new file mode 100644 index 00000000..7b87ccb3 --- /dev/null +++ b/openstackclient/tests/unit/volume/v1/test_qos_specs.py @@ -0,0 +1,472 @@ +# 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.unit import fakes +from openstackclient.tests.unit.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/unit/volume/v1/test_service.py b/openstackclient/tests/unit/volume/v1/test_service.py new file mode 100644 index 00000000..82d21bfc --- /dev/null +++ b/openstackclient/tests/unit/volume/v1/test_service.py @@ -0,0 +1,286 @@ +# +# 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.unit.volume.v1 import fakes as service_fakes +from openstackclient.volume.v1 import service + + +class TestService(service_fakes.TestVolumev1): + + 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/unit/volume/v1/test_transfer_request.py b/openstackclient/tests/unit/volume/v1/test_transfer_request.py new file mode 100644 index 00000000..f7980c34 --- /dev/null +++ b/openstackclient/tests/unit/volume/v1/test_transfer_request.py @@ -0,0 +1,114 @@ +# +# 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.unit.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/unit/volume/v1/test_volume.py b/openstackclient/tests/unit/volume/v1/test_volume.py new file mode 100644 index 00000000..f90566fd --- /dev/null +++ b/openstackclient/tests/unit/volume/v1/test_volume.py @@ -0,0 +1,716 @@ +# 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.unit import fakes +from openstackclient.tests.unit.identity.v2_0 import fakes as identity_fakes +from openstackclient.tests.unit.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/unit/volume/v2/__init__.py b/openstackclient/tests/unit/volume/v2/__init__.py new file mode 100644 index 00000000..e69de29b --- /dev/null +++ b/openstackclient/tests/unit/volume/v2/__init__.py diff --git a/openstackclient/tests/unit/volume/v2/fakes.py b/openstackclient/tests/unit/volume/v2/fakes.py new file mode 100644 index 00000000..a958c468 --- /dev/null +++ b/openstackclient/tests/unit/volume/v2/fakes.py @@ -0,0 +1,696 @@ +# +# 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.unit import fakes +from openstackclient.tests.unit.identity.v3 import fakes as identity_fakes +from openstackclient.tests.unit.image.v2 import fakes as image_fakes +from openstackclient.tests.unit 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/unit/volume/v2/test_backup.py b/openstackclient/tests/unit/volume/v2/test_backup.py new file mode 100644 index 00000000..45633870 --- /dev/null +++ b/openstackclient/tests/unit/volume/v2/test_backup.py @@ -0,0 +1,388 @@ +# +# 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.unit.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/unit/volume/v2/test_qos_specs.py b/openstackclient/tests/unit/volume/v2/test_qos_specs.py new file mode 100644 index 00000000..7597e852 --- /dev/null +++ b/openstackclient/tests/unit/volume/v2/test_qos_specs.py @@ -0,0 +1,453 @@ +# 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.unit.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/unit/volume/v2/test_service.py b/openstackclient/tests/unit/volume/v2/test_service.py new file mode 100644 index 00000000..3e9b2df9 --- /dev/null +++ b/openstackclient/tests/unit/volume/v2/test_service.py @@ -0,0 +1,286 @@ +# +# 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.unit.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/unit/volume/v2/test_snapshot.py b/openstackclient/tests/unit/volume/v2/test_snapshot.py new file mode 100644 index 00000000..333d8d72 --- /dev/null +++ b/openstackclient/tests/unit/volume/v2/test_snapshot.py @@ -0,0 +1,458 @@ +# +# 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.unit.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/unit/volume/v2/test_transfer_request.py b/openstackclient/tests/unit/volume/v2/test_transfer_request.py new file mode 100644 index 00000000..32108c02 --- /dev/null +++ b/openstackclient/tests/unit/volume/v2/test_transfer_request.py @@ -0,0 +1,114 @@ +# +# 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.unit.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/unit/volume/v2/test_type.py b/openstackclient/tests/unit/volume/v2/test_type.py new file mode 100644 index 00000000..84f87e3b --- /dev/null +++ b/openstackclient/tests/unit/volume/v2/test_type.py @@ -0,0 +1,575 @@ +# +# 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.unit.identity.v3 import fakes as identity_fakes +from openstackclient.tests.unit import utils as tests_utils +from openstackclient.tests.unit.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/unit/volume/v2/test_volume.py b/openstackclient/tests/unit/volume/v2/test_volume.py new file mode 100644 index 00000000..66f8f74d --- /dev/null +++ b/openstackclient/tests/unit/volume/v2/test_volume.py @@ -0,0 +1,966 @@ +# +# 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.unit.identity.v3 import fakes as identity_fakes +from openstackclient.tests.unit.image.v2 import fakes as image_fakes +from openstackclient.tests.unit.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) |