X-Git-Url: https://osm.etsi.org/gitweb/?a=blobdiff_plain;f=RO-VIM-openstack%2Fosm_rovim_openstack%2Ftests%2Ftest_vimconn_openstack.py;h=c48d1843a7b309fc63771a61374a8cfd738b5cfe;hb=refs%2Fchanges%2F28%2F12628%2F12;hp=09991244a2e748571bb39bc7bbc33993fb8b6cae;hpb=09dcc583ec5ab04e80e7a4349dcc4061ed794b47;p=osm%2FRO.git diff --git a/RO-VIM-openstack/osm_rovim_openstack/tests/test_vimconn_openstack.py b/RO-VIM-openstack/osm_rovim_openstack/tests/test_vimconn_openstack.py index 09991244..c48d1843 100644 --- a/RO-VIM-openstack/osm_rovim_openstack/tests/test_vimconn_openstack.py +++ b/RO-VIM-openstack/osm_rovim_openstack/tests/test_vimconn_openstack.py @@ -23,19 +23,50 @@ This module contains unit tests for the OpenStack VIM connector Run this directly with python2 or python3. """ - import copy +from copy import deepcopy import logging import unittest import mock +from mock import MagicMock, patch from neutronclient.v2_0.client import Client from osm_ro_plugin import vimconn +from osm_ro_plugin.vimconn import VimConnConnectionException, VimConnException from osm_rovim_openstack.vimconn_openstack import vimconnector __author__ = "Igor D.C." __date__ = "$23-aug-2017 23:59:59$" +# Variables Used in TestNewVmInstance Class +name = "basicvm" +description = "my firewall" +start = True +image_id = "408b73-e9cc-5a6a-t270-82cc4811bd4a" +flavor_id = "208b73-e9cc-5a6a-t270-82cc4811bd4a" +affinity_group_list = [] +net_list = [] +cloud_config = {} +disk_list = [] +disk_list2 = [ + {"size": 10, "image_id": image_id}, + {"size": 20}, +] +availability_zone_index = 0 +availability_zone_list = ["nova"] +floating_network_vim_id = "108b73-e9cc-5a6a-t270-82cc4811bd4a" +net_id = "83372685-f67f-49fd-8722-eabb7692fc22" +net2_id = "46472685-f67f-49fd-8722-eabb7692fc22" +mac_address = "00:00:5e:00:53:af" +port_id = "03372685-f67f-49fd-8722-eabb7692fc22" +time_return_value = 156570000 +port2_id = "17472685-f67f-49fd-8722-eabb7692fc22" +root_vol_id = "tc408b73-r9cc-5a6a-a270-82cc4811bd4a" +ip_addr1 = "20.3.4.5" +volume_id = "ac408b73-b9cc-4a6a-a270-82cc4811bd4a" +volume_id2 = "o4e0e83-b9uu-4akk-a234-89cc4811bd4a" +volume_id3 = "44e0e83-t9uu-4akk-a234-p9cc4811bd4a" + class TestSfcOperations(unittest.TestCase): @mock.patch("logging.getLogger", autospec=True) @@ -1057,5 +1088,3363 @@ class TestSfcOperations(unittest.TestCase): self.assertEqual(result, "638f957c-82df-11e7-b7c8-132706021464") +class Status: + def __init__(self, s): + self.status = s + + def __str__(self): + return self.status + + +class CopyingMock(MagicMock): + def __call__(self, *args, **kwargs): + args = deepcopy(args) + kwargs = deepcopy(kwargs) + return super(CopyingMock, self).__call__(*args, **kwargs) + + +class TestNewVmInstance(unittest.TestCase): + @patch("logging.getLogger", autospec=True) + def setUp(self, mock_logger): + # Instantiate dummy VIM connector so we can test it + # It throws exception because of dummy parameters, + # We are disabling the logging of exception not to print them to console. + mock_logger = logging.getLogger() + mock_logger.disabled = True + self.vimconn = vimconnector( + "123", + "openstackvim", + "456", + "789", + "http://dummy.url", + None, + "user", + "pass", + ) + self.vimconn.neutron = CopyingMock() + self.vimconn.nova = CopyingMock() + self.vimconn.cinder = CopyingMock() + self.server = MagicMock(object, autospec=True) + self.server.tenant_id = "408b73-r9cc-5a6a-a270-82cc4811bd4a" + self.server.id = "908b73-e9cc-5a6a-t270-82cc4811bd4a" + self.vimconn.config["security_groups"] = "default" + self.vimconn.config["keypair"] = "my_keypair" + self.vimconn.security_groups_id = "12345" + self.vimconn.nova.api_version.get_string.return_value = "2.32" + + @patch.object(vimconnector, "_get_ids_from_name") + def test_prepare_port_dict_security_security_groups_exists_in_config( + self, mock_get_ids + ): + """In VIM config security_groups exists, net port_security is True + no_port_security_extension does not exist. + """ + self.vimconn.config = {"security_groups": "example_security_group"} + net = {"port_security": True} + port_dict = {} + result_dict = {"security_groups": "12345"} + + self.vimconn._prepare_port_dict_security_groups(net, port_dict) + self.assertDictEqual(result_dict, port_dict) + mock_get_ids.assert_not_called() + + @patch.object(vimconnector, "_get_ids_from_name") + def test_prepare_port_dict_security_security_groups_exists_in_config_no_security_groups_id( + self, mock_get_ids + ): + """In VIM config Security_groups exists, net port_security is True, vim security_groups_id does not exist, + no_port_security_extension does not exist. + """ + self.vimconn.config = {"security_groups": "example_security_group"} + self.vimconn.security_groups_id = None + net = {"port_security": True} + port_dict = {} + result_dict = {"security_groups": None} + + self.vimconn._prepare_port_dict_security_groups(net, port_dict) + self.assertDictEqual(result_dict, port_dict) + mock_get_ids.assert_called() + + @patch.object(vimconnector, "_get_ids_from_name") + def test_prepare_port_dict_security_security_groups_exists_security_extension_true_in_config( + self, mock_get_ids + ): + """In VIM config security_groups exists, net port_security is True, in VIM security_groups_id exists, + no_port_security_extension set to True. + """ + self.vimconn.config = { + "security_groups": "example_security_group", + "no_port_security_extension": True, + } + net = {"port_security": True} + port_dict = {} + result_dict = {} + + self.vimconn._prepare_port_dict_security_groups(net, port_dict) + self.assertDictEqual(result_dict, port_dict) + mock_get_ids.assert_not_called() + + @patch.object(vimconnector, "_get_ids_from_name") + def test_prepare_port_dict_security_no_security_groups_in_config( + self, mock_get_ids + ): + """In VIM config security_group does not exist, net port_security True, in VIM security_groups_id exists, + no_port_security_extension does not exist.""" + self.vimconn.config = {} + net = {"port_security": True} + port_dict = {} + result_dict = {} + + self.vimconn._prepare_port_dict_security_groups(net, port_dict) + self.assertDictEqual(result_dict, port_dict) + mock_get_ids.assert_not_called() + + @patch.object(vimconnector, "_get_ids_from_name") + def test_prepare_port_dict_security_no_security_groups_security_extension_true_in_config( + self, mock_get_ids + ): + """Security_group does not exist, net port_security is True, in VIM security_groups_id exists, + no_port_security_extension set to True.""" + self.vimconn.config = {"no_port_security_extension": True} + net = {"port_security": True} + port_dict = {} + result_dict = {} + + self.vimconn._prepare_port_dict_security_groups(net, port_dict) + self.assertDictEqual(result_dict, port_dict) + mock_get_ids.assert_not_called() + + @patch.object(vimconnector, "_get_ids_from_name") + def test_prepare_port_dict_security_security_groups_exists_net_port_security_false( + self, mock_get_ids + ): + """In VIM config security_group exists, net port_security False, security_groups_id exists, + no_port_security_extension does not exist.""" + self.vimconn.config = {"security_groups": "example_security_group"} + net = {"port_security": False} + port_dict = {} + result_dict = {} + + self.vimconn._prepare_port_dict_security_groups(net, port_dict) + self.assertDictEqual(result_dict, port_dict) + mock_get_ids.assert_not_called() + + @patch.object(vimconnector, "_get_ids_from_name") + def test_prepare_port_dict_security_net_port_security_false_port_security_extension_true( + self, mock_get_ids + ): + """In VIM config security_group exists, net port_security False, security_groups_id exists, + no_port_security_extension set to True.""" + self.vimconn.config = { + "security_groups": "example_security_group", + "no_port_security_extension": True, + } + net = {"port_security": False} + port_dict = {} + result_dict = {} + + self.vimconn._prepare_port_dict_security_groups(net, port_dict) + self.assertDictEqual(result_dict, port_dict) + mock_get_ids.assert_not_called() + + def test_prepare_port_dict_binding_net_type_virtual(self): + """net type is virtual.""" + net = {"type": "virtual"} + port_dict = {} + result_dict = {} + self.vimconn._prepare_port_dict_binding(net, port_dict) + self.assertDictEqual(result_dict, port_dict) + + def test_prepare_port_dict_binding_net_type_vf(self): + """net type is VF, vim_type is not VIO.""" + net = {"type": "VF"} + self.vimconn.vim_type = None + port_dict = {} + result_dict = {"binding:vnic_type": "direct"} + self.vimconn._prepare_port_dict_binding(net, port_dict) + self.assertDictEqual(port_dict, result_dict) + + def test_prepare_port_dict_binding_net_type_sriov_vim_type_vio(self): + """net type is SR-IOV, vim_type is VIO.""" + net = {"type": "SR-IOV"} + self.vimconn.vim_type = "VIO" + port_dict = {} + result_dict = { + "binding:vnic_type": "direct", + "port_security_enabled": False, + "provider_security_groups": [], + "security_groups": [], + } + self.vimconn._prepare_port_dict_binding(net, port_dict) + self.assertDictEqual(port_dict, result_dict) + + def test_prepare_port_dict_binding_net_type_passthrough(self): + """net type is pci-passthrough.""" + net = {"type": "PCI-PASSTHROUGH"} + port_dict = {} + result_dict = { + "binding:vnic_type": "direct-physical", + } + self.vimconn._prepare_port_dict_binding(net, port_dict) + self.assertDictEqual(port_dict, result_dict) + + def test_prepare_port_dict_binding_no_net_type(self): + """net type is missing.""" + net = {} + port_dict = {} + with self.assertRaises(VimConnException) as err: + self.vimconn._prepare_port_dict_binding(net, port_dict) + self.assertEqual(str(err.exception), "Type is missing in the network details.") + + def test_set_fixed_ip(self): + """new_port has fixed ip.""" + net = {} + new_port = { + "port": { + "fixed_ips": [{"ip_address": "10.1.2.3"}, {"ip_address": "20.1.2.3"}] + } + } + result = {"ip": "10.1.2.3"} + self.vimconn._set_fixed_ip(new_port, net) + self.assertDictEqual(net, result) + + def test_set_fixed_ip_no_fixed_ip(self): + """new_port does not have fixed ip.""" + net = {} + new_port = {"port": {}} + result = {"ip": None} + self.vimconn._set_fixed_ip(new_port, net) + self.assertDictEqual(net, result) + + def test_set_fixed_ip_raise_exception(self): + """new_port does not have port details.""" + net = {} + new_port = {} + with self.assertRaises(Exception) as err: + self.vimconn._set_fixed_ip(new_port, net) + self.assertEqual(type(err.exception), KeyError) + + def test_prepare_port_dict_mac_ip_addr(self): + """mac address and ip address exist.""" + net = { + "mac_address": mac_address, + "ip_address": "10.0.1.5", + } + port_dict = {} + result_dict = { + "mac_address": mac_address, + "fixed_ips": [{"ip_address": "10.0.1.5"}], + } + self.vimconn._prepare_port_dict_mac_ip_addr(net, port_dict) + self.assertDictEqual(port_dict, result_dict) + + def test_prepare_port_dict_mac_ip_addr_no_mac_and_ip(self): + """mac address and ip address does not exist.""" + net = {} + port_dict = {} + result_dict = {} + self.vimconn._prepare_port_dict_mac_ip_addr(net, port_dict) + self.assertDictEqual(port_dict, result_dict) + + def test_create_new_port(self): + """new port has id and mac address.""" + new_port = { + "port": { + "id": port_id, + "mac_address": mac_address, + }, + } + self.vimconn.neutron.create_port.return_value = new_port + net, port_dict, created_items = {}, {}, {} + expected_result = new_port + expected_net = { + "mac_adress": mac_address, + "vim_id": port_id, + } + expected_created_items = {f"port:{port_id}": True} + result = self.vimconn._create_new_port(port_dict, created_items, net) + self.assertDictEqual(result, expected_result) + self.assertEqual(net, expected_net) + self.assertEqual(created_items, expected_created_items) + self.vimconn.neutron.create_port.assert_called_once_with({"port": port_dict}) + + def test_create_new_port_without_mac_or_id(self): + """new port does not have mac address or ID.""" + new_port = {} + self.vimconn.neutron.create_port.return_value = new_port + net, port_dict, created_items = {}, {}, {} + with self.assertRaises(KeyError): + self.vimconn._create_new_port(port_dict, created_items, net) + self.vimconn.neutron.create_port.assert_called_once_with({"port": port_dict}) + + def test_create_new_port_neutron_create_port_raises_exception(self): + """Neutron create port raises exception.""" + self.vimconn.neutron.create_port.side_effect = VimConnException( + "New port is not created." + ) + net, port_dict, created_items = {}, {}, {} + with self.assertRaises(VimConnException): + self.vimconn._create_new_port(port_dict, created_items, net) + self.vimconn.neutron.create_port.assert_called_once_with({"port": port_dict}) + + @patch.object(vimconnector, "_prepare_port_dict_security_groups") + @patch.object(vimconnector, "_prepare_port_dict_binding") + @patch.object(vimconnector, "_prepare_port_dict_mac_ip_addr") + @patch.object(vimconnector, "_create_new_port") + @patch.object(vimconnector, "_set_fixed_ip") + def test_create_port( + self, + mock_set_fixed_ip, + mock_create_new_port, + mock_prepare_port_dict_mac_ip_addr, + mock_prepare_port_dict_binding, + mock_prepare_port_dict_security_groups, + ): + """Net has name, type, net-id.""" + + net = { + "net_id": net_id, + "name": "management", + "type": "virtual", + } + created_items = {} + new_port = { + "port": { + "id": net_id, + "mac_address": mac_address, + "name": "management", + "fixed_ips": [{"ip_address": ip_addr1}], + }, + } + mock_create_new_port.return_value = new_port + expected_port = { + "port-id": net_id, + "tag": "management", + } + port_dict = { + "network_id": net_id, + "name": "management", + "admin_state_up": True, + } + + new_port_result, port_result = self.vimconn._create_port( + net, name, created_items + ) + + self.assertDictEqual(new_port_result, new_port) + self.assertDictEqual(port_result, expected_port) + + mock_prepare_port_dict_security_groups.assert_called_once_with(net, port_dict) + mock_prepare_port_dict_binding.assert_called_once_with(net, port_dict) + mock_prepare_port_dict_mac_ip_addr.assert_called_once_with(net, port_dict) + mock_create_new_port.assert_called_once_with(port_dict, created_items, net) + mock_set_fixed_ip.assert_called_once_with(new_port, net) + + @patch.object(vimconnector, "_prepare_port_dict_security_groups") + @patch.object(vimconnector, "_prepare_port_dict_binding") + @patch.object(vimconnector, "_prepare_port_dict_mac_ip_addr") + @patch.object(vimconnector, "_create_new_port") + @patch.object(vimconnector, "_set_fixed_ip") + def test_create_port_no_port_name( + self, + mock_set_fixed_ip, + mock_create_new_port, + mock_prepare_port_dict_mac_ip_addr, + mock_prepare_port_dict_binding, + mock_prepare_port_dict_security_groups, + ): + """Net has no name.""" + net = { + "net_id": net_id, + "type": "virtual", + } + created_items = {} + new_port = { + "port": { + "id": net_id, + "mac_address": mac_address, + "name": name, + "fixed_ips": [{"ip_address": ip_addr1}], + }, + } + mock_create_new_port.return_value = new_port + expected_port = { + "port-id": net_id, + "tag": name, + } + port_dict = { + "network_id": net_id, + "admin_state_up": True, + "name": name, + } + + new_port_result, port_result = self.vimconn._create_port( + net, name, created_items + ) + + self.assertDictEqual(new_port_result, new_port) + self.assertDictEqual(port_result, expected_port) + + mock_prepare_port_dict_security_groups.assert_called_once_with(net, port_dict) + mock_prepare_port_dict_binding.assert_called_once_with(net, port_dict) + mock_prepare_port_dict_mac_ip_addr.assert_called_once_with(net, port_dict) + mock_create_new_port.assert_called_once_with(port_dict, created_items, net) + mock_set_fixed_ip.assert_called_once_with(new_port, net) + + @patch.object(vimconnector, "_prepare_port_dict_security_groups") + @patch.object(vimconnector, "_prepare_port_dict_binding") + @patch.object(vimconnector, "_prepare_port_dict_mac_ip_addr") + @patch.object(vimconnector, "_create_new_port") + @patch.object(vimconnector, "_set_fixed_ip") + def test_create_port_nova_api_version_smaller_than_232( + self, + mock_set_fixed_ip, + mock_create_new_port, + mock_prepare_port_dict_mac_ip_addr, + mock_prepare_port_dict_binding, + mock_prepare_port_dict_security_groups, + ): + """Nova api version is smaller than 2.32.""" + self.vimconn.nova.api_version.get_string.return_value = "2.30" + net = { + "net_id": net_id, + "type": "virtual", + } + created_items = {} + new_port = { + "port": { + "id": net_id, + "mac_address": mac_address, + "name": name, + "fixed_ips": [{"ip_address": ip_addr1}], + }, + } + mock_create_new_port.return_value = new_port + expected_port = { + "port-id": net_id, + } + port_dict = { + "network_id": net_id, + "admin_state_up": True, + "name": name, + } + + new_port_result, port_result = self.vimconn._create_port( + net, name, created_items + ) + + self.assertDictEqual(new_port_result, new_port) + self.assertDictEqual(port_result, expected_port) + + mock_prepare_port_dict_security_groups.assert_called_once_with(net, port_dict) + mock_prepare_port_dict_binding.assert_called_once_with(net, port_dict) + mock_prepare_port_dict_mac_ip_addr.assert_called_once_with(net, port_dict) + mock_create_new_port.assert_called_once_with(port_dict, created_items, net) + mock_set_fixed_ip.assert_called_once_with(new_port, net) + + @patch.object(vimconnector, "_prepare_port_dict_security_groups") + @patch.object(vimconnector, "_prepare_port_dict_binding") + @patch.object(vimconnector, "_prepare_port_dict_mac_ip_addr") + @patch.object(vimconnector, "_create_new_port") + @patch.object(vimconnector, "_set_fixed_ip") + def test_create_port_create_new_port_raise_exception( + self, + mock_set_fixed_ip, + mock_create_new_port, + mock_prepare_port_dict_mac_ip_addr, + mock_prepare_port_dict_binding, + mock_prepare_port_dict_security_groups, + ): + """_create_new_port method raises exception.""" + net = { + "net_id": net_id, + "type": "virtual", + } + created_items = {} + mock_create_new_port.side_effect = Exception + port_dict = { + "network_id": net_id, + "admin_state_up": True, + "name": name, + } + + with self.assertRaises(Exception): + self.vimconn._create_port(net, name, created_items) + + mock_prepare_port_dict_security_groups.assert_called_once_with(net, port_dict) + mock_prepare_port_dict_binding.assert_called_once_with(net, port_dict) + mock_prepare_port_dict_mac_ip_addr.assert_called_once_with(net, port_dict) + mock_create_new_port.assert_called_once_with(port_dict, created_items, net) + mock_set_fixed_ip.assert_not_called() + + @patch.object(vimconnector, "_prepare_port_dict_security_groups") + @patch.object(vimconnector, "_prepare_port_dict_binding") + @patch.object(vimconnector, "_prepare_port_dict_mac_ip_addr") + @patch.object(vimconnector, "_create_new_port") + @patch.object(vimconnector, "_set_fixed_ip") + def test_create_port_create_sec_groups_raises_exception( + self, + mock_set_fixed_ip, + mock_create_new_port, + mock_prepare_port_dict_mac_ip_addr, + mock_prepare_port_dict_binding, + mock_prepare_port_dict_security_groups, + ): + """_prepare_port_dict_security_groups method raises exception.""" + net = { + "net_id": net_id, + "type": "virtual", + } + created_items = {} + mock_prepare_port_dict_security_groups.side_effect = Exception + port_dict = { + "network_id": net_id, + "admin_state_up": True, + "name": name, + } + + with self.assertRaises(Exception): + self.vimconn._create_port(net, name, created_items) + + mock_prepare_port_dict_security_groups.assert_called_once_with(net, port_dict) + + mock_prepare_port_dict_binding.assert_not_called() + mock_prepare_port_dict_mac_ip_addr.assert_not_called() + mock_create_new_port.assert_not_called() + mock_set_fixed_ip.assert_not_called() + + @patch.object(vimconnector, "_prepare_port_dict_security_groups") + @patch.object(vimconnector, "_prepare_port_dict_binding") + @patch.object(vimconnector, "_prepare_port_dict_mac_ip_addr") + @patch.object(vimconnector, "_create_new_port") + @patch.object(vimconnector, "_set_fixed_ip") + def test_create_port_create_port_dict_binding_raise_exception( + self, + mock_set_fixed_ip, + mock_create_new_port, + mock_prepare_port_dict_mac_ip_addr, + mock_prepare_port_dict_binding, + mock_prepare_port_dict_security_groups, + ): + """_prepare_port_dict_binding method raises exception.""" + + net = { + "net_id": net_id, + "type": "virtual", + } + created_items = {} + mock_prepare_port_dict_binding.side_effect = Exception + port_dict = { + "network_id": net_id, + "admin_state_up": True, + "name": name, + } + + with self.assertRaises(Exception): + self.vimconn._create_port(net, name, created_items) + + mock_prepare_port_dict_security_groups.assert_called_once_with(net, port_dict) + + mock_prepare_port_dict_binding.assert_called_once_with(net, port_dict) + + mock_prepare_port_dict_mac_ip_addr.assert_not_called() + mock_create_new_port.assert_not_called() + mock_set_fixed_ip.assert_not_called() + + @patch.object(vimconnector, "_prepare_port_dict_security_groups") + @patch.object(vimconnector, "_prepare_port_dict_binding") + @patch.object(vimconnector, "_prepare_port_dict_mac_ip_addr") + @patch.object(vimconnector, "_create_new_port") + @patch.object(vimconnector, "_set_fixed_ip") + def test_create_port_create_port_mac_ip_addr_raise_exception( + self, + mock_set_fixed_ip, + mock_create_new_port, + mock_prepare_port_dict_mac_ip_addr, + mock_prepare_port_dict_binding, + mock_prepare_port_dict_security_groups, + ): + """prepare_port_dict_mac_ip_addr method raises exception.""" + net = { + "net_id": net_id, + "type": "virtual", + } + created_items = {} + mock_prepare_port_dict_mac_ip_addr.side_effect = Exception + port_dict = { + "network_id": net_id, + "admin_state_up": True, + "name": name, + } + + with self.assertRaises(Exception): + self.vimconn._create_port(net, name, created_items) + + mock_prepare_port_dict_security_groups.assert_called_once_with(net, port_dict) + mock_prepare_port_dict_binding.assert_called_once_with(net, port_dict) + mock_prepare_port_dict_mac_ip_addr.assert_called_once_with(net, port_dict) + + mock_create_new_port.assert_not_called() + mock_set_fixed_ip.assert_not_called() + + @patch.object(vimconnector, "_prepare_port_dict_security_groups") + @patch.object(vimconnector, "_prepare_port_dict_binding") + @patch.object(vimconnector, "_prepare_port_dict_mac_ip_addr") + @patch.object(vimconnector, "_create_new_port") + @patch.object(vimconnector, "_set_fixed_ip") + def test_create_port_create_port_set_fixed_ip_raise_exception( + self, + mock_set_fixed_ip, + mock_create_new_port, + mock_prepare_port_dict_mac_ip_addr, + mock_prepare_port_dict_binding, + mock_prepare_port_dict_security_groups, + ): + """_set_fixed_ip method raises exception.""" + net = { + "net_id": net_id, + "type": "virtual", + } + created_items = {} + mock_set_fixed_ip.side_effect = VimConnException( + "Port detail is missing in new_port." + ) + port_dict = { + "network_id": net_id, + "admin_state_up": True, + "name": name, + } + new_port = { + "port": { + "id": net_id, + "mac_address": mac_address, + "name": name, + "fixed_ips": [{"ip_address": ip_addr1}], + }, + } + mock_create_new_port.return_value = new_port + + with self.assertRaises(VimConnException): + self.vimconn._create_port(net, name, created_items) + + mock_prepare_port_dict_security_groups.assert_called_once_with(net, port_dict) + mock_prepare_port_dict_binding.assert_called_once_with(net, port_dict) + mock_prepare_port_dict_mac_ip_addr.assert_called_once_with(net, port_dict) + mock_create_new_port.assert_called_once_with(port_dict, created_items, net) + mock_set_fixed_ip.assert_called_once_with(new_port, net) + + @patch.object(vimconnector, "_reload_connection") + @patch.object(vimconnector, "_create_port") + def test_prepare_network_for_vm_instance_no_net_id( + self, mock_create_port, mock_reload_connection + ): + """Nets do not have net_id""" + mock_reload_connection.side_effect = None + created_items = {} + net_list = [ + { + "use": "mgmt", + "port_security": False, + "exit_on_floating_ip_error": False, + "port_security_disable_strategy": "full", + }, + { + "port_security": True, + "exit_on_floating_ip_error": False, + "floating_ip": True, + }, + ] + net_list_vim = [] + external_network, no_secured_ports = [], [] + expected_external_network, expected_no_secured_ports = [], [] + expected_net_list_vim = [] + + self.vimconn._prepare_network_for_vminstance( + name, + net_list, + created_items, + net_list_vim, + external_network, + no_secured_ports, + ) + self.assertEqual(expected_net_list_vim, net_list_vim) + self.assertEqual(external_network, expected_external_network) + self.assertEqual(expected_no_secured_ports, no_secured_ports) + + mock_create_port.assert_not_called() + + @patch.object(vimconnector, "_reload_connection") + @patch.object(vimconnector, "_create_port") + def test_prepare_network_for_vm_instance_empty_net_list( + self, mock_create_port, mock_reload_connection + ): + """Net list is empty.""" + mock_reload_connection.side_effect = None + created_items = {} + net_list_vim = [] + external_network, no_secured_ports = [], [] + expected_external_network, expected_no_secured_ports = [], [] + expected_net_list_vim = [] + + self.vimconn._prepare_network_for_vminstance( + name, + net_list, + created_items, + net_list_vim, + external_network, + no_secured_ports, + ) + self.assertEqual(expected_net_list_vim, net_list_vim) + self.assertEqual(external_network, expected_external_network) + self.assertEqual(expected_no_secured_ports, no_secured_ports) + + mock_create_port.assert_not_called() + + @patch.object(vimconnector, "_reload_connection") + @patch.object(vimconnector, "_create_port") + def test_prepare_network_for_vm_instance_use_floating_ip_false_mgmt_net( + self, mock_create_port, mock_reload_connection + ): + """Nets have net-id, floating_ip False, mgmt network.""" + mock_reload_connection.side_effect = None + created_items = {} + net_list = [ + { + "net_id": net2_id, + "floating_ip": False, + "use": "mgmt", + } + ] + net_list_vim = [] + mock_create_port.side_effect = [ + ( + { + "port": { + "id": port2_id, + "mac_address": mac_address, + "name": name, + }, + }, + {"port-dict": port2_id}, + ), + ] + external_network, no_secured_ports = [], [] + expected_external_network, expected_no_secured_ports = [], [] + expected_net_list_vim = [{"port-dict": port2_id}] + self.vimconn._prepare_network_for_vminstance( + name, + net_list, + created_items, + net_list_vim, + external_network, + no_secured_ports, + ) + self.assertEqual(expected_net_list_vim, net_list_vim) + self.assertEqual(external_network, expected_external_network) + self.assertEqual(expected_no_secured_ports, no_secured_ports) + + mock_create_port.assert_called_once_with( + { + "net_id": net2_id, + "floating_ip": False, + "use": "mgmt", + }, + name, + created_items, + ) + + @patch.object(vimconnector, "_reload_connection") + def test_prepare_network_for_vm_instance_mgmt_net_net_port_security_and_floating_ip_true( + self, mock_reload_connection + ): + """Nets have net-id, use_floating_ip False in VIM config, mgmt network, net floating_ip is True.""" + self.vimconn.config["use_floating_ip"] = False + mock_create_port = CopyingMock() + mock_reload_connection.side_effect = None + created_items = {} + net_list = [ + { + "net_id": net2_id, + "floating_ip": True, + "use": "mgmt", + } + ] + net_list_vim = [] + mock_create_port.side_effect = [ + ( + { + "port": { + "id": port2_id, + "mac_address": mac_address, + "name": name, + }, + }, + {"port-dict": port2_id}, + ), + ] + external_network, no_secured_ports = [], [] + expected_external_network = [ + { + "net_id": net2_id, + "floating_ip": True, + "use": "mgmt", + "exit_on_floating_ip_error": True, + }, + ] + expected_no_secured_ports = [] + expected_net_list_vim = [{"port-dict": port2_id}] + with patch.object(vimconnector, "_create_port", mock_create_port): + self.vimconn._prepare_network_for_vminstance( + name, + net_list, + created_items, + net_list_vim, + external_network, + no_secured_ports, + ) + self.assertEqual(expected_net_list_vim, net_list_vim) + self.assertEqual(external_network, expected_external_network) + self.assertEqual(expected_no_secured_ports, no_secured_ports) + + mock_create_port.assert_called_once_with( + { + "net_id": net2_id, + "floating_ip": True, + "use": "mgmt", + }, + name, + created_items, + ) + + @patch.object(vimconnector, "_reload_connection") + def test_prepare_network_for_vm_instance_use_floating_ip_true_mgmt_net_port_security_false( + self, mock_reload_connection + ): + """Nets have net-id, use_floating_ip is True in VIM config, mgmt network, net port security is False.""" + mock_create_port = CopyingMock() + self.vimconn.config["use_floating_ip"] = True + self.vimconn.config["no_port_security_extension"] = False + mock_reload_connection.side_effect = None + created_items = {} + + net_list = [ + { + "net_id": net2_id, + "use": "mgmt", + "port_security": False, + "exit_on_floating_ip_error": False, + "port_security_disable_strategy": "full", + } + ] + net_list_vim = [] + mock_create_port.side_effect = [ + ( + { + "port": { + "id": port2_id, + "mac_address": mac_address, + "name": name, + }, + }, + {"port-dict": port2_id}, + ), + ] + external_network, no_secured_ports = [], [] + expected_external_network = [ + { + "net_id": net2_id, + "use": "mgmt", + "port_security": False, + "exit_on_floating_ip_error": False, + "port_security_disable_strategy": "full", + "floating_ip": True, + }, + ] + expected_no_secured_ports = [(port2_id, "full")] + expected_net_list_vim = [{"port-dict": port2_id}] + with patch.object(vimconnector, "_create_port", mock_create_port): + self.vimconn._prepare_network_for_vminstance( + name, + net_list, + created_items, + net_list_vim, + external_network, + no_secured_ports, + ) + + mock_create_port.assert_called_once_with( + { + "net_id": net2_id, + "use": "mgmt", + "port_security": False, + "exit_on_floating_ip_error": False, + "port_security_disable_strategy": "full", + }, + name, + created_items, + ) + self.assertEqual(expected_net_list_vim, net_list_vim) + self.assertEqual(external_network, expected_external_network) + self.assertEqual(expected_no_secured_ports, no_secured_ports) + + @patch.object(vimconnector, "_reload_connection") + def test_prepare_network_for_vm_instance_use_fip_true_non_mgmt_net_port_security_false( + self, mock_reload_connection + ): + """Nets have net-id, use_floating_ip True in VIM config, non-mgmt network, port security is False.""" + mock_create_port = CopyingMock() + self.vimconn.config["use_floating_ip"] = True + self.vimconn.config["no_port_security_extension"] = False + mock_reload_connection.side_effect = None + created_items = {} + + net_list = [ + { + "net_id": net2_id, + "use": "other", + "port_security": False, + "port_security_disable_strategy": "full", + } + ] + net_list_vim = [] + mock_create_port.side_effect = [ + ( + { + "port": { + "id": port2_id, + "mac_address": mac_address, + "name": name, + }, + }, + {"port-dict": port2_id}, + ), + ] + external_network, no_secured_ports = [], [] + expected_external_network = [] + expected_no_secured_ports = [(port2_id, "full")] + expected_net_list_vim = [{"port-dict": port2_id}] + with patch.object(vimconnector, "_create_port", mock_create_port): + self.vimconn._prepare_network_for_vminstance( + name, + net_list, + created_items, + net_list_vim, + external_network, + no_secured_ports, + ) + + mock_create_port.assert_called_once_with( + { + "net_id": net2_id, + "use": "other", + "port_security": False, + "port_security_disable_strategy": "full", + }, + name, + created_items, + ) + self.assertEqual(expected_net_list_vim, net_list_vim) + self.assertEqual(external_network, expected_external_network) + self.assertEqual(expected_no_secured_ports, no_secured_ports) + + @patch.object(vimconnector, "_reload_connection") + def test_prepare_network_for_vm_instance_use_fip_true_non_mgmt_net_port_security_true( + self, mock_reload_connection + ): + """Nets have net-id, use_floating_ip is True in VIM config, non-mgmt network, net port security is True.""" + mock_create_port = CopyingMock() + self.vimconn.config["use_floating_ip"] = True + self.vimconn.config["no_port_security_extension"] = True + mock_reload_connection.side_effect = None + created_items = {} + + net_list = [ + { + "net_id": net2_id, + "use": "other", + "port_security": True, + "port_security_disable_strategy": "full", + } + ] + net_list_vim = [] + mock_create_port.side_effect = [ + ( + { + "port": { + "id": port2_id, + "mac_address": mac_address, + "name": name, + }, + }, + {"port-dict": port2_id}, + ), + ] + external_network, no_secured_ports = [], [] + expected_external_network = [] + expected_no_secured_ports = [] + expected_net_list_vim = [{"port-dict": port2_id}] + with patch.object(vimconnector, "_create_port", mock_create_port): + self.vimconn._prepare_network_for_vminstance( + name, + net_list, + created_items, + net_list_vim, + external_network, + no_secured_ports, + ) + + mock_create_port.assert_called_once_with( + { + "net_id": net2_id, + "use": "other", + "port_security": True, + "port_security_disable_strategy": "full", + }, + name, + created_items, + ) + self.assertEqual(expected_net_list_vim, net_list_vim) + self.assertEqual(external_network, expected_external_network) + self.assertEqual(expected_no_secured_ports, no_secured_ports) + + @patch.object(vimconnector, "_reload_connection") + def test_prepare_network_for_vm_instance_create_port_raise_exception( + self, mock_reload_connection + ): + """_create_port method raise exception.""" + mock_create_port = CopyingMock() + self.vimconn.config["use_floating_ip"] = True + self.vimconn.config["no_port_security_extension"] = True + mock_reload_connection.side_effect = None + created_items = {} + + net_list = [ + { + "net_id": net2_id, + "use": "other", + "port_security": True, + "port_security_disable_strategy": "full", + } + ] + net_list_vim = [] + mock_create_port.side_effect = KeyError + external_network, no_secured_ports = [], [] + expected_external_network = [] + expected_no_secured_ports = [] + expected_net_list_vim = [] + with patch.object(vimconnector, "_create_port", mock_create_port): + with self.assertRaises(Exception) as err: + self.vimconn._prepare_network_for_vminstance( + name, + net_list, + created_items, + net_list_vim, + external_network, + no_secured_ports, + ) + + self.assertEqual(type(err.exception), KeyError) + + mock_create_port.assert_called_once_with( + { + "net_id": net2_id, + "use": "other", + "port_security": True, + "port_security_disable_strategy": "full", + }, + name, + created_items, + ) + self.assertEqual(expected_net_list_vim, net_list_vim) + self.assertEqual(external_network, expected_external_network) + self.assertEqual(expected_no_secured_ports, no_secured_ports) + + @patch.object(vimconnector, "_reload_connection") + def test_prepare_network_for_vm_instance_reload_connection_raise_exception( + self, mock_reload_connection + ): + """_reload_connection method raises exception.""" + mock_create_port = CopyingMock() + mock_reload_connection.side_effect = VimConnConnectionException( + "Connection failed." + ) + self.vimconn.config["use_floating_ip"] = True + self.vimconn.config["no_port_security_extension"] = True + created_items = {} + + net_list = [ + { + "net_id": net2_id, + "use": "other", + "port_security": True, + "port_security_disable_strategy": "full", + } + ] + net_list_vim = [] + mock_create_port.side_effect = None + external_network, no_secured_ports = [], [] + expected_external_network = [] + expected_no_secured_ports = [] + expected_net_list_vim = [] + with patch.object(vimconnector, "_create_port", mock_create_port): + with self.assertRaises(Exception) as err: + self.vimconn._prepare_network_for_vminstance( + name, + net_list, + created_items, + net_list_vim, + external_network, + no_secured_ports, + ) + + self.assertEqual(type(err.exception), VimConnConnectionException) + self.assertEqual(str(err.exception), "Connection failed.") + mock_reload_connection.assert_called_once() + mock_create_port.assert_not_called() + self.assertEqual(expected_net_list_vim, net_list_vim) + self.assertEqual(external_network, expected_external_network) + self.assertEqual(expected_no_secured_ports, no_secured_ports) + + def test_prepare_persistent_root_volumes_vim_using_volume_id(self): + """Existing persistent root volume with vim_volume_id.""" + vm_av_zone = ["nova"] + base_disk_index = ord("a") + disk = {"vim_volume_id": volume_id} + block_device_mapping = {} + existing_vim_volumes = [] + created_items = {} + expected_boot_vol_id = None + expected_block_device_mapping = {"vda": volume_id} + expected_existing_vim_volumes = [{"id": volume_id}] + boot_volume_id = self.vimconn._prepare_persistent_root_volumes( + name, + vm_av_zone, + disk, + base_disk_index, + block_device_mapping, + existing_vim_volumes, + created_items, + ) + self.assertEqual(boot_volume_id, expected_boot_vol_id) + self.assertDictEqual(block_device_mapping, expected_block_device_mapping) + self.assertEqual(existing_vim_volumes, expected_existing_vim_volumes) + self.vimconn.cinder.volumes.create.assert_not_called() + + def test_prepare_persistent_non_root_volumes_vim_using_volume_id(self): + """Existing persistent non root volume with vim_volume_id.""" + vm_av_zone = ["nova"] + base_disk_index = ord("b") + disk = {"vim_volume_id": volume_id} + block_device_mapping = {} + existing_vim_volumes = [] + created_items = {} + expected_block_device_mapping = {"vdb": volume_id} + expected_existing_vim_volumes = [{"id": volume_id}] + self.vimconn._prepare_non_root_persistent_volumes( + name, + disk, + vm_av_zone, + block_device_mapping, + base_disk_index, + existing_vim_volumes, + created_items, + ) + self.assertDictEqual(block_device_mapping, expected_block_device_mapping) + self.assertEqual(existing_vim_volumes, expected_existing_vim_volumes) + self.vimconn.cinder.volumes.create.assert_not_called() + + def test_prepare_persistent_root_volumes_using_vim_id(self): + """Existing persistent root volume with vim_id.""" + vm_av_zone = ["nova"] + base_disk_index = ord("a") + disk = {"vim_id": volume_id} + block_device_mapping = {} + existing_vim_volumes = [] + created_items = {} + expected_boot_vol_id = None + expected_block_device_mapping = {"vda": volume_id} + expected_existing_vim_volumes = [{"id": volume_id}] + boot_volume_id = self.vimconn._prepare_persistent_root_volumes( + name, + vm_av_zone, + disk, + base_disk_index, + block_device_mapping, + existing_vim_volumes, + created_items, + ) + self.assertEqual(boot_volume_id, expected_boot_vol_id) + self.assertDictEqual(block_device_mapping, expected_block_device_mapping) + self.assertEqual(existing_vim_volumes, expected_existing_vim_volumes) + self.vimconn.cinder.volumes.create.assert_not_called() + + def test_prepare_persistent_non_root_volumes_using_vim_id(self): + """Existing persistent root volume with vim_id.""" + vm_av_zone = ["nova"] + base_disk_index = ord("b") + disk = {"vim_id": volume_id} + block_device_mapping = {} + existing_vim_volumes = [] + created_items = {} + + expected_block_device_mapping = {"vdb": volume_id} + expected_existing_vim_volumes = [{"id": volume_id}] + self.vimconn._prepare_non_root_persistent_volumes( + name, + disk, + vm_av_zone, + block_device_mapping, + base_disk_index, + existing_vim_volumes, + created_items, + ) + + self.assertDictEqual(block_device_mapping, expected_block_device_mapping) + self.assertEqual(existing_vim_volumes, expected_existing_vim_volumes) + self.vimconn.cinder.volumes.create.assert_not_called() + + def test_prepare_persistent_root_volumes_create(self): + """Create persistent root volume.""" + self.vimconn.cinder.volumes.create.return_value.id = volume_id2 + vm_av_zone = ["nova"] + base_disk_index = ord("a") + disk = {"size": 10, "image_id": image_id} + block_device_mapping = {} + existing_vim_volumes = [] + created_items = {} + expected_boot_vol_id = volume_id2 + expected_block_device_mapping = {"vda": volume_id2} + expected_existing_vim_volumes = [] + boot_volume_id = self.vimconn._prepare_persistent_root_volumes( + name, + vm_av_zone, + disk, + base_disk_index, + block_device_mapping, + existing_vim_volumes, + created_items, + ) + self.assertEqual(boot_volume_id, expected_boot_vol_id) + self.assertDictEqual(block_device_mapping, expected_block_device_mapping) + self.assertEqual(existing_vim_volumes, expected_existing_vim_volumes) + self.vimconn.cinder.volumes.create.assert_called_once_with( + size=10, + name="basicvmvda", + imageRef=image_id, + availability_zone=["nova"], + ) + self.assertEqual(created_items, {f"volume:{volume_id2}": True}) + + def test_prepare_persistent_non_root_volumes_create(self): + """Create persistent non-root volume.""" + self.vimconn.cinder = CopyingMock() + self.vimconn.cinder.volumes.create.return_value.id = volume_id2 + vm_av_zone = ["nova"] + base_disk_index = ord("a") + disk = {"size": 10} + block_device_mapping = {} + existing_vim_volumes = [] + created_items = {} + expected_block_device_mapping = {"vda": volume_id2} + expected_existing_vim_volumes = [] + self.vimconn._prepare_non_root_persistent_volumes( + name, + disk, + vm_av_zone, + block_device_mapping, + base_disk_index, + existing_vim_volumes, + created_items, + ) + + self.assertDictEqual(block_device_mapping, expected_block_device_mapping) + self.assertEqual(existing_vim_volumes, expected_existing_vim_volumes) + self.vimconn.cinder.volumes.create.assert_called_once_with( + size=10, name="basicvmvda", availability_zone=["nova"] + ) + self.assertEqual(created_items, {f"volume:{volume_id2}": True}) + + def test_prepare_persistent_root_volumes_create_raise_exception(self): + """Create persistent root volume raise exception.""" + self.vimconn.cinder.volumes.create.side_effect = Exception + vm_av_zone = ["nova"] + base_disk_index = ord("a") + disk = {"size": 10, "image_id": image_id} + block_device_mapping = {} + existing_vim_volumes = [] + created_items = {} + + with self.assertRaises(Exception): + result = self.vimconn._prepare_persistent_root_volumes( + name, + vm_av_zone, + disk, + base_disk_index, + block_device_mapping, + existing_vim_volumes, + created_items, + ) + + self.assertEqual(result, None) + + self.vimconn.cinder.volumes.create.assert_called_once_with( + size=10, + name="basicvmvda", + imageRef=image_id, + availability_zone=["nova"], + ) + self.assertEqual(existing_vim_volumes, []) + self.assertEqual(block_device_mapping, {}) + self.assertEqual(created_items, {}) + + def test_prepare_persistent_non_root_volumes_create_raise_exception(self): + """Create persistent non-root volume raise exception.""" + self.vimconn.cinder.volumes.create.side_effect = Exception + vm_av_zone = ["nova"] + base_disk_index = ord("b") + disk = {"size": 10} + block_device_mapping = {} + existing_vim_volumes = [] + created_items = {} + + with self.assertRaises(Exception): + self.vimconn._prepare_non_root_persistent_volumes( + name, + disk, + vm_av_zone, + block_device_mapping, + base_disk_index, + existing_vim_volumes, + created_items, + ) + + self.vimconn.cinder.volumes.create.assert_called_once_with( + size=10, name="basicvmvdb", availability_zone=["nova"] + ) + self.assertEqual(existing_vim_volumes, []) + self.assertEqual(block_device_mapping, {}) + self.assertEqual(created_items, {}) + + @patch("time.sleep") + def test_wait_for_created_volumes_availability_volume_status_available( + self, mock_sleep + ): + """Created volume status is available.""" + elapsed_time = 5 + created_items = {f"volume:{volume_id2}": True} + self.vimconn.cinder.volumes.get.return_value.status = "available" + + result = self.vimconn._wait_for_created_volumes_availability( + elapsed_time, created_items + ) + self.assertEqual(result, elapsed_time) + self.vimconn.cinder.volumes.get.assert_called_with(volume_id2) + mock_sleep.assert_not_called() + + @patch("time.sleep") + def test_wait_for_existing_volumes_availability_volume_status_available( + self, mock_sleep + ): + """Existing volume status is available.""" + elapsed_time = 5 + existing_vim_volumes = [{"id": volume_id2}] + self.vimconn.cinder.volumes.get.return_value.status = "available" + + result = self.vimconn._wait_for_existing_volumes_availability( + elapsed_time, existing_vim_volumes + ) + self.assertEqual(result, elapsed_time) + self.vimconn.cinder.volumes.get.assert_called_with(volume_id2) + mock_sleep.assert_not_called() + + @patch("time.sleep") + def test_wait_for_created_volumes_availability_status_processing_multiple_volumes( + self, mock_sleep + ): + """Created volume status is processing.""" + elapsed_time = 5 + created_items = { + f"volume:{volume_id2}": True, + f"volume:{volume_id3}": True, + } + self.vimconn.cinder.volumes.get.side_effect = [ + Status("processing"), + Status("available"), + Status("available"), + ] + + result = self.vimconn._wait_for_created_volumes_availability( + elapsed_time, created_items + ) + self.assertEqual(result, 10) + _call_mock_get_volumes = self.vimconn.cinder.volumes.get.call_args_list + self.assertEqual(_call_mock_get_volumes[0][0], (volume_id2,)) + self.assertEqual(_call_mock_get_volumes[1][0], (volume_id2,)) + self.assertEqual(_call_mock_get_volumes[2][0], (volume_id3,)) + mock_sleep.assert_called_with(5) + self.assertEqual(1, mock_sleep.call_count) + + @patch("time.sleep") + def test_wait_for_existing_volumes_availability_status_processing_multiple_volumes( + self, mock_sleep + ): + """Existing volume status is processing.""" + elapsed_time = 5 + existing_vim_volumes = [ + {"id": volume_id2}, + {"id": "44e0e83-b9uu-4akk-t234-p9cc4811bd4a"}, + ] + self.vimconn.cinder.volumes.get.side_effect = [ + Status("processing"), + Status("available"), + Status("available"), + ] + + result = self.vimconn._wait_for_existing_volumes_availability( + elapsed_time, existing_vim_volumes + ) + self.assertEqual(result, 10) + _call_mock_get_volumes = self.vimconn.cinder.volumes.get.call_args_list + self.assertEqual(_call_mock_get_volumes[0][0], (volume_id2,)) + self.assertEqual(_call_mock_get_volumes[1][0], (volume_id2,)) + self.assertEqual( + _call_mock_get_volumes[2][0], ("44e0e83-b9uu-4akk-t234-p9cc4811bd4a",) + ) + mock_sleep.assert_called_with(5) + self.assertEqual(1, mock_sleep.call_count) + + @patch("time.sleep") + def test_wait_for_created_volumes_availability_volume_status_processing_timeout( + self, mock_sleep + ): + """Created volume status is processing, elapsed time greater than timeout (1800).""" + elapsed_time = 1805 + created_items = {f"volume:{volume_id2}": True} + self.vimconn.cinder.volumes.get.side_effect = [ + Status("processing"), + Status("processing"), + ] + with patch("time.sleep", mock_sleep): + result = self.vimconn._wait_for_created_volumes_availability( + elapsed_time, created_items + ) + self.assertEqual(result, 1805) + self.vimconn.cinder.volumes.get.assert_not_called() + mock_sleep.assert_not_called() + + @patch("time.sleep") + def test_wait_for_existing_volumes_availability_volume_status_processing_timeout( + self, mock_sleep + ): + """Exsiting volume status is processing, elapsed time greater than timeout (1800).""" + elapsed_time = 1805 + existing_vim_volumes = [{"id": volume_id2}] + self.vimconn.cinder.volumes.get.side_effect = [ + Status("processing"), + Status("processing"), + ] + + result = self.vimconn._wait_for_existing_volumes_availability( + elapsed_time, existing_vim_volumes + ) + self.assertEqual(result, 1805) + self.vimconn.cinder.volumes.get.assert_not_called() + mock_sleep.assert_not_called() + + @patch("time.sleep") + def test_wait_for_created_volumes_availability_cinder_raise_exception( + self, mock_sleep + ): + """Cinder get volumes raises exception for created volumes.""" + elapsed_time = 1000 + created_items = {f"volume:{volume_id2}": True} + self.vimconn.cinder.volumes.get.side_effect = Exception + with self.assertRaises(Exception): + result = self.vimconn._wait_for_created_volumes_availability( + elapsed_time, created_items + ) + self.assertEqual(result, 1000) + self.vimconn.cinder.volumes.get.assert_called_with(volume_id2) + mock_sleep.assert_not_called() + + @patch("time.sleep") + def test_wait_for_existing_volumes_availability_cinder_raise_exception( + self, mock_sleep + ): + """Cinder get volumes raises exception for existing volumes.""" + elapsed_time = 1000 + existing_vim_volumes = [{"id": volume_id2}] + self.vimconn.cinder.volumes.get.side_effect = Exception + with self.assertRaises(Exception): + result = self.vimconn._wait_for_existing_volumes_availability( + elapsed_time, existing_vim_volumes + ) + self.assertEqual(result, 1000) + self.vimconn.cinder.volumes.get.assert_called_with(volume_id2) + mock_sleep.assert_not_called() + + @patch("time.sleep") + def test_wait_for_created_volumes_availability_no_volume_in_created_items( + self, mock_sleep + ): + """Created_items dict does not have volume-id.""" + elapsed_time = 10 + created_items = {} + + self.vimconn.cinder.volumes.get.side_effect = [None] + + result = self.vimconn._wait_for_created_volumes_availability( + elapsed_time, created_items + ) + self.assertEqual(result, 10) + self.vimconn.cinder.volumes.get.assert_not_called() + mock_sleep.assert_not_called() + + @patch("time.sleep") + def test_wait_for_existing_volumes_availability_no_volume_in_existing_vim_volumes( + self, mock_sleep + ): + """Existing_vim_volumes list does not have volume.""" + elapsed_time = 10 + existing_vim_volumes = [] + + self.vimconn.cinder.volumes.get.side_effect = [None] + + result = self.vimconn._wait_for_existing_volumes_availability( + elapsed_time, existing_vim_volumes + ) + self.assertEqual(result, 10) + self.vimconn.cinder.volumes.get.assert_not_called() + mock_sleep.assert_not_called() + + @patch.object(vimconnector, "_prepare_persistent_root_volumes") + @patch.object(vimconnector, "_prepare_non_root_persistent_volumes") + @patch.object(vimconnector, "_wait_for_created_volumes_availability") + @patch.object(vimconnector, "_wait_for_existing_volumes_availability") + def test_prepare_disk_for_vm_instance( + self, + mock_existing_vol_availability, + mock_created_vol_availability, + mock_non_root_volumes, + mock_root_volumes, + ): + """Prepare disks for VM instance successfully.""" + existing_vim_volumes = [] + created_items = {} + vm_av_zone = ["nova"] + + mock_root_volumes.return_value = root_vol_id + mock_created_vol_availability.return_value = 10 + mock_existing_vol_availability.return_value = 15 + self.vimconn.cinder = CopyingMock() + + self.vimconn._prepare_disk_for_vminstance( + name, existing_vim_volumes, created_items, vm_av_zone, disk_list2 + ) + self.vimconn.cinder.volumes.set_bootable.assert_called_once_with( + root_vol_id, True + ) + mock_created_vol_availability.assert_called_once_with(0, created_items) + mock_existing_vol_availability.assert_called_once_with(10, existing_vim_volumes) + self.assertEqual(mock_root_volumes.call_count, 1) + self.assertEqual(mock_non_root_volumes.call_count, 1) + mock_root_volumes.assert_called_once_with( + name="basicvm", + vm_av_zone=["nova"], + disk={"size": 10, "image_id": image_id}, + base_disk_index=97, + block_device_mapping={}, + existing_vim_volumes=[], + created_items={}, + ) + mock_non_root_volumes.assert_called_once_with( + name="basicvm", + disk={"size": 20}, + vm_av_zone=["nova"], + base_disk_index=98, + block_device_mapping={}, + existing_vim_volumes=[], + created_items={}, + ) + + @patch.object(vimconnector, "_prepare_persistent_root_volumes") + @patch.object(vimconnector, "_prepare_non_root_persistent_volumes") + @patch.object(vimconnector, "_wait_for_created_volumes_availability") + @patch.object(vimconnector, "_wait_for_existing_volumes_availability") + def test_prepare_disk_for_vm_instance_timeout_exceeded( + self, + mock_existing_vol_availability, + mock_created_vol_availability, + mock_non_root_volumes, + mock_root_volumes, + ): + """Timeout exceeded while waiting for disks.""" + existing_vim_volumes = [] + created_items = {} + vm_av_zone = ["nova"] + + mock_root_volumes.return_value = root_vol_id + mock_created_vol_availability.return_value = 1700 + mock_existing_vol_availability.return_value = 1900 + + with self.assertRaises(VimConnException) as err: + self.vimconn._prepare_disk_for_vminstance( + name, existing_vim_volumes, created_items, vm_av_zone, disk_list2 + ) + self.assertEqual( + str(err.exception), "Timeout creating volumes for instance basicvm" + ) + self.vimconn.cinder.volumes.set_bootable.assert_not_called() + mock_created_vol_availability.assert_called_once_with(0, created_items) + mock_existing_vol_availability.assert_called_once_with( + 1700, existing_vim_volumes + ) + self.assertEqual(mock_root_volumes.call_count, 1) + self.assertEqual(mock_non_root_volumes.call_count, 1) + mock_root_volumes.assert_called_once_with( + name="basicvm", + vm_av_zone=["nova"], + disk={"size": 10, "image_id": image_id}, + base_disk_index=97, + block_device_mapping={}, + existing_vim_volumes=[], + created_items={}, + ) + mock_non_root_volumes.assert_called_once_with( + name="basicvm", + disk={"size": 20}, + vm_av_zone=["nova"], + base_disk_index=98, + block_device_mapping={}, + existing_vim_volumes=[], + created_items={}, + ) + + @patch.object(vimconnector, "_prepare_persistent_root_volumes") + @patch.object(vimconnector, "_prepare_non_root_persistent_volumes") + @patch.object(vimconnector, "_wait_for_created_volumes_availability") + @patch.object(vimconnector, "_wait_for_existing_volumes_availability") + def test_prepare_disk_for_vm_instance_empty_disk_list( + self, + mock_existing_vol_availability, + mock_created_vol_availability, + mock_non_root_volumes, + mock_root_volumes, + ): + """Disk list is empty.""" + existing_vim_volumes = [] + created_items = {} + vm_av_zone = ["nova"] + mock_created_vol_availability.return_value = 2 + mock_existing_vol_availability.return_value = 3 + + self.vimconn._prepare_disk_for_vminstance( + name, existing_vim_volumes, created_items, vm_av_zone, disk_list + ) + self.vimconn.cinder.volumes.set_bootable.assert_not_called() + mock_created_vol_availability.assert_called_once_with(0, created_items) + mock_existing_vol_availability.assert_called_once_with(2, existing_vim_volumes) + mock_root_volumes.assert_not_called() + mock_non_root_volumes.assert_not_called() + + @patch.object(vimconnector, "_prepare_persistent_root_volumes") + @patch.object(vimconnector, "_prepare_non_root_persistent_volumes") + @patch.object(vimconnector, "_wait_for_created_volumes_availability") + @patch.object(vimconnector, "_wait_for_existing_volumes_availability") + def test_prepare_disk_for_vm_instance_persistent_root_volume_error( + self, + mock_existing_vol_availability, + mock_created_vol_availability, + mock_non_root_volumes, + mock_root_volumes, + ): + """Persistent root volumes preparation raises error.""" + existing_vim_volumes = [] + created_items = {} + vm_av_zone = ["nova"] + + mock_root_volumes.side_effect = Exception() + mock_created_vol_availability.return_value = 10 + mock_existing_vol_availability.return_value = 15 + + with self.assertRaises(Exception): + self.vimconn._prepare_disk_for_vminstance( + name, existing_vim_volumes, created_items, vm_av_zone, disk_list2 + ) + self.vimconn.cinder.volumes.set_bootable.assert_not_called() + mock_created_vol_availability.assert_not_called() + mock_existing_vol_availability.assert_not_called() + mock_root_volumes.assert_called_once_with( + name="basicvm", + vm_av_zone=["nova"], + disk={"size": 10, "image_id": image_id}, + base_disk_index=97, + block_device_mapping={}, + existing_vim_volumes=[], + created_items={}, + ) + mock_non_root_volumes.assert_not_called() + + @patch.object(vimconnector, "_prepare_persistent_root_volumes") + @patch.object(vimconnector, "_prepare_non_root_persistent_volumes") + @patch.object(vimconnector, "_wait_for_created_volumes_availability") + @patch.object(vimconnector, "_wait_for_existing_volumes_availability") + def test_prepare_disk_for_vm_instance_non_root_volume_error( + self, + mock_existing_vol_availability, + mock_created_vol_availability, + mock_non_root_volumes, + mock_root_volumes, + ): + """Non-root volumes preparation raises error.""" + existing_vim_volumes = [] + created_items = {} + vm_av_zone = ["nova"] + + mock_root_volumes.return_value = root_vol_id + mock_non_root_volumes.side_effect = Exception + + with self.assertRaises(Exception): + self.vimconn._prepare_disk_for_vminstance( + name, existing_vim_volumes, created_items, vm_av_zone, disk_list2 + ) + self.vimconn.cinder.volumes.set_bootable.assert_not_called() + mock_created_vol_availability.assert_not_called() + mock_existing_vol_availability.assert_not_called() + self.assertEqual(mock_root_volumes.call_count, 1) + self.assertEqual(mock_non_root_volumes.call_count, 1) + mock_root_volumes.assert_called_once_with( + name="basicvm", + vm_av_zone=["nova"], + disk={"size": 10, "image_id": image_id}, + base_disk_index=97, + block_device_mapping={}, + existing_vim_volumes=[], + created_items={}, + ) + mock_non_root_volumes.assert_called_once_with( + name="basicvm", + disk={"size": 20}, + vm_av_zone=["nova"], + base_disk_index=98, + block_device_mapping={}, + existing_vim_volumes=[], + created_items={}, + ) + + def test_find_external_network_for_floating_ip_no_external_network(self): + """External network could not be found.""" + self.vimconn.neutron.list_networks.return_value = { + "networks": [ + {"id": "408b73-r9cc-5a6a-a270-82cc4811bd4a", "router:external": False} + ] + } + with self.assertRaises(VimConnException) as err: + self.vimconn._find_the_external_network_for_floating_ip() + self.assertEqual( + str(err.exception), + "Cannot create floating_ip automatically since no external network is present", + ) + + def test_find_external_network_for_floating_one_external_network(self): + """One external network has been found.""" + self.vimconn.neutron.list_networks.return_value = { + "networks": [ + {"id": "408b73-r9cc-5a6a-a270-82cc4811bd4a", "router:external": True} + ] + } + expected_result = "408b73-r9cc-5a6a-a270-82cc4811bd4a" + result = self.vimconn._find_the_external_network_for_floating_ip() + self.assertEqual(result, expected_result) + + def test_find_external_network_for_floating_neutron_raises_exception(self): + """Neutron list networks raises exception.""" + self.vimconn.neutron.list_networks.side_effect = Exception + with self.assertRaises(Exception): + self.vimconn._find_the_external_network_for_floating_ip() + + def test_find_external_network_for_floating_several_external_network(self): + """Several exernal networks has been found.""" + self.vimconn.neutron.list_networks.return_value = { + "networks": [ + {"id": "408b73-r9cc-5a6a-a270-82cc4811bd4a", "router:external": True}, + {"id": "608b73-y9cc-5a6a-a270-12cc4811bd4a", "router:external": True}, + ] + } + with self.assertRaises(VimConnException) as err: + self.vimconn._find_the_external_network_for_floating_ip() + self.assertEqual( + str(err.exception), + "Cannot create floating_ip automatically since multiple external networks are present", + ) + + def test_neutron_create_float_ip(self): + """Floating ip creation is successful.""" + param = {"net_id": "408b73-r9cc-5a6a-a270-p2cc4811bd9a"} + created_items = {} + self.vimconn.neutron.create_floatingip.return_value = { + "floatingip": {"id": "308b73-t9cc-1a6a-a270-12cc4811bd4a"} + } + expected_created_items = { + "floating_ip:308b73-t9cc-1a6a-a270-12cc4811bd4a": True + } + self.vimconn._neutron_create_float_ip(param, created_items) + self.assertEqual(created_items, expected_created_items) + + def test_neutron_create_float_ip_exception_occured(self): + """Floating ip could not be created.""" + param = { + "floatingip": { + "floating_network_id": "408b73-r9cc-5a6a-a270-p2cc4811bd9a", + "tenant_id": "308b73-19cc-8a6a-a270-02cc4811bd9a", + } + } + created_items = {} + self.vimconn.neutron = CopyingMock() + self.vimconn.neutron.create_floatingip.side_effect = Exception( + "Neutron floating ip create exception occured." + ) + with self.assertRaises(VimConnException) as err: + self.vimconn._neutron_create_float_ip(param, created_items) + self.assertEqual(created_items, {}) + self.assertEqual( + str(err.exception), + "Exception: Cannot create new floating_ip Neutron floating ip create exception occured.", + ) + + @patch.object(vimconnector, "_neutron_create_float_ip") + @patch.object(vimconnector, "_find_the_external_network_for_floating_ip") + def test_create_floating_ip_pool_id_available( + self, mock_find_ext_network, mock_create_float_ip + ): + """Floating ip creation, ip pool is available.""" + floating_network = {"floating_ip": "308b73-t9cc-1a6a-a270-12cc4811bd4a"} + created_items = {} + expected_param = { + "floatingip": { + "floating_network_id": "308b73-t9cc-1a6a-a270-12cc4811bd4a", + "tenant_id": "408b73-r9cc-5a6a-a270-82cc4811bd4a", + } + } + self.vimconn._create_floating_ip(floating_network, self.server, created_items) + mock_find_ext_network.assert_not_called() + mock_create_float_ip.assert_called_once_with(expected_param, {}) + + @patch.object(vimconnector, "_neutron_create_float_ip") + @patch.object(vimconnector, "_find_the_external_network_for_floating_ip") + def test_create_floating_ip_finding_pool_id( + self, mock_find_ext_network, mock_create_float_ip + ): + """Floating ip creation, pool id need to be found.""" + floating_network = {"floating_ip": True} + created_items = {} + mock_find_ext_network.return_value = "308b73-t9cc-1a6a-a270-12cc4811bd4a" + expected_param = { + "floatingip": { + "floating_network_id": "308b73-t9cc-1a6a-a270-12cc4811bd4a", + "tenant_id": "408b73-r9cc-5a6a-a270-82cc4811bd4a", + } + } + self.vimconn._create_floating_ip(floating_network, self.server, created_items) + mock_find_ext_network.assert_called_once() + mock_create_float_ip.assert_called_once_with(expected_param, {}) + + @patch.object(vimconnector, "_neutron_create_float_ip") + @patch.object(vimconnector, "_find_the_external_network_for_floating_ip") + def test_create_floating_ip_neutron_create_floating_ip_exception( + self, mock_find_ext_network, mock_create_float_ip + ): + """Neutron creat floating ip raises error.""" + floating_network = {"floating_ip": True} + created_items = {} + mock_create_float_ip.side_effect = VimConnException( + "Can not create floating ip." + ) + mock_find_ext_network.return_value = "308b73-t9cc-1a6a-a270-12cc4811bd4a" + expected_param = { + "floatingip": { + "floating_network_id": "308b73-t9cc-1a6a-a270-12cc4811bd4a", + "tenant_id": "408b73-r9cc-5a6a-a270-82cc4811bd4a", + } + } + + with self.assertRaises(VimConnException) as err: + self.vimconn._create_floating_ip( + floating_network, self.server, created_items + ) + self.assertEqual(str(err.exception), "Can not create floating ip.") + mock_find_ext_network.assert_called_once() + mock_create_float_ip.assert_called_once_with(expected_param, {}) + + @patch.object(vimconnector, "_neutron_create_float_ip") + @patch.object(vimconnector, "_find_the_external_network_for_floating_ip") + def test_create_floating_ip_can_not_find_pool_id( + self, mock_find_ext_network, mock_create_float_ip + ): + """Floating ip creation, pool id could not be found.""" + floating_network = {"floating_ip": True} + created_items = {} + mock_find_ext_network.side_effect = VimConnException( + "Cannot create floating_ip automatically since no external network is present" + ) + with self.assertRaises(VimConnException) as err: + self.vimconn._create_floating_ip( + floating_network, self.server, created_items + ) + self.assertEqual( + str(err.exception), + "Cannot create floating_ip automatically since no external network is present", + ) + mock_find_ext_network.assert_called_once() + mock_create_float_ip.assert_not_called() + + def test_find_floating_ip_get_free_floating_ip(self): + """Get free floating ips successfully.""" + floating_ips = [ + { + "tenant_id": "408b73-r9cc-5a6a-a270-82cc4811bd4a", + "floating_network_id": "308b73-t9cc-1a6a-a270-12cc4811bd4a", + "id": "508b73-o9cc-5a6a-a270-72cc4811bd8", + } + ] + floating_network = {"floating_ip": "308b73-t9cc-1a6a-a270-12cc4811bd4a"} + expected_result = "508b73-o9cc-5a6a-a270-72cc4811bd8" + + result = self.vimconn._find_floating_ip( + self.server, floating_ips, floating_network + ) + self.assertEqual(result, expected_result) + + def test_find_floating_ip_different_floating_network_id(self): + """Floating network id is different with floating_ip of floating network.""" + floating_ips = [ + { + "floating_network_id": "308b73-t9cc-1a6a-a270-12cc4811bd4a", + "id": "508b73-o9cc-5a6a-a270-72cc4811bd8", + } + ] + floating_network = {"floating_ip": "508b73-t9cc-1a6a-a270-12cc4811bd4a"} + + result = self.vimconn._find_floating_ip( + self.server, floating_ips, floating_network + ) + self.assertEqual(result, None) + + def test_find_floating_ip_different_fip_tenant(self): + """Items in floating_ips has port_id, tenant_is is not same with server tenant id.""" + floating_ips = [ + { + "port_id": "608b73-r9cc-5a6a-a270-82cc4811bd4a", + "floating_network_id": "308b73-t9cc-1a6a-a270-12cc4811bd4a", + "id": "508b73-o9cc-5a6a-a270-72cc4811bd8", + "tenant_id": self.server.id, + } + ] + floating_network = {"floating_ip": "308b73-t9cc-1a6a-a270-12cc4811bd4a"} + mock_create_floating_ip = CopyingMock() + with patch.object(vimconnector, "_create_floating_ip", mock_create_floating_ip): + result = self.vimconn._find_floating_ip( + self.server, floating_ips, floating_network + ) + self.assertEqual(result, None) + + @patch("time.sleep") + def test_assign_floating_ip(self, mock_sleep): + """Assign floating ip successfully.""" + free_floating_ip = "508b73-o9cc-5a6a-a270-72cc4811bd8" + floating_network = {"vim_id": floating_network_vim_id} + fip = { + "port_id": floating_network_vim_id, + "floating_network_id": "p08b73-e9cc-5a6a-t270-82cc4811bd4a", + "id": "508b73-o9cc-5a6a-a270-72cc4811bd8", + "tenant_id": "k08b73-e9cc-5a6a-t270-82cc4811bd4a", + } + self.vimconn.neutron.update_floatingip.side_effect = None + self.vimconn.neutron.show_floatingip.return_value = fip + expected_result = fip + + result = self.vimconn._assign_floating_ip(free_floating_ip, floating_network) + self.assertEqual(result, expected_result) + self.vimconn.neutron.update_floatingip.assert_called_once_with( + free_floating_ip, + {"floatingip": {"port_id": floating_network_vim_id}}, + ) + mock_sleep.assert_called_once_with(5) + self.vimconn.neutron.show_floatingip.assert_called_once_with(free_floating_ip) + + @patch("time.sleep") + def test_assign_floating_ip_update_floating_ip_exception(self, mock_sleep): + """Neutron update floating ip raises exception.""" + free_floating_ip = "508b73-o9cc-5a6a-a270-72cc4811bd8" + floating_network = {"vim_id": floating_network_vim_id} + self.vimconn.neutron = CopyingMock() + self.vimconn.neutron.update_floatingip.side_effect = Exception( + "Floating ip is not updated." + ) + + with self.assertRaises(Exception) as err: + result = self.vimconn._assign_floating_ip( + free_floating_ip, floating_network + ) + self.assertEqual(result, None) + self.assertEqual(str(err.exception), "Floating ip is not updated.") + + self.vimconn.neutron.update_floatingip.assert_called_once_with( + free_floating_ip, + {"floatingip": {"port_id": floating_network_vim_id}}, + ) + mock_sleep.assert_not_called() + self.vimconn.neutron.show_floatingip.assert_not_called() + + @patch("time.sleep") + def test_assign_floating_ip_show_floating_ip_exception(self, mock_sleep): + """Neutron show floating ip raises exception.""" + free_floating_ip = "508b73-o9cc-5a6a-a270-72cc4811bd8" + floating_network = {"vim_id": floating_network_vim_id} + self.vimconn.neutron.update_floatingip.side_effect = None + self.vimconn.neutron.show_floatingip.side_effect = Exception( + "Floating ip could not be shown." + ) + + with self.assertRaises(Exception) as err: + result = self.vimconn._assign_floating_ip( + free_floating_ip, floating_network + ) + self.assertEqual(result, None) + self.assertEqual(str(err.exception), "Floating ip could not be shown.") + self.vimconn.neutron.update_floatingip.assert_called_once_with( + free_floating_ip, + {"floatingip": {"port_id": floating_network_vim_id}}, + ) + mock_sleep.assert_called_once_with(5) + self.vimconn.neutron.show_floatingip.assert_called_once_with(free_floating_ip) + + @patch("random.shuffle") + @patch.object(vimconnector, "_find_floating_ip") + def test_get_free_floating_ip(self, mock_find_floating_ip, mock_shuffle): + """Get free floating ip successfully.""" + floating_network = {"floating_ip": "308b73-t9cc-1a6a-a270-12cc4811bd4a"} + created_items = {} + floating_ips = [ + { + "port_id": "608b73-r9cc-5a6a-a270-82cc4811bd4a", + "floating_network_id": "308b73-t9cc-1a6a-a270-12cc4811bd4a", + "id": "508b73-o9cc-5a6a-a270-72cc4811bd8", + "tenant_id": "208b73-e9cc-5a6a-t270-82cc4811bd4a", + }, + { + "port_id": "508b73-r9cc-5a6a-5270-o2cc4811bd4a", + "floating_network_id": "308b73-t9cc-1a6a-a270-12cc4811bd4a", + "id": "208b73-o9cc-5a6a-a270-52cc4811bd8", + "tenant_id": "208b73-e9cc-5a6a-t270-82cc4811bd4a", + }, + ] + self.vimconn.neutron.list_floatingips.return_value = { + "floatingips": floating_ips + } + mock_find_floating_ip.return_value = "508b73-o9cc-5a6a-a270-72cc4811bd8" + expected_result = "508b73-o9cc-5a6a-a270-72cc4811bd8" + + result = self.vimconn._get_free_floating_ip( + self.server, floating_network, created_items + ) + self.assertEqual(result, expected_result) + mock_shuffle.assert_called_once_with(floating_ips) + mock_find_floating_ip.assert_called_once_with( + self.server, floating_ips, floating_network, created_items + ) + + @patch("random.shuffle") + @patch.object(vimconnector, "_find_floating_ip") + def test_get_free_floating_ip_list_floating_ip_exception( + self, mock_find_floating_ip, mock_shuffle + ): + """Neutron list floating IPs raises exception.""" + floating_network = {"floating_ip": "308b73-t9cc-1a6a-a270-12cc4811bd4a"} + created_items = {} + self.vimconn.neutron = CopyingMock() + self.vimconn.neutron.list_floatingips.side_effect = Exception( + "Floating ips could not be listed." + ) + with self.assertRaises(Exception) as err: + result = self.vimconn._get_free_floating_ip( + self.server, floating_network, created_items + ) + self.assertEqual(result, None) + self.assertEqual(str(err.exception), "Floating ips could not be listed.") + mock_shuffle.assert_not_called() + mock_find_floating_ip.assert_not_called() + + @patch("random.shuffle") + @patch.object(vimconnector, "_find_floating_ip") + def test_get_free_floating_ip_find_floating_ip_exception( + self, mock_find_floating_ip, mock_shuffle + ): + """_find_floating_ip method raises exception.""" + floating_network = {"floating_ip": "308b73-t9cc-1a6a-a270-12cc4811bd4a"} + created_items = {} + floating_ips = [ + { + "port_id": "608b73-r9cc-5a6a-a270-82cc4811bd4a", + "floating_network_id": "308b73-t9cc-1a6a-a270-12cc4811bd4a", + "id": "508b73-o9cc-5a6a-a270-72cc4811bd8", + "tenant_id": "208b73-e9cc-5a6a-t270-82cc4811bd4a", + }, + { + "port_id": "508b73-r9cc-5a6a-5270-o2cc4811bd4a", + "floating_network_id": "308b73-t9cc-1a6a-a270-12cc4811bd4a", + "id": "208b73-o9cc-5a6a-a270-52cc4811bd8", + "tenant_id": "208b73-e9cc-5a6a-t270-82cc4811bd4a", + }, + ] + self.vimconn.neutron = CopyingMock() + self.vimconn.neutron.list_floatingips.return_value = { + "floatingips": floating_ips + } + mock_find_floating_ip.side_effect = Exception( + "Free floating ip could not be found." + ) + + with self.assertRaises(Exception) as err: + result = self.vimconn._get_free_floating_ip( + self.server, floating_network, created_items + ) + self.assertEqual(result, None) + self.assertEqual(str(err.exception), "Free floating ip could not be found.") + mock_shuffle.assert_called_once_with(floating_ips) + mock_find_floating_ip.assert_called_once_with( + self.server, floating_ips, floating_network, created_items + ) + + @patch.object(vimconnector, "_create_floating_ip") + @patch.object(vimconnector, "_get_free_floating_ip") + @patch.object(vimconnector, "_assign_floating_ip") + def test_prepare_external_network_for_vm_instance( + self, + mock_assign_floating_ip, + mock_get_free_floating_ip, + mock_create_floating_ip, + ): + """Prepare external network successfully.""" + external_network = [ + { + "floating_ip": "y08b73-o9cc-1a6a-a270-12cc4811bd4u", + "vim_id": "608b73-r9cc-5a6a-a270-82cc4811bd4a", + }, + ] + created_items = {} + vm_start_time = time_return_value + mock_get_free_floating_ip.side_effect = ["y08b73-o9cc-1a6a-a270-12cc4811bd4u"] + mock_assign_floating_ip.return_value = { + "floatingip": {"port_id": "608b73-r9cc-5a6a-a270-82cc4811bd4a"} + } + self.vimconn.neutron = CopyingMock() + self.vimconn.nova = CopyingMock() + self.vimconn.neutron.show_floatingip.return_value = { + "floatingip": {"port_id": ""} + } + + self.vimconn._prepare_external_network_for_vminstance( + external_network, self.server, created_items, vm_start_time + ) + + self.assertEqual(mock_get_free_floating_ip.call_count, 1) + mock_get_free_floating_ip.assert_called_once_with( + self.server, + { + "floating_ip": "y08b73-o9cc-1a6a-a270-12cc4811bd4u", + "vim_id": "608b73-r9cc-5a6a-a270-82cc4811bd4a", + }, + created_items, + ) + self.vimconn.neutron.show_floatingip.assert_called_once_with( + "y08b73-o9cc-1a6a-a270-12cc4811bd4u" + ) + self.vimconn.nova.servers.get.assert_not_called() + mock_create_floating_ip.assert_not_called() + mock_assign_floating_ip.assert_called_once_with( + "y08b73-o9cc-1a6a-a270-12cc4811bd4u", + { + "floating_ip": "y08b73-o9cc-1a6a-a270-12cc4811bd4u", + "vim_id": "608b73-r9cc-5a6a-a270-82cc4811bd4a", + }, + ) + + @patch("time.time") + @patch("time.sleep") + @patch.object(vimconnector, "_create_floating_ip") + @patch.object(vimconnector, "_get_free_floating_ip") + @patch.object(vimconnector, "_assign_floating_ip") + def test_prepare_external_network_for_vm_instance_no_free_floating_ip( + self, + mock_assign_floating_ip, + mock_get_free_floating_ip, + mock_create_floating_ip, + mock_sleep, + mock_time, + ): + """There is not any free floating ip.""" + floating_network = { + "floating_ip": "y08b73-o9cc-1a6a-a270-12cc4811bd4u", + "vim_id": "608b73-r9cc-5a6a-a270-82cc4811bd4a", + } + external_network = [floating_network] + + created_items = {} + vm_start_time = time_return_value + mock_get_free_floating_ip.return_value = None + mock_assign_floating_ip.return_value = {} + self.vimconn.nova.servers.get.return_value.status = "ERROR" + self.vimconn.neutron.show_floatingip.return_value = {} + + with self.assertRaises(KeyError): + self.vimconn._prepare_external_network_for_vminstance( + external_network, self.server, created_items, vm_start_time + ) + + self.assertEqual(mock_get_free_floating_ip.call_count, 4) + mock_get_free_floating_ip.assert_called_with( + self.server, + { + "floating_ip": "y08b73-o9cc-1a6a-a270-12cc4811bd4u", + "vim_id": "608b73-r9cc-5a6a-a270-82cc4811bd4a", + }, + created_items, + ) + self.vimconn.neutron.show_floatingip.assert_called_with(None) + mock_sleep.assert_not_called() + mock_time.assert_not_called() + self.assertEqual(self.vimconn.nova.servers.get.call_count, 4) + mock_create_floating_ip.assert_called_with( + floating_network, self.server, created_items + ) + self.assertEqual(mock_create_floating_ip.call_count, 4) + mock_assign_floating_ip.assert_not_called() + self.vimconn.nova.servers.get.assert_called_with(self.server.id) + + @patch("time.time") + @patch("time.sleep") + @patch.object(vimconnector, "_create_floating_ip") + @patch.object(vimconnector, "_get_free_floating_ip") + @patch.object(vimconnector, "_assign_floating_ip") + def test_prepare_external_network_for_vm_instance_no_free_fip_can_not_create_fip_exit_on_error_false( + self, + mock_assign_floating_ip, + mock_get_free_floating_ip, + mock_create_floating_ip, + mock_sleep, + mock_time, + ): + """There is not any free floating ip, create_floating ip method raise exception + exit_on_floating_ip_error set to False.""" + floating_network = { + "floating_ip": "y08b73-o9cc-1a6a-a270-12cc4811bd4u", + "vim_id": "608b73-r9cc-5a6a-a270-82cc4811bd4a", + "exit_on_floating_ip_error": False, + } + external_network = [floating_network] + + created_items = {} + vm_start_time = time_return_value + mock_get_free_floating_ip.return_value = None + mock_assign_floating_ip.return_value = {} + mock_create_floating_ip.side_effect = VimConnException( + "Can not create floating ip." + ) + self.vimconn.nova.servers.get.return_value.status = "ERROR" + self.vimconn.neutron.show_floatingip.return_value = {} + + self.vimconn._prepare_external_network_for_vminstance( + external_network, self.server, created_items, vm_start_time + ) + self.assertEqual(mock_get_free_floating_ip.call_count, 1) + mock_get_free_floating_ip.assert_called_with( + self.server, + { + "floating_ip": "y08b73-o9cc-1a6a-a270-12cc4811bd4u", + "vim_id": "608b73-r9cc-5a6a-a270-82cc4811bd4a", + "exit_on_floating_ip_error": False, + }, + created_items, + ) + self.vimconn.neutron.show_floatingip.assert_not_called() + mock_sleep.assert_not_called() + mock_time.assert_not_called() + self.vimconn.nova.servers.get.assert_not_called() + mock_create_floating_ip.assert_called_with( + floating_network, self.server, created_items + ) + self.assertEqual(mock_create_floating_ip.call_count, 1) + mock_assign_floating_ip.assert_not_called() + + @patch("time.time") + @patch("time.sleep") + @patch.object(vimconnector, "_create_floating_ip") + @patch.object(vimconnector, "_get_free_floating_ip") + @patch.object(vimconnector, "_assign_floating_ip") + def test_prepare_external_network_for_vm_instance_no_free_fip_can_not_create_fip_exit_on_error_true( + self, + mock_assign_floating_ip, + mock_get_free_floating_ip, + mock_create_floating_ip, + mock_sleep, + mock_time, + ): + """There is not any free floating ip, create_floating ip method raise exception + exit_on_floating_ip_error set to False.""" + floating_network = { + "floating_ip": "y08b73-o9cc-1a6a-a270-12cc4811bd4u", + "vim_id": "608b73-r9cc-5a6a-a270-82cc4811bd4a", + "exit_on_floating_ip_error": True, + } + external_network = [floating_network] + + created_items = {} + vm_start_time = time_return_value + mock_get_free_floating_ip.return_value = None + mock_assign_floating_ip.return_value = {} + mock_create_floating_ip.side_effect = VimConnException( + "Can not create floating ip." + ) + self.vimconn.nova.servers.get.return_value.status = "ERROR" + self.vimconn.neutron.show_floatingip.return_value = {} + with self.assertRaises(VimConnException): + self.vimconn._prepare_external_network_for_vminstance( + external_network, self.server, created_items, vm_start_time + ) + self.assertEqual(mock_get_free_floating_ip.call_count, 1) + mock_get_free_floating_ip.assert_called_with( + self.server, + { + "floating_ip": "y08b73-o9cc-1a6a-a270-12cc4811bd4u", + "vim_id": "608b73-r9cc-5a6a-a270-82cc4811bd4a", + "exit_on_floating_ip_error": True, + }, + created_items, + ) + self.vimconn.neutron.show_floatingip.assert_not_called() + mock_sleep.assert_not_called() + mock_time.assert_not_called() + self.vimconn.nova.servers.get.assert_not_called() + mock_create_floating_ip.assert_called_with( + floating_network, self.server, created_items + ) + self.assertEqual(mock_create_floating_ip.call_count, 1) + mock_assign_floating_ip.assert_not_called() + + @patch.object(vimconnector, "_create_floating_ip") + @patch.object(vimconnector, "_get_free_floating_ip") + @patch.object(vimconnector, "_assign_floating_ip") + def test_prepare_external_network_for_vm_instance_fip_has_port_id( + self, + mock_assign_floating_ip, + mock_get_free_floating_ip, + mock_create_floating_ip, + ): + """Neutron show floating ip return the fip with port_id and floating network vim_id + is different from port_id.""" + floating_network = { + "floating_ip": "y08b73-o9cc-1a6a-a270-12cc4811bd4u", + "vim_id": "608b73-r9cc-5a6a-a270-82cc4811bd4a", + } + external_network = [floating_network] + created_items = {} + vm_start_time = 150 + mock_get_free_floating_ip.side_effect = [ + "t08b73-o9cc-1a6a-a270-12cc4811bd4u", + "r08b73-o9cc-1a6a-a270-12cc4811bd4u", + "y08b73-o9cc-1a6a-a270-12cc4811bd4u", + ] + mock_assign_floating_ip.side_effect = [ + {"floatingip": {"port_id": "k08b73-r9cc-5a6a-a270-82cc4811bd4a"}}, + {"floatingip": {"port_id": "608b73-r9cc-5a6a-a270-82cc4811bd4a"}}, + ] + self.vimconn.neutron = CopyingMock() + self.vimconn.nova = CopyingMock() + self.vimconn.neutron.show_floatingip.side_effect = [ + {"floatingip": {"port_id": "608b73-r9cc-5a6a-a270-82cc4811bd4a"}}, + {"floatingip": {"port_id": ""}}, + {"floatingip": {"port_id": ""}}, + ] + self.vimconn._prepare_external_network_for_vminstance( + external_network, self.server, created_items, vm_start_time + ) + self.assertEqual(mock_get_free_floating_ip.call_count, 3) + _call_mock_get_free_floating_ip = mock_get_free_floating_ip.call_args_list + self.assertEqual( + _call_mock_get_free_floating_ip[0][0], + ( + self.server, + floating_network, + created_items, + ), + ) + self.assertEqual( + _call_mock_get_free_floating_ip[1][0], + ( + self.server, + floating_network, + created_items, + ), + ) + self.assertEqual( + _call_mock_get_free_floating_ip[2][0], + ( + self.server, + floating_network, + created_items, + ), + ) + self.assertEqual(self.vimconn.neutron.show_floatingip.call_count, 3) + self.vimconn.nova.servers.get.assert_not_called() + mock_create_floating_ip.assert_not_called() + self.assertEqual(mock_assign_floating_ip.call_count, 2) + _call_mock_assign_floating_ip = mock_assign_floating_ip.call_args_list + self.assertEqual( + _call_mock_assign_floating_ip[0][0], + ("r08b73-o9cc-1a6a-a270-12cc4811bd4u", floating_network), + ) + self.assertEqual( + _call_mock_assign_floating_ip[1][0], + ("y08b73-o9cc-1a6a-a270-12cc4811bd4u", floating_network), + ) + + @patch("time.time") + @patch("time.sleep") + @patch.object(vimconnector, "_create_floating_ip") + @patch.object(vimconnector, "_get_free_floating_ip") + @patch.object(vimconnector, "_assign_floating_ip") + def test_prepare_external_network_for_vm_instance_neutron_show_fip_exception_vm_status_in_error( + self, + mock_assign_floating_ip, + mock_get_free_floating_ip, + mock_create_floating_ip, + mock_sleep, + mock_time, + ): + """Neutron show floating ip gives exception, exit_on_floating_ip_error set to True, + VM status is in error.""" + floating_network = { + "floating_ip": "y08b73-o9cc-1a6a-a270-12cc4811bd4u", + "vim_id": "608b73-r9cc-5a6a-a270-82cc4811bd4a", + "exit_on_floating_ip_error": True, + } + external_network = [floating_network] + created_items = {} + vm_start_time = time_return_value + + mock_time.side_effect = [156570150, 156570800, 156571200] + + self.vimconn.nova.servers.get.return_value.status = "ERROR" + self.vimconn.neutron.show_floatingip.side_effect = [ + Exception("Floating ip could not be shown.") + ] * 4 + with self.assertRaises(Exception) as err: + self.vimconn._prepare_external_network_for_vminstance( + external_network, self.server, created_items, vm_start_time + ) + self.assertEqual( + str(err.exception), + "Cannot create floating_ip: Exception Floating ip could not be shown.", + ) + + self.assertEqual(mock_get_free_floating_ip.call_count, 4) + _call_mock_get_free_floating_ip = mock_get_free_floating_ip.call_args_list + self.assertEqual( + _call_mock_get_free_floating_ip[0][0], + ( + self.server, + floating_network, + created_items, + ), + ) + self.assertEqual( + _call_mock_get_free_floating_ip[1][0], + ( + self.server, + floating_network, + created_items, + ), + ) + self.assertEqual( + _call_mock_get_free_floating_ip[2][0], + ( + self.server, + floating_network, + created_items, + ), + ) + self.assertEqual( + _call_mock_get_free_floating_ip[3][0], + ( + self.server, + floating_network, + created_items, + ), + ) + + self.assertEqual(self.vimconn.neutron.show_floatingip.call_count, 4) + self.vimconn.nova.servers.get.assert_called_with(self.server.id) + mock_create_floating_ip.assert_not_called() + mock_assign_floating_ip.assert_not_called() + mock_time.assert_not_called() + mock_sleep.assert_not_called() + + @patch("time.time") + @patch("time.sleep") + @patch.object(vimconnector, "_create_floating_ip") + @patch.object(vimconnector, "_get_free_floating_ip") + @patch.object(vimconnector, "_assign_floating_ip") + def test_prepare_external_network_for_vm_instance_neutron_show_fip_exception_vm_status_in_active( + self, + mock_assign_floating_ip, + mock_get_free_floating_ip, + mock_create_floating_ip, + mock_sleep, + mock_time, + ): + """Neutron show floating ip gives exception, exit_on_floating_ip_error is set to False, + VM status is in active.""" + floating_network = { + "floating_ip": "y08b73-o9cc-1a6a-a270-12cc4811bd4u", + "vim_id": "608b73-r9cc-5a6a-a270-82cc4811bd4a", + "exit_on_floating_ip_error": False, + } + external_network = [floating_network] + created_items = {} + vm_start_time = time_return_value + + mock_time.side_effect = [156570150, 156570800, 156571200] + + self.vimconn.nova.servers.get.return_value.status = "ACTIVE" + self.vimconn.neutron.show_floatingip.side_effect = [ + Exception("Floating ip could not be shown.") + ] * 4 + + self.vimconn._prepare_external_network_for_vminstance( + external_network, self.server, created_items, vm_start_time + ) + # self.assertEqual(str(err.exception), "Cannot create floating_ip") + + self.assertEqual(mock_get_free_floating_ip.call_count, 4) + _call_mock_get_free_floating_ip = mock_get_free_floating_ip.call_args_list + self.assertEqual( + _call_mock_get_free_floating_ip[0][0], + ( + self.server, + floating_network, + created_items, + ), + ) + self.assertEqual( + _call_mock_get_free_floating_ip[1][0], + ( + self.server, + floating_network, + created_items, + ), + ) + self.assertEqual( + _call_mock_get_free_floating_ip[2][0], + ( + self.server, + floating_network, + created_items, + ), + ) + self.assertEqual( + _call_mock_get_free_floating_ip[3][0], + ( + self.server, + floating_network, + created_items, + ), + ) + + self.assertEqual(self.vimconn.neutron.show_floatingip.call_count, 4) + self.vimconn.nova.servers.get.assert_called_with(self.server.id) + mock_create_floating_ip.assert_not_called() + mock_assign_floating_ip.assert_not_called() + mock_time.assert_not_called() + mock_sleep.assert_not_called() + + @patch("time.time") + @patch("time.sleep") + @patch.object(vimconnector, "_create_floating_ip") + @patch.object(vimconnector, "_get_free_floating_ip") + @patch.object(vimconnector, "_assign_floating_ip") + def test_prepare_external_network_for_vm_instance_neutron_show_fip_exception_exit_on_error( + self, + mock_assign_floating_ip, + mock_get_free_floating_ip, + mock_create_floating_ip, + mock_sleep, + mock_time, + ): + """Neutron show floating ip gives exception, but exit_on_floating_ip_error is set to True. + VM status is not ACTIVE or ERROR, server timeout happened.""" + floating_network = { + "floating_ip": "y08b73-o9cc-1a6a-a270-12cc4811bd4u", + "vim_id": "608b73-r9cc-5a6a-a270-82cc4811bd4a", + "exit_on_floating_ip_error": True, + } + external_network = [floating_network] + created_items = {} + vm_start_time = time_return_value + mock_get_free_floating_ip.side_effect = None + mock_time.side_effect = [156571790, 156571795, 156571800, 156571805] + self.vimconn.nova.servers.get.return_value.status = "OTHER" + self.vimconn.neutron.show_floatingip.side_effect = [ + Exception("Floating ip could not be shown.") + ] * 5 + + with self.assertRaises(VimConnException) as err: + self.vimconn._prepare_external_network_for_vminstance( + external_network, self.server, created_items, vm_start_time + ) + self.assertEqual( + str(err.exception), + "Cannot create floating_ip: Exception Floating ip could not be shown.", + ) + + self.assertEqual(mock_get_free_floating_ip.call_count, 3) + _call_mock_get_free_floating_ip = mock_get_free_floating_ip.call_args_list + self.assertEqual( + _call_mock_get_free_floating_ip[0][0], + ( + self.server, + floating_network, + created_items, + ), + ) + self.assertEqual( + _call_mock_get_free_floating_ip[1][0], + ( + self.server, + floating_network, + created_items, + ), + ) + self.assertEqual( + _call_mock_get_free_floating_ip[2][0], + ( + self.server, + floating_network, + created_items, + ), + ) + + self.assertEqual(self.vimconn.neutron.show_floatingip.call_count, 3) + self.vimconn.nova.servers.get.assert_called_with(self.server.id) + mock_create_floating_ip.assert_not_called() + mock_assign_floating_ip.assert_not_called() + self.assertEqual(mock_time.call_count, 3) + self.assertEqual(mock_sleep.call_count, 2) + + @patch("time.time") + @patch("time.sleep") + @patch.object(vimconnector, "_create_floating_ip") + @patch.object(vimconnector, "_get_free_floating_ip") + @patch.object(vimconnector, "_assign_floating_ip") + def test_prepare_external_network_for_vm_instance_assign_floating_ip_exception_exit_on_error( + self, + mock_assign_floating_ip, + mock_get_free_floating_ip, + mock_create_floating_ip, + mock_sleep, + mock_time, + ): + """Assign floating ip method gives exception, exit_on_floating_ip_error is set to True. + VM status is in ERROR.""" + floating_network = { + "floating_ip": "y08b73-o9cc-1a6a-a270-12cc4811bd4u", + "vim_id": "608b73-r9cc-5a6a-a270-82cc4811bd4a", + "exit_on_floating_ip_error": True, + } + external_network = [floating_network] + created_items = {} + vm_start_time = time_return_value + + mock_get_free_floating_ip.side_effect = [ + "y08b73-o9cc-1a6a-a270-12cc4811bd4u" + ] * 4 + + mock_time.side_effect = [156571790, 156571795, 156571800, 156571805] + + mock_assign_floating_ip.side_effect = [ + Exception("Floating ip could not be assigned.") + ] * 4 + + self.vimconn.nova.servers.get.return_value.status = "ERROR" + self.vimconn.neutron.show_floatingip.side_effect = [ + {"floatingip": {"port_id": ""}} + ] * 4 + + with self.assertRaises(VimConnException) as err: + self.vimconn._prepare_external_network_for_vminstance( + external_network, self.server, created_items, vm_start_time + ) + self.assertEqual( + str(err.exception), + "Cannot create floating_ip: Exception Floating ip could not be assigned.", + ) + + self.assertEqual(mock_get_free_floating_ip.call_count, 4) + _call_mock_get_free_floating_ip = mock_get_free_floating_ip.call_args_list + self.assertEqual( + _call_mock_get_free_floating_ip[0][0], + ( + self.server, + floating_network, + created_items, + ), + ) + self.assertEqual( + _call_mock_get_free_floating_ip[1][0], + ( + self.server, + floating_network, + created_items, + ), + ) + self.assertEqual( + _call_mock_get_free_floating_ip[2][0], + ( + self.server, + floating_network, + created_items, + ), + ) + + self.assertEqual(self.vimconn.neutron.show_floatingip.call_count, 4) + self.vimconn.neutron.show_floatingip.assert_called_with( + "y08b73-o9cc-1a6a-a270-12cc4811bd4u" + ) + self.assertEqual(self.vimconn.nova.servers.get.call_count, 4) + self.vimconn.nova.servers.get.assert_called_with(self.server.id) + mock_time.assert_not_called() + mock_sleep.assert_not_called() + mock_create_floating_ip.assert_not_called() + + @patch("time.time") + @patch("time.sleep") + @patch.object(vimconnector, "_create_floating_ip") + @patch.object(vimconnector, "_get_free_floating_ip") + @patch.object(vimconnector, "_assign_floating_ip") + def test_prepare_external_network_for_vm_instance_empty_external_network_list( + self, + mock_assign_floating_ip, + mock_get_free_floating_ip, + mock_create_floating_ip, + mock_sleep, + mock_time, + ): + """External network list is empty.""" + external_network = [] + created_items = {} + vm_start_time = time_return_value + + self.vimconn._prepare_external_network_for_vminstance( + external_network, self.server, created_items, vm_start_time + ) + mock_create_floating_ip.assert_not_called() + mock_time.assert_not_called() + mock_sleep.assert_not_called() + mock_assign_floating_ip.assert_not_called() + mock_get_free_floating_ip.assert_not_called() + self.vimconn.neutron.show.show_floatingip.assert_not_called() + self.vimconn.nova.servers.get.assert_not_called() + + @patch.object(vimconnector, "_vimconnector__wait_for_vm") + def test_update_port_security_for_vm_instance(self, mock_wait_for_vm): + """no_secured_ports has port and the port has allow-address-pairs.""" + no_secured_ports = [(port2_id, "allow-address-pairs")] + + self.vimconn._update_port_security_for_vminstance(no_secured_ports, self.server) + + mock_wait_for_vm.assert_called_once_with(self.server.id, "ACTIVE") + + self.vimconn.neutron.update_port.assert_called_once_with( + port2_id, + {"port": {"allowed_address_pairs": [{"ip_address": "0.0.0.0/0"}]}}, + ) + + @patch.object(vimconnector, "_vimconnector__wait_for_vm") + def test_update_port_security_for_vm_instance_no_allowed_address_pairs( + self, mock_wait_for_vm + ): + """no_secured_ports has port and the port does not have allow-address-pairs.""" + no_secured_ports = [(port2_id, "something")] + + self.vimconn._update_port_security_for_vminstance(no_secured_ports, self.server) + + mock_wait_for_vm.assert_called_once_with(self.server.id, "ACTIVE") + + self.vimconn.neutron.update_port.assert_called_once_with( + port2_id, + {"port": {"port_security_enabled": False, "security_groups": None}}, + ) + + @patch.object(vimconnector, "_vimconnector__wait_for_vm") + def test_update_port_security_for_vm_instance_wait_for_vm_raise_exception( + self, mock_wait_for_vm + ): + """__wait_for_vm raises timeout exception.""" + no_secured_ports = [(port2_id, "something")] + + mock_wait_for_vm.side_effect = VimConnException("Timeout waiting for instance.") + + with self.assertRaises(VimConnException) as err: + self.vimconn._update_port_security_for_vminstance( + no_secured_ports, self.server + ) + self.assertEqual(str(err.exception), "Timeout waiting for instance.") + + mock_wait_for_vm.assert_called_once_with(self.server.id, "ACTIVE") + + self.vimconn.neutron.update_port.assert_not_called() + + @patch.object(vimconnector, "_vimconnector__wait_for_vm") + def test_update_port_security_for_vm_instance_neutron_update_port_raise_exception( + self, mock_wait_for_vm + ): + """neutron_update_port method raises exception.""" + no_secured_ports = [(port2_id, "something")] + + self.vimconn.neutron.update_port.side_effect = Exception( + "Port security could not be updated." + ) + + with self.assertRaises(VimConnException) as err: + self.vimconn._update_port_security_for_vminstance( + no_secured_ports, self.server + ) + self.assertEqual( + str(err.exception), + "It was not possible to disable port security for port 17472685-f67f-49fd-8722-eabb7692fc22", + ) + mock_wait_for_vm.assert_called_once_with(self.server.id, "ACTIVE") + + self.vimconn.neutron.update_port.assert_called_once_with( + port2_id, + {"port": {"port_security_enabled": False, "security_groups": None}}, + ) + + @patch.object(vimconnector, "_vimconnector__wait_for_vm") + def test_update_port_security_for_vm_instance_empty_port_list( + self, mock_wait_for_vm + ): + """no_secured_ports list does not have any ports.""" + no_secured_ports = [] + + self.vimconn._update_port_security_for_vminstance(no_secured_ports, self.server) + + mock_wait_for_vm.assert_not_called() + + self.vimconn.neutron.update_port.assert_not_called() + + @patch("time.time") + @patch.object(vimconnector, "_reload_connection") + @patch.object(vimconnector, "_prepare_network_for_vminstance") + @patch.object(vimconnector, "_create_user_data") + @patch.object(vimconnector, "_get_vm_availability_zone") + @patch.object(vimconnector, "_prepare_disk_for_vminstance") + @patch.object(vimconnector, "_update_port_security_for_vminstance") + @patch.object(vimconnector, "_prepare_external_network_for_vminstance") + @patch.object(vimconnector, "delete_vminstance") + @patch.object(vimconnector, "_format_exception") + def test_new_vm_instance( + self, + mock_format_exception, + mock_delete_vm_instance, + mock_prepare_external_network, + mock_update_port_security, + mock_prepare_disk_for_vm_instance, + mock_get_vm_availability_zone, + mock_create_user_data, + mock_prepare_network_for_vm_instance, + mock_reload_connection, + mock_time, + ): + """New VM instance creation is successful.""" + + mock_create_user_data.return_value = True, "userdata" + + mock_get_vm_availability_zone.return_value = "nova" + + self.vimconn.nova.servers.create.return_value = self.server + + mock_time.return_value = time_return_value + + expected_result = self.server.id, {} + + result = self.vimconn.new_vminstance( + name, + description, + start, + image_id, + flavor_id, + affinity_group_list, + net_list, + cloud_config, + disk_list2, + availability_zone_index, + availability_zone_list, + ) + self.assertEqual(result, expected_result) + + mock_reload_connection.assert_called_once() + mock_prepare_network_for_vm_instance.assert_called_once_with( + name=name, + net_list=net_list, + created_items={}, + net_list_vim=[], + external_network=[], + no_secured_ports=[], + ) + mock_create_user_data.assert_called_once_with(cloud_config) + mock_get_vm_availability_zone.assert_called_once_with( + availability_zone_index, availability_zone_list + ) + mock_prepare_disk_for_vm_instance.assert_called_once_with( + name=name, + existing_vim_volumes=[], + created_items={}, + vm_av_zone="nova", + disk_list=disk_list2, + ) + self.vimconn.nova.servers.create.assert_called_once_with( + name=name, + image=image_id, + flavor=flavor_id, + nics=[], + security_groups="default", + availability_zone="nova", + key_name="my_keypair", + userdata="userdata", + config_drive=True, + block_device_mapping=None, + scheduler_hints={}, + ) + mock_time.assert_called_once() + mock_update_port_security.assert_called_once_with([], self.server) + mock_prepare_external_network.assert_called_once_with( + external_network=[], + server=self.server, + created_items={}, + vm_start_time=time_return_value, + ) + mock_delete_vm_instance.assert_not_called() + mock_format_exception.assert_not_called() + + @patch("time.time") + @patch.object(vimconnector, "_reload_connection") + @patch.object(vimconnector, "_prepare_network_for_vminstance") + @patch.object(vimconnector, "_create_user_data") + @patch.object(vimconnector, "_get_vm_availability_zone") + @patch.object(vimconnector, "_prepare_disk_for_vminstance") + @patch.object(vimconnector, "_update_port_security_for_vminstance") + @patch.object(vimconnector, "_prepare_external_network_for_vminstance") + @patch.object(vimconnector, "delete_vminstance") + @patch.object(vimconnector, "_format_exception") + def test_new_vm_instance_create_user_data_fails( + self, + mock_format_exception, + mock_delete_vm_instance, + mock_prepare_external_network, + mock_update_port_security, + mock_prepare_disk_for_vm_instance, + mock_get_vm_availability_zone, + mock_create_user_data, + mock_prepare_network_for_vm_instance, + mock_reload_connection, + mock_time, + ): + """New VM instance creation failed because of user data creation failure.""" + + mock_create_user_data.side_effect = Exception( + "User data could not be retrieved." + ) + + mock_get_vm_availability_zone.return_value = "nova" + + self.vimconn.nova.servers.create.return_value = self.server + + mock_time.return_value = time_return_value + + self.vimconn.new_vminstance( + name, + description, + start, + image_id, + flavor_id, + affinity_group_list, + net_list, + cloud_config, + disk_list, + availability_zone_index, + availability_zone_list, + ) + + mock_reload_connection.assert_called_once() + mock_prepare_network_for_vm_instance.assert_called_once_with( + name=name, + net_list=net_list, + created_items={}, + net_list_vim=[], + external_network=[], + no_secured_ports=[], + ) + mock_create_user_data.assert_called_once_with(cloud_config) + mock_get_vm_availability_zone.assert_not_called() + mock_prepare_disk_for_vm_instance.assert_not_called() + self.vimconn.nova.servers.create.assert_not_called() + mock_time.assert_not_called() + mock_update_port_security.assert_not_called() + mock_prepare_external_network.assert_not_called() + mock_delete_vm_instance.assert_called_once_with(None, {}) + mock_format_exception.assert_called_once() + arg = mock_format_exception.call_args[0][0] + self.assertEqual(str(arg), "User data could not be retrieved.") + + @patch("time.time") + @patch.object(vimconnector, "_reload_connection") + @patch.object(vimconnector, "_prepare_network_for_vminstance") + @patch.object(vimconnector, "_create_user_data") + @patch.object(vimconnector, "_get_vm_availability_zone") + @patch.object(vimconnector, "_prepare_disk_for_vminstance") + @patch.object(vimconnector, "_update_port_security_for_vminstance") + @patch.object(vimconnector, "_prepare_external_network_for_vminstance") + @patch.object(vimconnector, "delete_vminstance") + @patch.object(vimconnector, "_format_exception") + def test_new_vm_instance_external_network_exception( + self, + mock_format_exception, + mock_delete_vm_instance, + mock_prepare_external_network, + mock_update_port_security, + mock_prepare_disk_for_vm_instance, + mock_get_vm_availability_zone, + mock_create_user_data, + mock_prepare_network_for_vm_instance, + mock_reload_connection, + mock_time, + ): + """New VM instance creation, external network connection has failed as floating + ip could not be created.""" + + mock_create_user_data.return_value = True, "userdata" + + mock_get_vm_availability_zone.return_value = "nova" + + self.vimconn.nova.servers.create.return_value = self.server + + mock_time.return_value = time_return_value + + mock_prepare_external_network.side_effect = VimConnException( + "Can not create floating ip." + ) + + self.vimconn.new_vminstance( + name, + description, + start, + image_id, + flavor_id, + affinity_group_list, + net_list, + cloud_config, + disk_list2, + availability_zone_index, + availability_zone_list, + ) + + mock_reload_connection.assert_called_once() + mock_prepare_network_for_vm_instance.assert_called_once_with( + name=name, + net_list=net_list, + created_items={}, + net_list_vim=[], + external_network=[], + no_secured_ports=[], + ) + mock_create_user_data.assert_called_once_with(cloud_config) + mock_get_vm_availability_zone.assert_called_once_with( + availability_zone_index, availability_zone_list + ) + mock_prepare_disk_for_vm_instance.assert_called_once_with( + name=name, + existing_vim_volumes=[], + created_items={}, + vm_av_zone="nova", + disk_list=disk_list2, + ) + self.vimconn.nova.servers.create.assert_called_once_with( + name=name, + image=image_id, + flavor=flavor_id, + nics=[], + security_groups="default", + availability_zone="nova", + key_name="my_keypair", + userdata="userdata", + config_drive=True, + block_device_mapping=None, + scheduler_hints={}, + ) + mock_time.assert_called_once() + mock_update_port_security.assert_called_once_with([], self.server) + mock_prepare_external_network.assert_called_once_with( + external_network=[], + server=self.server, + created_items={}, + vm_start_time=time_return_value, + ) + mock_delete_vm_instance.assert_called_once_with(self.server.id, {}) + mock_format_exception.assert_called_once() + arg = mock_format_exception.call_args[0][0] + self.assertEqual(str(arg), "Can not create floating ip.") + + @patch("time.time") + @patch.object(vimconnector, "_reload_connection") + @patch.object(vimconnector, "_prepare_network_for_vminstance") + @patch.object(vimconnector, "_create_user_data") + @patch.object(vimconnector, "_get_vm_availability_zone") + @patch.object(vimconnector, "_prepare_disk_for_vminstance") + @patch.object(vimconnector, "_update_port_security_for_vminstance") + @patch.object(vimconnector, "_prepare_external_network_for_vminstance") + @patch.object(vimconnector, "delete_vminstance") + @patch.object(vimconnector, "_format_exception") + def test_new_vm_instance_with_affinity_group( + self, + mock_format_exception, + mock_delete_vm_instance, + mock_prepare_external_network, + mock_update_port_security, + mock_prepare_disk_for_vm_instance, + mock_get_vm_availability_zone, + mock_create_user_data, + mock_prepare_network_for_vm_instance, + mock_reload_connection, + mock_time, + ): + """New VM creation with affinity group.""" + affinity_group_list = [ + {"affinity_group_id": "38b73-e9cc-5a6a-t270-82cc4811bd4a"} + ] + mock_create_user_data.return_value = True, "userdata" + mock_get_vm_availability_zone.return_value = "nova" + self.vimconn.nova.servers.create.return_value = self.server + mock_time.return_value = time_return_value + expected_result = self.server.id, {} + + result = self.vimconn.new_vminstance( + name, + description, + start, + image_id, + flavor_id, + affinity_group_list, + net_list, + cloud_config, + disk_list2, + availability_zone_index, + availability_zone_list, + ) + self.assertEqual(result, expected_result) + + mock_reload_connection.assert_called_once() + mock_prepare_network_for_vm_instance.assert_called_once_with( + name=name, + net_list=net_list, + created_items={}, + net_list_vim=[], + external_network=[], + no_secured_ports=[], + ) + mock_create_user_data.assert_called_once_with(cloud_config) + mock_get_vm_availability_zone.assert_called_once_with( + availability_zone_index, availability_zone_list + ) + mock_prepare_disk_for_vm_instance.assert_called_once_with( + name=name, + existing_vim_volumes=[], + created_items={}, + vm_av_zone="nova", + disk_list=disk_list2, + ) + self.vimconn.nova.servers.create.assert_called_once_with( + name=name, + image=image_id, + flavor=flavor_id, + nics=[], + security_groups="default", + availability_zone="nova", + key_name="my_keypair", + userdata="userdata", + config_drive=True, + block_device_mapping=None, + scheduler_hints={"group": "38b73-e9cc-5a6a-t270-82cc4811bd4a"}, + ) + mock_time.assert_called_once() + mock_update_port_security.assert_called_once_with([], self.server) + mock_prepare_external_network.assert_called_once_with( + external_network=[], + server=self.server, + created_items={}, + vm_start_time=time_return_value, + ) + mock_delete_vm_instance.assert_not_called() + mock_format_exception.assert_not_called() + + @patch("time.time") + @patch.object(vimconnector, "_reload_connection") + @patch.object(vimconnector, "_prepare_network_for_vminstance") + @patch.object(vimconnector, "_create_user_data") + @patch.object(vimconnector, "_get_vm_availability_zone") + @patch.object(vimconnector, "_prepare_disk_for_vminstance") + @patch.object(vimconnector, "_update_port_security_for_vminstance") + @patch.object(vimconnector, "_prepare_external_network_for_vminstance") + @patch.object(vimconnector, "delete_vminstance") + @patch.object(vimconnector, "_format_exception") + def test_new_vm_instance_nova_server_create_failed( + self, + mock_format_exception, + mock_delete_vm_instance, + mock_prepare_external_network, + mock_update_port_security, + mock_prepare_disk_for_vm_instance, + mock_get_vm_availability_zone, + mock_create_user_data, + mock_prepare_network_for_vm_instance, + mock_reload_connection, + mock_time, + ): + """New VM(server) creation failed.""" + + mock_create_user_data.return_value = True, "userdata" + + mock_get_vm_availability_zone.return_value = "nova" + + self.vimconn.nova.servers.create.side_effect = Exception( + "Server could not be created." + ) + + mock_time.return_value = time_return_value + + self.vimconn.new_vminstance( + name, + description, + start, + image_id, + flavor_id, + affinity_group_list, + net_list, + cloud_config, + disk_list2, + availability_zone_index, + availability_zone_list, + ) + + mock_reload_connection.assert_called_once() + mock_prepare_network_for_vm_instance.assert_called_once_with( + name=name, + net_list=net_list, + created_items={}, + net_list_vim=[], + external_network=[], + no_secured_ports=[], + ) + mock_create_user_data.assert_called_once_with(cloud_config) + mock_get_vm_availability_zone.assert_called_once_with( + availability_zone_index, availability_zone_list + ) + mock_prepare_disk_for_vm_instance.assert_called_once_with( + name=name, + existing_vim_volumes=[], + created_items={}, + vm_av_zone="nova", + disk_list=disk_list2, + ) + + self.vimconn.nova.servers.create.assert_called_once_with( + name=name, + image=image_id, + flavor=flavor_id, + nics=[], + security_groups="default", + availability_zone="nova", + key_name="my_keypair", + userdata="userdata", + config_drive=True, + block_device_mapping=None, + scheduler_hints={}, + ) + mock_time.assert_not_called() + mock_update_port_security.assert_not_called() + mock_prepare_external_network.assert_not_called() + mock_delete_vm_instance.assert_called_once_with(None, {}) + mock_format_exception.assert_called_once() + arg = mock_format_exception.call_args[0][0] + self.assertEqual(str(arg), "Server could not be created.") + + @patch("time.time") + @patch.object(vimconnector, "_reload_connection") + @patch.object(vimconnector, "_prepare_network_for_vminstance") + @patch.object(vimconnector, "_create_user_data") + @patch.object(vimconnector, "_get_vm_availability_zone") + @patch.object(vimconnector, "_prepare_disk_for_vminstance") + @patch.object(vimconnector, "_update_port_security_for_vminstance") + @patch.object(vimconnector, "_prepare_external_network_for_vminstance") + @patch.object(vimconnector, "delete_vminstance") + @patch.object(vimconnector, "_format_exception") + def test_new_vm_instance_connection_exception( + self, + mock_format_exception, + mock_delete_vm_instance, + mock_prepare_external_network, + mock_update_port_security, + mock_prepare_disk_for_vm_instance, + mock_get_vm_availability_zone, + mock_create_user_data, + mock_prepare_network_for_vm_instance, + mock_reload_connection, + mock_time, + ): + """Connection to Cloud API has failed.""" + mock_reload_connection.side_effect = Exception("Can not connect to Cloud APIs.") + mock_create_user_data.return_value = True, "userdata" + mock_get_vm_availability_zone.return_value = "nova" + self.vimconn.nova.servers.create.return_value = self.server + mock_time.return_value = time_return_value + + self.vimconn.new_vminstance( + name, + description, + start, + image_id, + flavor_id, + affinity_group_list, + net_list, + cloud_config, + disk_list, + availability_zone_index, + availability_zone_list, + ) + mock_format_exception.assert_called_once() + arg = mock_format_exception.call_args[0][0] + self.assertEqual(str(arg), "Can not connect to Cloud APIs.") + mock_reload_connection.assert_called_once() + mock_prepare_network_for_vm_instance.assert_not_called() + mock_create_user_data.assert_not_called() + mock_get_vm_availability_zone.assert_not_called() + mock_prepare_disk_for_vm_instance.assert_not_called() + self.vimconn.nova.servers.create.assert_not_called() + mock_time.assert_not_called() + mock_update_port_security.assert_not_called() + mock_prepare_external_network.assert_not_called() + mock_delete_vm_instance.assert_called_once_with(None, {}) + + if __name__ == "__main__": unittest.main()