diff options
| -rw-r--r-- | functest/tests/unit/utils/test_openstack_clean.py | 671 | ||||
| -rw-r--r-- | functest/tests/unit/utils/test_openstack_snapshot.py | 235 | ||||
| -rw-r--r-- | functest/tests/unit/utils/test_openstack_tacker.py | 455 |
3 files changed, 1361 insertions, 0 deletions
diff --git a/functest/tests/unit/utils/test_openstack_clean.py b/functest/tests/unit/utils/test_openstack_clean.py new file mode 100644 index 00000000..28eab4f8 --- /dev/null +++ b/functest/tests/unit/utils/test_openstack_clean.py @@ -0,0 +1,671 @@ +#!/usr/bin/env python + +# All rights reserved. This program and the accompanying materials +# are made available under the terms of the Apache License, Version 2.0 +# which accompanies this distribution, and is available at +# http://www.apache.org/licenses/LICENSE-2.0 + +import logging +import mock +import unittest + +from functest.utils import openstack_clean +from functest.tests.unit import test_utils + + +class OSCleanTesting(unittest.TestCase): + + logging.disable(logging.CRITICAL) + + def _get_instance(self, key): + mock_obj = mock.Mock() + attrs = {'id': 'id' + str(key), 'name': 'name' + str(key), + 'ip': 'ip' + str(key)} + mock_obj.configure_mock(**attrs) + return mock_obj + + def setUp(self): + self.client = mock.Mock() + self.test_list = [self._get_instance(1), self._get_instance(2)] + self.update_list = {'id1': 'name1', 'id2': 'name2'} + self.remove_list = {'id3': 'name3', 'id4': 'name4'} + self.test_dict_list = [{'id': 'id1', 'name': 'name1', 'ip': 'ip1', + 'router:external': False, + 'external_gateway_info': None}, + {'id': 'id2', 'name': 'name2', 'ip': 'ip2', + 'router:external': False, + 'external_gateway_info': None}] + self.routers = [mock.Mock()] + self.ports = [mock.Mock()] + + @mock.patch('functest.utils.openstack_clean.logger.debug') + def test_separator(self, mock_logger_debug): + openstack_clean.separator() + mock_logger_debug.assert_called_once_with("-----------------" + "-----------------" + "---------") + + @mock.patch('functest.utils.openstack_clean.logger.debug') + def test_remove_instances(self, mock_logger_debug): + with mock.patch('functest.utils.openstack_clean.os_utils' + '.get_instances', return_value=self.test_list): + openstack_clean.remove_instances(self.client, self.update_list) + mock_logger_debug.assert_any_call("Removing Nova instances...") + mock_logger_debug.assert_any_call(" > this is a default " + "instance and will " + "NOT be deleted.") + + @mock.patch('functest.utils.openstack_clean.logger.debug') + def test_remove_instances_missing_instances(self, mock_logger_debug): + with mock.patch('functest.utils.openstack_clean.os_utils' + '.get_instances', return_value=[]): + openstack_clean.remove_instances(self.client, self.update_list) + mock_logger_debug.assert_any_call("Removing Nova instances...") + mock_logger_debug.assert_any_call("No instances found.") + + @mock.patch('functest.utils.openstack_clean.logger.debug') + def test_remove_instances_delete_success(self, mock_logger_debug): + with mock.patch('functest.utils.openstack_clean.os_utils' + '.get_instances', return_value=self.test_list), \ + mock.patch('functest.utils.openstack_clean.os_utils' + '.delete_instance', return_value=True): + openstack_clean.remove_instances(self.client, self.remove_list) + mock_logger_debug.assert_any_call("Removing Nova instances...") + mock_logger_debug.assert_any_call(" > Request sent.") + mock_logger_debug.assert_any_call(test_utils.RegexMatch("Removing" + " instance" + " '\s*\S+'" + " ...")) + + @mock.patch('functest.utils.openstack_clean.logger.error') + @mock.patch('functest.utils.openstack_clean.logger.debug') + def test_remove_instances_delete_failed(self, mock_logger_debug, + mock_logger_error): + with mock.patch('functest.utils.openstack_clean.os_utils' + '.get_instances', return_value=self.test_list), \ + mock.patch('functest.utils.openstack_clean.os_utils' + '.delete_instance', return_value=False): + openstack_clean.remove_instances(self.client, self.remove_list) + mock_logger_debug.assert_any_call("Removing Nova instances...") + mock_logger_error.assert_any_call(test_utils. + RegexMatch("There has been a " + "problem removing " + "the instance \s*\S+" + "...")) + mock_logger_debug.assert_any_call(test_utils.RegexMatch("Removing" + " instance" + " '\s*\S+'" + " ...")) + + @mock.patch('functest.utils.openstack_clean.logger.debug') + def test_remove_images(self, mock_logger_debug): + with mock.patch('functest.utils.openstack_clean.os_utils' + '.get_images', return_value=self.test_list): + openstack_clean.remove_images(self.client, self.update_list) + mock_logger_debug.assert_any_call("Removing Glance images...") + mock_logger_debug.assert_any_call(" > this is a default " + "image and will " + "NOT be deleted.") + + @mock.patch('functest.utils.openstack_clean.logger.debug') + def test_remove_images_missing_images(self, mock_logger_debug): + with mock.patch('functest.utils.openstack_clean.os_utils' + '.get_images', return_value=[]): + openstack_clean.remove_images(self.client, self.update_list) + mock_logger_debug.assert_any_call("Removing Glance images...") + mock_logger_debug.assert_any_call("No images found.") + + @mock.patch('functest.utils.openstack_clean.logger.debug') + def test_remove_images_delete_success(self, mock_logger_debug): + with mock.patch('functest.utils.openstack_clean.os_utils' + '.get_images', return_value=self.test_list), \ + mock.patch('functest.utils.openstack_clean.os_utils' + '.delete_glance_image', return_value=True): + openstack_clean.remove_images(self.client, self.remove_list) + mock_logger_debug.assert_any_call("Removing Glance images...") + mock_logger_debug.assert_any_call(" > Done!") + mock_logger_debug.assert_any_call(test_utils. + RegexMatch("Removing image " + "\s*\S+," + " ID=\s*\S+ ...")) + + @mock.patch('functest.utils.openstack_clean.logger.error') + @mock.patch('functest.utils.openstack_clean.logger.debug') + def test_remove_images_delete_failed(self, mock_logger_debug, + mock_logger_error): + with mock.patch('functest.utils.openstack_clean.os_utils' + '.get_images', return_value=self.test_list), \ + mock.patch('functest.utils.openstack_clean.os_utils' + '.delete_glance_image', return_value=False): + openstack_clean.remove_images(self.client, self.remove_list) + mock_logger_debug.assert_any_call("Removing Glance images...") + mock_logger_error.assert_any_call(test_utils. + RegexMatch("There has been a " + "problem removing the" + "image \s*\S+...")) + mock_logger_debug.assert_any_call(test_utils. + RegexMatch("Removing image " + "\s*\S+," + " ID=\s*\S+ ...")) + + @mock.patch('functest.utils.openstack_clean.logger.debug') + def test_remove_volumes(self, mock_logger_debug): + with mock.patch('functest.utils.openstack_clean.os_utils' + '.get_volumes', return_value=self.test_list): + openstack_clean.remove_volumes(self.client, self.update_list) + mock_logger_debug.assert_any_call("Removing Cinder volumes...") + mock_logger_debug.assert_any_call(" > this is a default " + "volume and will " + "NOT be deleted.") + + @mock.patch('functest.utils.openstack_clean.logger.debug') + def test_remove_volumes_missing_volumes(self, mock_logger_debug): + with mock.patch('functest.utils.openstack_clean.os_utils' + '.get_volumes', return_value=[]): + openstack_clean.remove_volumes(self.client, self.update_list) + mock_logger_debug.assert_any_call("Removing Cinder volumes...") + mock_logger_debug.assert_any_call("No volumes found.") + + @mock.patch('functest.utils.openstack_clean.logger.debug') + def test_remove_volumes_delete_success(self, mock_logger_debug): + with mock.patch('functest.utils.openstack_clean.os_utils' + '.get_volumes', return_value=self.test_list), \ + mock.patch('functest.utils.openstack_clean.os_utils' + '.delete_volume', return_value=True): + openstack_clean.remove_volumes(self.client, self.remove_list) + mock_logger_debug.assert_any_call("Removing Cinder volumes...") + mock_logger_debug.assert_any_call(" > Done!") + mock_logger_debug.assert_any_call(test_utils. + RegexMatch("Removing cinder " + "volume \s*\S+ ...")) + + @mock.patch('functest.utils.openstack_clean.logger.error') + @mock.patch('functest.utils.openstack_clean.logger.debug') + def test_remove_volumes_delete_failed(self, mock_logger_debug, + mock_logger_error): + with mock.patch('functest.utils.openstack_clean.os_utils' + '.get_volumes', return_value=self.test_list), \ + mock.patch('functest.utils.openstack_clean.os_utils' + '.delete_volume', return_value=False): + openstack_clean.remove_volumes(self.client, self.remove_list) + mock_logger_debug.assert_any_call("Removing Cinder volumes...") + mock_logger_error.assert_any_call(test_utils. + RegexMatch("There has been a " + "problem removing " + "the " + "volume \s*\S+...")) + mock_logger_debug.assert_any_call(test_utils. + RegexMatch("Removing cinder " + "volume \s*\S+ ...")) + + @mock.patch('functest.utils.openstack_clean.logger.debug') + def test_remove_floatingips(self, mock_logger_debug): + with mock.patch('functest.utils.openstack_clean.os_utils' + '.get_floating_ips', return_value=self.test_list): + openstack_clean.remove_floatingips(self.client, self.update_list) + mock_logger_debug.assert_any_call("Removing floating IPs...") + mock_logger_debug.assert_any_call(" > this is a default " + "floating IP and will " + "NOT be deleted.") + + @mock.patch('functest.utils.openstack_clean.logger.debug') + def test_remove_floatingips_missing_floatingips(self, mock_logger_debug): + with mock.patch('functest.utils.openstack_clean.os_utils' + '.get_floating_ips', return_value=[]): + openstack_clean.remove_floatingips(self.client, self.update_list) + mock_logger_debug.assert_any_call("Removing floating IPs...") + mock_logger_debug.assert_any_call("No floating IPs found.") + + @mock.patch('functest.utils.openstack_clean.logger.debug') + def test_remove_floatingips_delete_success(self, mock_logger_debug): + with mock.patch('functest.utils.openstack_clean.os_utils' + '.get_floating_ips', return_value=self.test_list), \ + mock.patch('functest.utils.openstack_clean.os_utils' + '.delete_volume', return_value=True): + openstack_clean.remove_floatingips(self.client, self.remove_list) + mock_logger_debug.assert_any_call("Removing floating IPs...") + mock_logger_debug.assert_any_call(" > Done!") + mock_logger_debug.assert_any_call(test_utils. + RegexMatch("Removing floating " + "IP \s*\S+ ...")) + + @mock.patch('functest.utils.openstack_clean.logger.error') + @mock.patch('functest.utils.openstack_clean.logger.debug') + def test_remove_floatingips_delete_failed(self, mock_logger_debug, + mock_logger_error): + with mock.patch('functest.utils.openstack_clean.os_utils' + '.get_floating_ips', return_value=self.test_list), \ + mock.patch('functest.utils.openstack_clean.os_utils' + '.delete_floating_ip', return_value=False): + openstack_clean.remove_floatingips(self.client, self.remove_list) + mock_logger_debug.assert_any_call("Removing floating IPs...") + mock_logger_error.assert_any_call(test_utils. + RegexMatch("There has been a " + "problem removing " + "the floating IP " + "\s*\S+...")) + mock_logger_debug.assert_any_call(test_utils. + RegexMatch("Removing floating " + "IP \s*\S+ ...")) + + @mock.patch('functest.utils.openstack_clean.remove_routers') + @mock.patch('functest.utils.openstack_clean.remove_ports') + @mock.patch('functest.utils.openstack_clean.logger.debug') + def test_remove_networks(self, mock_logger_debug, + mock_remove_ports, + mock_remove_routers): + with mock.patch('functest.utils.openstack_clean.os_utils' + '.get_network_list', + return_value=self.test_dict_list), \ + mock.patch('functest.utils.openstack_clean.os_utils' + '.get_port_list', return_value=self.ports), \ + mock.patch('functest.utils.openstack_clean.os_utils' + '.get_router_list', return_value=self.routers): + openstack_clean.remove_networks(self.client, self.update_list, + self.update_list) + mock_logger_debug.assert_any_call("Removing Neutron objects") + mock_logger_debug.assert_any_call(" > this is a default " + "network and will " + "NOT be deleted.") + mock_remove_ports.assert_called_once_with(self.client, self.ports, + []) + mock_remove_routers.assert_called_once_with(self.client, + self.routers, + self.update_list) + + @mock.patch('functest.utils.openstack_clean.remove_routers') + @mock.patch('functest.utils.openstack_clean.remove_ports') + @mock.patch('functest.utils.openstack_clean.logger.debug') + def test_remove_networks_missing_networks(self, mock_logger_debug, + mock_remove_ports, + mock_remove_routers): + with mock.patch('functest.utils.openstack_clean.os_utils' + '.get_network_list', return_value=None), \ + mock.patch('functest.utils.openstack_clean.os_utils' + '.get_port_list', return_value=self.ports), \ + mock.patch('functest.utils.openstack_clean.os_utils' + '.get_router_list', return_value=self.routers): + openstack_clean.remove_networks(self.client, self.update_list, + self.update_list) + mock_logger_debug.assert_any_call("Removing Neutron objects") + mock_logger_debug.assert_any_call("There are no networks in the" + " deployment. ") + mock_remove_ports.assert_called_once_with(self.client, self.ports, + []) + mock_remove_routers.assert_called_once_with(self.client, + self.routers, + self.update_list) + + @mock.patch('functest.utils.openstack_clean.remove_routers') + @mock.patch('functest.utils.openstack_clean.remove_ports') + @mock.patch('functest.utils.openstack_clean.logger.debug') + def test_remove_networks_delete_success(self, mock_logger_debug, + mock_remove_ports, + mock_remove_routers): + + with mock.patch('functest.utils.openstack_clean.os_utils' + '.get_network_list', + return_value=self.test_dict_list), \ + mock.patch('functest.utils.openstack_clean.os_utils' + '.delete_neutron_net', return_value=True), \ + mock.patch('functest.utils.openstack_clean.os_utils' + '.get_port_list', return_value=self.ports), \ + mock.patch('functest.utils.openstack_clean.os_utils' + '.get_router_list', return_value=self.routers): + openstack_clean.remove_networks(self.client, self.remove_list, + self.remove_list) + mock_logger_debug.assert_any_call("Removing Neutron objects") + mock_logger_debug.assert_any_call(" > this network will be " + "deleted.") + mock_logger_debug.assert_any_call(" > Done!") + mock_logger_debug.assert_any_call(test_utils. + RegexMatch("Removing network " + "\s*\S+ ...")) + mock_remove_ports.assert_called_once_with(self.client, self.ports, + ['id1', 'id2']) + mock_remove_routers.assert_called_once_with(self.client, + self.routers, + self.remove_list) + + @mock.patch('functest.utils.openstack_clean.remove_routers') + @mock.patch('functest.utils.openstack_clean.remove_ports') + @mock.patch('functest.utils.openstack_clean.logger.error') + @mock.patch('functest.utils.openstack_clean.logger.debug') + def test_remove_networks_delete_failed(self, mock_logger_debug, + mock_logger_error, + mock_remove_ports, + mock_remove_routers): + with mock.patch('functest.utils.openstack_clean.os_utils' + '.get_network_list', + return_value=self.test_dict_list), \ + mock.patch('functest.utils.openstack_clean.os_utils' + '.delete_neutron_net', return_value=False), \ + mock.patch('functest.utils.openstack_clean.os_utils' + '.get_port_list', return_value=self.ports), \ + mock.patch('functest.utils.openstack_clean.os_utils' + '.get_router_list', return_value=self.routers): + openstack_clean.remove_networks(self.client, self.remove_list, + self.remove_list) + mock_logger_debug.assert_any_call("Removing Neutron objects") + mock_logger_error.assert_any_call(test_utils. + RegexMatch("There has been a" + " problem removing" + " the network \s*\S+" + "...")) + mock_logger_debug.assert_any_call(test_utils. + RegexMatch("Removing network " + "\s*\S+ ...")) + mock_remove_ports.assert_called_once_with(self.client, self.ports, + ['id1', 'id2']) + mock_remove_routers.assert_called_once_with(self.client, + self.routers, + self.remove_list) + + # TODO: ports + @mock.patch('functest.utils.openstack_clean.os_utils.update_neutron_port') + @mock.patch('functest.utils.openstack_clean.logger.error') + @mock.patch('functest.utils.openstack_clean.logger.debug') + def test_force_remove_port(self, mock_logger_debug, + mock_logger_error, + mock_update_neutron_port): + with mock.patch('functest.utils.openstack_clean.os_utils' + '.delete_neutron_port', + return_value=True): + openstack_clean.force_remove_port(self.client, 'id') + mock_logger_debug.assert_any_call(" > Done!") + mock_logger_debug.assert_any_call(test_utils. + RegexMatch("Clearing device_" + "owner for port " + "\s*\S+ ...")) + + @mock.patch('functest.utils.openstack_clean.os_utils.update_neutron_port') + @mock.patch('functest.utils.openstack_clean.logger.error') + @mock.patch('functest.utils.openstack_clean.logger.debug') + def test_force_remove_port_failed(self, mock_logger_debug, + mock_logger_error, + mock_update_neutron_port): + with mock.patch('functest.utils.openstack_clean.os_utils' + '.delete_neutron_port', + return_value=False): + openstack_clean.force_remove_port(self.client, 'id') + mock_logger_error.assert_any_call("There has been a " + "problem removing " + "the port id...") + mock_logger_debug.assert_any_call(test_utils. + RegexMatch("Clearing device_" + "owner for port " + "\s*\S+ ...")) + + @mock.patch('functest.utils.openstack_clean.logger.debug') + def test_remove_routers_missing_routers(self, mock_logger_debug): + with mock.patch('functest.utils.openstack_clean.os_utils' + '.delete_neutron_router', + return_value=True): + openstack_clean.remove_routers(self.client, self.test_dict_list, + self.remove_list) + mock_logger_debug.assert_any_call("Router is not connected" + " to anything." + "Ready to remove...") + mock_logger_debug.assert_any_call(" > Done!") + mock_logger_debug.assert_any_call(test_utils. + RegexMatch("Removing router " + "\s*\S+(\s*\S+) ...")) + + @mock.patch('functest.utils.openstack_clean.logger.error') + @mock.patch('functest.utils.openstack_clean.logger.debug') + def test_remove_routers_failed(self, mock_logger_debug, + mock_logger_error): + with mock.patch('functest.utils.openstack_clean.os_utils' + '.delete_neutron_router', + return_value=False): + openstack_clean.remove_routers(self.client, self.test_dict_list, + self.remove_list) + mock_logger_debug.assert_any_call("Router is not connected" + " to anything." + "Ready to remove...") + mock_logger_debug.assert_any_call(test_utils. + RegexMatch("Removing router " + "\s*\S+(\s*\S+) ...")) + mock_logger_error.assert_any_call(test_utils. + RegexMatch("There has been " + "a problem" + " removing the " + "router \s*\S+(" + "\s*\S+)...")) + + @mock.patch('functest.utils.openstack_clean.logger.error') + @mock.patch('functest.utils.openstack_clean.logger.debug') + def test_remove_missing_external_gateway(self, mock_logger_debug, + mock_logger_error): + with mock.patch('functest.utils.openstack_clean.os_utils' + '.delete_neutron_router', + return_value=False), \ + mock.patch('functest.utils.openstack_clean.os_utils' + '.remove_gateway_router', + return_value=False): + self.test_dict_list[0]['external_gateway_info'] = mock.Mock() + openstack_clean.remove_routers(self.client, self.test_dict_list, + self.remove_list) + mock_logger_debug.assert_any_call("Router has gateway to external" + " network.Removing link...") + mock_logger_error.assert_any_call("There has been a problem " + "removing the gateway...") + mock_logger_debug.assert_any_call(test_utils. + RegexMatch("Removing router " + "\s*\S+(\s*\S+) ...")) + mock_logger_error.assert_any_call(test_utils. + RegexMatch("There has been " + "a problem" + " removing the " + "router \s*\S+(" + "\s*\S+)...")) + + @mock.patch('functest.utils.openstack_clean.logger.debug') + def remove_security_groups(self, mock_logger_debug): + with mock.patch('functest.utils.openstack_clean.os_utils' + '.get_security_groups', + return_value=self.test_dict_list): + openstack_clean.remove_security_groups(self.client, + self.update_list) + mock_logger_debug.assert_any_call("Removing Security groups...") + mock_logger_debug.assert_any_call(" > this is a default " + "security group and will NOT " + "be deleted.") + + @mock.patch('functest.utils.openstack_clean.logger.debug') + def test_remove_security_groups_missing_sec_group(self, mock_logger_debug): + with mock.patch('functest.utils.openstack_clean.os_utils' + '.get_security_groups', return_value=[]): + openstack_clean.remove_security_groups(self.client, + self.update_list) + mock_logger_debug.assert_any_call("Removing Security groups...") + mock_logger_debug.assert_any_call("No security groups found.") + + @mock.patch('functest.utils.openstack_clean.logger.debug') + def test_remove_security_groups_delete_success(self, mock_logger_debug): + with mock.patch('functest.utils.openstack_clean.os_utils' + '.get_security_groups', + return_value=self.test_dict_list), \ + mock.patch('functest.utils.openstack_clean.os_utils' + '.delete_security_group', return_value=True): + openstack_clean.remove_security_groups(self.client, + self.remove_list) + mock_logger_debug.assert_any_call("Removing Security groups...") + mock_logger_debug.assert_any_call(" > Done!") + mock_logger_debug.assert_any_call(test_utils. + RegexMatch(" Removing \s*\S+" + "...")) + + @mock.patch('functest.utils.openstack_clean.logger.error') + @mock.patch('functest.utils.openstack_clean.logger.debug') + def test_remove_security_groups_delete_failed(self, mock_logger_debug, + mock_logger_error): + with mock.patch('functest.utils.openstack_clean.os_utils' + '.get_security_groups', + return_value=self.test_dict_list), \ + mock.patch('functest.utils.openstack_clean.os_utils' + '.delete_security_group', return_value=False): + openstack_clean.remove_security_groups(self.client, + self.remove_list) + mock_logger_debug.assert_any_call("Removing Security groups...") + mock_logger_error.assert_any_call(test_utils. + RegexMatch("There has been a " + "problem removing " + "the security group" + " \s*\S+...")) + mock_logger_debug.assert_any_call(test_utils. + RegexMatch(" Removing \s*\S+" + "...")) + + @mock.patch('functest.utils.openstack_clean.logger.debug') + def test_remove_users(self, mock_logger_debug): + with mock.patch('functest.utils.openstack_clean.os_utils' + '.get_users', return_value=self.test_list): + openstack_clean.remove_users(self.client, self.update_list) + mock_logger_debug.assert_any_call("Removing Users...") + mock_logger_debug.assert_any_call(" > this is a default " + "user and will " + "NOT be deleted.") + + @mock.patch('functest.utils.openstack_clean.logger.debug') + def test_remove_users_missing_users(self, mock_logger_debug): + with mock.patch('functest.utils.openstack_clean.os_utils' + '.get_users', return_value=None): + openstack_clean.remove_users(self.client, self.update_list) + mock_logger_debug.assert_any_call("Removing Users...") + mock_logger_debug.assert_any_call("There are no users in" + " the deployment. ") + + @mock.patch('functest.utils.openstack_clean.logger.debug') + def test_remove_users_delete_success(self, mock_logger_debug): + with mock.patch('functest.utils.openstack_clean.os_utils' + '.get_users', return_value=self.test_list), \ + mock.patch('functest.utils.openstack_clean.os_utils' + '.delete_user', return_value=True): + openstack_clean.remove_users(self.client, self.remove_list) + mock_logger_debug.assert_any_call("Removing Users...") + mock_logger_debug.assert_any_call(" > Done!") + mock_logger_debug.assert_any_call(test_utils. + RegexMatch(" Removing " + "\s*\S+...")) + + @mock.patch('functest.utils.openstack_clean.logger.error') + @mock.patch('functest.utils.openstack_clean.logger.debug') + def test_remove_users_delete_failed(self, mock_logger_debug, + mock_logger_error): + with mock.patch('functest.utils.openstack_clean.os_utils' + '.get_users', return_value=self.test_list), \ + mock.patch('functest.utils.openstack_clean.os_utils' + '.delete_user', return_value=False): + openstack_clean.remove_users(self.client, self.remove_list) + mock_logger_debug.assert_any_call("Removing Users...") + mock_logger_error.assert_any_call(test_utils. + RegexMatch("There has been a " + "problem removing " + "the user \s*\S+" + "...")) + mock_logger_debug.assert_any_call(test_utils. + RegexMatch(" Removing " + "\s*\S+...")) + + @mock.patch('functest.utils.openstack_clean.logger.debug') + def test_remove_tenants(self, mock_logger_debug): + with mock.patch('functest.utils.openstack_clean.os_utils' + '.get_tenants', return_value=self.test_list): + openstack_clean.remove_tenants(self.client, self.update_list) + mock_logger_debug.assert_any_call("Removing Tenants...") + mock_logger_debug.assert_any_call(" > this is a default" + " tenant and will " + "NOT be deleted.") + + @mock.patch('functest.utils.openstack_clean.logger.debug') + def test_remove_tenants_missing_tenants(self, mock_logger_debug): + with mock.patch('functest.utils.openstack_clean.os_utils' + '.get_tenants', return_value=None): + openstack_clean.remove_tenants(self.client, self.update_list) + mock_logger_debug.assert_any_call("Removing Tenants...") + mock_logger_debug.assert_any_call("There are no tenants in" + " the deployment. ") + + @mock.patch('functest.utils.openstack_clean.logger.debug') + def test_remove_tenants_delete_success(self, mock_logger_debug): + with mock.patch('functest.utils.openstack_clean.os_utils' + '.get_tenants', return_value=self.test_list), \ + mock.patch('functest.utils.openstack_clean.os_utils' + '.delete_tenant', return_value=True): + openstack_clean.remove_tenants(self.client, self.remove_list) + mock_logger_debug.assert_any_call("Removing Tenants...") + mock_logger_debug.assert_any_call(" > Done!") + mock_logger_debug.assert_any_call(test_utils. + RegexMatch(" Removing " + "\s*\S+...")) + + @mock.patch('functest.utils.openstack_clean.logger.error') + @mock.patch('functest.utils.openstack_clean.logger.debug') + def test_remove_tenants_delete_failed(self, mock_logger_debug, + mock_logger_error): + with mock.patch('functest.utils.openstack_clean.os_utils' + '.get_tenants', return_value=self.test_list), \ + mock.patch('functest.utils.openstack_clean.os_utils' + '.delete_tenant', return_value=False): + openstack_clean.remove_tenants(self.client, self.remove_list) + mock_logger_debug.assert_any_call("Removing Tenants...") + mock_logger_error.assert_any_call(test_utils. + RegexMatch("There has been a " + "problem removing " + "the tenant \s*\S+" + "...")) + mock_logger_debug.assert_any_call(test_utils. + RegexMatch(" Removing " + "\s*\S+...")) + + @mock.patch('functest.utils.openstack_clean.os_utils.get_cinder_client') + @mock.patch('functest.utils.openstack_clean.os_utils' + '.get_keystone_client') + @mock.patch('functest.utils.openstack_clean.os_utils' + '.get_neutron_client') + @mock.patch('functest.utils.openstack_clean.os_utils.get_nova_client') + @mock.patch('functest.utils.openstack_clean.os_utils.check_credentials', + return_value=True) + @mock.patch('functest.utils.openstack_clean.logger.info') + @mock.patch('functest.utils.openstack_clean.logger.debug') + def test_main_default(self, mock_logger_debug, mock_logger_info, + mock_creds, mock_nova, mock_neutron, + mock_keystone, mock_cinder): + + with mock.patch('functest.utils.openstack_clean.remove_instances') \ + as mock_remove_instances, \ + mock.patch('functest.utils.openstack_clean.remove_images') \ + as mock_remove_images, \ + mock.patch('functest.utils.openstack_clean.remove_volumes') \ + as mock_remove_volumes, \ + mock.patch('functest.utils.openstack_clean.remove_floatingips') \ + as mock_remove_floatingips, \ + mock.patch('functest.utils.openstack_clean.remove_networks') \ + as mock_remove_networks, \ + mock.patch('functest.utils.openstack_clean.' + 'remove_security_groups') \ + as mock_remove_security_groups, \ + mock.patch('functest.utils.openstack_clean.remove_users') \ + as mock_remove_users, \ + mock.patch('functest.utils.openstack_clean.remove_tenants') \ + as mock_remove_tenants, \ + mock.patch('functest.utils.openstack_clean.yaml.safe_load', + return_value=mock.Mock()), \ + mock.patch('__builtin__.open', mock.mock_open()) as m: + openstack_clean.main() + self.assertTrue(mock_remove_instances) + self.assertTrue(mock_remove_images) + self.assertTrue(mock_remove_volumes) + self.assertTrue(mock_remove_floatingips) + self.assertTrue(mock_remove_networks) + self.assertTrue(mock_remove_security_groups) + self.assertTrue(mock_remove_users) + self.assertTrue(mock_remove_tenants) + m.assert_called_once_with(openstack_clean.OS_SNAPSHOT_FILE) + mock_logger_info.assert_called_once_with("Cleaning OpenStack " + "resources...") + + +if __name__ == "__main__": + unittest.main(verbosity=2) diff --git a/functest/tests/unit/utils/test_openstack_snapshot.py b/functest/tests/unit/utils/test_openstack_snapshot.py new file mode 100644 index 00000000..52744db1 --- /dev/null +++ b/functest/tests/unit/utils/test_openstack_snapshot.py @@ -0,0 +1,235 @@ +#!/usr/bin/env python + +# All rights reserved. This program and the accompanying materials +# are made available under the terms of the Apache License, Version 2.0 +# which accompanies this distribution, and is available at +# http://www.apache.org/licenses/LICENSE-2.0 + +import logging +import mock +import unittest + +from functest.utils import openstack_snapshot + + +class OSTackerTesting(unittest.TestCase): + + logging.disable(logging.CRITICAL) + + def _get_instance(self, key): + mock_obj = mock.Mock() + attrs = {'id': 'id' + str(key), 'name': 'name' + str(key), + 'ip': 'ip' + str(key)} + mock_obj.configure_mock(**attrs) + return mock_obj + + def setUp(self): + self.client = mock.Mock() + self.test_list = [self._get_instance(1), self._get_instance(2)] + self.update_list = {'id1': 'name1', 'id2': 'name2'} + self.update_floatingips = {'id1': 'ip1', 'id2': 'ip2'} + self.test_dict_list = [{'id': 'id1', 'name': 'name1', 'ip': 'ip1'}, + {'id': 'id2', 'name': 'name2', 'ip': 'ip2'}] + + @mock.patch('functest.utils.openstack_snapshot.logger.info') + def test_separator(self, mock_logger_info): + openstack_snapshot.separator() + mock_logger_info.assert_called_once_with("-----------------" + "-----------------" + "---------") + + @mock.patch('functest.utils.openstack_snapshot.logger.debug') + def test_get_instances(self, mock_logger_debug): + with mock.patch('functest.utils.openstack_snapshot.os_utils' + '.get_instances', return_value=self.test_list): + resp = openstack_snapshot.get_instances(self.client) + mock_logger_debug.assert_called_once_with("Getting instances...") + self.assertDictEqual(resp, {'instances': self.update_list}) + + @mock.patch('functest.utils.openstack_snapshot.logger.debug') + def test_get_instances_missing_instances(self, mock_logger_debug): + with mock.patch('functest.utils.openstack_snapshot.os_utils' + '.get_instances', return_value=[]): + resp = openstack_snapshot.get_instances(self.client) + mock_logger_debug.assert_called_once_with("Getting instances...") + self.assertDictEqual(resp, {'instances': {}}) + + @mock.patch('functest.utils.openstack_snapshot.logger.debug') + def test_get_images(self, mock_logger_debug): + with mock.patch('functest.utils.openstack_snapshot.os_utils' + '.get_images', return_value=self.test_list): + resp = openstack_snapshot.get_images(self.client) + mock_logger_debug.assert_called_once_with("Getting images...") + self.assertDictEqual(resp, {'images': self.update_list}) + + @mock.patch('functest.utils.openstack_snapshot.logger.debug') + def test_get_images_missing_images(self, mock_logger_debug): + with mock.patch('functest.utils.openstack_snapshot.os_utils' + '.get_images', return_value=[]): + resp = openstack_snapshot.get_images(self.client) + mock_logger_debug.assert_called_once_with("Getting images...") + self.assertDictEqual(resp, {'images': {}}) + + @mock.patch('functest.utils.openstack_snapshot.logger.debug') + def test_get_volumes(self, mock_logger_debug): + with mock.patch('functest.utils.openstack_snapshot.os_utils' + '.get_volumes', return_value=self.test_list): + resp = openstack_snapshot.get_volumes(self.client) + mock_logger_debug.assert_called_once_with("Getting volumes...") + self.assertDictEqual(resp, {'volumes': self.update_list}) + + @mock.patch('functest.utils.openstack_snapshot.logger.debug') + def test_get_volumes_missing_volumes(self, mock_logger_debug): + with mock.patch('functest.utils.openstack_snapshot.os_utils' + '.get_volumes', return_value=[]): + resp = openstack_snapshot.get_volumes(self.client) + mock_logger_debug.assert_called_once_with("Getting volumes...") + self.assertDictEqual(resp, {'volumes': {}}) + + @mock.patch('functest.utils.openstack_snapshot.logger.debug') + def test_get_networks(self, mock_logger_debug): + with mock.patch('functest.utils.openstack_snapshot.os_utils' + '.get_network_list', return_value=self.test_dict_list): + resp = openstack_snapshot.get_networks(self.client) + mock_logger_debug.assert_called_once_with("Getting networks") + self.assertDictEqual(resp, {'networks': self.update_list}) + + @mock.patch('functest.utils.openstack_snapshot.logger.debug') + def test_get_networks_missing_networks(self, mock_logger_debug): + with mock.patch('functest.utils.openstack_snapshot.os_utils' + '.get_network_list', return_value=[]): + resp = openstack_snapshot.get_networks(self.client) + mock_logger_debug.assert_called_once_with("Getting networks") + self.assertDictEqual(resp, {'networks': {}}) + + @mock.patch('functest.utils.openstack_snapshot.logger.debug') + def test_get_routers(self, mock_logger_debug): + with mock.patch('functest.utils.openstack_snapshot.os_utils' + '.get_router_list', return_value=self.test_dict_list): + resp = openstack_snapshot.get_routers(self.client) + mock_logger_debug.assert_called_once_with("Getting routers") + self.assertDictEqual(resp, {'routers': self.update_list}) + + @mock.patch('functest.utils.openstack_snapshot.logger.debug') + def test_get_routers_missing_routers(self, mock_logger_debug): + with mock.patch('functest.utils.openstack_snapshot.os_utils' + '.get_router_list', return_value=[]): + resp = openstack_snapshot.get_routers(self.client) + mock_logger_debug.assert_called_once_with("Getting routers") + self.assertDictEqual(resp, {'routers': {}}) + + @mock.patch('functest.utils.openstack_snapshot.logger.debug') + def test_get_secgroups(self, mock_logger_debug): + with mock.patch('functest.utils.openstack_snapshot.os_utils' + '.get_security_groups', + return_value=self.test_dict_list): + resp = openstack_snapshot.get_security_groups(self.client) + mock_logger_debug.assert_called_once_with("Getting Security " + "groups...") + self.assertDictEqual(resp, {'secgroups': self.update_list}) + + @mock.patch('functest.utils.openstack_snapshot.logger.debug') + def test_get_secgroups_missing_secgroups(self, mock_logger_debug): + with mock.patch('functest.utils.openstack_snapshot.os_utils' + '.get_security_groups', return_value=[]): + resp = openstack_snapshot.get_security_groups(self.client) + mock_logger_debug.assert_called_once_with("Getting Security " + "groups...") + self.assertDictEqual(resp, {'secgroups': {}}) + + @mock.patch('functest.utils.openstack_snapshot.logger.debug') + def test_get_floatingips(self, mock_logger_debug): + with mock.patch('functest.utils.openstack_snapshot.os_utils' + '.get_floating_ips', return_value=self.test_list): + resp = openstack_snapshot.get_floatinips(self.client) + mock_logger_debug.assert_called_once_with("Getting Floating " + "IPs...") + self.assertDictEqual(resp, {'floatingips': + self.update_floatingips}) + + @mock.patch('functest.utils.openstack_snapshot.logger.debug') + def test_get_floatingips_missing_floatingips(self, mock_logger_debug): + with mock.patch('functest.utils.openstack_snapshot.os_utils' + '.get_floating_ips', return_value=[]): + resp = openstack_snapshot.get_floatinips(self.client) + mock_logger_debug.assert_called_once_with("Getting Floating " + "IPs...") + self.assertDictEqual(resp, {'floatingips': {}}) + + @mock.patch('functest.utils.openstack_snapshot.logger.debug') + def test_get_users(self, mock_logger_debug): + with mock.patch('functest.utils.openstack_snapshot.os_utils' + '.get_users', return_value=self.test_list): + resp = openstack_snapshot.get_users(self.client) + mock_logger_debug.assert_called_once_with("Getting users...") + self.assertDictEqual(resp, {'users': self.update_list}) + + @mock.patch('functest.utils.openstack_snapshot.logger.debug') + def test_get_users_missing_users(self, mock_logger_debug): + with mock.patch('functest.utils.openstack_snapshot.os_utils' + '.get_users', return_value=[]): + resp = openstack_snapshot.get_users(self.client) + mock_logger_debug.assert_called_once_with("Getting users...") + self.assertDictEqual(resp, {'users': {}}) + + @mock.patch('functest.utils.openstack_snapshot.logger.debug') + def test_get_tenants(self, mock_logger_debug): + with mock.patch('functest.utils.openstack_snapshot.os_utils' + '.get_tenants', return_value=self.test_list): + resp = openstack_snapshot.get_tenants(self.client) + mock_logger_debug.assert_called_once_with("Getting tenants...") + self.assertDictEqual(resp, {'tenants': self.update_list}) + + @mock.patch('functest.utils.openstack_snapshot.logger.debug') + def test_get_tenants_missing_tenants(self, mock_logger_debug): + with mock.patch('functest.utils.openstack_snapshot.os_utils' + '.get_tenants', return_value=[]): + resp = openstack_snapshot.get_tenants(self.client) + mock_logger_debug.assert_called_once_with("Getting tenants...") + self.assertDictEqual(resp, {'tenants': {}}) + + @mock.patch('functest.utils.openstack_snapshot.os_utils.get_cinder_client') + @mock.patch('functest.utils.openstack_snapshot.os_utils' + '.get_keystone_client') + @mock.patch('functest.utils.openstack_snapshot.os_utils' + '.get_neutron_client') + @mock.patch('functest.utils.openstack_snapshot.os_utils.get_nova_client') + @mock.patch('functest.utils.openstack_snapshot.os_utils.check_credentials') + @mock.patch('functest.utils.openstack_snapshot.logger.info') + @mock.patch('functest.utils.openstack_snapshot.logger.debug') + def test_main_default(self, mock_logger_debug, mock_logger_info, + mock_creds, mock_nova, mock_neutron, + mock_keystone, mock_cinder): + with mock.patch('functest.utils.openstack_snapshot.get_instances', + return_value=self.update_list), \ + mock.patch('functest.utils.openstack_snapshot.get_images', + return_value=self.update_list), \ + mock.patch('functest.utils.openstack_snapshot.get_images', + return_value=self.update_list), \ + mock.patch('functest.utils.openstack_snapshot.get_volumes', + return_value=self.update_list), \ + mock.patch('functest.utils.openstack_snapshot.get_networks', + return_value=self.update_list), \ + mock.patch('functest.utils.openstack_snapshot.get_routers', + return_value=self.update_list), \ + mock.patch('functest.utils.openstack_snapshot.get_security_groups', + return_value=self.update_list), \ + mock.patch('functest.utils.openstack_snapshot.get_floatinips', + return_value=self.update_floatingips), \ + mock.patch('functest.utils.openstack_snapshot.get_users', + return_value=self.update_list), \ + mock.patch('functest.utils.openstack_snapshot.get_tenants', + return_value=self.update_list), \ + mock.patch('__builtin__.open', mock.mock_open()) as m: + openstack_snapshot.main() + mock_logger_info.assert_called_once_with("Generating OpenStack " + "snapshot...") + m.assert_called_once_with(openstack_snapshot.OS_SNAPSHOT_FILE, + 'w+') + mock_logger_debug.assert_any_call("NOTE: These objects will " + "NOT be deleted after " + + "running the test.") + + +if __name__ == "__main__": + unittest.main(verbosity=2) diff --git a/functest/tests/unit/utils/test_openstack_tacker.py b/functest/tests/unit/utils/test_openstack_tacker.py new file mode 100644 index 00000000..a8330c0e --- /dev/null +++ b/functest/tests/unit/utils/test_openstack_tacker.py @@ -0,0 +1,455 @@ +#!/usr/bin/env python + +# All rights reserved. This program and the accompanying materials +# are made available under the terms of the Apache License, Version 2.0 +# which accompanies this distribution, and is available at +# http://www.apache.org/licenses/LICENSE-2.0 + +import logging +import mock +import unittest + +from functest.utils import openstack_tacker +from functest.tests.unit import test_utils + + +class OSTackerTesting(unittest.TestCase): + + logging.disable(logging.CRITICAL) + + def setUp(self): + self.tacker_client = mock.Mock() + self.getresponse = {'vnfds': [{'id': 'test_id'}], + 'vnfs': [{'id': 'test_id'}], + 'sfcs': [{'id': 'test_id'}]} + self.vnfdlist = {'vnfds': [{'id': 'test_vnfd1'}, {'id': 'test_vnfd2'}]} + self.vnflist = {'vnfs': [{'id': 'test_vnf1'}, {'id': 'test_vnf2'}]} + self.sfclist = {'sfcs': [{'id': 'test_sfc1'}, {'id': 'test_sfc2'}]} + self.sfc_classifierlist = {'sfc_classifiers': [{'id': 'test_sfc_cl1'}, + {'id': 'test_sfc_cl2'}]} + + self.createvnfd = {"vnfd": {"attributes": {"vnfd": 'vnfd_body'}}} + self.createvnf = {"vnf": {"attributes": {"vnf": 'vnf_body'}}} + self.createsfc = {"sfc": {"attributes": {"sfc": 'sfc_body'}}} + self.createsfc_clf = {"sfc_classifier": {"attributes": + {"sfc_clf": 'sfc_clf_body'}}} + + self.resource_type = 'vnfd' + self.resource_name = 'resource_name' + self.tosca_file = 'test_tosca_file' + self.vnfd = 'test_vnfd' + self.vnf = 'test_vnf' + self.sfc = 'test_sfc' + self.sfc_clf = 'test_sfc_clf' + + def _get_creds(self): + cred_dict = { + 'OS_USERNAME': 'username', + 'OS_PASSWORD': 'password', + 'OS_AUTH_URL': 'auth_url', + 'OS_TENANT_NAME': 'tenant_name', + 'OS_USER_DOMAIN_NAME': 'user_domain_name', + 'OS_PROJECT_DOMAIN_NAME': 'project_domain_name', + 'OS_PROJECT_NAME': 'project_name', + 'OS_ENDPOINT_TYPE': 'endpoint_type', + 'OS_REGION_NAME': 'region_name' + } + return cred_dict + + def test_get_id_from_name(self): + with mock.patch.object(self.tacker_client, 'get', + return_value=self.getresponse): + resp = openstack_tacker.get_id_from_name(self.tacker_client, + self.resource_type, + self.resource_name) + self.assertEqual(resp, 'test_id') + + @mock.patch('functest.utils.openstack_tacker.logger.error') + def test_get_id_from_name_exception(self, mock_logger_error): + with mock.patch.object(self.tacker_client, 'get', + side_effect=Exception): + resp = openstack_tacker.get_id_from_name(self.tacker_client, + self.resource_type, + self.resource_name) + self.assertIsNone(resp) + mock_logger_error.assert_called_once_with(test_utils. + SubstrMatch("Error [get" + "_id_from_" + "name(tacker" + "_client" + ", resource_" + "type, " + "resource_" + "name)]:")) + + @mock.patch('functest.utils.openstack_tacker.get_id_from_name') + def test_get_vnfd_id(self, mock_get_id_from_name): + openstack_tacker.get_vnfd_id(self.tacker_client, self.resource_name) + mock_get_id_from_name.assert_called_once_with(self.tacker_client, + 'vnfd', + self.resource_name) + + @mock.patch('functest.utils.openstack_tacker.get_id_from_name') + def test_get_vnf_id(self, mock_get_id_from_name): + openstack_tacker.get_vnf_id(self.tacker_client, self.resource_name) + mock_get_id_from_name.assert_called_once_with(self.tacker_client, + 'vnf', + self.resource_name) + + @mock.patch('functest.utils.openstack_tacker.get_id_from_name') + def test_get_sfc_id(self, mock_get_id_from_name): + openstack_tacker.get_sfc_id(self.tacker_client, self.resource_name) + mock_get_id_from_name.assert_called_once_with(self.tacker_client, + 'sfc', + self.resource_name) + + @mock.patch('functest.utils.openstack_tacker.get_id_from_name') + def test_get_sfc_classifier_id(self, mock_get_id_from_name): + openstack_tacker.get_sfc_classifier_id(self.tacker_client, + self.resource_name) + mock_get_id_from_name.assert_called_once_with(self.tacker_client, + 'sfc-classifier', + self.resource_name) + + def test_list_vnfds(self): + with mock.patch.object(self.tacker_client, 'list_vnfds', + return_value=self.vnfdlist): + resp = openstack_tacker.list_vnfds(self.tacker_client, + verbose=False) + self.assertEqual(resp, ['test_vnfd1', 'test_vnfd2']) + + def test_list_vnfds_verbose(self): + with mock.patch.object(self.tacker_client, 'list_vnfds', + return_value=self.vnfdlist): + resp = openstack_tacker.list_vnfds(self.tacker_client, + verbose=True) + self.assertEqual(resp, self.vnfdlist) + + @mock.patch('functest.utils.openstack_tacker.logger.error') + def test_list_vnfds_exception(self, mock_logger_error): + with mock.patch.object(self.tacker_client, 'list_vnfds', + side_effect=Exception): + resp = openstack_tacker.list_vnfds(self.tacker_client, + verbose=False) + mock_logger_error.assert_called_once_with(test_utils. + SubstrMatch("Error" + " [list" + "_vnfds(" + "tacker_" + "client)]:")) + self.assertIsNone(resp) + + def test_create_vnfd_missing_file(self): + with mock.patch.object(self.tacker_client, 'create_vnfd', + return_value=self.createvnfd): + resp = openstack_tacker.create_vnfd(self.tacker_client, + tosca_file=None) + self.assertEqual(resp, self.createvnfd) + + @mock.patch('functest.utils.openstack_tacker.logger.error') + def test_create_vnfd_default(self, mock_logger_error): + with mock.patch.object(self.tacker_client, 'create_vnfd', + return_value=self.createvnfd), \ + mock.patch('__builtin__.open', mock.mock_open(read_data='1')) \ + as m: + resp = openstack_tacker.create_vnfd(self.tacker_client, + tosca_file=self.tosca_file) + m.assert_called_once_with(self.tosca_file) + mock_logger_error.assert_called_once_with('1') + self.assertEqual(resp, self.createvnfd) + + @mock.patch('functest.utils.openstack_tacker.logger.exception') + def test_create_vnfd_exception(self, mock_logger_excep): + with mock.patch.object(self.tacker_client, 'create_vnfd', + side_effect=Exception): + resp = openstack_tacker.create_vnfd(self.tacker_client, + tosca_file=self.tosca_file) + mock_logger_excep.assert_called_once_with(test_utils. + SubstrMatch("Error" + " [create" + "_vnfd(" + "tacker_" + "client, " + "'%s')]" + % self. + tosca_file)) + self.assertIsNone(resp) + + def test_delete_vnfd(self): + with mock.patch('functest.utils.openstack_tacker.get_vnfd_id', + return_value=self.vnfd), \ + mock.patch.object(self.tacker_client, 'delete_vnfd', + return_value=self.vnfd): + resp = openstack_tacker.delete_vnfd(self.tacker_client, + vnfd_id='vnfd_id', + vnfd_name=self.vnfd) + self.assertEqual(resp, self.vnfd) + + # TODO: Exception('You need to provide an VNFD' + # 'id or name') AssertionError + + @mock.patch('functest.utils.openstack_tacker.logger.error') + def test_delete_vnfd_exception(self, mock_logger_error): + with mock.patch('functest.utils.openstack_tacker.get_vnfd_id', + return_value=self.vnfd), \ + mock.patch.object(self.tacker_client, 'delete_vnfd', + side_effect=Exception): + resp = openstack_tacker.delete_vnfd(self.tacker_client, + vnfd_id=None, + vnfd_name=None) + self.assertIsNone(resp) + self.assertTrue(mock_logger_error.called) + + def test_list_vnfs(self): + with mock.patch.object(self.tacker_client, 'list_vnfs', + return_value=self.vnflist): + resp = openstack_tacker.list_vnfs(self.tacker_client, + verbose=False) + self.assertEqual(resp, ['test_vnf1', 'test_vnf2']) + + def test_list_vnfs_verbose(self): + with mock.patch.object(self.tacker_client, 'list_vnfs', + return_value=self.vnflist): + resp = openstack_tacker.list_vnfs(self.tacker_client, + verbose=True) + self.assertEqual(resp, self.vnflist) + + @mock.patch('functest.utils.openstack_tacker.logger.error') + def test_list_vnfs_exception(self, mock_logger_error): + with mock.patch.object(self.tacker_client, 'list_vnfs', + side_effect=Exception): + resp = openstack_tacker.list_vnfs(self.tacker_client, + verbose=False) + mock_logger_error.assert_called_once_with(test_utils. + SubstrMatch("Error" + " [list" + "_vnfs(" + "tacker_" + "client)]:")) + self.assertIsNone(resp) + + def test_create_vnf_default(self): + with mock.patch.object(self.tacker_client, 'create_vnf', + return_value=self.createvnf), \ + mock.patch('functest.utils.openstack_tacker.get_vnfd_id', + return_value=self.vnf): + resp = openstack_tacker.create_vnf(self.tacker_client, + vnf_name=self.vnf, + vnfd_id='vnfd_id', + vnfd_name=self.vnfd) + self.assertEqual(resp, self.createvnf) + + @mock.patch('functest.utils.openstack_tacker.logger.error') + def test_create_vnf_exception(self, mock_logger_error): + with mock.patch.object(self.tacker_client, 'create_vnf', + side_effect=Exception): + resp = openstack_tacker.create_vnf(self.tacker_client, + vnf_name=self.vnf, + vnfd_id='vnfd_id', + vnfd_name=self.vnfd) + mock_logger_error.assert_called_once_with(test_utils. + SubstrMatch("error" + " [create" + "_vnf(" + "tacker_" + "client")) + self.assertIsNone(resp) + + # TODO: wait_for_vnf + + def test_delete_vnf(self): + with mock.patch('functest.utils.openstack_tacker.get_vnf_id', + return_value=self.vnf), \ + mock.patch.object(self.tacker_client, 'delete_vnf', + return_value=self.vnf): + resp = openstack_tacker.delete_vnf(self.tacker_client, + vnf_id='vnf_id', + vnf_name=self.vnf) + self.assertEqual(resp, self.vnf) + + # TODO: Exception('You need to provide an VNF' + # 'classifier id or name') AssertionError + + @mock.patch('functest.utils.openstack_tacker.logger.error') + def test_delete_vnf_exception(self, mock_logger_error): + with mock.patch('functest.utils.openstack_tacker.get_vnf_id', + return_value=self.vnf), \ + mock.patch.object(self.tacker_client, 'delete_vnf', + side_effect=Exception): + resp = openstack_tacker.delete_vnf(self.tacker_client, + vnf_id=None, + vnf_name=None) + self.assertIsNone(resp) + self.assertTrue(mock_logger_error.called) + + def test_list_sfcs(self): + with mock.patch.object(self.tacker_client, 'list_sfcs', + return_value=self.sfclist): + resp = openstack_tacker.list_sfcs(self.tacker_client, + verbose=False) + self.assertEqual(resp, ['test_sfc1', 'test_sfc2']) + + def test_list_sfcs_verbose(self): + with mock.patch.object(self.tacker_client, 'list_sfcs', + return_value=self.sfclist): + resp = openstack_tacker.list_sfcs(self.tacker_client, + verbose=True) + self.assertEqual(resp, self.sfclist) + + @mock.patch('functest.utils.openstack_tacker.logger.error') + def test_list_sfcs_exception(self, mock_logger_error): + with mock.patch.object(self.tacker_client, 'list_sfcs', + side_effect=Exception): + resp = openstack_tacker.list_sfcs(self.tacker_client, + verbose=False) + mock_logger_error.assert_called_once_with(test_utils. + SubstrMatch("Error" + " [list" + "_sfcs(" + "tacker_" + "client)]:")) + self.assertIsNone(resp) + + def test_create_sfc_default(self): + with mock.patch.object(self.tacker_client, 'create_sfc', + return_value=self.createsfc), \ + mock.patch('functest.utils.openstack_tacker.get_vnf_id', + return_value=self.vnf): + resp = openstack_tacker.create_sfc(self.tacker_client, + sfc_name=self.sfc, + chain_vnf_ids=['chain_vnf_id'], + chain_vnf_names=[self.vnf]) + self.assertEqual(resp, self.createsfc) + + @mock.patch('functest.utils.openstack_tacker.logger.error') + def test_create_sfc_exception(self, mock_logger_error): + with mock.patch.object(self.tacker_client, 'create_sfc', + side_effect=Exception): + resp = openstack_tacker.create_sfc(self.tacker_client, + sfc_name=self.sfc, + chain_vnf_ids=['chain_vnf_id'], + chain_vnf_names=[self.vnf]) + mock_logger_error.assert_called_once_with(test_utils. + SubstrMatch("error" + " [create" + "_sfc(" + "tacker_" + "client")) + self.assertIsNone(resp) + + def test_delete_sfc(self): + with mock.patch('functest.utils.openstack_tacker.get_sfc_id', + return_value=self.sfc), \ + mock.patch.object(self.tacker_client, 'delete_sfc', + return_value=self.sfc): + resp = openstack_tacker.delete_sfc(self.tacker_client, + sfc_id='sfc_id', + sfc_name=self.sfc) + self.assertEqual(resp, self.sfc) + + # TODO: Exception('You need to provide an SFC' + # 'id or name') AssertionError + + @mock.patch('functest.utils.openstack_tacker.logger.error') + def test_delete_sfc_exception(self, mock_logger_error): + with mock.patch('functest.utils.openstack_tacker.get_sfc_id', + return_value=self.sfc), \ + mock.patch.object(self.tacker_client, 'delete_sfc', + side_effect=Exception): + resp = openstack_tacker.delete_sfc(self.tacker_client, + sfc_id=None, + sfc_name=None) + self.assertIsNone(resp) + self.assertTrue(mock_logger_error.called) + + def test_list_sfc_classifiers(self): + with mock.patch.object(self.tacker_client, 'list_sfc_classifiers', + return_value=self.sfc_classifierlist): + resp = openstack_tacker.list_sfc_classifiers(self.tacker_client, + verbose=False) + self.assertEqual(resp, ['test_sfc_cl1', 'test_sfc_cl2']) + + def test_list_sfc_classifiers_verbose(self): + with mock.patch.object(self.tacker_client, 'list_sfc_classifiers', + return_value=self.sfc_classifierlist): + resp = openstack_tacker.list_sfc_classifiers(self.tacker_client, + verbose=True) + self.assertEqual(resp, self.sfc_classifierlist) + + @mock.patch('functest.utils.openstack_tacker.logger.error') + def test_list_sfc_classifiers_exception(self, mock_logger_error): + with mock.patch.object(self.tacker_client, 'list_sfc_classifiers', + side_effect=Exception): + resp = openstack_tacker.list_sfc_classifiers(self.tacker_client, + verbose=False) + mock_logger_error.assert_called_once_with(test_utils. + SubstrMatch("Error" + " [list" + "_sfc_cl" + "assifiers(" + "tacker_" + "client)]:")) + self.assertIsNone(resp) + + def test_create_sfc_classifier_default(self): + with mock.patch.object(self.tacker_client, 'create_sfc_classifier', + return_value=self.createsfc_clf), \ + mock.patch('functest.utils.openstack_tacker.get_sfc_id', + return_value=self.sfc): + cl = self.sfc_clf + resp = openstack_tacker.create_sfc_classifier(self.tacker_client, + sfc_clf_name=cl, + sfc_id='sfc_id', + sfc_name=self.sfc) + self.assertEqual(resp, self.createsfc_clf) + + @mock.patch('functest.utils.openstack_tacker.logger.error') + def test_sfc_classifier_exception(self, mock_logger_error): + with mock.patch.object(self.tacker_client, 'create_sfc_classifier', + side_effect=Exception): + cl = self.sfc_clf + resp = openstack_tacker.create_sfc_classifier(self.tacker_client, + sfc_clf_name=cl, + sfc_id='sfc_id', + sfc_name=self.sfc) + mock_logger_error.assert_called_once_with(test_utils. + SubstrMatch("error" + " [create" + "_sfc_cl" + "assifier(" + "tacker_" + "client")) + self.assertIsNone(resp) + + def test_delete_sfc_classifier(self): + with mock.patch('functest.utils.openstack_tacker.get_sfc_' + 'classifier_id', + return_value=self.sfc_clf), \ + mock.patch.object(self.tacker_client, 'delete_sfc_classifier', + return_value=self.sfc_clf): + cl = self.sfc_clf + resp = openstack_tacker.delete_sfc_classifier(self.tacker_client, + sfc_clf_id='sfc_id', + sfc_clf_name=cl) + self.assertEqual(resp, cl) + + # TODO: Exception('You need to provide an SFC' + # 'classifier id or name') AssertionError + + @mock.patch('functest.utils.openstack_tacker.logger.error') + def test_delete_sfc_classifier_exception(self, mock_logger_error): + with mock.patch('functest.utils.openstack_tacker.get_sfc_' + 'classifier_id', + return_value=self.sfc_clf), \ + mock.patch.object(self.tacker_client, 'delete_sfc_classifier', + side_effect=Exception): + cl = self.sfc_clf + resp = openstack_tacker.delete_sfc_classifier(self.tacker_client, + sfc_clf_id='sfc_id', + sfc_clf_name=cl) + self.assertIsNone(resp) + self.assertTrue(mock_logger_error.called) + + +if __name__ == "__main__": + unittest.main(verbosity=2) |