summaryrefslogtreecommitdiff
path: root/openstackclient/tests/unit
diff options
context:
space:
mode:
Diffstat (limited to 'openstackclient/tests/unit')
-rw-r--r--openstackclient/tests/unit/common/test_quota.py72
-rw-r--r--openstackclient/tests/unit/compute/v2/test_console.py28
-rw-r--r--openstackclient/tests/unit/compute/v2/test_server.py146
-rw-r--r--openstackclient/tests/unit/fakes.py3
-rw-r--r--openstackclient/tests/unit/network/v2/fakes.py67
-rw-r--r--openstackclient/tests/unit/network/v2/test_network_flavor.py407
-rw-r--r--openstackclient/tests/unit/volume/v1/fakes.py31
-rw-r--r--openstackclient/tests/unit/volume/v1/test_qos_specs.py21
-rw-r--r--openstackclient/tests/unit/volume/v1/test_type.py237
-rw-r--r--openstackclient/tests/unit/volume/v2/fakes.py31
-rw-r--r--openstackclient/tests/unit/volume/v2/test_type.py276
-rw-r--r--openstackclient/tests/unit/volume/v2/test_volume_host.py31
12 files changed, 1274 insertions, 76 deletions
diff --git a/openstackclient/tests/unit/common/test_quota.py b/openstackclient/tests/unit/common/test_quota.py
index 244d74d2..7dd23373 100644
--- a/openstackclient/tests/unit/common/test_quota.py
+++ b/openstackclient/tests/unit/common/test_quota.py
@@ -13,8 +13,6 @@
import copy
import mock
-from openstack.network.v2 import quota as _quota
-
from openstackclient.common import quota
from openstackclient.tests.unit.compute.v2 import fakes as compute_fakes
from openstackclient.tests.unit import fakes
@@ -284,32 +282,27 @@ class TestQuotaSet(TestQuota):
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
- # TODO(huanxuan): Remove this if condition once the fixed
- # SDK Quota class is the minimum required version.
- # This is expected to be SDK release 0.9.13
- if not hasattr(_quota.Quota, 'allow_get'):
- # Just run this when sdk <= 0.9.10
- result = self.cmd.take_action(parsed_args)
- kwargs = {
- 'subnet': network_fakes.QUOTA['subnet'],
- 'network': network_fakes.QUOTA['network'],
- 'floatingip': network_fakes.QUOTA['floatingip'],
- 'subnetpool': network_fakes.QUOTA['subnetpool'],
- 'security_group_rule':
- network_fakes.QUOTA['security_group_rule'],
- 'security_group': network_fakes.QUOTA['security_group'],
- 'router': network_fakes.QUOTA['router'],
- 'rbac_policy': network_fakes.QUOTA['rbac_policy'],
- 'port': network_fakes.QUOTA['port'],
- 'vip': network_fakes.QUOTA['vip'],
- 'healthmonitor': network_fakes.QUOTA['healthmonitor'],
- 'l7policy': network_fakes.QUOTA['l7policy'],
- }
- self.network_mock.update_quota.assert_called_once_with(
- identity_fakes.project_id,
- **kwargs
- )
- self.assertIsNone(result)
+ result = self.cmd.take_action(parsed_args)
+ kwargs = {
+ 'subnet': network_fakes.QUOTA['subnet'],
+ 'network': network_fakes.QUOTA['network'],
+ 'floatingip': network_fakes.QUOTA['floatingip'],
+ 'subnetpool': network_fakes.QUOTA['subnetpool'],
+ 'security_group_rule':
+ network_fakes.QUOTA['security_group_rule'],
+ 'security_group': network_fakes.QUOTA['security_group'],
+ 'router': network_fakes.QUOTA['router'],
+ 'rbac_policy': network_fakes.QUOTA['rbac_policy'],
+ 'port': network_fakes.QUOTA['port'],
+ 'vip': network_fakes.QUOTA['vip'],
+ 'healthmonitor': network_fakes.QUOTA['healthmonitor'],
+ 'l7policy': network_fakes.QUOTA['l7policy'],
+ }
+ self.network_mock.update_quota.assert_called_once_with(
+ identity_fakes.project_id,
+ **kwargs
+ )
+ self.assertIsNone(result)
def test_quota_set_with_class(self):
arglist = [
@@ -483,20 +476,15 @@ class TestQuotaShow(TestQuota):
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
- # TODO(huanxuan): Remove this if condition once the fixed
- # SDK QuotaDefault class is the minimum required version.
- # This is expected to be SDK release 0.9.13
- if not hasattr(_quota.QuotaDefault, 'project'):
- # Just run this when sdk <= 0.9.10
- self.cmd.take_action(parsed_args)
-
- self.quotas_mock.defaults.assert_called_once_with(
- identity_fakes.project_id)
- self.volume_quotas_mock.defaults.assert_called_once_with(
- identity_fakes.project_id)
- self.network.get_quota_default.assert_called_once_with(
- identity_fakes.project_id)
- self.assertNotCalled(self.network.get_quota)
+ self.cmd.take_action(parsed_args)
+
+ self.quotas_mock.defaults.assert_called_once_with(
+ identity_fakes.project_id)
+ self.volume_quotas_mock.defaults.assert_called_once_with(
+ identity_fakes.project_id)
+ self.network.get_quota_default.assert_called_once_with(
+ identity_fakes.project_id)
+ self.assertNotCalled(self.network.get_quota)
def test_quota_show_with_class(self):
arglist = [
diff --git a/openstackclient/tests/unit/compute/v2/test_console.py b/openstackclient/tests/unit/compute/v2/test_console.py
index d53d241e..3c708aae 100644
--- a/openstackclient/tests/unit/compute/v2/test_console.py
+++ b/openstackclient/tests/unit/compute/v2/test_console.py
@@ -35,11 +35,7 @@ class TestConsoleUrlShow(TestConsole):
'protocol': 'fake_protocol',
'type': 'fake_type'}}
methods = {
- 'get_vnc_console': fake_console_data,
- 'get_spice_console': fake_console_data,
- 'get_serial_console': fake_console_data,
- 'get_rdp_console': fake_console_data,
- 'get_mks_console': fake_console_data,
+ 'get_console_url': fake_console_data
}
self.fake_server = compute_fakes.FakeServer.create_one_server(
methods=methods)
@@ -68,7 +64,7 @@ class TestConsoleUrlShow(TestConsole):
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
columns, data = self.cmd.take_action(parsed_args)
- self.fake_server.get_vnc_console.assert_called_once_with('novnc')
+ self.fake_server.get_console_url.assert_called_once_with('novnc')
self.assertEqual(self.columns, columns)
self.assertEqual(self.data, data)
@@ -83,7 +79,7 @@ class TestConsoleUrlShow(TestConsole):
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
columns, data = self.cmd.take_action(parsed_args)
- self.fake_server.get_vnc_console.assert_called_once_with('novnc')
+ self.fake_server.get_console_url.assert_called_once_with('novnc')
self.assertEqual(self.columns, columns)
self.assertEqual(self.data, data)
@@ -98,7 +94,7 @@ class TestConsoleUrlShow(TestConsole):
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
columns, data = self.cmd.take_action(parsed_args)
- self.fake_server.get_vnc_console.assert_called_once_with('xvpvnc')
+ self.fake_server.get_console_url.assert_called_once_with('xvpvnc')
self.assertEqual(self.columns, columns)
self.assertEqual(self.data, data)
@@ -113,14 +109,14 @@ class TestConsoleUrlShow(TestConsole):
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
columns, data = self.cmd.take_action(parsed_args)
- self.fake_server.get_spice_console.assert_called_once_with(
+ self.fake_server.get_console_url.assert_called_once_with(
'spice-html5')
self.assertEqual(self.columns, columns)
self.assertEqual(self.data, data)
def test_console_url_show_compatible(self):
methods = {
- 'get_vnc_console': {'console': {'url': 'http://localhost',
+ 'get_console_url': {'console': {'url': 'http://localhost',
'type': 'fake_type'}},
}
old_fake_server = compute_fakes.FakeServer.create_one_server(
@@ -130,8 +126,8 @@ class TestConsoleUrlShow(TestConsole):
'url',
)
old_data = (
- methods['get_vnc_console']['console']['type'],
- methods['get_vnc_console']['console']['url']
+ methods['get_console_url']['console']['type'],
+ methods['get_console_url']['console']['url']
)
arglist = [
'foo_vm',
@@ -144,7 +140,7 @@ class TestConsoleUrlShow(TestConsole):
with mock.patch.object(self.servers_mock, 'get',
return_value=old_fake_server):
columns, data = self.cmd.take_action(parsed_args)
- old_fake_server.get_vnc_console.assert_called_once_with('novnc')
+ old_fake_server.get_console_url.assert_called_once_with('novnc')
self.assertEqual(old_columns, columns)
self.assertEqual(old_data, data)
@@ -159,7 +155,7 @@ class TestConsoleUrlShow(TestConsole):
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
columns, data = self.cmd.take_action(parsed_args)
- self.fake_server.get_rdp_console.assert_called_once_with(
+ self.fake_server.get_console_url.assert_called_once_with(
'rdp-html5')
self.assertEqual(self.columns, columns)
self.assertEqual(self.data, data)
@@ -175,7 +171,7 @@ class TestConsoleUrlShow(TestConsole):
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
columns, data = self.cmd.take_action(parsed_args)
- self.fake_server.get_serial_console.assert_called_once_with(
+ self.fake_server.get_console_url.assert_called_once_with(
'serial')
self.assertEqual(self.columns, columns)
self.assertEqual(self.data, data)
@@ -191,6 +187,6 @@ class TestConsoleUrlShow(TestConsole):
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
columns, data = self.cmd.take_action(parsed_args)
- self.fake_server.get_mks_console.assert_called_once_with()
+ self.fake_server.get_console_url.assert_called_once_with('webmks')
self.assertEqual(self.columns, columns)
self.assertEqual(self.data, data)
diff --git a/openstackclient/tests/unit/compute/v2/test_server.py b/openstackclient/tests/unit/compute/v2/test_server.py
index 4cac990e..249902bc 100644
--- a/openstackclient/tests/unit/compute/v2/test_server.py
+++ b/openstackclient/tests/unit/compute/v2/test_server.py
@@ -467,6 +467,152 @@ class TestServerCreate(TestServer):
self.assertEqual(self.columns, columns)
self.assertEqual(self.datalist(), data)
+ def test_server_create_with_auto_network(self):
+ arglist = [
+ '--image', 'image1',
+ '--flavor', 'flavor1',
+ '--nic', 'auto',
+ self.new_server.name,
+ ]
+ verifylist = [
+ ('image', 'image1'),
+ ('flavor', 'flavor1'),
+ ('nic', ['auto']),
+ ('config_drive', False),
+ ('server_name', self.new_server.name),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ columns, data = self.cmd.take_action(parsed_args)
+
+ # Set expected values
+ kwargs = dict(
+ meta=None,
+ files={},
+ reservation_id=None,
+ min_count=1,
+ max_count=1,
+ security_groups=[],
+ userdata=None,
+ key_name=None,
+ availability_zone=None,
+ block_device_mapping_v2=[],
+ nics='auto',
+ scheduler_hints={},
+ config_drive=None,
+ )
+ # ServerManager.create(name, image, flavor, **kwargs)
+ self.servers_mock.create.assert_called_with(
+ self.new_server.name,
+ self.image,
+ self.flavor,
+ **kwargs
+ )
+
+ self.assertEqual(self.columns, columns)
+ self.assertEqual(self.datalist(), data)
+
+ def test_server_create_with_none_network(self):
+ arglist = [
+ '--image', 'image1',
+ '--flavor', 'flavor1',
+ '--nic', 'none',
+ self.new_server.name,
+ ]
+ verifylist = [
+ ('image', 'image1'),
+ ('flavor', 'flavor1'),
+ ('nic', ['none']),
+ ('config_drive', False),
+ ('server_name', self.new_server.name),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ columns, data = self.cmd.take_action(parsed_args)
+
+ # Set expected values
+ kwargs = dict(
+ meta=None,
+ files={},
+ reservation_id=None,
+ min_count=1,
+ max_count=1,
+ security_groups=[],
+ userdata=None,
+ key_name=None,
+ availability_zone=None,
+ block_device_mapping_v2=[],
+ nics='none',
+ scheduler_hints={},
+ config_drive=None,
+ )
+ # ServerManager.create(name, image, flavor, **kwargs)
+ self.servers_mock.create.assert_called_with(
+ self.new_server.name,
+ self.image,
+ self.flavor,
+ **kwargs
+ )
+
+ self.assertEqual(self.columns, columns)
+ self.assertEqual(self.datalist(), data)
+
+ def test_server_create_with_conflict_network_options(self):
+ arglist = [
+ '--image', 'image1',
+ '--flavor', 'flavor1',
+ '--nic', 'none',
+ '--nic', 'auto',
+ '--nic', 'port-id=port1',
+ self.new_server.name,
+ ]
+ verifylist = [
+ ('image', 'image1'),
+ ('flavor', 'flavor1'),
+ ('nic', ['none', 'auto', 'port-id=port1']),
+ ('config_drive', False),
+ ('server_name', self.new_server.name),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ get_endpoints = mock.Mock()
+ get_endpoints.return_value = {'network': []}
+ self.app.client_manager.auth_ref = mock.Mock()
+ self.app.client_manager.auth_ref.service_catalog = mock.Mock()
+ self.app.client_manager.auth_ref.service_catalog.get_endpoints = (
+ get_endpoints)
+
+ find_port = mock.Mock()
+ network_client = self.app.client_manager.network
+ network_client.find_port = find_port
+ port_resource = mock.Mock()
+ port_resource.id = 'port1_uuid'
+ find_port.return_value = port_resource
+
+ self.assertRaises(exceptions.CommandError,
+ self.cmd.take_action, parsed_args)
+ self.assertNotCalled(self.servers_mock.create)
+
+ def test_server_create_with_invalid_network_options(self):
+ arglist = [
+ '--image', 'image1',
+ '--flavor', 'flavor1',
+ '--nic', 'abcdefgh',
+ self.new_server.name,
+ ]
+ verifylist = [
+ ('image', 'image1'),
+ ('flavor', 'flavor1'),
+ ('nic', ['abcdefgh']),
+ ('config_drive', False),
+ ('server_name', self.new_server.name),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ self.assertRaises(exceptions.CommandError,
+ self.cmd.take_action, parsed_args)
+ self.assertNotCalled(self.servers_mock.create)
+
@mock.patch.object(common_utils, 'wait_for_status', return_value=True)
def test_server_create_with_wait_ok(self, mock_wait_for_status):
arglist = [
diff --git a/openstackclient/tests/unit/fakes.py b/openstackclient/tests/unit/fakes.py
index 626b466d..f28f9103 100644
--- a/openstackclient/tests/unit/fakes.py
+++ b/openstackclient/tests/unit/fakes.py
@@ -161,6 +161,9 @@ class FakeModule(object):
def __init__(self, name, version):
self.name = name
self.__version__ = version
+ # Workaround for openstacksdk case
+ self.version = mock.Mock()
+ self.version.__version__ = version
class FakeResource(object):
diff --git a/openstackclient/tests/unit/network/v2/fakes.py b/openstackclient/tests/unit/network/v2/fakes.py
index 4b266efb..6a73b7e9 100644
--- a/openstackclient/tests/unit/network/v2/fakes.py
+++ b/openstackclient/tests/unit/network/v2/fakes.py
@@ -71,6 +71,10 @@ class TestNetworkV2(utils.TestCommand):
token=fakes.AUTH_TOKEN,
)
+ self.app.client_manager.sdk_connection = mock.Mock()
+ self.app.client_manager.sdk_connection.network = \
+ self.app.client_manager.network
+
self.app.client_manager.identity = (
identity_fakes_v3.FakeIdentityv3Client(
endpoint=fakes.AUTH_URL,
@@ -386,6 +390,69 @@ class FakeNetwork(object):
return mock.Mock(side_effect=networks)
+class FakeNetworkFlavor(object):
+ """Fake Network Flavor."""
+
+ @staticmethod
+ def create_one_network_flavor(attrs=None):
+ """Create a fake network flavor.
+
+ :param Dictionary attrs:
+ A dictionary with all attributes
+ :return:
+ A FakeResource object faking the network flavor
+ """
+ attrs = attrs or {}
+
+ fake_uuid = uuid.uuid4().hex
+ network_flavor_attrs = {
+ 'description': 'network-flavor-description-' + fake_uuid,
+ 'enabled': True,
+ 'id': 'network-flavor-id-' + fake_uuid,
+ 'name': 'network-flavor-name-' + fake_uuid,
+ 'service_type': 'vpn',
+ 'tenant_id': 'project-id-' + uuid.uuid4().hex,
+ }
+
+ # Overwrite default attributes.
+ network_flavor_attrs.update(attrs)
+
+ network_flavor = fakes.FakeResource(
+ info=copy.deepcopy(network_flavor_attrs),
+ loaded=True
+ )
+
+ network_flavor.project_id = network_flavor_attrs['tenant_id']
+ network_flavor.is_enabled = network_flavor_attrs['enabled']
+
+ return network_flavor
+
+ @staticmethod
+ def create_flavor(attrs=None, count=2):
+ """Create multiple fake network flavors.
+
+ :param Dictionary attrs:
+ A dictionary with all attributes
+ :param int count:
+ The number of network flavors to fake
+ :return:
+ A list of FakeResource objects faking the network falvors
+ """
+ network_flavors = []
+ for i in range(0, count):
+ network_flavors.append(
+ FakeNetworkFlavor.create_one_network_flavor(attrs)
+ )
+ return network_flavors
+
+ @staticmethod
+ def get_flavor(network_flavors=None, count=2):
+ """Get a list of flavors."""
+ if network_flavors is None:
+ network_flavors = (FakeNetworkFlavor.create_flavor(count))
+ return mock.Mock(side_effect=network_flavors)
+
+
class FakeNetworkSegment(object):
"""Fake one or more network segments."""
diff --git a/openstackclient/tests/unit/network/v2/test_network_flavor.py b/openstackclient/tests/unit/network/v2/test_network_flavor.py
new file mode 100644
index 00000000..11e27841
--- /dev/null
+++ b/openstackclient/tests/unit/network/v2/test_network_flavor.py
@@ -0,0 +1,407 @@
+# Copyright (c) 2016, Intel Corporation.
+# All Rights Reserved.
+#
+# 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 openstackclient.network.v2 import network_flavor
+from openstackclient.tests.unit.identity.v3 import fakes as identity_fakes_v3
+from openstackclient.tests.unit.network.v2 import fakes as network_fakes
+from openstackclient.tests.unit import utils as tests_utils
+
+
+class TestNetworkFlavor(network_fakes.TestNetworkV2):
+
+ def setUp(self):
+ super(TestNetworkFlavor, self).setUp()
+
+ # Get a shortcut to the network client
+ self.network = self.app.client_manager.network
+ # Get a shortcut to the ProjectManager Mock
+ self.projects_mock = self.app.client_manager.identity.projects
+ # Get a shortcut to the DomainManager Mock
+ self.domains_mock = self.app.client_manager.identity.domains
+
+
+class TestCreateNetworkFlavor(TestNetworkFlavor):
+
+ project = identity_fakes_v3.FakeProject.create_one_project()
+ domain = identity_fakes_v3.FakeDomain.create_one_domain()
+ # The new network flavor created.
+ new_network_flavor = (
+ network_fakes.FakeNetworkFlavor.create_one_network_flavor())
+ columns = (
+ 'description',
+ 'enabled',
+ 'id',
+ 'name',
+ 'project_id',
+ 'service_type'
+ )
+ data = (
+ new_network_flavor.description,
+ new_network_flavor.enabled,
+ new_network_flavor.id,
+ new_network_flavor.name,
+ new_network_flavor.project_id,
+ new_network_flavor.service_type,
+ )
+
+ def setUp(self):
+ super(TestCreateNetworkFlavor, self).setUp()
+ self.network.create_flavor = mock.Mock(
+ return_value=self.new_network_flavor)
+
+ # Get the command object to test
+ self.cmd = network_flavor.CreateNetworkFlavor(self.app, self.namespace)
+
+ self.projects_mock.get.return_value = self.project
+ self.domains_mock.get.return_value = self.domain
+
+ def test_create_no_options(self):
+ arglist = []
+ verifylist = []
+
+ # Missing required args should bail here
+ self.assertRaises(tests_utils.ParserException, self.check_parser,
+ self.cmd, arglist, verifylist)
+
+ def test_create_default_options(self):
+ arglist = [
+ '--service-type', self.new_network_flavor.service_type,
+ self.new_network_flavor.name,
+ ]
+ verifylist = [
+ ('service_type', self.new_network_flavor.service_type),
+ ('name', self.new_network_flavor.name),
+ ]
+
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+ columns, data = (self.cmd.take_action(parsed_args))
+
+ self.network.create_flavor.assert_called_once_with(**{
+ 'service_type': self.new_network_flavor.service_type,
+ 'name': self.new_network_flavor.name,
+ })
+ self.assertEqual(self.columns, columns)
+ self.assertEqual(self.data, data)
+
+ def test_create_all_options(self):
+ arglist = [
+ '--description', self.new_network_flavor.description,
+ '--enable',
+ '--project', self.new_network_flavor.project_id,
+ '--project-domain', self.domain.name,
+ '--service-type', self.new_network_flavor.service_type,
+ self.new_network_flavor.name,
+ ]
+ verifylist = [
+ ('description', self.new_network_flavor.description),
+ ('enable', True),
+ ('project', self.new_network_flavor.project_id),
+ ('project_domain', self.domain.name),
+ ('service_type', self.new_network_flavor.service_type),
+ ('name', self.new_network_flavor.name),
+ ]
+
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+ columns, data = (self.cmd.take_action(parsed_args))
+
+ self.network.create_flavor.assert_called_once_with(**{
+ 'description': self.new_network_flavor.description,
+ 'enabled': True,
+ 'tenant_id': self.project.id,
+ 'service_type': self.new_network_flavor.service_type,
+ 'name': self.new_network_flavor.name,
+ })
+ self.assertEqual(self.columns, columns)
+ self.assertEqual(self.data, data)
+
+ def test_create_disable(self):
+ arglist = [
+ '--disable',
+ '--service-type', self.new_network_flavor.service_type,
+ self.new_network_flavor.name,
+ ]
+ verifylist = [
+ ('disable', True),
+ ('service_type', self.new_network_flavor.service_type),
+ ('name', self.new_network_flavor.name),
+ ]
+
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+ columns, data = self.cmd.take_action(parsed_args)
+
+ self.network.create_flavor.assert_called_once_with(**{
+ 'enabled': False,
+ 'service_type': self.new_network_flavor.service_type,
+ 'name': self.new_network_flavor.name,
+ })
+ self.assertEqual(self.columns, columns)
+ self.assertEqual(self.data, data)
+
+
+class TestDeleteNetworkFlavor(TestNetworkFlavor):
+
+ # The network flavor to delete.
+ _network_flavors = (
+ network_fakes.FakeNetworkFlavor.create_flavor(count=2))
+
+ def setUp(self):
+ super(TestDeleteNetworkFlavor, self).setUp()
+ self.network.delete_flavor = mock.Mock(return_value=None)
+ self.network.find_flavor = (
+ network_fakes.FakeNetworkFlavor.get_flavor(
+ network_flavors=self._network_flavors)
+ )
+
+ # Get the command object to test
+ self.cmd = network_flavor.DeleteNetworkFlavor(self.app, self.namespace)
+
+ def test_network_flavor_delete(self):
+ arglist = [
+ self._network_flavors[0].name,
+ ]
+ verifylist = [
+ ('flavor', [self._network_flavors[0].name]),
+ ]
+
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ result = self.cmd.take_action(parsed_args)
+ self.network.find_flavor.assert_called_once_with(
+ self._network_flavors[0].name, ignore_missing=False)
+ self.network.delete_flavor.assert_called_once_with(
+ self._network_flavors[0])
+ self.assertIsNone(result)
+
+ def test_multi_network_flavors_delete(self):
+ arglist = []
+ verifylist = []
+
+ for a in self._network_flavors:
+ arglist.append(a.name)
+ verifylist = [
+ ('flavor', arglist),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ result = self.cmd.take_action(parsed_args)
+
+ calls = []
+ for a in self._network_flavors:
+ calls.append(mock.call(a))
+ self.network.delete_flavor.assert_has_calls(calls)
+ self.assertIsNone(result)
+
+ def test_multi_network_flavors_delete_with_exception(self):
+ arglist = [
+ self._network_flavors[0].name,
+ 'unexist_network_flavor',
+ ]
+ verifylist = [
+ ('flavor',
+ [self._network_flavors[0].name, 'unexist_network_flavor']),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ find_mock_result = [self._network_flavors[0], exceptions.CommandError]
+ self.network.find_flavor = (
+ mock.Mock(side_effect=find_mock_result)
+ )
+
+ try:
+ self.cmd.take_action(parsed_args)
+ self.fail('CommandError should be raised.')
+ except exceptions.CommandError as e:
+ self.assertEqual('1 of 2 flavors failed to delete.', str(e))
+
+ self.network.find_flavor.assert_any_call(
+ self._network_flavors[0].name, ignore_missing=False)
+ self.network.find_flavor.assert_any_call(
+ 'unexist_network_flavor', ignore_missing=False)
+ self.network.delete_flavor.assert_called_once_with(
+ self._network_flavors[0]
+ )
+
+
+class TestListNetworkFlavor(TestNetworkFlavor):
+
+ # The network flavors to list up.
+ _network_flavors = (
+ network_fakes.FakeNetworkFlavor.create_flavor(count=2))
+ columns = (
+ 'ID',
+ 'Name',
+ 'Enabled',
+ 'Service Type',
+ 'Description',
+ )
+ data = []
+ for flavor in _network_flavors:
+ data.append((
+ flavor.id,
+ flavor.name,
+ flavor.enabled,
+ flavor.service_type,
+ flavor.description,
+ ))
+
+ def setUp(self):
+ super(TestListNetworkFlavor, self).setUp()
+ self.network.flavors = mock.Mock(
+ return_value=self._network_flavors)
+
+ # Get the command object to test
+ self.cmd = network_flavor.ListNetworkFlavor(self.app, self.namespace)
+
+ def test_network_flavor_list(self):
+ arglist = []
+ verifylist = []
+
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+ columns, data = self.cmd.take_action(parsed_args)
+
+ self.network.flavors.assert_called_once_with(**{})
+ self.assertEqual(self.columns, columns)
+ self.assertEqual(self.data, list(data))
+
+
+class TestShowNetworkFlavor(TestNetworkFlavor):
+
+ # The network flavor to show.
+ new_network_flavor = (
+ network_fakes.FakeNetworkFlavor.create_one_network_flavor())
+ columns = (
+ 'description',
+ 'enabled',
+ 'id',
+ 'name',
+ 'project_id',
+ 'service_type'
+ )
+ data = (
+ new_network_flavor.description,
+ new_network_flavor.enabled,
+ new_network_flavor.id,
+ new_network_flavor.name,
+ new_network_flavor.project_id,
+ new_network_flavor.service_type,
+ )
+
+ def setUp(self):
+ super(TestShowNetworkFlavor, self).setUp()
+ self.network.find_flavor = mock.Mock(
+ return_value=self.new_network_flavor)
+
+ # Get the command object to test
+ self.cmd = network_flavor.ShowNetworkFlavor(self.app, self.namespace)
+
+ def test_show_no_options(self):
+ arglist = []
+ verifylist = []
+
+ # Missing required args should bail here
+ self.assertRaises(tests_utils.ParserException, self.check_parser,
+ self.cmd, arglist, verifylist)
+
+ def test_show_all_options(self):
+ arglist = [
+ self.new_network_flavor.name,
+ ]
+ verifylist = [
+ ('flavor', self.new_network_flavor.name),
+ ]
+
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+ columns, data = self.cmd.take_action(parsed_args)
+
+ self.network.find_flavor.assert_called_once_with(
+ self.new_network_flavor.name, ignore_missing=False)
+ self.assertEqual(self.columns, columns)
+ self.assertEqual(self.data, data)
+
+
+class TestSetNetworkFlavor(TestNetworkFlavor):
+
+ # The network flavor to set.
+ new_network_flavor = (
+ network_fakes.FakeNetworkFlavor.create_one_network_flavor())
+
+ def setUp(self):
+ super(TestSetNetworkFlavor, self).setUp()
+ self.network.update_flavor = mock.Mock(return_value=None)
+ self.network.find_flavor = mock.Mock(
+ return_value=self.new_network_flavor)
+
+ # Get the command object to test
+ self.cmd = network_flavor.SetNetworkFlavor(self.app, self.namespace)
+
+ def test_set_nothing(self):
+ arglist = [self.new_network_flavor.name, ]
+ verifylist = [
+ ('flavor', self.new_network_flavor.name),
+ ]
+
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+ result = self.cmd.take_action(parsed_args)
+
+ attrs = {}
+ self.network.update_flavor.assert_called_with(
+ self.new_network_flavor, **attrs)
+ self.assertIsNone(result)
+
+ def test_set_name_and_enable(self):
+ arglist = [
+ '--name', 'new_network_flavor',
+ '--enable',
+ self.new_network_flavor.name,
+ ]
+ verifylist = [
+ ('name', 'new_network_flavor'),
+ ('enable', True),
+ ('flavor', self.new_network_flavor.name),
+ ]
+
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+ result = self.cmd.take_action(parsed_args)
+ attrs = {
+ 'name': "new_network_flavor",
+ 'enabled': True,
+ }
+ self.network.update_flavor.assert_called_with(
+ self.new_network_flavor, **attrs)
+ self.assertIsNone(result)
+
+ def test_set_disable(self):
+ arglist = [
+ '--disable',
+ self.new_network_flavor.name,
+ ]
+ verifylist = [
+ ('disable', True),
+ ('flavor', self.new_network_flavor.name),
+ ]
+
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+ result = self.cmd.take_action(parsed_args)
+ attrs = {
+ 'enabled': False,
+ }
+ self.network.update_flavor.assert_called_with(
+ self.new_network_flavor, **attrs)
+ self.assertIsNone(result)
diff --git a/openstackclient/tests/unit/volume/v1/fakes.py b/openstackclient/tests/unit/volume/v1/fakes.py
index 78a8227e..fff5181d 100644
--- a/openstackclient/tests/unit/volume/v1/fakes.py
+++ b/openstackclient/tests/unit/volume/v1/fakes.py
@@ -364,6 +364,9 @@ class FakeVolumev1Client(object):
self.qos_specs.resource_class = fakes.FakeResource(None, {})
self.volume_types = mock.Mock()
self.volume_types.resource_class = fakes.FakeResource(None, {})
+ self.volume_encryption_types = mock.Mock()
+ self.volume_encryption_types.resource_class = (
+ fakes.FakeResource(None, {}))
self.transfers = mock.Mock()
self.transfers.resource_class = fakes.FakeResource(None, {})
self.volume_snapshots = mock.Mock()
@@ -470,6 +473,34 @@ class FakeType(object):
return mock.Mock(side_effect=types)
+ @staticmethod
+ def create_one_encryption_type(attrs=None):
+ """Create a fake encryption type.
+
+ :param Dictionary attrs:
+ A dictionary with all attributes
+ :return:
+ A FakeResource object with volume_type_id etc.
+ """
+ attrs = attrs or {}
+
+ # Set default attributes.
+ encryption_info = {
+ "volume_type_id": 'type-id-' + uuid.uuid4().hex,
+ 'provider': 'LuksEncryptor',
+ 'cipher': None,
+ 'key_size': None,
+ 'control_location': 'front-end',
+ }
+
+ # Overwrite default attributes.
+ encryption_info.update(attrs)
+
+ encryption_type = fakes.FakeResource(
+ info=copy.deepcopy(encryption_info),
+ loaded=True)
+ return encryption_type
+
class FakeSnapshot(object):
"""Fake one or more snapshot."""
diff --git a/openstackclient/tests/unit/volume/v1/test_qos_specs.py b/openstackclient/tests/unit/volume/v1/test_qos_specs.py
index 464038e7..e3dc1e78 100644
--- a/openstackclient/tests/unit/volume/v1/test_qos_specs.py
+++ b/openstackclient/tests/unit/volume/v1/test_qos_specs.py
@@ -70,23 +70,22 @@ class TestQosAssociate(TestQos):
class TestQosCreate(TestQos):
- new_qos_spec = volume_fakes.FakeQos.create_one_qos()
columns = (
'consumer',
'id',
'name',
- 'specs'
- )
- datalist = (
- new_qos_spec.consumer,
- new_qos_spec.id,
- new_qos_spec.name,
- new_qos_spec.specs
+ 'properties'
)
def setUp(self):
super(TestQosCreate, self).setUp()
-
+ self.new_qos_spec = volume_fakes.FakeQos.create_one_qos()
+ self.datalist = (
+ self.new_qos_spec.consumer,
+ self.new_qos_spec.id,
+ self.new_qos_spec.name,
+ utils.format_dict(self.new_qos_spec.specs)
+ )
self.qos_mock.create.return_value = self.new_qos_spec
# Get the command object to test
self.cmd = qos_specs.CreateQos(self.app, None)
@@ -336,7 +335,7 @@ class TestQosList(TestQos):
'Name',
'Consumer',
'Associations',
- 'Specs',
+ 'Properties',
)
self.assertEqual(collist, columns)
datalist = ((
@@ -413,7 +412,7 @@ class TestQosShow(TestQos):
'consumer',
'id',
'name',
- 'specs'
+ 'properties'
)
self.assertEqual(collist, columns)
datalist = (
diff --git a/openstackclient/tests/unit/volume/v1/test_type.py b/openstackclient/tests/unit/volume/v1/test_type.py
index 81ad8301..dcdd3d56 100644
--- a/openstackclient/tests/unit/volume/v1/test_type.py
+++ b/openstackclient/tests/unit/volume/v1/test_type.py
@@ -31,6 +31,10 @@ class TestType(volume_fakes.TestVolumev1):
self.types_mock = self.app.client_manager.volume.volume_types
self.types_mock.reset_mock()
+ self.encryption_types_mock = (
+ self.app.client_manager.volume.volume_encryption_types)
+ self.encryption_types_mock.reset_mock()
+
class TestTypeCreate(TestType):
@@ -75,6 +79,67 @@ class TestTypeCreate(TestType):
self.assertEqual(self.columns, columns)
self.assertEqual(self.data, data)
+ def test_type_create_with_encryption(self):
+ encryption_info = {
+ 'provider': 'LuksEncryptor',
+ 'cipher': 'aes-xts-plain64',
+ 'key_size': '128',
+ 'control_location': 'front-end',
+ }
+ encryption_type = volume_fakes.FakeType.create_one_encryption_type(
+ attrs=encryption_info
+ )
+ self.new_volume_type = volume_fakes.FakeType.create_one_type(
+ attrs={'encryption': encryption_info})
+ self.types_mock.create.return_value = self.new_volume_type
+ self.encryption_types_mock.create.return_value = encryption_type
+ encryption_columns = (
+ 'description',
+ 'encryption',
+ 'id',
+ 'is_public',
+ 'name',
+ )
+ encryption_data = (
+ self.new_volume_type.description,
+ utils.format_dict(encryption_info),
+ self.new_volume_type.id,
+ True,
+ self.new_volume_type.name,
+ )
+ arglist = [
+ '--encryption-provider', 'LuksEncryptor',
+ '--encryption-cipher', 'aes-xts-plain64',
+ '--encryption-key-size', '128',
+ '--encryption-control-location', 'front-end',
+ self.new_volume_type.name,
+ ]
+ verifylist = [
+ ('encryption_provider', 'LuksEncryptor'),
+ ('encryption_cipher', 'aes-xts-plain64'),
+ ('encryption_key_size', 128),
+ ('encryption_control_location', 'front-end'),
+ ('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,
+ )
+ body = {
+ 'provider': 'LuksEncryptor',
+ 'cipher': 'aes-xts-plain64',
+ 'key_size': 128,
+ 'control_location': 'front-end',
+ }
+ self.encryption_types_mock.create.assert_called_with(
+ self.new_volume_type,
+ body,
+ )
+ self.assertEqual(encryption_columns, columns)
+ self.assertEqual(encryption_data, data)
+
class TestTypeDelete(TestType):
@@ -156,17 +221,17 @@ class TestTypeList(TestType):
volume_types = volume_fakes.FakeType.create_types()
- columns = (
+ columns = [
"ID",
"Name",
"Is Public",
- )
- columns_long = (
+ ]
+ columns_long = [
"ID",
"Name",
"Is Public",
"Properties"
- )
+ ]
data = []
for t in volume_types:
@@ -188,6 +253,8 @@ class TestTypeList(TestType):
super(TestTypeList, self).setUp()
self.types_mock.list.return_value = self.volume_types
+ self.encryption_types_mock.create.return_value = None
+ self.encryption_types_mock.update.return_value = None
# get the command to test
self.cmd = volume_type.ListVolumeType(self.app, None)
@@ -195,6 +262,7 @@ class TestTypeList(TestType):
arglist = []
verifylist = [
("long", False),
+ ("encryption_type", False),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
@@ -217,6 +285,47 @@ class TestTypeList(TestType):
self.assertEqual(self.columns_long, columns)
self.assertEqual(self.data_long, list(data))
+ def test_type_list_with_encryption(self):
+ encryption_type = volume_fakes.FakeType.create_one_encryption_type(
+ attrs={'volume_type_id': self.volume_types[0].id})
+ encryption_info = {
+ 'provider': 'LuksEncryptor',
+ 'cipher': None,
+ 'key_size': None,
+ 'control_location': 'front-end',
+ }
+ encryption_columns = self.columns + [
+ "Encryption",
+ ]
+ encryption_data = []
+ encryption_data.append((
+ self.volume_types[0].id,
+ self.volume_types[0].name,
+ self.volume_types[0].is_public,
+ utils.format_dict(encryption_info),
+ ))
+ encryption_data.append((
+ self.volume_types[1].id,
+ self.volume_types[1].name,
+ self.volume_types[1].is_public,
+ '-',
+ ))
+
+ self.encryption_types_mock.list.return_value = [encryption_type]
+ arglist = [
+ "--encryption-type",
+ ]
+ verifylist = [
+ ("encryption_type", True),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ columns, data = self.cmd.take_action(parsed_args)
+ self.encryption_types_mock.list.assert_called_once_with()
+ self.types_mock.list.assert_called_once_with()
+ self.assertEqual(encryption_columns, columns)
+ self.assertEqual(encryption_data, list(data))
+
class TestTypeSet(TestType):
@@ -260,6 +369,60 @@ class TestTypeSet(TestType):
{'myprop': 'myvalue'})
self.assertIsNone(result)
+ def test_type_set_new_encryption(self):
+ arglist = [
+ '--encryption-provider', 'LuksEncryptor',
+ '--encryption-cipher', 'aes-xts-plain64',
+ '--encryption-key-size', '128',
+ '--encryption-control-location', 'front-end',
+ self.volume_type.id,
+ ]
+ verifylist = [
+ ('encryption_provider', 'LuksEncryptor'),
+ ('encryption_cipher', 'aes-xts-plain64'),
+ ('encryption_key_size', 128),
+ ('encryption_control_location', 'front-end'),
+ ('volume_type', self.volume_type.id),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ result = self.cmd.take_action(parsed_args)
+ body = {
+ 'provider': 'LuksEncryptor',
+ 'cipher': 'aes-xts-plain64',
+ 'key_size': 128,
+ 'control_location': 'front-end',
+ }
+ self.encryption_types_mock.create.assert_called_with(
+ self.volume_type,
+ body,
+ )
+ self.assertIsNone(result)
+
+ def test_type_set_new_encryption_without_provider(self):
+ arglist = [
+ '--encryption-cipher', 'aes-xts-plain64',
+ '--encryption-key-size', '128',
+ '--encryption-control-location', 'front-end',
+ self.volume_type.id,
+ ]
+ verifylist = [
+ ('encryption_cipher', 'aes-xts-plain64'),
+ ('encryption_key_size', 128),
+ ('encryption_control_location', 'front-end'),
+ ('volume_type', self.volume_type.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("Command Failed: One or more of"
+ " the operations failed",
+ str(e))
+ self.encryption_types_mock.create.assert_not_called()
+ self.encryption_types_mock.update.assert_not_called()
+
class TestTypeShow(TestType):
@@ -293,7 +456,8 @@ class TestTypeShow(TestType):
self.volume_type.id
]
verifylist = [
- ("volume_type", self.volume_type.id)
+ ("volume_type", self.volume_type.id),
+ ("encryption_type", False),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
@@ -303,6 +467,50 @@ class TestTypeShow(TestType):
self.assertEqual(self.columns, columns)
self.assertEqual(self.data, data)
+ def test_type_show_with_encryption(self):
+ encryption_type = volume_fakes.FakeType.create_one_encryption_type()
+ encryption_info = {
+ 'provider': 'LuksEncryptor',
+ 'cipher': None,
+ 'key_size': None,
+ 'control_location': 'front-end',
+ }
+ self.volume_type = volume_fakes.FakeType.create_one_type(
+ attrs={'encryption': encryption_info})
+ self.types_mock.get.return_value = self.volume_type
+ self.encryption_types_mock.get.return_value = encryption_type
+ encryption_columns = (
+ 'description',
+ 'encryption',
+ 'id',
+ 'is_public',
+ 'name',
+ 'properties',
+ )
+ encryption_data = (
+ self.volume_type.description,
+ utils.format_dict(encryption_info),
+ self.volume_type.id,
+ True,
+ self.volume_type.name,
+ utils.format_dict(self.volume_type.extra_specs)
+ )
+ arglist = [
+ '--encryption-type',
+ self.volume_type.id
+ ]
+ verifylist = [
+ ('encryption_type', True),
+ ("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.encryption_types_mock.get.assert_called_with(self.volume_type.id)
+ self.assertEqual(encryption_columns, columns)
+ self.assertEqual(encryption_data, data)
+
class TestTypeUnset(TestType):
@@ -317,13 +525,14 @@ class TestTypeUnset(TestType):
# Get the command object to test
self.cmd = volume_type.UnsetVolumeType(self.app, None)
- def test_type_unset(self):
+ def test_type_unset_property(self):
arglist = [
'--property', 'property',
'--property', 'multi_property',
self.volume_type.id,
]
verifylist = [
+ ('encryption_type', False),
('property', ['property', 'multi_property']),
('volume_type', self.volume_type.id),
]
@@ -333,6 +542,7 @@ class TestTypeUnset(TestType):
result = self.cmd.take_action(parsed_args)
self.volume_type.unset_keys.assert_called_once_with(
['property', 'multi_property'])
+ self.encryption_types_mock.delete.assert_not_called()
self.assertIsNone(result)
def test_type_unset_failed_with_missing_volume_type_argument(self):
@@ -362,3 +572,18 @@ class TestTypeUnset(TestType):
result = self.cmd.take_action(parsed_args)
self.assertIsNone(result)
+
+ def test_type_unset_encryption_type(self):
+ arglist = [
+ '--encryption-type',
+ self.volume_type.id,
+ ]
+ verifylist = [
+ ('encryption_type', True),
+ ('volume_type', self.volume_type.id),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ result = self.cmd.take_action(parsed_args)
+ self.encryption_types_mock.delete.assert_called_with(self.volume_type)
+ self.assertIsNone(result)
diff --git a/openstackclient/tests/unit/volume/v2/fakes.py b/openstackclient/tests/unit/volume/v2/fakes.py
index a6676403..d54faec7 100644
--- a/openstackclient/tests/unit/volume/v2/fakes.py
+++ b/openstackclient/tests/unit/volume/v2/fakes.py
@@ -208,6 +208,9 @@ class FakeVolumeClient(object):
self.volume_types.resource_class = fakes.FakeResource(None, {})
self.volume_type_access = mock.Mock()
self.volume_type_access.resource_class = fakes.FakeResource(None, {})
+ self.volume_encryption_types = mock.Mock()
+ self.volume_encryption_types.resource_class = (
+ fakes.FakeResource(None, {}))
self.restores = mock.Mock()
self.restores.resource_class = fakes.FakeResource(None, {})
self.qos_specs = mock.Mock()
@@ -923,3 +926,31 @@ class FakeType(object):
types = FakeType.create_types(count)
return mock.Mock(side_effect=types)
+
+ @staticmethod
+ def create_one_encryption_type(attrs=None):
+ """Create a fake encryption type.
+
+ :param Dictionary attrs:
+ A dictionary with all attributes
+ :return:
+ A FakeResource object with volume_type_id etc.
+ """
+ attrs = attrs or {}
+
+ # Set default attributes.
+ encryption_info = {
+ "volume_type_id": 'type-id-' + uuid.uuid4().hex,
+ 'provider': 'LuksEncryptor',
+ 'cipher': None,
+ 'key_size': None,
+ 'control_location': 'front-end',
+ }
+
+ # Overwrite default attributes.
+ encryption_info.update(attrs)
+
+ encryption_type = fakes.FakeResource(
+ info=copy.deepcopy(encryption_info),
+ loaded=True)
+ return encryption_type
diff --git a/openstackclient/tests/unit/volume/v2/test_type.py b/openstackclient/tests/unit/volume/v2/test_type.py
index cec01bd8..4023d55b 100644
--- a/openstackclient/tests/unit/volume/v2/test_type.py
+++ b/openstackclient/tests/unit/volume/v2/test_type.py
@@ -36,6 +36,10 @@ class TestType(volume_fakes.TestVolume):
self.app.client_manager.volume.volume_type_access)
self.types_access_mock.reset_mock()
+ self.encryption_types_mock = (
+ self.app.client_manager.volume.volume_encryption_types)
+ self.encryption_types_mock.reset_mock()
+
self.projects_mock = self.app.client_manager.identity.projects
self.projects_mock.reset_mock()
@@ -131,6 +135,68 @@ class TestTypeCreate(TestType):
self.cmd.take_action,
parsed_args)
+ def test_type_create_with_encryption(self):
+ encryption_info = {
+ 'provider': 'LuksEncryptor',
+ 'cipher': 'aes-xts-plain64',
+ 'key_size': '128',
+ 'control_location': 'front-end',
+ }
+ encryption_type = volume_fakes.FakeType.create_one_encryption_type(
+ attrs=encryption_info
+ )
+ self.new_volume_type = volume_fakes.FakeType.create_one_type(
+ attrs={'encryption': encryption_info})
+ self.types_mock.create.return_value = self.new_volume_type
+ self.encryption_types_mock.create.return_value = encryption_type
+ encryption_columns = (
+ 'description',
+ 'encryption',
+ 'id',
+ 'is_public',
+ 'name',
+ )
+ encryption_data = (
+ self.new_volume_type.description,
+ utils.format_dict(encryption_info),
+ self.new_volume_type.id,
+ True,
+ self.new_volume_type.name,
+ )
+ arglist = [
+ '--encryption-provider', 'LuksEncryptor',
+ '--encryption-cipher', 'aes-xts-plain64',
+ '--encryption-key-size', '128',
+ '--encryption-control-location', 'front-end',
+ self.new_volume_type.name,
+ ]
+ verifylist = [
+ ('encryption_provider', 'LuksEncryptor'),
+ ('encryption_cipher', 'aes-xts-plain64'),
+ ('encryption_key_size', 128),
+ ('encryption_control_location', 'front-end'),
+ ('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=None,
+ )
+ body = {
+ 'provider': 'LuksEncryptor',
+ 'cipher': 'aes-xts-plain64',
+ 'key_size': 128,
+ 'control_location': 'front-end',
+ }
+ self.encryption_types_mock.create.assert_called_with(
+ self.new_volume_type,
+ body,
+ )
+ self.assertEqual(encryption_columns, columns)
+ self.assertEqual(encryption_data, data)
+
class TestTypeDelete(TestType):
@@ -305,6 +371,7 @@ class TestTypeList(TestType):
"--default",
]
verifylist = [
+ ("encryption_type", False),
("long", False),
("private", False),
("public", False),
@@ -317,6 +384,47 @@ class TestTypeList(TestType):
self.assertEqual(self.columns, columns)
self.assertEqual(self.data_with_default_type, list(data))
+ def test_type_list_with_encryption(self):
+ encryption_type = volume_fakes.FakeType.create_one_encryption_type(
+ attrs={'volume_type_id': self.volume_types[0].id})
+ encryption_info = {
+ 'provider': 'LuksEncryptor',
+ 'cipher': None,
+ 'key_size': None,
+ 'control_location': 'front-end',
+ }
+ encryption_columns = self.columns + [
+ "Encryption",
+ ]
+ encryption_data = []
+ encryption_data.append((
+ self.volume_types[0].id,
+ self.volume_types[0].name,
+ self.volume_types[0].is_public,
+ utils.format_dict(encryption_info),
+ ))
+ encryption_data.append((
+ self.volume_types[1].id,
+ self.volume_types[1].name,
+ self.volume_types[1].is_public,
+ '-',
+ ))
+
+ self.encryption_types_mock.list.return_value = [encryption_type]
+ arglist = [
+ "--encryption-type",
+ ]
+ verifylist = [
+ ("encryption_type", True),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ columns, data = self.cmd.take_action(parsed_args)
+ self.encryption_types_mock.list.assert_called_once_with()
+ self.types_mock.list.assert_called_once_with(is_public=None)
+ self.assertEqual(encryption_columns, columns)
+ self.assertEqual(encryption_data, list(data))
+
class TestTypeSet(TestType):
@@ -331,6 +439,8 @@ class TestTypeSet(TestType):
# Return a project
self.projects_mock.get.return_value = self.project
+ self.encryption_types_mock.create.return_value = None
+ self.encryption_types_mock.update.return_value = None
# Get the command object to test
self.cmd = volume_type.SetVolumeType(self.app, None)
@@ -454,6 +564,107 @@ class TestTypeSet(TestType):
self.project.id,
)
+ def test_type_set_new_encryption(self):
+ self.encryption_types_mock.update.side_effect = (
+ exceptions.NotFound('NotFound'))
+ arglist = [
+ '--encryption-provider', 'LuksEncryptor',
+ '--encryption-cipher', 'aes-xts-plain64',
+ '--encryption-key-size', '128',
+ '--encryption-control-location', 'front-end',
+ self.volume_type.id,
+ ]
+ verifylist = [
+ ('encryption_provider', 'LuksEncryptor'),
+ ('encryption_cipher', 'aes-xts-plain64'),
+ ('encryption_key_size', 128),
+ ('encryption_control_location', 'front-end'),
+ ('volume_type', self.volume_type.id),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ result = self.cmd.take_action(parsed_args)
+ body = {
+ 'provider': 'LuksEncryptor',
+ 'cipher': 'aes-xts-plain64',
+ 'key_size': 128,
+ 'control_location': 'front-end',
+ }
+ self.encryption_types_mock.update.assert_called_with(
+ self.volume_type,
+ body,
+ )
+ self.encryption_types_mock.create.assert_called_with(
+ self.volume_type,
+ body,
+ )
+ self.assertIsNone(result)
+
+ @mock.patch.object(utils, 'find_resource')
+ def test_type_set_existing_encryption(self, mock_find):
+ mock_find.side_effect = [self.volume_type,
+ "existing_encryption_type"]
+ arglist = [
+ '--encryption-provider', 'LuksEncryptor',
+ '--encryption-cipher', 'aes-xts-plain64',
+ '--encryption-control-location', 'front-end',
+ self.volume_type.id,
+ ]
+ verifylist = [
+ ('encryption_provider', 'LuksEncryptor'),
+ ('encryption_cipher', 'aes-xts-plain64'),
+ ('encryption_control_location', 'front-end'),
+ ('volume_type', self.volume_type.id),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ result = self.cmd.take_action(parsed_args)
+ body = {
+ 'provider': 'LuksEncryptor',
+ 'cipher': 'aes-xts-plain64',
+ 'control_location': 'front-end',
+ }
+ self.encryption_types_mock.update.assert_called_with(
+ self.volume_type,
+ body,
+ )
+ self.encryption_types_mock.create.assert_not_called()
+ self.assertIsNone(result)
+
+ def test_type_set_new_encryption_without_provider(self):
+ self.encryption_types_mock.update.side_effect = (
+ exceptions.NotFound('NotFound'))
+ arglist = [
+ '--encryption-cipher', 'aes-xts-plain64',
+ '--encryption-key-size', '128',
+ '--encryption-control-location', 'front-end',
+ self.volume_type.id,
+ ]
+ verifylist = [
+ ('encryption_cipher', 'aes-xts-plain64'),
+ ('encryption_key_size', 128),
+ ('encryption_control_location', 'front-end'),
+ ('volume_type', self.volume_type.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("Command Failed: One or more of"
+ " the operations failed",
+ str(e))
+ body = {
+ 'cipher': 'aes-xts-plain64',
+ 'key_size': 128,
+ 'control_location': 'front-end',
+ }
+ self.encryption_types_mock.update.assert_called_with(
+ self.volume_type,
+ body,
+ )
+ self.encryption_types_mock.create.assert_not_called()
+
class TestTypeShow(TestType):
@@ -489,6 +700,7 @@ class TestTypeShow(TestType):
self.volume_type.id
]
verifylist = [
+ ("encryption_type", False),
("volume_type", self.volume_type.id)
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
@@ -564,6 +776,52 @@ class TestTypeShow(TestType):
)
self.assertEqual(private_type_data, data)
+ def test_type_show_with_encryption(self):
+ encryption_type = volume_fakes.FakeType.create_one_encryption_type()
+ encryption_info = {
+ 'provider': 'LuksEncryptor',
+ 'cipher': None,
+ 'key_size': None,
+ 'control_location': 'front-end',
+ }
+ self.volume_type = volume_fakes.FakeType.create_one_type(
+ attrs={'encryption': encryption_info})
+ self.types_mock.get.return_value = self.volume_type
+ self.encryption_types_mock.get.return_value = encryption_type
+ encryption_columns = (
+ 'access_project_ids',
+ 'description',
+ 'encryption',
+ 'id',
+ 'is_public',
+ 'name',
+ 'properties',
+ )
+ encryption_data = (
+ None,
+ self.volume_type.description,
+ utils.format_dict(encryption_info),
+ self.volume_type.id,
+ True,
+ self.volume_type.name,
+ utils.format_dict(self.volume_type.extra_specs)
+ )
+ arglist = [
+ '--encryption-type',
+ self.volume_type.id
+ ]
+ verifylist = [
+ ('encryption_type', True),
+ ("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.encryption_types_mock.get.assert_called_with(self.volume_type.id)
+ self.assertEqual(encryption_columns, columns)
+ self.assertEqual(encryption_data, data)
+
class TestTypeUnset(TestType):
@@ -625,6 +883,7 @@ class TestTypeUnset(TestType):
self.volume_type.id,
]
verifylist = [
+ ('encryption_type', False),
('project', ''),
('volume_type', self.volume_type.id),
]
@@ -633,7 +892,7 @@ class TestTypeUnset(TestType):
result = self.cmd.take_action(parsed_args)
self.assertIsNone(result)
-
+ self.encryption_types_mock.delete.assert_not_called()
self.assertFalse(self.types_access_mock.remove_project_access.called)
def test_type_unset_failed_with_missing_volume_type_argument(self):
@@ -649,3 +908,18 @@ class TestTypeUnset(TestType):
self.cmd,
arglist,
verifylist)
+
+ def test_type_unset_encryption_type(self):
+ arglist = [
+ '--encryption-type',
+ self.volume_type.id,
+ ]
+ verifylist = [
+ ('encryption_type', True),
+ ('volume_type', self.volume_type.id),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ result = self.cmd.take_action(parsed_args)
+ self.encryption_types_mock.delete.assert_called_with(self.volume_type)
+ self.assertIsNone(result)
diff --git a/openstackclient/tests/unit/volume/v2/test_volume_host.py b/openstackclient/tests/unit/volume/v2/test_volume_host.py
index aad7bb0b..b024329a 100644
--- a/openstackclient/tests/unit/volume/v2/test_volume_host.py
+++ b/openstackclient/tests/unit/volume/v2/test_volume_host.py
@@ -35,6 +35,7 @@ class TestVolumeHostSet(TestVolumeHost):
self.host_mock.freeze_host.return_value = None
self.host_mock.thaw_host.return_value = None
+ # Get the command object to mock
self.cmd = volume_host.SetVolumeHost(self.app, None)
def test_volume_host_set_nothing(self):
@@ -84,3 +85,33 @@ class TestVolumeHostSet(TestVolumeHost):
self.host_mock.freeze_host.assert_called_with(self.service.host)
self.host_mock.thaw_host.assert_not_called()
self.assertIsNone(result)
+
+
+class TestVolumeHostFailover(TestVolumeHost):
+
+ service = host_fakes.FakeService.create_one_service()
+
+ def setUp(self):
+ super(TestVolumeHostFailover, self).setUp()
+
+ self.host_mock.failover_host.return_value = None
+
+ # Get the command object to mock
+ self.cmd = volume_host.FailoverVolumeHost(self.app, None)
+
+ def test_volume_host_failover(self):
+ arglist = [
+ '--volume-backend', 'backend_test',
+ self.service.host,
+ ]
+ verifylist = [
+ ('volume_backend', 'backend_test'),
+ ('host', self.service.host),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ result = self.cmd.take_action(parsed_args)
+
+ self.host_mock.failover_host.assert_called_with(
+ self.service.host, 'backend_test')
+ self.assertIsNone(result)
View Lorry Controller Status