diff options
Diffstat (limited to 'snaps/openstack/utils/tests')
| -rw-r--r-- | snaps/openstack/utils/tests/__init__.py | 15 | ||||
| -rw-r--r-- | snaps/openstack/utils/tests/glance_utils_tests.py | 115 | ||||
| -rw-r--r-- | snaps/openstack/utils/tests/keystone_utils_tests.py | 100 | ||||
| -rw-r--r-- | snaps/openstack/utils/tests/neutron_utils_tests.py | 651 | ||||
| -rw-r--r-- | snaps/openstack/utils/tests/nova_utils_tests.py | 208 |
5 files changed, 1089 insertions, 0 deletions
diff --git a/snaps/openstack/utils/tests/__init__.py b/snaps/openstack/utils/tests/__init__.py new file mode 100644 index 0000000..7f92908 --- /dev/null +++ b/snaps/openstack/utils/tests/__init__.py @@ -0,0 +1,15 @@ +# Copyright (c) 2016 Cable Television Laboratories, Inc. ("CableLabs") +# and others. 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. +__author__ = 'spisarski'
\ No newline at end of file diff --git a/snaps/openstack/utils/tests/glance_utils_tests.py b/snaps/openstack/utils/tests/glance_utils_tests.py new file mode 100644 index 0000000..d13908b --- /dev/null +++ b/snaps/openstack/utils/tests/glance_utils_tests.py @@ -0,0 +1,115 @@ +# Copyright (c) 2016 Cable Television Laboratories, Inc. ("CableLabs") +# and others. 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 os +import shutil +import uuid + +from snaps import file_utils +from snaps.openstack.tests import openstack_tests + +from snaps.openstack.utils import nova_utils +from snaps.openstack.tests import validation_utils +from snaps.openstack.tests.os_source_file_test import OSComponentTestCase +from snaps.openstack.utils import glance_utils + +__author__ = 'spisarski' + + +class GlanceSmokeTests(OSComponentTestCase): + """ + Tests to ensure that the neutron client can communicate with the cloud + """ + + def test_glance_connect_success(self): + """ + Tests to ensure that the proper credentials can connect. + """ + glance = glance_utils.glance_client(self.os_creds) + + users = glance.images.list() + self.assertIsNotNone(users) + + def test_glance_connect_fail(self): + """ + Tests to ensure that the improper credentials cannot connect. + """ + from snaps.openstack.os_credentials import OSCreds + + with self.assertRaises(Exception): + neutron = glance_utils.glance_client(OSCreds('user', 'pass', 'url', 'project')) + neutron.list_networks() + + +class GlanceUtilsTests(OSComponentTestCase): + """ + Test for the CreateImage class defined in create_image.py + """ + + def setUp(self): + """ + Instantiates the CreateImage object that is responsible for downloading and creating an OS image file + within OpenStack + """ + guid = uuid.uuid4() + self.image_name = self.__class__.__name__ + '-' + str(guid) + self.image = None + self.nova = nova_utils.nova_client(self.os_creds) + self.glance = glance_utils.glance_client(self.os_creds) + + self.tmp_dir = 'tmp/' + str(guid) + if not os.path.exists(self.tmp_dir): + os.makedirs(self.tmp_dir) + + def tearDown(self): + """ + Cleans the remote OpenStack objects + """ + if self.image: + glance_utils.delete_image(self.glance, self.image) + + if os.path.exists(self.tmp_dir) and os.path.isdir(self.tmp_dir): + shutil.rmtree(self.tmp_dir) + + def test_create_image_minimal_url(self): + """ + Tests the glance_utils.create_image() function with a URL + """ + os_image_settings = openstack_tests.cirros_url_image(name=self.image_name) + + self.image = glance_utils.create_image(self.glance, os_image_settings) + self.assertIsNotNone(self.image) + + self.assertEqual(self.image_name, self.image.name) + + image = glance_utils.get_image(self.nova, self.glance, os_image_settings.name) + self.assertIsNotNone(image) + + validation_utils.objects_equivalent(self.image, image) + + def test_create_image_minimal_file(self): + """ + Tests the glance_utils.create_image() function with a file + """ + url_image_settings = openstack_tests.cirros_url_image('foo') + image_file = file_utils.download(url_image_settings.url, self.tmp_dir) + file_image_settings = openstack_tests.file_image_test_settings(name=self.image_name, file_path=image_file.name) + + self.image = glance_utils.create_image(self.glance, file_image_settings) + self.assertIsNotNone(self.image) + self.assertEqual(self.image_name, self.image.name) + + image = glance_utils.get_image(self.nova, self.glance, file_image_settings.name) + self.assertIsNotNone(image) + validation_utils.objects_equivalent(self.image, image) diff --git a/snaps/openstack/utils/tests/keystone_utils_tests.py b/snaps/openstack/utils/tests/keystone_utils_tests.py new file mode 100644 index 0000000..76a43ef --- /dev/null +++ b/snaps/openstack/utils/tests/keystone_utils_tests.py @@ -0,0 +1,100 @@ +# Copyright (c) 2016 Cable Television Laboratories, Inc. ("CableLabs") +# and others. 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 uuid + +from snaps.openstack.create_project import ProjectSettings +from snaps.openstack.create_user import UserSettings +from snaps.openstack.tests.os_source_file_test import OSComponentTestCase +from snaps.openstack.utils import keystone_utils + +__author__ = 'spisarski' + + +class KeystoneSmokeTests(OSComponentTestCase): + """ + Tests to ensure that the neutron client can communicate with the cloud + """ + + def test_keystone_connect_success(self): + """ + Tests to ensure that the proper credentials can connect. + """ + keystone = keystone_utils.keystone_client(self.os_creds) + + users = keystone.users.list() + self.assertIsNotNone(users) + + def test_keystone_connect_fail(self): + """ + Tests to ensure that the improper credentials cannot connect. + """ + from snaps.openstack.os_credentials import OSCreds + + with self.assertRaises(Exception): + keystone = keystone_utils.keystone_client(OSCreds('user', 'pass', 'url', 'project')) + keystone.users.list() + + +class KeystoneUtilsTests(OSComponentTestCase): + """ + Test for the CreateImage class defined in create_image.py + """ + + def setUp(self): + """ + Instantiates the CreateImage object that is responsible for downloading and creating an OS image file + within OpenStack + """ + guid = uuid.uuid4() + self.username = self.__class__.__name__ + '-' + str(guid) + self.user = None + + self.project_name = self.__class__.__name__ + '-' + str(guid) + self.project = None + self.keystone = keystone_utils.keystone_client(self.os_creds) + + def tearDown(self): + """ + Cleans the remote OpenStack objects + """ + if self.project: + keystone_utils.delete_project(self.keystone, self.project) + + if self.user: + keystone_utils.delete_user(self.keystone, self.user) + + def test_create_user_minimal(self): + """ + Tests the keystone_utils.create_user() function + """ + user_settings = UserSettings(name=self.username, password='test123') + self.user = keystone_utils.create_user(self.keystone, user_settings) + self.assertEqual(self.username, self.user.name) + + user = keystone_utils.get_user(self.keystone, self.username) + self.assertIsNotNone(user) + self.assertEqual(self.user, user) + + def test_create_project_minimal(self): + """ + Tests the keyston_utils.create_project() funtion + """ + project_settings = ProjectSettings(name=self.project_name) + self.project = keystone_utils.create_project(self.keystone, project_settings) + self.assertEquals(self.project_name, self.project.name) + + project = keystone_utils.get_project(keystone=self.keystone, project_name=project_settings.name) + self.assertIsNotNone(project) + self.assertEquals(self.project_name, self.project.name) diff --git a/snaps/openstack/utils/tests/neutron_utils_tests.py b/snaps/openstack/utils/tests/neutron_utils_tests.py new file mode 100644 index 0000000..5f95fc9 --- /dev/null +++ b/snaps/openstack/utils/tests/neutron_utils_tests.py @@ -0,0 +1,651 @@ +# Copyright (c) 2016 Cable Television Laboratories, Inc. ("CableLabs") +# and others. 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 uuid + +from snaps.openstack.utils import keystone_utils +from snaps.openstack.create_security_group import SecurityGroupSettings, SecurityGroupRuleSettings, Direction +from snaps.openstack.tests import openstack_tests +from snaps.openstack.utils import neutron_utils +from snaps.openstack.create_network import NetworkSettings, SubnetSettings, PortSettings +from snaps.openstack import create_router +from snaps.openstack.tests.os_source_file_test import OSComponentTestCase +from snaps.openstack.tests import validation_utils + +__author__ = 'spisarski' + +ip_1 = '10.55.1.100' +ip_2 = '10.55.1.200' + + +class NeutronSmokeTests(OSComponentTestCase): + """ + Tests to ensure that the neutron client can communicate with the cloud + """ + + def test_neutron_connect_success(self): + """ + Tests to ensure that the proper credentials can connect. + """ + neutron = neutron_utils.neutron_client(self.os_creds) + + networks = neutron.list_networks() + + found = False + networks = networks.get('networks') + for network in networks: + if network.get('name') == self.ext_net_name: + found = True + self.assertTrue(found) + + def test_neutron_connect_fail(self): + """ + Tests to ensure that the improper credentials cannot connect. + """ + from snaps.openstack.os_credentials import OSCreds + + with self.assertRaises(Exception): + neutron = neutron_utils.neutron_client( + OSCreds(username='user', password='pass', auth_url='url', project_name='project')) + neutron.list_networks() + + def test_retrieve_ext_network_name(self): + """ + Tests the neutron_utils.get_external_network_names to ensure the configured self.ext_net_name is contained + within the returned list + :return: + """ + neutron = neutron_utils.neutron_client(self.os_creds) + ext_networks = neutron_utils.get_external_networks(neutron) + found = False + for network in ext_networks: + if network['network']['name'] == self.ext_net_name: + found = True + break + self.assertTrue(found) + + +class NeutronUtilsNetworkTests(OSComponentTestCase): + """ + Test for creating networks via neutron_utils.py + """ + + def setUp(self): + guid = self.__class__.__name__ + '-' + str(uuid.uuid4()) + self.port_name = str(guid) + '-port' + self.neutron = neutron_utils.neutron_client(self.os_creds) + self.network = None + self.net_config = openstack_tests.get_pub_net_config(net_name=guid + '-pub-net') + + def tearDown(self): + """ + Cleans the remote OpenStack objects + """ + if self.network: + neutron_utils.delete_network(self.neutron, self.network) + validate_network(self.neutron, self.network['network']['name'], False) + + def test_create_network(self): + """ + Tests the neutron_utils.create_neutron_net() function + """ + self.network = neutron_utils.create_network(self.neutron, self.os_creds, self.net_config.network_settings) + self.assertEqual(self.net_config.network_settings.name, self.network['network']['name']) + self.assertTrue(validate_network(self.neutron, self.net_config.network_settings.name, True)) + + def test_create_network_empty_name(self): + """ + Tests the neutron_utils.create_neutron_net() function with an empty network name + """ + with self.assertRaises(Exception): + self.network = neutron_utils.create_network(self.neutron, NetworkSettings(name='')) + + def test_create_network_null_name(self): + """ + Tests the neutron_utils.create_neutron_net() function when the network name is None + """ + with self.assertRaises(Exception): + self.network = neutron_utils.create_network(self.neutron, NetworkSettings()) + + +class NeutronUtilsSubnetTests(OSComponentTestCase): + """ + Test for creating networks with subnets via neutron_utils.py + """ + + def setUp(self): + guid = self.__class__.__name__ + '-' + str(uuid.uuid4()) + self.port_name = str(guid) + '-port' + self.neutron = neutron_utils.neutron_client(self.os_creds) + self.network = None + self.subnet = None + self.net_config = openstack_tests.get_pub_net_config( + net_name=guid + '-pub-net', subnet_name=guid + '-pub-subnet', external_net=self.ext_net_name) + + def tearDown(self): + """ + Cleans the remote OpenStack objects + """ + if self.subnet: + neutron_utils.delete_subnet(self.neutron, self.subnet) + validate_subnet(self.neutron, self.subnet.get('name'), + self.net_config.network_settings.subnet_settings[0].cidr, False) + + if self.network: + neutron_utils.delete_network(self.neutron, self.network) + validate_network(self.neutron, self.network['network']['name'], False) + + def test_create_subnet(self): + """ + Tests the neutron_utils.create_neutron_net() function + """ + self.network = neutron_utils.create_network(self.neutron, self.os_creds, self.net_config.network_settings) + self.assertEqual(self.net_config.network_settings.name, self.network['network']['name']) + self.assertTrue(validate_network(self.neutron, self.net_config.network_settings.name, True)) + + self.subnet = neutron_utils.create_subnet(self.neutron, self.net_config.network_settings.subnet_settings[0], + self.os_creds, network=self.network) + validate_subnet(self.neutron, self.net_config.network_settings.subnet_settings[0].name, + self.net_config.network_settings.subnet_settings[0].cidr, True) + + def test_create_subnet_null_name(self): + """ + Tests the neutron_utils.create_neutron_subnet() function for an Exception when the subnet name is None + """ + self.network = neutron_utils.create_network(self.neutron, self.os_creds, self.net_config.network_settings) + self.assertEqual(self.net_config.network_settings.name, self.network['network']['name']) + self.assertTrue(validate_network(self.neutron, self.net_config.network_settings.name, True)) + + with self.assertRaises(Exception): + SubnetSettings(cidr=self.net_config.subnet_cidr) + + def test_create_subnet_empty_name(self): + """ + Tests the neutron_utils.create_neutron_net() function with an empty name + """ + self.network = neutron_utils.create_network(self.neutron, self.os_creds, self.net_config.network_settings) + self.assertEqual(self.net_config.network_settings.name, self.network['network']['name']) + self.assertTrue(validate_network(self.neutron, self.net_config.network_settings.name, True)) + + neutron_utils.create_subnet(self.neutron, self.net_config.network_settings.subnet_settings[0], + self.os_creds, network=self.network) + validate_subnet(self.neutron, '', self.net_config.network_settings.subnet_settings[0].cidr, True) + + def test_create_subnet_null_cidr(self): + """ + Tests the neutron_utils.create_neutron_subnet() function for an Exception when the subnet CIDR value is None + """ + self.network = neutron_utils.create_network(self.neutron, self.os_creds, self.net_config.network_settings) + self.assertEqual(self.net_config.network_settings.name, self.network['network']['name']) + self.assertTrue(validate_network(self.neutron, self.net_config.network_settings.name, True)) + + with self.assertRaises(Exception): + sub_sets = SubnetSettings(cidr=None, name=self.net_config.subnet_name) + neutron_utils.create_subnet(self.neutron, sub_sets, self.os_creds, network=self.network) + + def test_create_subnet_empty_cidr(self): + """ + Tests the neutron_utils.create_neutron_subnet() function for an Exception when the subnet CIDR value is empty + """ + self.network = neutron_utils.create_network(self.neutron, self.os_creds, self.net_config.network_settings) + self.assertEqual(self.net_config.network_settings.name, self.network['network']['name']) + self.assertTrue(validate_network(self.neutron, self.net_config.network_settings.name, True)) + + with self.assertRaises(Exception): + sub_sets = SubnetSettings(cidr='', name=self.net_config.subnet_name) + neutron_utils.create_subnet(self.neutron, sub_sets, self.os_creds, network=self.network) + + +class NeutronUtilsRouterTests(OSComponentTestCase): + """ + Test for creating routers via neutron_utils.py + """ + + def setUp(self): + guid = self.__class__.__name__ + '-' + str(uuid.uuid4()) + self.port_name = str(guid) + '-port' + self.neutron = neutron_utils.neutron_client(self.os_creds) + self.network = None + self.subnet = None + self.port = None + self.router = None + self.interface_router = None + self.net_config = openstack_tests.get_pub_net_config( + net_name=guid + '-pub-net', subnet_name=guid + '-pub-subnet', + router_name=guid + '-pub-router', external_net=self.ext_net_name) + + def tearDown(self): + """ + Cleans the remote OpenStack objects + """ + if self.interface_router: + neutron_utils.remove_interface_router(self.neutron, self.router, self.subnet) + + if self.router: + neutron_utils.delete_router(self.neutron, self.router) + validate_router(self.neutron, self.router.get('name'), False) + + if self.port: + neutron_utils.delete_port(self.neutron, self.port) + + if self.subnet: + neutron_utils.delete_subnet(self.neutron, self.subnet) + validate_subnet(self.neutron, self.subnet.get('name'), + self.net_config.network_settings.subnet_settings[0].cidr, False) + + if self.network: + neutron_utils.delete_network(self.neutron, self.network) + validate_network(self.neutron, self.network['network']['name'], False) + + def test_create_router_simple(self): + """ + Tests the neutron_utils.create_neutron_net() function when an external gateway is requested + """ + self.router = neutron_utils.create_router(self.neutron, self.os_creds, self.net_config.router_settings) + validate_router(self.neutron, self.net_config.router_settings.name, True) + + def test_create_router_with_public_interface(self): + """ + Tests the neutron_utils.create_neutron_net() function when an external gateway is requested + """ + self.net_config = openstack_tests.OSNetworkConfig( + self.net_config.network_settings.name, + self.net_config.network_settings.subnet_settings[0].name, + self.net_config.network_settings.subnet_settings[0].cidr, self.net_config.router_settings.name, + self.ext_net_name) + self.router = neutron_utils.create_router(self.neutron, self.os_creds, self.net_config.router_settings) + validate_router(self.neutron, self.net_config.router_settings.name, True) + # TODO - Add validation that the router gatway has been set + + def test_create_router_empty_name(self): + """ + Tests the neutron_utils.create_neutron_net() function + """ + with self.assertRaises(Exception): + this_router_settings = create_router.RouterSettings(name='') + self.router = neutron_utils.create_router(self.neutron, self.os_creds, this_router_settings) + + def test_create_router_null_name(self): + """ + Tests the neutron_utils.create_neutron_subnet() function when the subnet CIDR value is None + """ + with self.assertRaises(Exception): + this_router_settings = create_router.RouterSettings() + self.router = neutron_utils.create_router(self.neutron, self.os_creds, this_router_settings) + validate_router(self.neutron, None, True) + + def test_add_interface_router(self): + """ + Tests the neutron_utils.add_interface_router() function + """ + self.network = neutron_utils.create_network(self.neutron, self.os_creds, self.net_config.network_settings) + self.assertEqual(self.net_config.network_settings.name, self.network['network']['name']) + self.assertTrue(validate_network(self.neutron, self.net_config.network_settings.name, True)) + + self.subnet = neutron_utils.create_subnet(self.neutron, self.net_config.network_settings.subnet_settings[0], + self.os_creds, self.network) + validate_subnet(self.neutron, self.net_config.network_settings.subnet_settings[0].name, + self.net_config.network_settings.subnet_settings[0].cidr, True) + + self.router = neutron_utils.create_router(self.neutron, self.os_creds, self.net_config.router_settings) + validate_router(self.neutron, self.net_config.router_settings.name, True) + + self.interface_router = neutron_utils.add_interface_router(self.neutron, self.router, self.subnet) + validate_interface_router(self.interface_router, self.router, self.subnet) + + def test_add_interface_router_null_router(self): + """ + Tests the neutron_utils.add_interface_router() function for an Exception when the router value is None + """ + self.network = neutron_utils.create_network(self.neutron, self.os_creds, self.net_config.network_settings) + self.assertEqual(self.net_config.network_settings.name, self.network['network']['name']) + self.assertTrue(validate_network(self.neutron, self.net_config.network_settings.name, True)) + + self.subnet = neutron_utils.create_subnet(self.neutron, self.net_config.network_settings.subnet_settings[0], + self.os_creds, self.network) + validate_subnet(self.neutron, self.net_config.network_settings.subnet_settings[0].name, + self.net_config.network_settings.subnet_settings[0].cidr, True) + + with self.assertRaises(Exception): + self.interface_router = neutron_utils.add_interface_router(self.neutron, self.router, self.subnet) + + def test_add_interface_router_null_subnet(self): + """ + Tests the neutron_utils.add_interface_router() function for an Exception when the subnet value is None + """ + self.network = neutron_utils.create_network(self.neutron, self.os_creds, self.net_config.network_settings) + self.assertEqual(self.net_config.network_settings.name, self.network['network']['name']) + self.assertTrue(validate_network(self.neutron, self.net_config.network_settings.name, True)) + + self.router = neutron_utils.create_router(self.neutron, self.os_creds, self.net_config.router_settings) + validate_router(self.neutron, self.net_config.router_settings.name, True) + + with self.assertRaises(Exception): + self.interface_router = neutron_utils.add_interface_router(self.neutron, self.router, self.subnet) + + def test_create_port(self): + """ + Tests the neutron_utils.create_port() function + """ + self.network = neutron_utils.create_network(self.neutron, self.os_creds, self.net_config.network_settings) + self.assertEqual(self.net_config.network_settings.name, self.network['network']['name']) + self.assertTrue(validate_network(self.neutron, self.net_config.network_settings.name, True)) + + self.subnet = neutron_utils.create_subnet(self.neutron, self.net_config.network_settings.subnet_settings[0], + self.os_creds, self.network) + validate_subnet(self.neutron, self.net_config.network_settings.subnet_settings[0].name, + self.net_config.network_settings.subnet_settings[0].cidr, True) + + self.port = neutron_utils.create_port( + self.neutron, self.os_creds, PortSettings( + name=self.port_name, + ip_addrs=[{'subnet_name': self.net_config.network_settings.subnet_settings[0].name, 'ip': ip_1}], + network_name=self.net_config.network_settings.name)) + validate_port(self.neutron, self.port, self.port_name) + + def test_create_port_empty_name(self): + """ + Tests the neutron_utils.create_port() function + """ + self.network = neutron_utils.create_network(self.neutron, self.os_creds, self.net_config.network_settings) + self.assertEqual(self.net_config.network_settings.name, self.network['network']['name']) + self.assertTrue(validate_network(self.neutron, self.net_config.network_settings.name, True)) + + self.subnet = neutron_utils.create_subnet(self.neutron, self.net_config.network_settings.subnet_settings[0], + self.os_creds, self.network) + validate_subnet(self.neutron, self.net_config.network_settings.subnet_settings[0].name, + self.net_config.network_settings.subnet_settings[0].cidr, True) + + self.port = neutron_utils.create_port( + self.neutron, self.os_creds, PortSettings( + name=self.port_name, network_name=self.net_config.network_settings.name, + ip_addrs=[{'subnet_name': self.net_config.network_settings.subnet_settings[0].name, 'ip': ip_1}])) + validate_port(self.neutron, self.port, self.port_name) + + def test_create_port_null_name(self): + """ + Tests the neutron_utils.create_port() function for an Exception when the port name value is None + """ + self.network = neutron_utils.create_network(self.neutron, self.os_creds, self.net_config.network_settings) + self.assertEqual(self.net_config.network_settings.name, self.network['network']['name']) + self.assertTrue(validate_network(self.neutron, self.net_config.network_settings.name, True)) + + self.subnet = neutron_utils.create_subnet(self.neutron, self.net_config.network_settings.subnet_settings[0], + self.os_creds, self.network) + validate_subnet(self.neutron, self.net_config.network_settings.subnet_settings[0].name, + self.net_config.network_settings.subnet_settings[0].cidr, True) + + with self.assertRaises(Exception): + self.port = neutron_utils.create_port(self.neutron, self.os_creds, PortSettings( + network_name=self.net_config.network_settings.name, + ip_addrs=[{'subnet_name': self.net_config.network_settings.subnet_settings.name, 'ip': ip_1}])) + + def test_create_port_null_network_object(self): + """ + Tests the neutron_utils.create_port() function for an Exception when the network object is None + """ + self.network = neutron_utils.create_network(self.neutron, self.os_creds, self.net_config.network_settings) + self.assertEqual(self.net_config.network_settings.name, self.network['network']['name']) + self.assertTrue(validate_network(self.neutron, self.net_config.network_settings.name, True)) + + self.subnet = neutron_utils.create_subnet(self.neutron, self.net_config.network_settings.subnet_settings[0], + self.os_creds, self.network) + validate_subnet(self.neutron, self.net_config.network_settings.subnet_settings[0].name, + self.net_config.network_settings.subnet_settings[0].cidr, True) + + with self.assertRaises(Exception): + self.port = neutron_utils.create_port(self.neutron, self.os_creds, PortSettings( + self.neutron, self.port_name, self.net_config.network_settings.name, + ip_addrs=[{'subnet_name': self.net_config.network_settings.subnet_settings.name, 'ip': ip_1}])) + + def test_create_port_null_ip(self): + """ + Tests the neutron_utils.create_port() function for an Exception when the IP value is None + """ + self.network = neutron_utils.create_network(self.neutron, self.os_creds, self.net_config.network_settings) + self.assertEqual(self.net_config.network_settings.name, self.network['network']['name']) + self.assertTrue(validate_network(self.neutron, self.net_config.network_settings.name, True)) + + self.subnet = neutron_utils.create_subnet(self.neutron, self.net_config.network_settings.subnet_settings[0], + self.os_creds, self.network) + validate_subnet(self.neutron, self.net_config.network_settings.subnet_settings[0].name, + self.net_config.network_settings.subnet_settings[0].cidr, True) + + with self.assertRaises(Exception): + self.port = neutron_utils.create_port(self.neutron, self.os_creds, PortSettings( + name=self.port_name, network_name=self.net_config.network_settings.name, + ip_addrs=[{'subnet_name': self.net_config.network_settings.subnet_settings.name, 'ip': None}])) + + def test_create_port_invalid_ip(self): + """ + Tests the neutron_utils.create_port() function for an Exception when the IP value is None + """ + self.network = neutron_utils.create_network(self.neutron, self.os_creds, self.net_config.network_settings) + self.assertEqual(self.net_config.network_settings.name, self.network['network']['name']) + self.assertTrue(validate_network(self.neutron, self.net_config.network_settings.name, True)) + + self.subnet = neutron_utils.create_subnet(self.neutron, self.net_config.network_settings.subnet_settings[0], + self.os_creds, self.network) + validate_subnet(self.neutron, self.net_config.network_settings.subnet_settings[0].name, + self.net_config.network_settings.subnet_settings[0].cidr, True) + + with self.assertRaises(Exception): + self.port = neutron_utils.create_port(self.neutron, self.os_creds, PortSettings( + name=self.port_name, network_name=self.net_config.network_settings.name, + ip_addrs=[{'subnet_name': self.net_config.network_settings.subnet_settings.name, 'ip': 'foo'}])) + + def test_create_port_invalid_ip_to_subnet(self): + """ + Tests the neutron_utils.create_port() function for an Exception when the IP value is None + """ + self.network = neutron_utils.create_network(self.neutron, self.os_creds, self.net_config.network_settings) + self.assertEqual(self.net_config.network_settings.name, self.network['network']['name']) + self.assertTrue(validate_network(self.neutron, self.net_config.network_settings.name, True)) + + self.subnet = neutron_utils.create_subnet(self.neutron, self.net_config.network_settings.subnet_settings[0], + self.os_creds, self.network) + validate_subnet(self.neutron, self.net_config.network_settings.subnet_settings[0].name, + self.net_config.network_settings.subnet_settings[0].cidr, True) + + with self.assertRaises(Exception): + self.port = neutron_utils.create_port(self.neutron, self.os_creds, PortSettings( + name=self.port_name, network_name=self.net_config.network_settings.name, + ip_addrs=[{'subnet_name': self.net_config.network_settings.subnet_settings.name, + 'ip': '10.197.123.100'}])) + + +class NeutronUtilsSecurityGroupTests(OSComponentTestCase): + """ + Test for creating security groups via neutron_utils.py + """ + + def setUp(self): + guid = self.__class__.__name__ + '-' + str(uuid.uuid4()) + self.sec_grp_name = guid + 'name' + + self.security_group = None + self.security_group_rules = list() + self.neutron = neutron_utils.neutron_client(self.os_creds) + self.keystone = keystone_utils.keystone_client(self.os_creds) + + def tearDown(self): + """ + Cleans the remote OpenStack objects + """ + for rule in self.security_group_rules: + neutron_utils.delete_security_group_rule(self.neutron, rule) + + if self.security_group: + neutron_utils.delete_security_group(self.neutron, self.security_group) + + def test_create_delete_simple_sec_grp(self): + """ + Tests the neutron_utils.create_security_group() function + """ + sec_grp_settings = SecurityGroupSettings(name=self.sec_grp_name) + self.security_group = neutron_utils.create_security_group(self.neutron, self.keystone, sec_grp_settings) + + self.assertTrue(sec_grp_settings.name, self.security_group['security_group']['name']) + + sec_grp_get = neutron_utils.get_security_group(self.neutron, sec_grp_settings.name) + self.assertIsNotNone(sec_grp_get) + self.assertTrue(validation_utils.objects_equivalent( + self.security_group['security_group'], sec_grp_get['security_group'])) + + neutron_utils.delete_security_group(self.neutron, self.security_group) + sec_grp_get = neutron_utils.get_security_group(self.neutron, sec_grp_settings.name) + self.assertIsNone(sec_grp_get) + self.security_group = None + + def test_create_sec_grp_no_name(self): + """ + Tests the SecurityGroupSettings constructor and neutron_utils.create_security_group() function to ensure + that attempting to create a security group without a name will raise an exception + """ + with self.assertRaises(Exception): + sec_grp_settings = SecurityGroupSettings() + self.security_group = neutron_utils.create_security_group(self.neutron, self.keystone, sec_grp_settings) + + def test_create_sec_grp_no_rules(self): + """ + Tests the neutron_utils.create_security_group() function + """ + sec_grp_settings = SecurityGroupSettings(name=self.sec_grp_name, description='hello group') + self.security_group = neutron_utils.create_security_group(self.neutron, self.keystone, sec_grp_settings) + + self.assertTrue(sec_grp_settings.name, self.security_group['security_group']['name']) + self.assertTrue(sec_grp_settings.description, self.security_group['security_group']['description']) + + sec_grp_get = neutron_utils.get_security_group(self.neutron, sec_grp_settings.name) + self.assertIsNotNone(sec_grp_get) + self.assertTrue(validation_utils.objects_equivalent( + self.security_group['security_group'], sec_grp_get['security_group'])) + + def test_create_sec_grp_one_rule(self): + """ + Tests the neutron_utils.create_security_group() function + """ + + sec_grp_rule_settings = SecurityGroupRuleSettings(sec_grp_name=self.sec_grp_name, direction=Direction.ingress) + sec_grp_settings = SecurityGroupSettings(name=self.sec_grp_name, description='hello group', + rule_settings=[sec_grp_rule_settings]) + + self.security_group = neutron_utils.create_security_group(self.neutron, self.keystone, sec_grp_settings) + free_rules = neutron_utils.get_rules_by_security_group(self.neutron, self.security_group) + for free_rule in free_rules: + self.security_group_rules.append(free_rule) + + self.security_group_rules.append( + neutron_utils.create_security_group_rule(self.neutron, sec_grp_settings.rule_settings[0])) + + # Refresh object so it is populated with the newly added rule + self.security_group = neutron_utils.get_security_group(self.neutron, sec_grp_settings.name) + + rules = neutron_utils.get_rules_by_security_group(self.neutron, self.security_group) + + self.assertTrue(validation_utils.objects_equivalent(self.security_group_rules, rules)) + + self.assertTrue(sec_grp_settings.name, self.security_group['security_group']['name']) + self.assertTrue(sec_grp_settings.description, self.security_group['security_group']['description']) + + sec_grp_get = neutron_utils.get_security_group(self.neutron, sec_grp_settings.name) + self.assertIsNotNone(sec_grp_get) + self.assertTrue(validation_utils.objects_equivalent( + self.security_group['security_group'], sec_grp_get['security_group'])) + + +""" +Validation routines +""" + + +def validate_network(neutron, name, exists): + """ + Returns true if a network for a given name DOES NOT exist if the exists parameter is false conversely true. + Returns false if a network for a given name DOES exist if the exists parameter is true conversely false. + :param neutron: The neutron client + :param name: The expected network name + :param exists: Whether or not the network name should exist or not + :return: True/False + """ + network = neutron_utils.get_network(neutron, name) + if exists and network: + return True + if not exists and not network: + return True + return False + + +def validate_subnet(neutron, name, cidr, exists): + """ + Returns true if a subnet for a given name DOES NOT exist if the exists parameter is false conversely true. + Returns false if a subnet for a given name DOES exist if the exists parameter is true conversely false. + :param neutron: The neutron client + :param name: The expected subnet name + :param cidr: The expected CIDR value + :param exists: Whether or not the network name should exist or not + :return: True/False + """ + subnet = neutron_utils.get_subnet_by_name(neutron, name) + if exists and subnet: + return subnet.get('cidr') == cidr + if not exists and not subnet: + return True + return False + + +def validate_router(neutron, name, exists): + """ + Returns true if a router for a given name DOES NOT exist if the exists parameter is false conversely true. + Returns false if a router for a given name DOES exist if the exists parameter is true conversely false. + :param neutron: The neutron client + :param name: The expected router name + :param exists: Whether or not the network name should exist or not + :return: True/False + """ + router = neutron_utils.get_router_by_name(neutron, name) + if exists and router: + return True + return False + + +def validate_interface_router(interface_router, router, subnet): + """ + Returns true if the router ID & subnet ID have been properly included into the interface router object + :param interface_router: the object to validate + :param router: to validate against the interface_router + :param subnet: to validate against the interface_router + :return: True if both IDs match else False + """ + subnet_id = interface_router.get('subnet_id') + router_id = interface_router.get('port_id') + + return subnet.get('id') == subnet_id and router.get('id') == router_id + + +def validate_port(neutron, port_obj, this_port_name): + """ + Returns true if a port for a given name DOES NOT exist if the exists parameter is false conversely true. + Returns false if a port for a given name DOES exist if the exists parameter is true conversely false. + :param neutron: The neutron client + :param port_obj: The port object to lookup + :param this_port_name: The expected router name + :return: True/False + """ + ports = neutron.list_ports() + for port, port_insts in ports.iteritems(): + for inst in port_insts: + if inst['id'] == port_obj['port']['id']: + return inst['name'] == this_port_name + return False diff --git a/snaps/openstack/utils/tests/nova_utils_tests.py b/snaps/openstack/utils/tests/nova_utils_tests.py new file mode 100644 index 0000000..f6c9156 --- /dev/null +++ b/snaps/openstack/utils/tests/nova_utils_tests.py @@ -0,0 +1,208 @@ +# Copyright (c) 2016 Cable Television Laboratories, Inc. ("CableLabs") +# and others. 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 logging +import os +import uuid + +from Crypto.PublicKey import RSA + +from snaps.openstack.utils import nova_utils +from snaps.openstack.tests.os_source_file_test import OSComponentTestCase +from snaps.openstack.create_flavor import FlavorSettings + +__author__ = 'spisarski' + +logger = logging.getLogger('nova_utils_tests') + + +class NovaSmokeTests(OSComponentTestCase): + """ + Tests to ensure that the nova client can communicate with the cloud + """ + + def test_nova_connect_success(self): + """ + Tests to ensure that the proper credentials can connect. + """ + nova = nova_utils.nova_client(self.os_creds) + + # This should not throw an exception + nova.flavors.list() + + def test_nova_connect_fail(self): + """ + Tests to ensure that the improper credentials cannot connect. + """ + from snaps.openstack.os_credentials import OSCreds + + nova = nova_utils.nova_client( + OSCreds(username='user', password='pass', auth_url=self.os_creds.auth_url, + project_name=self.os_creds.project_name, proxy_settings=self.os_creds.proxy_settings)) + + # This should throw an exception + with self.assertRaises(Exception): + nova.flavors.list() + + +class NovaUtilsKeypairTests(OSComponentTestCase): + """ + Test basic nova keypair functionality + """ + + def setUp(self): + """ + Instantiates the CreateImage object that is responsible for downloading and creating an OS image file + within OpenStack + """ + guid = self.__class__.__name__ + '-' + str(uuid.uuid4()) + self.priv_key_file_path = 'tmp/' + guid + self.pub_key_file_path = self.priv_key_file_path + '.pub' + + self.nova = nova_utils.nova_client(self.os_creds) + self.keys = RSA.generate(1024) + self.public_key = self.keys.publickey().exportKey('OpenSSH') + self.keypair_name = guid + self.keypair = None + self.floating_ip = None + + def tearDown(self): + """ + Cleans the image and downloaded image file + """ + if self.keypair: + try: + nova_utils.delete_keypair(self.nova, self.keypair) + except: + pass + + try: + os.remove(self.priv_key_file_path) + except: + pass + + try: + os.remove(self.pub_key_file_path) + except: + pass + + if self.floating_ip: + nova_utils.delete_floating_ip(self.nova, self.floating_ip) + + def test_create_keypair(self): + """ + Tests the creation of an OpenStack keypair that does not exist. + """ + self.keypair = nova_utils.upload_keypair(self.nova, self.keypair_name, self.public_key) + result = nova_utils.keypair_exists(self.nova, self.keypair) + self.assertEquals(self.keypair, result) + keypair = nova_utils.get_keypair_by_name(self.nova, self.keypair_name) + self.assertEquals(self.keypair, keypair) + + def test_create_delete_keypair(self): + """ + Tests the creation of an OpenStack keypair that does not exist. + """ + self.keypair = nova_utils.upload_keypair(self.nova, self.keypair_name, self.public_key) + result = nova_utils.keypair_exists(self.nova, self.keypair) + self.assertEquals(self.keypair, result) + nova_utils.delete_keypair(self.nova, self.keypair) + result2 = nova_utils.keypair_exists(self.nova, self.keypair) + self.assertIsNone(result2) + + def test_create_key_from_file(self): + """ + Tests that the generated RSA keys are properly saved to files + :return: + """ + nova_utils.save_keys_to_files(self.keys, self.pub_key_file_path, self.priv_key_file_path) + self.keypair = nova_utils.upload_keypair_file(self.nova, self.keypair_name, self.pub_key_file_path) + pub_key = open(os.path.expanduser(self.pub_key_file_path)).read() + self.assertEquals(self.keypair.public_key, pub_key) + + def test_floating_ips(self): + """ + Tests the creation of a floating IP + :return: + """ + ips = nova_utils.get_floating_ips(self.nova) + self.assertIsNotNone(ips) + + self.floating_ip = nova_utils.create_floating_ip(self.nova, self.ext_net_name) + returned = nova_utils.get_floating_ip(self.nova, self.floating_ip) + self.assertEquals(self.floating_ip, returned) + + +class NovaUtilsFlavorTests(OSComponentTestCase): + """ + Test basic nova flavor functionality + """ + + def setUp(self): + """ + Instantiates the CreateImage object that is responsible for downloading and creating an OS image file + within OpenStack + """ + guid = self.__class__.__name__ + '-' + str(uuid.uuid4()) + self.flavor_settings = FlavorSettings(name=guid + '-name', flavor_id=guid + '-id', ram=1, disk=1, vcpus=1, + ephemeral=1, swap=2, rxtx_factor=3.0, is_public=False) + self.nova = nova_utils.nova_client(self.os_creds) + self.flavor = None + + def tearDown(self): + """ + Cleans the image and downloaded image file + """ + if self.flavor: + try: + nova_utils.delete_flavor(self.nova, self.flavor) + except: + pass + + def test_create_flavor(self): + """ + Tests the creation of an OpenStack keypair that does not exist. + """ + self.flavor = nova_utils.create_flavor(self.nova, self.flavor_settings) + self.validate_flavor() + + def test_create_delete_flavor(self): + """ + Tests the creation of an OpenStack keypair that does not exist. + """ + self.flavor = nova_utils.create_flavor(self.nova, self.flavor_settings) + self.validate_flavor() + nova_utils.delete_flavor(self.nova, self.flavor) + flavor = nova_utils.get_flavor_by_name(self.nova, self.flavor_settings.name) + self.assertIsNone(flavor) + + def validate_flavor(self): + """ + Validates the flavor_settings against the OpenStack flavor object + """ + self.assertIsNotNone(self.flavor) + self.assertEquals(self.flavor_settings.name, self.flavor.name) + self.assertEquals(self.flavor_settings.flavor_id, self.flavor.id) + self.assertEquals(self.flavor_settings.ram, self.flavor.ram) + self.assertEquals(self.flavor_settings.disk, self.flavor.disk) + self.assertEquals(self.flavor_settings.vcpus, self.flavor.vcpus) + self.assertEquals(self.flavor_settings.ephemeral, self.flavor.ephemeral) + + if self.flavor_settings.swap == 0: + self.assertEquals('', self.flavor.swap) + else: + self.assertEquals(self.flavor_settings.swap, self.flavor.swap) + + self.assertEquals(self.flavor_settings.rxtx_factor, self.flavor.rxtx_factor) + self.assertEquals(self.flavor_settings.is_public, self.flavor.is_public) |