summaryrefslogtreecommitdiff
path: root/openstackclient/tests/volume/v1
diff options
context:
space:
mode:
authorSteve Martinelli <s.martinelli@gmail.com>2016-09-05 22:14:33 -0700
committerSteve Martinelli <s.martinelli@gmail.com>2016-09-08 15:19:50 -0700
commit39839def2e356e8d145be89380c73a71423cf06d (patch)
treef53c090b7ded46554866436e9d687492d2cef487 /openstackclient/tests/volume/v1
parent7d1a5d0854c732681a130255ddc6abb2e9721a7a (diff)
downloadpython-openstackclient-39839def2e356e8d145be89380c73a71423cf06d.tar.gz
move unit tests to new "unit" test module
this will better isolate the unit tests from the functional tests. unfortunately, the "integration" tests had to be lumped into the "unit" tests since we need the separation in testr.conf Change-Id: Ifd12198c1f90e4e3c951c73bfa1884ab300d8ded
Diffstat (limited to 'openstackclient/tests/volume/v1')
-rw-r--r--openstackclient/tests/volume/v1/__init__.py0
-rw-r--r--openstackclient/tests/volume/v1/fakes.py280
-rw-r--r--openstackclient/tests/volume/v1/test_qos_specs.py472
-rw-r--r--openstackclient/tests/volume/v1/test_transfer_request.py114
-rw-r--r--openstackclient/tests/volume/v1/test_volume.py716
5 files changed, 0 insertions, 1582 deletions
diff --git a/openstackclient/tests/volume/v1/__init__.py b/openstackclient/tests/volume/v1/__init__.py
deleted file mode 100644
index e69de29b..00000000
--- a/openstackclient/tests/volume/v1/__init__.py
+++ /dev/null
diff --git a/openstackclient/tests/volume/v1/fakes.py b/openstackclient/tests/volume/v1/fakes.py
deleted file mode 100644
index b96f925d..00000000
--- a/openstackclient/tests/volume/v1/fakes.py
+++ /dev/null
@@ -1,280 +0,0 @@
-# Copyright 2013 Nebula Inc.
-#
-# Licensed under the Apache License, Version 2.0 (the "License"); you may
-# not use this file except in compliance with the License. You may obtain
-# a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
-# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
-# License for the specific language governing permissions and limitations
-# under the License.
-#
-
-import mock
-
-from openstackclient.tests import fakes
-from openstackclient.tests.identity.v2_0 import fakes as identity_fakes
-from openstackclient.tests import utils
-
-
-volume_id = 'vvvvvvvv-vvvv-vvvv-vvvvvvvv'
-volume_name = 'nigel'
-volume_description = 'Nigel Tufnel'
-volume_status = 'available'
-volume_size = 120
-volume_type = 'to-eleven'
-volume_zone = 'stonehenge'
-volume_metadata = {
- 'Alpha': 'a',
- 'Beta': 'b',
- 'Gamma': 'g',
-}
-volume_metadata_str = "Alpha='a', Beta='b', Gamma='g'"
-
-VOLUME = {
- 'id': volume_id,
- 'display_name': volume_name,
- 'display_description': volume_description,
- 'size': volume_size,
- 'status': volume_status,
- 'attach_status': 'detached',
- 'availability_zone': volume_zone,
- 'volume_type': volume_type,
- 'metadata': volume_metadata,
-}
-
-extension_name = 'SchedulerHints'
-extension_namespace = 'http://docs.openstack.org/'\
- 'block-service/ext/scheduler-hints/api/v2'
-extension_description = 'Pass arbitrary key/value'\
- 'pairs to the scheduler.'
-extension_updated = '2014-02-07T12:00:0-00:00'
-extension_alias = 'OS-SCH-HNT'
-extension_links = '[{"href":'\
- '"https://github.com/openstack/block-api", "type":'\
- ' "text/html", "rel": "describedby"}]'
-
-EXTENSION = {
- 'name': extension_name,
- 'namespace': extension_namespace,
- 'description': extension_description,
- 'updated': extension_updated,
- 'alias': extension_alias,
- 'links': extension_links,
-}
-
-# NOTE(dtroyer): duplicating here the minimum image info needed to test
-# volume create --image until circular references can be
-# avoided by refactoring the test fakes.
-
-image_id = 'im1'
-image_name = 'graven'
-
-
-IMAGE = {
- 'id': image_id,
- 'name': image_name,
-}
-
-type_id = "5520dc9e-6f9b-4378-a719-729911c0f407"
-type_name = "fake-lvmdriver-1"
-
-TYPE = {
- 'id': type_id,
- 'name': type_name
-}
-
-qos_id = '6f2be1de-997b-4230-b76c-a3633b59e8fb'
-qos_consumer = 'front-end'
-qos_default_consumer = 'both'
-qos_name = "fake-qos-specs"
-qos_specs = {
- 'foo': 'bar',
- 'iops': '9001'
-}
-qos_association = {
- 'association_type': 'volume_type',
- 'name': type_name,
- 'id': type_id
-}
-
-QOS = {
- 'id': qos_id,
- 'consumer': qos_consumer,
- 'name': qos_name
-}
-
-QOS_DEFAULT_CONSUMER = {
- 'id': qos_id,
- 'consumer': qos_default_consumer,
- 'name': qos_name
-}
-
-QOS_WITH_SPECS = {
- 'id': qos_id,
- 'consumer': qos_consumer,
- 'name': qos_name,
- 'specs': qos_specs
-}
-
-QOS_WITH_ASSOCIATIONS = {
- 'id': qos_id,
- 'consumer': qos_consumer,
- 'name': qos_name,
- 'specs': qos_specs,
- 'associations': [qos_association]
-}
-
-
-class FakeTransfer(object):
- """Fake one or more Transfer."""
-
- @staticmethod
- def create_one_transfer(attrs=None):
- """Create a fake transfer.
-
- :param Dictionary attrs:
- A dictionary with all attributes of Transfer Request
- :return:
- A FakeResource object with volume_id, name, id.
- """
- # Set default attribute
- transfer_info = {
- 'volume_id': 'ce26708d-a7f8-4b4b-9861-4a80256615a7',
- 'name': 'fake_transfer_name',
- 'id': '731a7f53-aa92-4fbd-9de3-6f7d729c926b'
- }
-
- # Overwrite default attributes if there are some attributes set
- attrs = attrs or {}
-
- transfer_info.update(attrs)
-
- transfer = fakes.FakeResource(
- None,
- transfer_info,
- loaded=True)
-
- return transfer
-
-
-class FakeService(object):
- """Fake one or more Services."""
-
- @staticmethod
- def create_one_service(attrs=None):
- """Create a fake service.
-
- :param Dictionary attrs:
- A dictionary with all attributes of service
- :return:
- A FakeResource object with host, status, etc.
- """
- # Set default attribute
- service_info = {
- 'host': 'host_test',
- 'binary': 'cinder_test',
- 'status': 'enabled',
- 'disabled_reason': 'LongHoliday-GoldenWeek',
- 'zone': 'fake_zone',
- 'updated_at': 'fake_date',
- 'state': 'fake_state',
- }
-
- # Overwrite default attributes if there are some attributes set
- attrs = attrs or {}
-
- service_info.update(attrs)
-
- service = fakes.FakeResource(
- None,
- service_info,
- loaded=True)
-
- return service
-
- @staticmethod
- def create_services(attrs=None, count=2):
- """Create multiple fake services.
-
- :param Dictionary attrs:
- A dictionary with all attributes of service
- :param Integer count:
- The number of services to be faked
- :return:
- A list of FakeResource objects
- """
- services = []
- for n in range(0, count):
- services.append(FakeService.create_one_service(attrs))
-
- return services
-
- @staticmethod
- def get_services(services=None, count=2):
- """Get an iterable MagicMock object with a list of faked services.
-
- If services list is provided, then initialize the Mock object with the
- list. Otherwise create one.
-
- :param List services:
- A list of FakeResource objects faking services
- :param Integer count:
- The number of services to be faked
- :return
- An iterable Mock object with side_effect set to a list of faked
- services
- """
- if services is None:
- services = FakeService.create_services(count)
-
- return mock.MagicMock(side_effect=services)
-
-
-class FakeImagev1Client(object):
-
- def __init__(self, **kwargs):
- self.images = mock.Mock()
-
-
-class FakeVolumev1Client(object):
-
- def __init__(self, **kwargs):
- self.volumes = mock.Mock()
- self.volumes.resource_class = fakes.FakeResource(None, {})
- self.services = mock.Mock()
- self.services.resource_class = fakes.FakeResource(None, {})
- self.extensions = mock.Mock()
- self.extensions.resource_class = fakes.FakeResource(None, {})
- self.qos_specs = mock.Mock()
- self.qos_specs.resource_class = fakes.FakeResource(None, {})
- self.volume_types = mock.Mock()
- self.volume_types.resource_class = fakes.FakeResource(None, {})
- self.transfers = mock.Mock()
- self.transfers.resource_class = fakes.FakeResource(None, {})
- self.auth_token = kwargs['token']
- self.management_url = kwargs['endpoint']
-
-
-class TestVolumev1(utils.TestCommand):
-
- def setUp(self):
- super(TestVolumev1, self).setUp()
-
- self.app.client_manager.volume = FakeVolumev1Client(
- endpoint=fakes.AUTH_URL,
- token=fakes.AUTH_TOKEN,
- )
-
- self.app.client_manager.identity = identity_fakes.FakeIdentityv2Client(
- endpoint=fakes.AUTH_URL,
- token=fakes.AUTH_TOKEN,
- )
-
- self.app.client_manager.image = FakeImagev1Client(
- endpoint=fakes.AUTH_URL,
- token=fakes.AUTH_TOKEN,
- )
diff --git a/openstackclient/tests/volume/v1/test_qos_specs.py b/openstackclient/tests/volume/v1/test_qos_specs.py
deleted file mode 100644
index 4e1733fd..00000000
--- a/openstackclient/tests/volume/v1/test_qos_specs.py
+++ /dev/null
@@ -1,472 +0,0 @@
-# Copyright 2015 iWeb Technologies Inc.
-#
-# Licensed under the Apache License, Version 2.0 (the "License"); you may
-# not use this file except in compliance with the License. You may obtain
-# a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
-# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
-# License for the specific language governing permissions and limitations
-# under the License.
-#
-
-import copy
-
-from osc_lib import utils
-
-from openstackclient.tests import fakes
-from openstackclient.tests.volume.v1 import fakes as volume_fakes
-from openstackclient.volume.v1 import qos_specs
-
-
-class TestQos(volume_fakes.TestVolumev1):
-
- def setUp(self):
- super(TestQos, self).setUp()
-
- self.qos_mock = self.app.client_manager.volume.qos_specs
- self.qos_mock.reset_mock()
-
- self.types_mock = self.app.client_manager.volume.volume_types
- self.types_mock.reset_mock()
-
-
-class TestQosAssociate(TestQos):
-
- def setUp(self):
- super(TestQosAssociate, self).setUp()
-
- # Get the command object to test
- self.cmd = qos_specs.AssociateQos(self.app, None)
-
- def test_qos_associate(self):
- self.qos_mock.get.return_value = fakes.FakeResource(
- None,
- copy.deepcopy(volume_fakes.QOS),
- loaded=True
- )
- self.types_mock.get.return_value = fakes.FakeResource(
- None,
- copy.deepcopy(volume_fakes.TYPE),
- loaded=True
- )
- arglist = [
- volume_fakes.qos_id,
- volume_fakes.type_id
- ]
- verifylist = [
- ('qos_spec', volume_fakes.qos_id),
- ('volume_type', volume_fakes.type_id)
- ]
- parsed_args = self.check_parser(self.cmd, arglist, verifylist)
-
- result = self.cmd.take_action(parsed_args)
-
- self.qos_mock.associate.assert_called_with(
- volume_fakes.qos_id,
- volume_fakes.type_id
- )
- self.assertIsNone(result)
-
-
-class TestQosCreate(TestQos):
-
- columns = (
- 'consumer',
- 'id',
- 'name',
- )
- datalist = (
- volume_fakes.qos_consumer,
- volume_fakes.qos_id,
- volume_fakes.qos_name
- )
-
- def setUp(self):
- super(TestQosCreate, self).setUp()
-
- # Get the command object to test
- self.cmd = qos_specs.CreateQos(self.app, None)
-
- def test_qos_create_without_properties(self):
- self.qos_mock.create.return_value = fakes.FakeResource(
- None,
- copy.deepcopy(volume_fakes.QOS_DEFAULT_CONSUMER),
- loaded=True
- )
-
- arglist = [
- volume_fakes.qos_name,
- ]
- verifylist = [
- ('name', volume_fakes.qos_name),
- ]
- parsed_args = self.check_parser(self.cmd, arglist, verifylist)
-
- columns, data = self.cmd.take_action(parsed_args)
-
- self.qos_mock.create.assert_called_with(
- volume_fakes.qos_name,
- {'consumer': volume_fakes.qos_default_consumer}
- )
-
- self.assertEqual(self.columns, columns)
- datalist = (
- volume_fakes.qos_default_consumer,
- volume_fakes.qos_id,
- volume_fakes.qos_name
- )
- self.assertEqual(datalist, data)
-
- def test_qos_create_with_consumer(self):
- self.qos_mock.create.return_value = fakes.FakeResource(
- None,
- copy.deepcopy(volume_fakes.QOS),
- loaded=True
- )
-
- arglist = [
- volume_fakes.qos_name,
- '--consumer', volume_fakes.qos_consumer
- ]
- verifylist = [
- ('name', volume_fakes.qos_name),
- ('consumer', volume_fakes.qos_consumer)
- ]
- parsed_args = self.check_parser(self.cmd, arglist, verifylist)
-
- columns, data = self.cmd.take_action(parsed_args)
-
- self.qos_mock.create.assert_called_with(
- volume_fakes.qos_name,
- {'consumer': volume_fakes.qos_consumer}
- )
- self.assertEqual(self.columns, columns)
- self.assertEqual(self.datalist, data)
-
- def test_qos_create_with_properties(self):
- self.qos_mock.create.return_value = fakes.FakeResource(
- None,
- copy.deepcopy(volume_fakes.QOS_WITH_SPECS),
- loaded=True
- )
-
- arglist = [
- volume_fakes.qos_name,
- '--consumer', volume_fakes.qos_consumer,
- '--property', 'foo=bar',
- '--property', 'iops=9001'
- ]
- verifylist = [
- ('name', volume_fakes.qos_name),
- ('consumer', volume_fakes.qos_consumer),
- ('property', volume_fakes.qos_specs)
- ]
- parsed_args = self.check_parser(self.cmd, arglist, verifylist)
-
- columns, data = self.cmd.take_action(parsed_args)
-
- specs = volume_fakes.qos_specs.copy()
- specs.update({'consumer': volume_fakes.qos_consumer})
- self.qos_mock.create.assert_called_with(
- volume_fakes.qos_name,
- specs
- )
-
- columns = self.columns + (
- 'specs',
- )
- self.assertEqual(columns, columns)
- datalist = self.datalist + (
- volume_fakes.qos_specs,
- )
- self.assertEqual(datalist, data)
-
-
-class TestQosDelete(TestQos):
-
- def setUp(self):
- super(TestQosDelete, self).setUp()
-
- self.qos_mock.get.return_value = fakes.FakeResource(
- None,
- copy.deepcopy(volume_fakes.QOS),
- loaded=True,
- )
-
- # Get the command object to test
- self.cmd = qos_specs.DeleteQos(self.app, None)
-
- def test_qos_delete_with_id(self):
- arglist = [
- volume_fakes.qos_id
- ]
- verifylist = [
- ('qos_specs', [volume_fakes.qos_id])
- ]
- parsed_args = self.check_parser(self.cmd, arglist, verifylist)
-
- result = self.cmd.take_action(parsed_args)
-
- self.qos_mock.delete.assert_called_with(volume_fakes.qos_id, False)
- self.assertIsNone(result)
-
- def test_qos_delete_with_name(self):
- arglist = [
- volume_fakes.qos_name
- ]
- verifylist = [
- ('qos_specs', [volume_fakes.qos_name])
- ]
- parsed_args = self.check_parser(self.cmd, arglist, verifylist)
-
- result = self.cmd.take_action(parsed_args)
-
- self.qos_mock.delete.assert_called_with(volume_fakes.qos_id, False)
- self.assertIsNone(result)
-
- def test_qos_delete_with_force(self):
- arglist = [
- '--force',
- volume_fakes.qos_id
- ]
- verifylist = [
- ('force', True),
- ('qos_specs', [volume_fakes.qos_id])
- ]
- parsed_args = self.check_parser(self.cmd, arglist, verifylist)
-
- result = self.cmd.take_action(parsed_args)
-
- self.qos_mock.delete.assert_called_with(volume_fakes.qos_id, True)
- self.assertIsNone(result)
-
-
-class TestQosDisassociate(TestQos):
-
- def setUp(self):
- super(TestQosDisassociate, self).setUp()
-
- # Get the command object to test
- self.cmd = qos_specs.DisassociateQos(self.app, None)
-
- def test_qos_disassociate_with_volume_type(self):
- self.qos_mock.get.return_value = fakes.FakeResource(
- None,
- copy.deepcopy(volume_fakes.QOS),
- loaded=True
- )
- self.types_mock.get.return_value = fakes.FakeResource(
- None,
- copy.deepcopy(volume_fakes.TYPE),
- loaded=True
- )
- arglist = [
- volume_fakes.qos_id,
- '--volume-type', volume_fakes.type_id
- ]
- verifylist = [
- ('qos_spec', volume_fakes.qos_id),
- ('volume_type', volume_fakes.type_id)
- ]
- parsed_args = self.check_parser(self.cmd, arglist, verifylist)
-
- result = self.cmd.take_action(parsed_args)
-
- self.qos_mock.disassociate.assert_called_with(
- volume_fakes.qos_id,
- volume_fakes.type_id
- )
- self.assertIsNone(result)
-
- def test_qos_disassociate_with_all_volume_types(self):
- self.qos_mock.get.return_value = fakes.FakeResource(
- None,
- copy.deepcopy(volume_fakes.QOS),
- loaded=True
- )
-
- arglist = [
- volume_fakes.qos_id,
- '--all'
- ]
- verifylist = [
- ('qos_spec', volume_fakes.qos_id)
- ]
- parsed_args = self.check_parser(self.cmd, arglist, verifylist)
-
- result = self.cmd.take_action(parsed_args)
-
- self.qos_mock.disassociate_all.assert_called_with(volume_fakes.qos_id)
- self.assertIsNone(result)
-
-
-class TestQosList(TestQos):
-
- def setUp(self):
- super(TestQosList, self).setUp()
-
- self.qos_mock.get.return_value = fakes.FakeResource(
- None,
- copy.deepcopy(volume_fakes.QOS_WITH_ASSOCIATIONS),
- loaded=True,
- )
- self.qos_mock.list.return_value = [self.qos_mock.get.return_value]
- self.qos_mock.get_associations.return_value = [fakes.FakeResource(
- None,
- copy.deepcopy(volume_fakes.qos_association),
- loaded=True,
- )]
-
- # Get the command object to test
- self.cmd = qos_specs.ListQos(self.app, None)
-
- def test_qos_list(self):
- arglist = []
- verifylist = []
-
- parsed_args = self.check_parser(self.cmd, arglist, verifylist)
-
- columns, data = self.cmd.take_action(parsed_args)
- self.qos_mock.list.assert_called_with()
-
- collist = (
- 'ID',
- 'Name',
- 'Consumer',
- 'Associations',
- 'Specs',
- )
- self.assertEqual(collist, columns)
- datalist = ((
- volume_fakes.qos_id,
- volume_fakes.qos_name,
- volume_fakes.qos_consumer,
- volume_fakes.type_name,
- utils.format_dict(volume_fakes.qos_specs),
- ), )
- self.assertEqual(datalist, tuple(data))
-
-
-class TestQosSet(TestQos):
-
- def setUp(self):
- super(TestQosSet, self).setUp()
-
- # Get the command object to test
- self.cmd = qos_specs.SetQos(self.app, None)
-
- def test_qos_set_with_properties_with_id(self):
- self.qos_mock.get.return_value = fakes.FakeResource(
- None,
- copy.deepcopy(volume_fakes.QOS_WITH_SPECS),
- loaded=True
- )
- arglist = [
- volume_fakes.qos_id,
- '--property', 'foo=bar',
- '--property', 'iops=9001'
- ]
- verifylist = [
- ('qos_spec', volume_fakes.qos_id),
- ('property', volume_fakes.qos_specs)
- ]
- parsed_args = self.check_parser(self.cmd, arglist, verifylist)
-
- result = self.cmd.take_action(parsed_args)
-
- self.qos_mock.set_keys.assert_called_with(
- volume_fakes.qos_id,
- volume_fakes.qos_specs
- )
- self.assertIsNone(result)
-
-
-class TestQosShow(TestQos):
-
- def setUp(self):
- super(TestQosShow, self).setUp()
-
- self.qos_mock.get.return_value = fakes.FakeResource(
- None,
- copy.deepcopy(volume_fakes.QOS_WITH_ASSOCIATIONS),
- loaded=True,
- )
- self.qos_mock.get_associations.return_value = [fakes.FakeResource(
- None,
- copy.deepcopy(volume_fakes.qos_association),
- loaded=True,
- )]
-
- # Get the command object to test
- self.cmd = qos_specs.ShowQos(self.app, None)
-
- def test_qos_show(self):
- arglist = [
- volume_fakes.qos_id
- ]
- verifylist = [
- ('qos_spec', volume_fakes.qos_id)
- ]
-
- parsed_args = self.check_parser(self.cmd, arglist, verifylist)
-
- columns, data = self.cmd.take_action(parsed_args)
- self.qos_mock.get.assert_called_with(
- volume_fakes.qos_id
- )
-
- collist = (
- 'associations',
- 'consumer',
- 'id',
- 'name',
- 'specs'
- )
- self.assertEqual(collist, columns)
- datalist = (
- volume_fakes.type_name,
- volume_fakes.qos_consumer,
- volume_fakes.qos_id,
- volume_fakes.qos_name,
- utils.format_dict(volume_fakes.qos_specs),
- )
- self.assertEqual(datalist, tuple(data))
-
-
-class TestQosUnset(TestQos):
-
- def setUp(self):
- super(TestQosUnset, self).setUp()
-
- # Get the command object to test
- self.cmd = qos_specs.UnsetQos(self.app, None)
-
- def test_qos_unset_with_properties(self):
- self.qos_mock.get.return_value = fakes.FakeResource(
- None,
- copy.deepcopy(volume_fakes.QOS),
- loaded=True
- )
- arglist = [
- volume_fakes.qos_id,
- '--property', 'iops',
- '--property', 'foo'
- ]
-
- verifylist = [
- ('qos_spec', volume_fakes.qos_id),
- ('property', ['iops', 'foo'])
- ]
- parsed_args = self.check_parser(self.cmd, arglist, verifylist)
-
- result = self.cmd.take_action(parsed_args)
-
- self.qos_mock.unset_keys.assert_called_with(
- volume_fakes.qos_id,
- ['iops', 'foo']
- )
- self.assertIsNone(result)
diff --git a/openstackclient/tests/volume/v1/test_transfer_request.py b/openstackclient/tests/volume/v1/test_transfer_request.py
deleted file mode 100644
index 91f5f4b2..00000000
--- a/openstackclient/tests/volume/v1/test_transfer_request.py
+++ /dev/null
@@ -1,114 +0,0 @@
-#
-# Licensed under the Apache License, Version 2.0 (the "License"); you may
-# not use this file except in compliance with the License. You may obtain
-# a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
-# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
-# License for the specific language governing permissions and limitations
-# under the License.
-#
-
-
-from openstackclient.tests.volume.v1 import fakes as transfer_fakes
-from openstackclient.volume.v1 import volume_transfer_request
-
-
-class TestTransfer(transfer_fakes.TestVolumev1):
-
- def setUp(self):
- super(TestTransfer, self).setUp()
-
- # Get a shortcut to the TransferManager Mock
- self.transfer_mock = self.app.client_manager.volume.transfers
- self.transfer_mock.reset_mock()
-
-
-class TestTransferList(TestTransfer):
-
- # The Transfers to be listed
- volume_transfers = transfer_fakes.FakeTransfer.create_one_transfer()
-
- def setUp(self):
- super(TestTransferList, self).setUp()
-
- self.transfer_mock.list.return_value = [self.volume_transfers]
-
- # Get the command object to test
- self.cmd = volume_transfer_request.ListTransferRequests(self.app, None)
-
- def test_transfer_list_without_argument(self):
- arglist = []
- verifylist = []
- parsed_args = self.check_parser(self.cmd, arglist, verifylist)
-
- # In base command class Lister in cliff, abstract method take_action()
- # returns a tuple containing the column names and an iterable
- # containing the data to be listed.
- columns, data = self.cmd.take_action(parsed_args)
-
- expected_columns = [
- 'ID',
- 'Volume',
- 'Name',
- ]
-
- # confirming if all expected columns are present in the result.
- self.assertEqual(expected_columns, columns)
-
- datalist = ((
- self.volume_transfers.id,
- self.volume_transfers.volume_id,
- self.volume_transfers.name,
- ), )
-
- # confirming if all expected values are present in the result.
- self.assertEqual(datalist, tuple(data))
-
- # checking if proper call was made to list volume_transfers
- self.transfer_mock.list.assert_called_with(
- detailed=True,
- search_opts={'all_tenants': 0}
- )
-
- def test_transfer_list_with_argument(self):
- arglist = [
- "--all-projects"
- ]
- verifylist = [
- ("all_projects", True)
- ]
-
- parsed_args = self.check_parser(self.cmd, arglist, verifylist)
-
- # In base command class Lister in cliff, abstract method take_action()
- # returns a tuple containing the column names and an iterable
- # containing the data to be listed.
- columns, data = self.cmd.take_action(parsed_args)
-
- expected_columns = [
- 'ID',
- 'Volume',
- 'Name',
- ]
-
- # confirming if all expected columns are present in the result.
- self.assertEqual(expected_columns, columns)
-
- datalist = ((
- self.volume_transfers.id,
- self.volume_transfers.volume_id,
- self.volume_transfers.name,
- ), )
-
- # confirming if all expected values are present in the result.
- self.assertEqual(datalist, tuple(data))
-
- # checking if proper call was made to list volume_transfers
- self.transfer_mock.list.assert_called_with(
- detailed=True,
- search_opts={'all_tenants': 1}
- )
diff --git a/openstackclient/tests/volume/v1/test_volume.py b/openstackclient/tests/volume/v1/test_volume.py
deleted file mode 100644
index e41c2a72..00000000
--- a/openstackclient/tests/volume/v1/test_volume.py
+++ /dev/null
@@ -1,716 +0,0 @@
-# Copyright 2013 Nebula Inc.
-#
-# Licensed under the Apache License, Version 2.0 (the "License"); you may
-# not use this file except in compliance with the License. You may obtain
-# a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
-# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
-# License for the specific language governing permissions and limitations
-# under the License.
-#
-
-import copy
-import mock
-
-from openstackclient.tests import fakes
-from openstackclient.tests.identity.v2_0 import fakes as identity_fakes
-from openstackclient.tests.volume.v1 import fakes as volume_fakes
-from openstackclient.volume.v1 import volume
-
-
-class TestVolume(volume_fakes.TestVolumev1):
-
- def setUp(self):
- super(TestVolume, self).setUp()
-
- # Get a shortcut to the VolumeManager Mock
- self.volumes_mock = self.app.client_manager.volume.volumes
- self.volumes_mock.reset_mock()
-
- # Get a shortcut to the TenantManager Mock
- self.projects_mock = self.app.client_manager.identity.tenants
- self.projects_mock.reset_mock()
-
- # Get a shortcut to the UserManager Mock
- self.users_mock = self.app.client_manager.identity.users
- self.users_mock.reset_mock()
-
- # Get a shortcut to the ImageManager Mock
- self.images_mock = self.app.client_manager.image.images
- self.images_mock.reset_mock()
-
-
-# TODO(dtroyer): The volume create tests are incomplete, only the minimal
-# options and the options that require additional processing
-# are implemented at this time.
-
-class TestVolumeCreate(TestVolume):
-
- project = identity_fakes.FakeProject.create_one_project()
- user = identity_fakes.FakeUser.create_one_user()
-
- columns = (
- 'attach_status',
- 'availability_zone',
- 'display_description',
- 'display_name',
- 'id',
- 'properties',
- 'size',
- 'status',
- 'type',
- )
- datalist = (
- 'detached',
- volume_fakes.volume_zone,
- volume_fakes.volume_description,
- volume_fakes.volume_name,
- volume_fakes.volume_id,
- volume_fakes.volume_metadata_str,
- volume_fakes.volume_size,
- volume_fakes.volume_status,
- volume_fakes.volume_type,
- )
-
- def setUp(self):
- super(TestVolumeCreate, self).setUp()
-
- self.volumes_mock.create.return_value = fakes.FakeResource(
- None,
- copy.deepcopy(volume_fakes.VOLUME),
- loaded=True,
- )
-
- # Get the command object to test
- self.cmd = volume.CreateVolume(self.app, None)
-
- def test_volume_create_min_options(self):
- arglist = [
- '--size', str(volume_fakes.volume_size),
- volume_fakes.volume_name,
- ]
- verifylist = [
- ('size', volume_fakes.volume_size),
- ('name', volume_fakes.volume_name),
- ]
- parsed_args = self.check_parser(self.cmd, arglist, verifylist)
-
- # In base command class ShowOne in cliff, abstract method take_action()
- # returns a two-part tuple with a tuple of column names and a tuple of
- # data to be shown.
- columns, data = self.cmd.take_action(parsed_args)
-
- # VolumeManager.create(size, snapshot_id=, source_volid=,
- # display_name=, display_description=,
- # volume_type=, user_id=,
- # project_id=, availability_zone=,
- # metadata=, imageRef=)
- self.volumes_mock.create.assert_called_with(
- volume_fakes.volume_size,
- None,
- None,
- volume_fakes.volume_name,
- None,
- None,
- None,
- None,
- None,
- None,
- None,
- )
- self.assertEqual(self.columns, columns)
- self.assertEqual(self.datalist, data)
-
- def test_volume_create_options(self):
- arglist = [
- '--size', str(volume_fakes.volume_size),
- '--description', volume_fakes.volume_description,
- '--type', volume_fakes.volume_type,
- '--availability-zone', volume_fakes.volume_zone,
- volume_fakes.volume_name,
- ]
- verifylist = [
- ('size', volume_fakes.volume_size),
- ('description', volume_fakes.volume_description),
- ('type', volume_fakes.volume_type),
- ('availability_zone', volume_fakes.volume_zone),
- ('name', volume_fakes.volume_name),
- ]
- parsed_args = self.check_parser(self.cmd, arglist, verifylist)
-
- # In base command class ShowOne in cliff, abstract method take_action()
- # returns a two-part tuple with a tuple of column names and a tuple of
- # data to be shown.
- columns, data = self.cmd.take_action(parsed_args)
-
- # VolumeManager.create(size, snapshot_id=, source_volid=,
- # display_name=, display_description=,
- # volume_type=, user_id=,
- # project_id=, availability_zone=,
- # metadata=, imageRef=)
- self.volumes_mock.create.assert_called_with(
- volume_fakes.volume_size,
- None,
- None,
- volume_fakes.volume_name,
- volume_fakes.volume_description,
- volume_fakes.volume_type,
- None,
- None,
- volume_fakes.volume_zone,
- None,
- None,
- )
-
- self.assertEqual(self.columns, columns)
- self.assertEqual(self.datalist, data)
-
- def test_volume_create_user_project_id(self):
- # Return a project
- self.projects_mock.get.return_value = self.project
- # Return a user
- self.users_mock.get.return_value = self.user
-
- arglist = [
- '--size', str(volume_fakes.volume_size),
- '--project', self.project.id,
- '--user', self.user.id,
- volume_fakes.volume_name,
- ]
- verifylist = [
- ('size', volume_fakes.volume_size),
- ('project', self.project.id),
- ('user', self.user.id),
- ('name', volume_fakes.volume_name),
- ]
- parsed_args = self.check_parser(self.cmd, arglist, verifylist)
-
- # In base command class ShowOne in cliff, abstract method take_action()
- # returns a two-part tuple with a tuple of column names and a tuple of
- # data to be shown.
- columns, data = self.cmd.take_action(parsed_args)
-
- # VolumeManager.create(size, snapshot_id=, source_volid=,
- # display_name=, display_description=,
- # volume_type=, user_id=,
- # project_id=, availability_zone=,
- # metadata=, imageRef=)
- self.volumes_mock.create.assert_called_with(
- volume_fakes.volume_size,
- None,
- None,
- volume_fakes.volume_name,
- None,
- None,
- self.user.id,
- self.project.id,
- None,
- None,
- None,
- )
-
- self.assertEqual(self.columns, columns)
- self.assertEqual(self.datalist, data)
-
- def test_volume_create_user_project_name(self):
- # Return a project
- self.projects_mock.get.return_value = self.project
- # Return a user
- self.users_mock.get.return_value = self.user
-
- arglist = [
- '--size', str(volume_fakes.volume_size),
- '--project', self.project.name,
- '--user', self.user.name,
- volume_fakes.volume_name,
- ]
- verifylist = [
- ('size', volume_fakes.volume_size),
- ('project', self.project.name),
- ('user', self.user.name),
- ('name', volume_fakes.volume_name),
- ]
- parsed_args = self.check_parser(self.cmd, arglist, verifylist)
-
- # In base command class ShowOne in cliff, abstract method take_action()
- # returns a two-part tuple with a tuple of column names and a tuple of
- # data to be shown.
- columns, data = self.cmd.take_action(parsed_args)
-
- # VolumeManager.create(size, snapshot_id=, source_volid=,
- # display_name=, display_description=,
- # volume_type=, user_id=,
- # project_id=, availability_zone=,
- # metadata=, imageRef=)
- self.volumes_mock.create.assert_called_with(
- volume_fakes.volume_size,
- None,
- None,
- volume_fakes.volume_name,
- None,
- None,
- self.user.id,
- self.project.id,
- None,
- None,
- None,
- )
-
- self.assertEqual(self.columns, columns)
- self.assertEqual(self.datalist, data)
-
- def test_volume_create_properties(self):
- arglist = [
- '--property', 'Alpha=a',
- '--property', 'Beta=b',
- '--size', str(volume_fakes.volume_size),
- volume_fakes.volume_name,
- ]
- verifylist = [
- ('property', {'Alpha': 'a', 'Beta': 'b'}),
- ('size', volume_fakes.volume_size),
- ('name', volume_fakes.volume_name),
- ]
- parsed_args = self.check_parser(self.cmd, arglist, verifylist)
-
- # In base command class ShowOne in cliff, abstract method take_action()
- # returns a two-part tuple with a tuple of column names and a tuple of
- # data to be shown.
- columns, data = self.cmd.take_action(parsed_args)
-
- # VolumeManager.create(size, snapshot_id=, source_volid=,
- # display_name=, display_description=,
- # volume_type=, user_id=,
- # project_id=, availability_zone=,
- # metadata=, imageRef=)
- self.volumes_mock.create.assert_called_with(
- volume_fakes.volume_size,
- None,
- None,
- volume_fakes.volume_name,
- None,
- None,
- None,
- None,
- None,
- {'Alpha': 'a', 'Beta': 'b'},
- None,
- )
-
- self.assertEqual(self.columns, columns)
- self.assertEqual(self.datalist, data)
-
- def test_volume_create_image_id(self):
- self.images_mock.get.return_value = fakes.FakeResource(
- None,
- copy.deepcopy(volume_fakes.IMAGE),
- loaded=True,
- )
-
- arglist = [
- '--image', volume_fakes.image_id,
- '--size', str(volume_fakes.volume_size),
- volume_fakes.volume_name,
- ]
- verifylist = [
- ('image', volume_fakes.image_id),
- ('size', volume_fakes.volume_size),
- ('name', volume_fakes.volume_name),
- ]
- parsed_args = self.check_parser(self.cmd, arglist, verifylist)
-
- # In base command class ShowOne in cliff, abstract method take_action()
- # returns a two-part tuple with a tuple of column names and a tuple of
- # data to be shown.
- columns, data = self.cmd.take_action(parsed_args)
-
- # VolumeManager.create(size, snapshot_id=, source_volid=,
- # display_name=, display_description=,
- # volume_type=, user_id=,
- # project_id=, availability_zone=,
- # metadata=, imageRef=)
- self.volumes_mock.create.assert_called_with(
- volume_fakes.volume_size,
- None,
- None,
- volume_fakes.volume_name,
- None,
- None,
- None,
- None,
- None,
- None,
- volume_fakes.image_id,
- )
-
- self.assertEqual(self.columns, columns)
- self.assertEqual(self.datalist, data)
-
- def test_volume_create_image_name(self):
- self.images_mock.get.return_value = fakes.FakeResource(
- None,
- copy.deepcopy(volume_fakes.IMAGE),
- loaded=True,
- )
-
- arglist = [
- '--image', volume_fakes.image_name,
- '--size', str(volume_fakes.volume_size),
- volume_fakes.volume_name,
- ]
- verifylist = [
- ('image', volume_fakes.image_name),
- ('size', volume_fakes.volume_size),
- ('name', volume_fakes.volume_name),
- ]
- parsed_args = self.check_parser(self.cmd, arglist, verifylist)
-
- # In base command class ShowOne in cliff, abstract method take_action()
- # returns a two-part tuple with a tuple of column names and a tuple of
- # data to be shown.
- columns, data = self.cmd.take_action(parsed_args)
-
- # VolumeManager.create(size, snapshot_id=, source_volid=,
- # display_name=, display_description=,
- # volume_type=, user_id=,
- # project_id=, availability_zone=,
- # metadata=, imageRef=)
- self.volumes_mock.create.assert_called_with(
- volume_fakes.volume_size,
- None,
- None,
- volume_fakes.volume_name,
- None,
- None,
- None,
- None,
- None,
- None,
- volume_fakes.image_id,
- )
-
- self.assertEqual(self.columns, columns)
- self.assertEqual(self.datalist, data)
-
-
-class TestVolumeList(TestVolume):
-
- columns = (
- 'ID',
- 'Display Name',
- 'Status',
- 'Size',
- 'Attached to',
- )
- datalist = (
- (
- volume_fakes.volume_id,
- volume_fakes.volume_name,
- volume_fakes.volume_status,
- volume_fakes.volume_size,
- '',
- ),
- )
-
- def setUp(self):
- super(TestVolumeList, self).setUp()
-
- self.volumes_mock.list.return_value = [
- fakes.FakeResource(
- None,
- copy.deepcopy(volume_fakes.VOLUME),
- loaded=True,
- ),
- ]
-
- # Get the command object to test
- self.cmd = volume.ListVolume(self.app, None)
-
- def test_volume_list_no_options(self):
- arglist = []
- verifylist = [
- ('long', False),
- ('all_projects', False),
- ('name', None),
- ('status', None),
- ]
- parsed_args = self.check_parser(self.cmd, arglist, verifylist)
-
- columns, data = self.cmd.take_action(parsed_args)
-
- self.assertEqual(self.columns, columns)
- self.assertEqual(self.datalist, tuple(data))
-
- def test_volume_list_name(self):
- arglist = [
- '--name', volume_fakes.volume_name,
- ]
- verifylist = [
- ('long', False),
- ('all_projects', False),
- ('name', volume_fakes.volume_name),
- ('status', None),
- ]
- parsed_args = self.check_parser(self.cmd, arglist, verifylist)
-
- columns, data = self.cmd.take_action(parsed_args)
- self.assertEqual(self.columns, tuple(columns))
- self.assertEqual(self.datalist, tuple(data))
-
- def test_volume_list_status(self):
- arglist = [
- '--status', volume_fakes.volume_status,
- ]
- verifylist = [
- ('long', False),
- ('all_projects', False),
- ('name', None),
- ('status', volume_fakes.volume_status),
- ]
- parsed_args = self.check_parser(self.cmd, arglist, verifylist)
-
- columns, data = self.cmd.take_action(parsed_args)
- self.assertEqual(self.columns, tuple(columns))
- self.assertEqual(self.datalist, tuple(data))
-
- def test_volume_list_all_projects(self):
- arglist = [
- '--all-projects',
- ]
- verifylist = [
- ('long', False),
- ('all_projects', True),
- ('name', None),
- ('status', None),
- ]
- parsed_args = self.check_parser(self.cmd, arglist, verifylist)
-
- columns, data = self.cmd.take_action(parsed_args)
- self.assertEqual(self.columns, tuple(columns))
- self.assertEqual(self.datalist, tuple(data))
-
- def test_volume_list_long(self):
- arglist = [
- '--long',
- ]
- verifylist = [
- ('long', True),
- ('all_projects', False),
- ('name', None),
- ('status', None),
- ]
-
- parsed_args = self.check_parser(self.cmd, arglist, verifylist)
-
- columns, data = self.cmd.take_action(parsed_args)
-
- collist = (
- 'ID',
- 'Display Name',
- 'Status',
- 'Size',
- 'Type',
- 'Bootable',
- 'Attached to',
- 'Properties',
- )
- self.assertEqual(collist, columns)
-
- datalist = ((
- volume_fakes.volume_id,
- volume_fakes.volume_name,
- volume_fakes.volume_status,
- volume_fakes.volume_size,
- volume_fakes.volume_type,
- '',
- '',
- "Alpha='a', Beta='b', Gamma='g'",
- ), )
- self.assertEqual(datalist, tuple(data))
-
-
-class TestVolumeSet(TestVolume):
-
- def setUp(self):
- super(TestVolumeSet, self).setUp()
-
- self.volumes_mock.get.return_value = fakes.FakeResource(
- None,
- copy.deepcopy(volume_fakes.VOLUME),
- loaded=True,
- )
-
- self.volumes_mock.update.return_value = fakes.FakeResource(
- None,
- copy.deepcopy(volume_fakes.VOLUME),
- loaded=True,
- )
- # Get the command object to test
- self.cmd = volume.SetVolume(self.app, None)
-
- def test_volume_set_no_options(self):
- arglist = [
- volume_fakes.volume_name,
- ]
- verifylist = [
- ('name', None),
- ('description', None),
- ('size', None),
- ('property', None),
- ('volume', volume_fakes.volume_name),
- ]
- parsed_args = self.check_parser(self.cmd, arglist, verifylist)
-
- result = self.cmd.take_action(parsed_args)
- self.assertIsNone(result)
-
- def test_volume_set_name(self):
- arglist = [
- '--name', 'qwerty',
- volume_fakes.volume_name,
- ]
- verifylist = [
- ('name', 'qwerty'),
- ('description', None),
- ('size', None),
- ('property', None),
- ('volume', volume_fakes.volume_name),
- ]
- parsed_args = self.check_parser(self.cmd, arglist, verifylist)
-
- result = self.cmd.take_action(parsed_args)
-
- # Set expected values
- kwargs = {
- 'display_name': 'qwerty',
- }
- self.volumes_mock.update.assert_called_with(
- volume_fakes.volume_id,
- **kwargs
- )
- self.assertIsNone(result)
-
- def test_volume_set_description(self):
- arglist = [
- '--description', 'new desc',
- volume_fakes.volume_name,
- ]
- verifylist = [
- ('name', None),
- ('description', 'new desc'),
- ('size', None),
- ('property', None),
- ('volume', volume_fakes.volume_name),
- ]
- parsed_args = self.check_parser(self.cmd, arglist, verifylist)
-
- result = self.cmd.take_action(parsed_args)
-
- # Set expected values
- kwargs = {
- 'display_description': 'new desc',
- }
- self.volumes_mock.update.assert_called_with(
- volume_fakes.volume_id,
- **kwargs
- )
- self.assertIsNone(result)
-
- def test_volume_set_size(self):
- arglist = [
- '--size', '130',
- volume_fakes.volume_name,
- ]
- verifylist = [
- ('name', None),
- ('description', None),
- ('size', 130),
- ('property', None),
- ('volume', volume_fakes.volume_name),
- ]
- parsed_args = self.check_parser(self.cmd, arglist, verifylist)
-
- result = self.cmd.take_action(parsed_args)
-
- # Set expected values
- size = 130
- self.volumes_mock.extend.assert_called_with(
- volume_fakes.volume_id,
- size
- )
- self.assertIsNone(result)
-
- @mock.patch.object(volume.LOG, 'error')
- def test_volume_set_size_smaller(self, mock_log_error):
- arglist = [
- '--size', '100',
- volume_fakes.volume_name,
- ]
- verifylist = [
- ('name', None),
- ('description', None),
- ('size', 100),
- ('property', None),
- ('volume', volume_fakes.volume_name),
- ]
- parsed_args = self.check_parser(self.cmd, arglist, verifylist)
-
- result = self.cmd.take_action(parsed_args)
-
- mock_log_error.assert_called_with("New size must be greater "
- "than %s GB",
- volume_fakes.volume_size)
- self.assertIsNone(result)
-
- @mock.patch.object(volume.LOG, 'error')
- def test_volume_set_size_not_available(self, mock_log_error):
- self.volumes_mock.get.return_value.status = 'error'
- arglist = [
- '--size', '130',
- volume_fakes.volume_name,
- ]
- verifylist = [
- ('name', None),
- ('description', None),
- ('size', 130),
- ('property', None),
- ('volume', volume_fakes.volume_name),
- ]
- parsed_args = self.check_parser(self.cmd, arglist, verifylist)
-
- result = self.cmd.take_action(parsed_args)
-
- mock_log_error.assert_called_with("Volume is in %s state, it must be "
- "available before size can be "
- "extended", 'error')
- self.assertIsNone(result)
-
- def test_volume_set_property(self):
- arglist = [
- '--property', 'myprop=myvalue',
- volume_fakes.volume_name,
- ]
- verifylist = [
- ('name', None),
- ('description', None),
- ('size', None),
- ('property', {'myprop': 'myvalue'}),
- ('volume', volume_fakes.volume_name),
- ]
- parsed_args = self.check_parser(self.cmd, arglist, verifylist)
-
- result = self.cmd.take_action(parsed_args)
-
- # Set expected values
- metadata = {
- 'myprop': 'myvalue'
- }
- self.volumes_mock.set_metadata.assert_called_with(
- volume_fakes.volume_id,
- metadata
- )
- self.assertIsNone(result)