# Licensed under the Apache License, Version 2.0 (the "License"); you may # not use this file except in compliance with the License. You may obtain # a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the # License for the specific language governing permissions and limitations # under the License. import copy import json import os import time from unittest import mock from keystoneauth1 import loading as ks_loading import openstack from openstack.connection import exceptions as openstack_exc from oslo_utils import uuidutils import tenacity from ironic.common import context from ironic.common import exception from ironic.common import neutron from ironic.conductor import task_manager from ironic.tests import base from ironic.tests.unit.db import base as db_base from ironic.tests.unit.objects import utils as object_utils from ironic.tests.unit import stubs @mock.patch('ironic.common.keystone.get_service_auth', autospec=True, return_value=mock.sentinel.sauth) @mock.patch('ironic.common.keystone.get_auth', autospec=True, return_value=mock.sentinel.auth) @mock.patch('ironic.common.keystone.get_adapter', autospec=True) @mock.patch('ironic.common.keystone.get_session', autospec=True, return_value=mock.sentinel.session) @mock.patch.object(openstack.connection, "Connection", autospec=True) class TestNeutronClient(base.TestCase): def setUp(self): super(TestNeutronClient, self).setUp() # NOTE(pas-ha) register keystoneauth dynamic options manually plugin = ks_loading.get_plugin_loader('password') opts = ks_loading.get_auth_plugin_conf_options(plugin) self.cfg_fixture.register_opts(opts, group='neutron') self.config(retries=2, group='neutron') self.config(username='test-admin-user', project_name='test-admin-tenant', password='test-admin-password', auth_url='test-auth-uri', auth_type='password', interface='internal', service_type='network', timeout=10, group='neutron') # force-reset the global session object neutron._NEUTRON_SESSION = None self.context = context.RequestContext(global_request_id='global') def _call_and_assert_client(self, client_mock, url, auth=mock.sentinel.auth): neutron.get_client(context=self.context) client_mock.assert_called_once_with(oslo_conf=mock.ANY, session=mock.sentinel.session) @mock.patch('ironic.common.context.RequestContext', autospec=True) def test_get_neutron_client_with_token(self, mock_ctxt, mock_client_init, mock_session, mock_adapter, mock_auth, mock_sauth): mock_ctxt.return_value = ctxt = mock.Mock() ctxt.auth_token = 'test-token-123' neutron.get_client(token='test-token-123') mock_ctxt.assert_called_once_with(auth_token='test-token-123') mock_client_init.assert_called_once_with(oslo_conf=mock.ANY, session=mock.sentinel.session) # testing handling of default url_timeout mock_session.assert_has_calls( [mock.call('neutron', timeout=10), mock.call('neutron', auth=mock.sentinel.sauth, timeout=10)]) def test_get_neutron_client_with_context(self, mock_client_init, mock_session, mock_adapter, mock_auth, mock_sauth): self.context = context.RequestContext(global_request_id='global', auth_token='test-token-123') self._call_and_assert_client(mock_client_init, 'neutron_url') # testing handling of default url_timeout mock_session.assert_has_calls( [mock.call('neutron', timeout=10), mock.call('neutron', auth=mock.sentinel.sauth, timeout=10)]) def test_get_neutron_client_without_token(self, mock_client_init, mock_session, mock_adapter, mock_auth, mock_sauth): self._call_and_assert_client(mock_client_init, 'neutron_url') mock_session.assert_has_calls( [mock.call('neutron', timeout=10), mock.call('neutron', auth=mock.sentinel.auth, timeout=10)]) def test_get_neutron_client_noauth(self, mock_client_init, mock_session, mock_adapter, mock_auth, mock_sauth): self.config(endpoint_override='neutron_url', auth_type='none', timeout=10, group='neutron') self._call_and_assert_client(mock_client_init, 'neutron_url') self.assertEqual('none', neutron.CONF.neutron.auth_type) mock_session.assert_has_calls( [mock.call('neutron', timeout=10), mock.call('neutron', auth=mock.sentinel.auth, timeout=10)]) def test_get_neutron_client_auth_from_config(self, mock_client_init, mock_session, mock_adapter, mock_auth, mock_sauth): self.context = context.RequestContext(global_request_id='global', auth_token='test-token-123') neutron.get_client(context=self.context, auth_from_config=True) mock_client_init.assert_called_once_with(oslo_conf=mock.ANY, session=mock.sentinel.session) mock_sauth.assert_not_called() # testing handling of default url_timeout mock_session.assert_has_calls( [mock.call('neutron', timeout=10), mock.call('neutron', auth=mock.sentinel.auth, timeout=10)]) class TestUpdateNeutronPort(base.TestCase): def setUp(self): super(TestUpdateNeutronPort, self).setUp() self.uuid = uuidutils.generate_uuid() self.context = context.RequestContext() self.port_attr = {'name': 'name_it'} @mock.patch.object(neutron, 'get_client', autospec=True) def test_update_neutron_port(self, client_mock): client_mock.return_value.get_port.return_value = {} client_mock.return_value.update_port.return_value = {'name': 'name_it'} neutron.update_neutron_port(self.context, self.uuid, self.port_attr) client_mock.assert_any_call(context=self.context) client_mock.assert_any_call(context=self.context, auth_from_config=True) client_mock.return_value.get_port.assert_called_once_with(self.uuid) client_mock.return_value.update_port.assert_called_once_with( self.uuid, **self.port_attr) @mock.patch.object(neutron, 'get_client', autospec=True) def test_update_neutron_port_with_client(self, client_mock): client_mock.return_value.get_port.return_value = {} client_mock.return_value.update_port.return_value = { 'name': 'name_it'} client = mock.Mock() client.update_port.return_value = {} neutron.update_neutron_port(self.context, self.uuid, self.port_attr, client) self.assertFalse(client_mock.called) client.update_port.assert_called_once_with(self.uuid, **self.port_attr) @mock.patch.object(neutron, 'get_client', autospec=True) def test_update_neutron_port_with_exception(self, client_mock): client_mock.return_value.get_port.side_effect = \ openstack_exc.OpenStackCloudException client_mock.return_value.update_port.return_value = {} self.assertRaises( openstack_exc.OpenStackCloudException, neutron.update_neutron_port, self.context, self.uuid, self.port_attr) client_mock.assert_called_once_with(context=self.context) client_mock.return_value.get_port.assert_called_once_with(self.uuid) class TestNeutronNetworkActions(db_base.DbTestCase): _CLIENT_ID = ( '20:00:55:04:01:fe:80:00:00:00:00:00:00:00:02:c9:02:00:23:13:92') def setUp(self): super(TestNeutronNetworkActions, self).setUp() self.node = object_utils.create_test_node(self.context) self.ports = [object_utils.create_test_port( self.context, node_id=self.node.id, uuid='1be26c0b-03f2-4d2e-ae87-c02d7f33c782', address='52:54:00:cf:2d:32', extra={'vif_port_id': uuidutils.generate_uuid()} )] self.neutron_port = stubs.FakeNeutronPort( id='132f871f-eaec-4fed-9475-0d54465e0f00', mac_address='52:54:00:cf:2d:32', fixed_ips=[]) self.network_uuid = uuidutils.generate_uuid() self.client_mock = mock.Mock() patcher = mock.patch('ironic.common.neutron.get_client', return_value=self.client_mock, autospec=True) patcher.start() self.addCleanup(patcher.stop) port_show_file = os.path.join( os.path.dirname(__file__), 'json_samples', 'neutron_port_show.json') with open(port_show_file, 'rb') as fl: self.port_data = json.load(fl) self.client_mock.get_port.return_value = stubs.FakeNeutronPort( **self.port_data['port']) network_show_file = os.path.join( os.path.dirname(__file__), 'json_samples', 'neutron_network_show.json') with open(network_show_file, 'rb') as fl: self.network_data = json.load(fl) self.client_mock.get_network.return_value = stubs.FakeNeutronNetwork( **self.network_data['network']) subnet_show_file = os.path.join( os.path.dirname(__file__), 'json_samples', 'neutron_subnet_show.json') with open(subnet_show_file, 'rb') as fl: self.subnet_data = json.load(fl) self.client_mock.get_subnet.return_value = stubs.FakeNeutronSubnet( **self.subnet_data['subnet']) @mock.patch.object(neutron, 'update_neutron_port', autospec=True) def _test_add_ports_to_network(self, update_mock, is_client_id, security_groups=None, add_all_ports=False): # Ports will be created only if pxe_enabled is True self.node.network_interface = 'neutron' self.node.save() port2 = object_utils.create_test_port( self.context, node_id=self.node.id, uuid=uuidutils.generate_uuid(), address='54:00:00:cf:2d:22', pxe_enabled=False ) if add_all_ports: self.config(add_all_ports=True, group="neutron") port = self.ports[0] if is_client_id: extra = port.extra extra['client-id'] = self._CLIENT_ID port.extra = extra port.save() expected_create_attrs = { 'network_id': self.network_uuid, 'admin_state_up': True, 'binding:vnic_type': 'baremetal', 'device_id': self.node.uuid } expected_update_attrs = { 'device_owner': 'baremetal:none', 'binding:host_id': self.node.uuid, 'mac_address': port.address, 'binding:profile': { 'local_link_information': [port.local_link_connection] } } if security_groups: expected_create_attrs['security_groups'] = security_groups if is_client_id: expected_create_attrs['extra_dhcp_opts'] = ( [{'opt_name': '61', 'opt_value': self._CLIENT_ID}]) if add_all_ports: expected_create_attrs2 = copy.deepcopy(expected_create_attrs) expected_update_attrs2 = copy.deepcopy(expected_update_attrs) expected_update_attrs2['mac_address'] = port2.address expected_create_attrs2['fixed_ips'] = [] neutron_port2 = stubs.FakeNeutronPort( id='132f871f-eaec-4fed-9475-0d54465e0f01', mac_address=port2.address, fixed_ips=[]) self.client_mock.create_port.side_effect = [self.neutron_port, neutron_port2] update_mock.side_effect = [self.neutron_port, neutron_port2] expected = {port.uuid: self.neutron_port.id, port2.uuid: neutron_port2.id} else: self.client_mock.create_port.return_value = self.neutron_port update_mock.return_value = self.neutron_port expected = {port.uuid: self.neutron_port['id']} with task_manager.acquire(self.context, self.node.uuid) as task: ports = neutron.add_ports_to_network( task, self.network_uuid, security_groups=security_groups) self.assertEqual(expected, ports) if add_all_ports: create_calls = [mock.call(**expected_create_attrs), mock.call(**expected_create_attrs2)] update_calls = [ mock.call(self.context, self.neutron_port['id'], expected_update_attrs), mock.call(self.context, neutron_port2['id'], expected_update_attrs2)] self.client_mock.create_port.assert_has_calls(create_calls) update_mock.assert_has_calls(update_calls) else: self.client_mock.create_port.assert_called_once_with( **expected_create_attrs) update_mock.assert_called_once_with( self.context, self.neutron_port['id'], expected_update_attrs) def test_add_ports_to_network(self): self._test_add_ports_to_network(is_client_id=False, security_groups=None) def test_add_ports_to_network_all_ports(self): self._test_add_ports_to_network(is_client_id=False, security_groups=None, add_all_ports=True) @mock.patch.object(neutron, '_verify_security_groups', autospec=True) def test_add_ports_to_network_with_sg(self, verify_mock): sg_ids = [] for i in range(2): sg_ids.append(uuidutils.generate_uuid()) self._test_add_ports_to_network(is_client_id=False, security_groups=sg_ids) @mock.patch.object(neutron, 'update_neutron_port', autospec=True) def test__add_ip_addresses_for_ipv6_stateful(self, mock_update): subnet_id = uuidutils.generate_uuid() self.client_mock.get_subnet.return_value = stubs.FakeNeutronSubnet( id=subnet_id, ip_version=6, ipv6_address_mode='dhcpv6-stateful') self.neutron_port.fixed_ips = [{'subnet_id': subnet_id, 'ip_address': '2001:db8::1'}] expected_body = { 'fixed_ips': [ {'subnet_id': subnet_id, 'ip_address': '2001:db8::1'}, {'subnet_id': subnet_id}, {'subnet_id': subnet_id}, {'subnet_id': subnet_id} ] } neutron._add_ip_addresses_for_ipv6_stateful( self.context, self.neutron_port, self.client_mock) mock_update.assert_called_once_with( self.context, self.neutron_port.id, expected_body, client=self.client_mock) def test_verify_sec_groups(self): sg_ids = [] expected_vals = [] for i in range(2): uuid = uuidutils.generate_uuid() sg_ids.append(uuid) expected_vals.append(stubs.FakeNeutronSecurityGroup(id=uuid)) client = mock.MagicMock() client.security_groups.return_value = iter(expected_vals) self.assertIsNone( neutron._verify_security_groups(sg_ids, client)) client.security_groups.assert_called_once_with(id=sg_ids) def test_verify_sec_groups_less_than_configured(self): sg_ids = [] expected_vals = [] for i in range(2): uuid = uuidutils.generate_uuid() sg_ids.append(uuid) expected_vals.append(stubs.FakeNeutronSecurityGroup(id=uuid)) client = mock.MagicMock() client.security_groups.return_value = iter(expected_vals) self.assertIsNone( neutron._verify_security_groups(sg_ids[:1], client)) client.security_groups.assert_called_once_with(id=sg_ids[:1]) def test_verify_sec_groups_more_than_configured(self): sg_ids = [] for i in range(1): sg_ids.append(uuidutils.generate_uuid()) client = mock.MagicMock() client.get.return_value = iter([]) self.assertRaises( exception.NetworkError, neutron._verify_security_groups, sg_ids, client) client.security_groups.assert_called_once_with(id=sg_ids) def test_verify_sec_groups_no_sg_from_neutron(self): sg_ids = [] for i in range(1): sg_ids.append(uuidutils.generate_uuid()) client = mock.MagicMock() client.security_groups.return_value = iter([]) self.assertRaises( exception.NetworkError, neutron._verify_security_groups, sg_ids, client) client.security_groups.assert_called_once_with(id=sg_ids) def test_verify_sec_groups_exception_by_neutronclient(self): sg_ids = [] for i in range(2): sg_ids.append(uuidutils.generate_uuid()) client = mock.MagicMock() client.security_groups.side_effect = ( openstack_exc.OpenStackCloudException) self.assertRaisesRegex( exception.NetworkError, "Could not retrieve security groups", neutron._verify_security_groups, sg_ids, client) client.security_groups.assert_called_once_with(id=sg_ids) def test_add_ports_with_client_id_to_network(self): self._test_add_ports_to_network(is_client_id=True) @mock.patch.object(neutron, 'update_neutron_port', autospec=True) @mock.patch.object(neutron, 'validate_port_info', autospec=True) def test_add_ports_to_network_instance_uuid(self, vpi_mock, update_mock): self.node.instance_uuid = uuidutils.generate_uuid() self.node.network_interface = 'neutron' self.node.save() port = self.ports[0] expected_create_body = { 'network_id': self.network_uuid, 'admin_state_up': True, 'binding:vnic_type': 'baremetal', 'device_id': self.node.instance_uuid } expected_update_body = { 'device_owner': 'baremetal:none', 'binding:host_id': self.node.uuid, 'mac_address': port.address, 'binding:profile': { 'local_link_information': [port.local_link_connection] } } vpi_mock.return_value = True # Ensure we can create ports self.client_mock.create_port.return_value = self.neutron_port update_mock.return_value = self.neutron_port expected = {port.uuid: self.neutron_port.id} with task_manager.acquire(self.context, self.node.uuid) as task: ports = neutron.add_ports_to_network(task, self.network_uuid) self.assertEqual(expected, ports) self.client_mock.create_port.assert_called_once_with( **expected_create_body) update_mock.assert_called_once_with(self.context, self.neutron_port['id'], expected_update_body) self.assertTrue(vpi_mock.called) @mock.patch.object(neutron, 'rollback_ports', autospec=True) def test_add_network_all_ports_fail(self, rollback_mock): # Check that if creating a port fails, the ports are cleaned up self.client_mock.create_port.side_effect = \ openstack_exc.OpenStackCloudException with task_manager.acquire(self.context, self.node.uuid) as task: self.assertRaises( exception.NetworkError, neutron.add_ports_to_network, task, self.network_uuid) rollback_mock.assert_called_once_with(task, self.network_uuid) @mock.patch.object(neutron, 'update_neutron_port', autospec=True) @mock.patch.object(neutron, 'LOG', autospec=True) def test_add_network_create_some_ports_fail(self, log_mock, update_mock): object_utils.create_test_port( self.context, node_id=self.node.id, uuid=uuidutils.generate_uuid(), address='52:54:55:cf:2d:32', extra={'vif_port_id': uuidutils.generate_uuid()} ) self.client_mock.create_port.side_effect = [ self.neutron_port, openstack_exc.OpenStackCloudException] update_mock.return_value = self.neutron_port with task_manager.acquire(self.context, self.node.uuid) as task: neutron.add_ports_to_network(task, self.network_uuid) self.assertIn("Could not create neutron port for node's", log_mock.warning.call_args_list[0][0][0]) self.assertIn("Some errors were encountered when updating", log_mock.warning.call_args_list[1][0][0]) def test_add_network_no_port(self): # No port registered node = object_utils.create_test_node(self.context, uuid=uuidutils.generate_uuid()) with task_manager.acquire(self.context, node.uuid) as task: self.assertEqual([], task.ports) self.assertRaisesRegex(exception.NetworkError, 'No available', neutron.add_ports_to_network, task, self.network_uuid) def test_add_network_no_pxe_enabled_ports(self): # Have port but no PXE enabled port = self.ports[0] port.pxe_enabled = False port.save() with task_manager.acquire(self.context, self.node.uuid) as task: self.assertFalse(task.ports[0].pxe_enabled) self.assertRaisesRegex(exception.NetworkError, 'No available', neutron.add_ports_to_network, task, self.network_uuid) @mock.patch.object(neutron, 'remove_neutron_ports', autospec=True) def test_remove_ports_from_network(self, remove_mock): with task_manager.acquire(self.context, self.node.uuid) as task: neutron.remove_ports_from_network(task, self.network_uuid) remove_mock.assert_called_once_with( task, {'network_id': self.network_uuid, 'mac_address': [self.ports[0].address]} ) @mock.patch.object(neutron, 'remove_neutron_ports', autospec=True) def test_remove_ports_from_network_not_all_pxe_enabled(self, remove_mock): object_utils.create_test_port( self.context, node_id=self.node.id, uuid=uuidutils.generate_uuid(), address='52:54:55:cf:2d:32', pxe_enabled=False ) with task_manager.acquire(self.context, self.node.uuid) as task: neutron.remove_ports_from_network(task, self.network_uuid) remove_mock.assert_called_once_with( task, {'network_id': self.network_uuid, 'mac_address': [self.ports[0].address]} ) @mock.patch.object(neutron, 'remove_neutron_ports', autospec=True) def test_remove_ports_from_network_not_all_pxe_enabled_all_ports( self, remove_mock): self.config(add_all_ports=True, group="neutron") object_utils.create_test_port( self.context, node_id=self.node.id, uuid=uuidutils.generate_uuid(), address='52:54:55:cf:2d:32', pxe_enabled=False ) with task_manager.acquire(self.context, self.node.uuid) as task: neutron.remove_ports_from_network(task, self.network_uuid) calls = [ mock.call(task, {'network_id': self.network_uuid, 'mac_address': [task.ports[0].address, task.ports[1].address]}), ] remove_mock.assert_has_calls(calls) def test_remove_neutron_ports(self): with task_manager.acquire(self.context, self.node.uuid) as task: self.client_mock.ports.return_value = iter([self.neutron_port]) neutron.remove_neutron_ports(task, {'param': 'value'}) self.client_mock.ports.assert_called_once_with(**{'param': 'value'}) self.client_mock.delete_port.assert_called_once_with(self.neutron_port) def test_remove_neutron_ports_list_fail(self): with task_manager.acquire(self.context, self.node.uuid) as task: self.client_mock.ports.side_effect = ( openstack_exc.OpenStackCloudException) self.assertRaisesRegex( exception.NetworkError, 'Could not get given network VIF', neutron.remove_neutron_ports, task, {'param': 'value'}) self.client_mock.ports.assert_called_once_with(**{'param': 'value'}) def test_remove_neutron_ports_delete_fail(self): with task_manager.acquire(self.context, self.node.uuid) as task: self.client_mock.delete_port.side_effect = \ openstack_exc.OpenStackCloudException self.client_mock.ports.return_value = iter([self.neutron_port]) self.assertRaisesRegex( exception.NetworkError, 'Could not remove VIF', neutron.remove_neutron_ports, task, {'param': 'value'}) self.client_mock.ports.assert_called_once_with(**{'param': 'value'}) self.client_mock.delete_port.assert_called_once_with(self.neutron_port) def test_remove_neutron_ports_delete_race(self): with task_manager.acquire(self.context, self.node.uuid) as task: self.client_mock.delete_port.side_effect = \ openstack_exc.ResourceNotFound self.client_mock.ports.return_value = iter([self.neutron_port]) neutron.remove_neutron_ports(task, {'param': 'value'}) self.client_mock.ports.assert_called_once_with(**{'param': 'value'}) self.client_mock.delete_port.assert_called_once_with(self.neutron_port) def test__uncidr_ipv4(self): network, netmask = neutron._uncidr('10.0.0.0/24') self.assertEqual('10.0.0.0', network) self.assertEqual('255.255.255.0', netmask) def test__uncidr_ipv6(self): network, netmask = neutron._uncidr('::1/64', ipv6=True) self.assertEqual('::', network) self.assertEqual('ffff:ffff:ffff:ffff::', netmask) def test_get_neutron_port_data(self): network_data = neutron.get_neutron_port_data('port0', 'vif0') expected_port = { 'id': 'port0', 'type': 'vif', 'ethernet_mac_address': 'fa:16:3e:23:fd:d7', 'vif_id': '46d4bfb9-b26e-41f3-bd2e-e6dcc1ccedb2', 'mtu': 1500 } self.assertEqual(expected_port, network_data['links'][0]) expected_network = { 'id': 'a0304c3a-4f08-4c43-88af-d796509c97d2', 'network_id': 'a87cc70a-3e15-4acf-8205-9b711a3531b7', 'type': 'ipv4', 'link': 'port0', 'ip_address': '10.0.0.2', 'netmask': '255.255.255.0', 'routes': [ {'gateway': '10.0.0.1', 'netmask': '0.0.0.0', 'network': '0.0.0.0'} ] } self.assertEqual(expected_network, network_data['networks'][0]) def load_ipv6_files(self): port_show_file = os.path.join( os.path.dirname(__file__), 'json_samples', 'neutron_port_show_ipv6.json') with open(port_show_file, 'rb') as fl: self.port_data = json.load(fl) self.client_mock.get_port.return_value = stubs.FakeNeutronPort( **self.port_data['port']) network_show_file = os.path.join( os.path.dirname(__file__), 'json_samples', 'neutron_network_show_ipv6.json') with open(network_show_file, 'rb') as fl: self.network_data = json.load(fl) self.client_mock.get_network.return_value = stubs.FakeNeutronNetwork( **self.network_data['network']) subnet_show_file = os.path.join( os.path.dirname(__file__), 'json_samples', 'neutron_subnet_show_ipv6.json') with open(subnet_show_file, 'rb') as fl: self.subnet_data = json.load(fl) self.client_mock.get_subnet.return_value = stubs.FakeNeutronSubnet( **self.subnet_data['subnet']) def test_get_neutron_port_data_ipv6(self): self.load_ipv6_files() network_data = neutron.get_neutron_port_data('port1', 'vif1') print(network_data) expected_port = { 'id': 'port1', 'type': 'vif', 'ethernet_mac_address': '52:54:00:4f:ef:b7', 'vif_id': '96d4bfb9-b26e-41f3-bd2e-e6dcc1ccedb8', 'mtu': 1500 } self.assertEqual(expected_port, network_data['links'][0]) expected_network = { 'id': '906e685a-b964-4d58-9939-9cf3af197c67', 'network_id': 'a87cc70a-3e15-4acf-8205-9b711a3531b7', 'type': 'ipv6', 'link': 'port1', 'ip_address': 'fd00:203:0:113::2', 'netmask': 'ffff:ffff:ffff:ffff::', 'routes': [ {'gateway': 'fd00:203:0:113::1', 'netmask': '::0', 'network': '::0'} ] } self.assertEqual(expected_network, network_data['networks'][0]) def test_get_node_portmap(self): with task_manager.acquire(self.context, self.node.uuid) as task: portmap = neutron.get_node_portmap(task) self.assertEqual( {self.ports[0].uuid: self.ports[0].local_link_connection}, portmap ) def test_get_local_group_information(self): pg = object_utils.create_test_portgroup( self.context, node_id=self.node.id, uuid=uuidutils.generate_uuid(), address='52:54:55:cf:2d:32', mode='802.3ad', properties={'bond_opt1': 'foo', 'opt2': 'bar'}, name='test-pg' ) expected = { 'id': pg.uuid, 'name': pg.name, 'bond_mode': pg.mode, 'bond_properties': {'bond_opt1': 'foo', 'bond_opt2': 'bar'}, } with task_manager.acquire(self.context, self.node.uuid) as task: res = neutron.get_local_group_information(task, pg) self.assertEqual(expected, res) @mock.patch.object(neutron, 'remove_ports_from_network', autospec=True) def test_rollback_ports(self, remove_mock): with task_manager.acquire(self.context, self.node.uuid) as task: neutron.rollback_ports(task, self.network_uuid) remove_mock.assert_called_once_with(task, self.network_uuid) @mock.patch.object(neutron, 'LOG', autospec=True) @mock.patch.object(neutron, 'remove_ports_from_network', autospec=True) def test_rollback_ports_exception(self, remove_mock, log_mock): remove_mock.side_effect = exception.NetworkError('boom') with task_manager.acquire(self.context, self.node.uuid) as task: neutron.rollback_ports(task, self.network_uuid) self.assertTrue(log_mock.exception.called) @mock.patch.object(neutron, 'LOG', autospec=True) def test_validate_port_info_neutron_interface(self, log_mock): self.node.network_interface = 'neutron' self.node.save() port = object_utils.create_test_port( self.context, node_id=self.node.id, uuid=uuidutils.generate_uuid(), address='52:54:00:cf:2d:33') res = neutron.validate_port_info(self.node, port) self.assertTrue(res) self.assertFalse(log_mock.warning.called) @mock.patch.object(neutron, 'LOG', autospec=True) def test_validate_port_info_neutron_interface_missed_info(self, log_mock): self.node.network_interface = 'neutron' self.node.save() llc = {} port = object_utils.create_test_port( self.context, node_id=self.node.id, uuid=uuidutils.generate_uuid(), address='52:54:00:cf:2d:33', local_link_connection=llc) res = neutron.validate_port_info(self.node, port) self.assertFalse(res) self.assertTrue(log_mock.warning.called) @mock.patch.object(neutron, 'LOG', autospec=True) def test_validate_port_info_flat_interface(self, log_mock): self.node.network_interface = 'flat' self.node.save() llc = {} port = object_utils.create_test_port( self.context, node_id=self.node.id, uuid=uuidutils.generate_uuid(), address='52:54:00:cf:2d:33', local_link_connection=llc) res = neutron.validate_port_info(self.node, port) self.assertTrue(res) self.assertFalse(log_mock.warning.called) @mock.patch.object(neutron, 'LOG', autospec=True) def test_validate_port_info_flat_interface_with_client_id(self, log_mock): self.node.network_interface = 'flat' self.node.save() llc = {} port = object_utils.create_test_port( self.context, node_id=self.node.id, uuid=uuidutils.generate_uuid(), address='52:54:00:cf:2d:33', local_link_connection=llc, extra={'client-id': self._CLIENT_ID}) res = neutron.validate_port_info(self.node, port) self.assertTrue(res) self.assertFalse(log_mock.warning.called) @mock.patch.object(neutron, 'LOG', autospec=True) def test_validate_port_info_neutron_interface_with_client_id( self, log_mock): self.node.network_interface = 'neutron' self.node.save() llc = {} port = object_utils.create_test_port( self.context, node_id=self.node.id, uuid=uuidutils.generate_uuid(), address='52:54:00:cf:2d:33', local_link_connection=llc, extra={'client-id': self._CLIENT_ID}) res = neutron.validate_port_info(self.node, port) self.assertTrue(res) self.assertFalse(log_mock.warning.called) @mock.patch.object(neutron, 'LOG', autospec=True) def test_validate_port_info_neutron_with_smartnic_and_link_info( self, log_mock): self.node.network_interface = 'neutron' self.node.save() llc = {'hostname': 'host1', 'port_id': 'rep0-0'} port = object_utils.create_test_port( self.context, node_id=self.node.id, uuid=uuidutils.generate_uuid(), address='52:54:00:cf:2d:33', local_link_connection=llc, is_smartnic=True) res = neutron.validate_port_info(self.node, port) self.assertTrue(res) self.assertFalse(log_mock.error.called) @mock.patch.object(neutron, 'LOG', autospec=True) def test_validate_port_info_neutron_with_no_smartnic_and_link_info( self, log_mock): self.node.network_interface = 'neutron' self.node.save() llc = {'hostname': 'host1', 'port_id': 'rep0-0'} port = object_utils.create_test_port( self.context, node_id=self.node.id, uuid=uuidutils.generate_uuid(), address='52:54:00:cf:2d:33', local_link_connection=llc, is_smartnic=False) res = neutron.validate_port_info(self.node, port) self.assertFalse(res) self.assertTrue(log_mock.error.called) @mock.patch.object(neutron, 'LOG', autospec=True) def test_validate_port_info_neutron_with_smartnic_and_no_link_info( self, log_mock): self.node.network_interface = 'neutron' self.node.save() llc = {'switch_id': 'switch', 'port_id': 'rep0-0'} port = object_utils.create_test_port( self.context, node_id=self.node.id, uuid=uuidutils.generate_uuid(), address='52:54:00:cf:2d:33', local_link_connection=llc, is_smartnic=True) res = neutron.validate_port_info(self.node, port) self.assertFalse(res) self.assertTrue(log_mock.error.called) @mock.patch.object(neutron, 'LOG', autospec=True) def test_validate_port_info_neutron_with_network_type_unmanaged( self, log_mock): self.node.network_interface = 'neutron' self.node.save() llc = {'network_type': 'unmanaged'} port = object_utils.create_test_port( self.context, node_id=self.node.id, uuid=uuidutils.generate_uuid(), address='52:54:00:cf:2d:33', local_link_connection=llc) res = neutron.validate_port_info(self.node, port) self.assertTrue(res) self.assertFalse(log_mock.warning.called) def test_validate_agent_up(self): self.client_mock.agents.return_value = iter([ stubs.FakeNeutronAgent(alive=True) ]) self.assertTrue(neutron._validate_agent(self.client_mock)) def test_validate_agent_down(self): self.client_mock.agents.return_value = iter([ stubs.FakeNeutronAgent(alive=False) ]) self.assertFalse(neutron._validate_agent(self.client_mock)) def test_is_smartnic_port_true(self): port = self.ports[0] port.is_smartnic = True self.assertTrue(neutron.is_smartnic_port(port)) def test_is_smartnic_port_false(self): port = self.ports[0] self.assertFalse(neutron.is_smartnic_port(port)) @mock.patch.object(neutron, '_validate_agent', autospec=True) @mock.patch.object(time, 'sleep', autospec=True) def test_wait_for_host_agent_up_target_state_up( self, sleep_mock, validate_agent_mock): validate_agent_mock.return_value = True self.assertTrue(neutron.wait_for_host_agent( self.client_mock, 'hostname')) validate_agent_mock.assert_called_once() @mock.patch.object(neutron, '_validate_agent', autospec=True) @mock.patch.object(time, 'sleep', autospec=True) def test_wait_for_host_agent_down_target_state_up( self, sleep_mock, validate_agent_mock): neutron.wait_for_host_agent.retry.stop = ( tenacity.stop_after_attempt(3)) validate_agent_mock.return_value = False self.assertRaises(exception.NetworkError, neutron.wait_for_host_agent, self.client_mock, 'hostname') @mock.patch.object(neutron, '_validate_agent', autospec=True) @mock.patch.object(time, 'sleep', autospec=True) def test_wait_for_host_agent_up_target_state_down( self, sleep_mock, validate_agent_mock): neutron.wait_for_host_agent.retry.stop = ( tenacity.stop_after_attempt(3)) validate_agent_mock.return_value = True self.assertRaises(exception.NetworkError, neutron.wait_for_host_agent, self.client_mock, 'hostname', target_state='down') @mock.patch.object(neutron, '_validate_agent', autospec=True) def test_wait_for_host_agent_down_target_state_down( self, validate_agent_mock): validate_agent_mock.return_value = False self.assertTrue( neutron.wait_for_host_agent(self.client_mock, 'hostname', target_state='down')) sleep_mock = mock.Mock() neutron.wait_for_host_agent.retry.sleep = sleep_mock sleep_mock.assert_not_called() validate_agent_mock.assert_called_once() @mock.patch.object(neutron, '_get_port_by_uuid', autospec=True) @mock.patch.object(time, 'sleep', autospec=True) def test_wait_for_port_status_up(self, sleep_mock, get_port_mock): get_port_mock.return_value = stubs.FakeNeutronPort(status='ACTIVE') neutron.wait_for_port_status(self.client_mock, 'port_id', 'ACTIVE') get_port_mock.assert_called_once() @mock.patch.object(neutron, '_get_port_by_uuid', autospec=True) @mock.patch.object(time, 'sleep', autospec=True) def test_wait_for_port_status_down(self, sleep_mock, get_port_mock): get_port_mock.side_effect = [stubs.FakeNeutronPort(status='DOWN'), stubs.FakeNeutronPort(status='ACTIVE')] neutron.wait_for_port_status(self.client_mock, 'port_id', 'ACTIVE') self.assertEqual(get_port_mock.call_count, 2) @mock.patch.object(neutron, '_get_port_by_uuid', autospec=True) @mock.patch.object(time, 'sleep', autospec=True) def test_wait_for_port_status_active_max_retry( self, sleep_mock, get_port_mock): neutron.wait_for_port_status.retry.stop = ( tenacity.stop_after_attempt(3)) get_port_mock.return_value = stubs.FakeNeutronPort(status='DOWN') self.assertRaises(exception.NetworkError, neutron.wait_for_port_status, self.client_mock, 'port_id', 'ACTIVE') @mock.patch.object(neutron, '_get_port_by_uuid', autospec=True) @mock.patch.object(time, 'sleep', autospec=True) def test_wait_for_port_status_down_max_retry( self, sleep_mock, get_port_mock): neutron.wait_for_port_status.retry.stop = ( tenacity.stop_after_attempt(3)) get_port_mock.return_value = stubs.FakeNeutronPort(status='ACTIVE') self.assertRaises(exception.NetworkError, neutron.wait_for_port_status, self.client_mock, 'port_id', 'DOWN') @mock.patch.object(neutron, 'update_neutron_port', autospec=True) @mock.patch.object(neutron, 'wait_for_host_agent', autospec=True) @mock.patch.object(neutron, 'wait_for_port_status', autospec=True) def test_add_smartnic_port_to_network( self, wait_port_mock, wait_agent_mock, update_mock): # Ports will be created only if pxe_enabled is True self.node.network_interface = 'neutron' self.node.save() object_utils.create_test_port( self.context, node_id=self.node.id, uuid=uuidutils.generate_uuid(), address='52:54:00:cf:2d:22', pxe_enabled=False ) port = self.ports[0] local_link_connection = port.local_link_connection local_link_connection['hostname'] = 'hostname' port.local_link_connection = local_link_connection port.is_smartnic = True port.save() expected_create_attrs = { 'network_id': self.network_uuid, 'admin_state_up': True, 'binding:vnic_type': 'smart-nic', 'device_id': self.node.uuid, } expected_update_attrs = { 'device_owner': 'baremetal:none', 'binding:host_id': port.local_link_connection['hostname'], 'mac_address': port.address, 'binding:profile': { 'local_link_information': [port.local_link_connection] }, } # Ensure we can create ports self.client_mock.create_port.return_value = self.neutron_port update_mock.return_value = self.neutron_port expected = {port.uuid: self.neutron_port.id} with task_manager.acquire(self.context, self.node.uuid) as task: ports = neutron.add_ports_to_network(task, self.network_uuid) self.assertEqual(expected, ports) self.client_mock.create_port.assert_called_once_with( **expected_create_attrs) update_mock.assert_called_once_with( self.context, self.neutron_port.id, expected_update_attrs) wait_agent_mock.assert_called_once_with( self.client_mock, 'hostname') wait_port_mock.assert_called_once_with( self.client_mock, self.neutron_port.id, 'ACTIVE') @mock.patch.object(neutron, 'is_smartnic_port', autospec=True) @mock.patch.object(neutron, 'wait_for_host_agent', autospec=True) def test_remove_neutron_smartnic_ports( self, wait_agent_mock, is_smartnic_mock): with task_manager.acquire(self.context, self.node.uuid) as task: is_smartnic_mock.return_value = True self.neutron_port['binding:host_id'] = 'hostname' self.client_mock.ports.return_value = iter([self.neutron_port]) neutron.remove_neutron_ports(task, {'param': 'value'}) self.client_mock.ports.assert_called_once_with( **{'param': 'value'}) self.client_mock.delete_port.assert_called_once_with( self.neutron_port) is_smartnic_mock.assert_called_once_with(self.neutron_port) wait_agent_mock.assert_called_once_with(self.client_mock, 'hostname') @mock.patch.object(neutron, 'get_client', autospec=True) class TestValidateNetwork(base.TestCase): def setUp(self): super(TestValidateNetwork, self).setUp() self.uuid = uuidutils.generate_uuid() self.context = context.RequestContext() def test_by_uuid(self, client_mock): net_mock = client_mock.return_value.find_network net_mock.return_value = stubs.FakeNeutronNetwork(id=self.uuid) self.assertEqual(self.uuid, neutron.validate_network( self.uuid, context=self.context)) net_mock.assert_called_once_with(self.uuid, ignore_missing=False) def test_by_name(self, client_mock): net_mock = client_mock.return_value.find_network net_mock.return_value = stubs.FakeNeutronNetwork(id=self.uuid, name='name') self.assertEqual(self.uuid, neutron.validate_network( 'name', context=self.context)) net_mock.assert_called_once_with('name', ignore_missing=False) def test_not_found(self, client_mock): net_mock = client_mock.return_value.find_network net_mock.side_effect = openstack_exc.ResourceNotFound() self.assertRaisesRegex(exception.InvalidParameterValue, 'was not found', neutron.validate_network, self.uuid, context=self.context) net_mock.assert_called_once_with(self.uuid, ignore_missing=False) def test_failure(self, client_mock): net_mock = client_mock.return_value.find_network net_mock.side_effect = openstack_exc.OpenStackCloudException('foo') self.assertRaisesRegex(exception.NetworkError, 'foo', neutron.validate_network, 'name', context=self.context) net_mock.assert_called_once_with('name', ignore_missing=False) def test_duplicate(self, client_mock): net_mock = client_mock.return_value.find_network net_mock.side_effect = openstack_exc.DuplicateResource() self.assertRaisesRegex(exception.InvalidParameterValue, 'More than one network', neutron.validate_network, 'name', context=self.context) net_mock.assert_called_once_with('name', ignore_missing=False) @mock.patch.object(neutron, 'get_client', autospec=True) class TestUpdatePortAddress(base.TestCase): def setUp(self): super(TestUpdatePortAddress, self).setUp() self.context = context.RequestContext() @mock.patch.object(neutron, 'update_neutron_port', autospec=True) def test_update_port_address(self, mock_unp, mock_client): address = 'fe:54:00:77:07:d9' port_id = 'fake-port-id' expected = {'mac_address': address} mock_client.return_value.get_port.return_value = {} neutron.update_port_address(port_id, address, context=self.context) mock_unp.assert_called_once_with(self.context, port_id, expected) @mock.patch.object(neutron, 'update_neutron_port', autospec=True) @mock.patch.object(neutron, 'unbind_neutron_port', autospec=True) def test_update_port_address_with_binding(self, mock_unp, mock_update, mock_client): address = 'fe:54:00:77:07:d9' port_id = 'fake-port-id' mock_client.return_value.get_port.return_value = { 'binding:host_id': 'host', 'binding:profile': 'foo'} calls = [mock.call(self.context, port_id, {'mac_address': address}), mock.call(self.context, port_id, {'binding:host_id': 'host', 'binding:profile': 'foo'})] neutron.update_port_address(port_id, address, context=self.context) mock_unp.assert_called_once_with( port_id, context=self.context) mock_update.assert_has_calls(calls) @mock.patch.object(neutron, 'update_neutron_port', autospec=True) @mock.patch.object(neutron, 'unbind_neutron_port', autospec=True) def test_update_port_address_without_binding(self, mock_unp, mock_update, mock_client): address = 'fe:54:00:77:07:d9' port_id = 'fake-port-id' expected = {'mac_address': address} mock_client.return_value.get_port.return_value = { 'binding:profile': 'foo'} neutron.update_port_address(port_id, address, context=self.context) self.assertFalse(mock_unp.called) mock_update.assert_any_call(self.context, port_id, expected) def test_update_port_address_show_failed(self, mock_client): address = 'fe:54:00:77:07:d9' port_id = 'fake-port-id' mock_client.return_value.get_port.side_effect = ( openstack_exc.OpenStackCloudException()) self.assertRaises(exception.FailedToUpdateMacOnPort, neutron.update_port_address, port_id, address, context=self.context) self.assertFalse(mock_client.return_value.update_port.called) @mock.patch.object(neutron, 'unbind_neutron_port', autospec=True) def test_update_port_address_unbind_port_failed(self, mock_unp, mock_client): address = 'fe:54:00:77:07:d9' port_id = 'fake-port-id' mock_client.return_value.get_port.return_value = { 'binding:profile': 'foo', 'binding:host_id': 'host'} mock_unp.side_effect = (exception.NetworkError('boom')) self.assertRaises(exception.FailedToUpdateMacOnPort, neutron.update_port_address, port_id, address, context=self.context) mock_unp.assert_called_once_with( port_id, context=self.context) self.assertFalse(mock_client.return_value.update_port.called) @mock.patch.object(neutron, 'update_neutron_port', autospec=True) @mock.patch.object(neutron, 'unbind_neutron_port', autospec=True) def test_update_port_address_with_exception(self, mock_unp, mock_update, mock_client): address = 'fe:54:00:77:07:d9' port_id = 'fake-port-id' mock_client.return_value.get_port.return_value = {} mock_update.side_effect = ( openstack_exc.OpenStackCloudException()) self.assertRaises(exception.FailedToUpdateMacOnPort, neutron.update_port_address, port_id, address, context=self.context) @mock.patch.object(neutron, 'update_neutron_port', autospec=True) class TestUnbindPort(base.TestCase): def setUp(self): super(TestUnbindPort, self).setUp() self.context = context.RequestContext() def test_unbind_neutron_port_client_passed(self, mock_unp): port_id = 'fake-port-id' attr_unbind = {'binding:host_id': '', 'binding:profile': {}} attr_reset_mac = {'mac_address': None} client = mock.MagicMock() update_calls = [ mock.call(self.context, port_id, attr_unbind, client), mock.call(self.context, port_id, attr_reset_mac, client) ] neutron.unbind_neutron_port(port_id, client, context=self.context) self.assertEqual(2, mock_unp.call_count) mock_unp.assert_has_calls(update_calls) @mock.patch.object(neutron, 'LOG', autospec=True) def test_unbind_neutron_port_failure(self, mock_log, mock_unp): mock_unp.side_effect = openstack_exc.OpenStackCloudException() attr = {'binding:host_id': '', 'binding:profile': {}} port_id = 'fake-port-id' self.assertRaises(exception.NetworkError, neutron.unbind_neutron_port, port_id, context=self.context) mock_unp.assert_called_once_with(self.context, port_id, attr, None) mock_log.exception.assert_called_once() def test_unbind_neutron_port(self, mock_unp): port_id = 'fake-port-id' attr_unbind = {'binding:host_id': '', 'binding:profile': {}} attr_reset_mac = {'mac_address': None} update_calls = [ mock.call(self.context, port_id, attr_unbind, None), mock.call(self.context, port_id, attr_reset_mac, None) ] neutron.unbind_neutron_port(port_id, context=self.context) mock_unp.assert_has_calls(update_calls) def test_unbind_neutron_port_not_reset_mac(self, mock_unp): port_id = 'fake-port-id' attr_unbind = {'binding:host_id': '', 'binding:profile': {}} neutron.unbind_neutron_port(port_id, context=self.context, reset_mac=False) mock_unp.assert_called_once_with(self.context, port_id, attr_unbind, None) @mock.patch.object(neutron, 'LOG', autospec=True) def test_unbind_neutron_port_not_found(self, mock_log, mock_unp): port_id = 'fake-port-id' mock_unp.side_effect = ( openstack_exc.ResourceNotFound()) attr = {'binding:host_id': '', 'binding:profile': {}} neutron.unbind_neutron_port(port_id, context=self.context) mock_unp.assert_called_once_with(self.context, port_id, attr, None) mock_log.info.assert_called_once_with('Port %s was not found while ' 'unbinding.', port_id) class TestGetNetworkByUUIDOrName(base.TestCase): def setUp(self): super(TestGetNetworkByUUIDOrName, self).setUp() self.client = mock.MagicMock() def test__get_network_by_uuid_or_name_uuid(self): network_uuid = '9acb0256-2c1b-420a-b9d7-62bee90b6ed7' network = stubs.FakeNeutronNetwork(id=network_uuid) self.client.find_network.return_value = network result = neutron._get_network_by_uuid_or_name( self.client, network_uuid) self.client.find_network.assert_called_once_with( network_uuid, ignore_missing=False) self.assertEqual(network, result) def test__get_network_by_uuid_or_name_name(self): network_name = 'test-net' network = stubs.FakeNeutronNetwork(name=network_name) self.client.find_network.return_value = network result = neutron._get_network_by_uuid_or_name( self.client, network_name) self.client.find_network.assert_called_once_with( network_name, ignore_missing=False) self.assertEqual(network, result) def test__get_network_by_uuid_or_name_failure(self): network_uuid = '9acb0256-2c1b-420a-b9d7-62bee90b6ed7' self.client.find_network.side_effect = ( openstack_exc.OpenStackCloudException()) self.assertRaises(exception.NetworkError, neutron._get_network_by_uuid_or_name, self.client, network_uuid) self.client.find_network.assert_called_once_with( network_uuid, ignore_missing=False) def test__get_network_by_uuid_or_name_missing(self): network_uuid = '9acb0256-2c1b-420a-b9d7-62bee90b6ed7' self.client.find_network.side_effect = ( openstack_exc.ResourceNotFound()) self.assertRaises(exception.InvalidParameterValue, neutron._get_network_by_uuid_or_name, self.client, network_uuid) self.client.find_network.assert_called_once_with( network_uuid, ignore_missing=False) def test__get_network_by_uuid_or_name_duplicate(self): network_name = 'test-net' self.client.find_network.side_effect = ( openstack_exc.DuplicateResource()) self.assertRaises(exception.InvalidParameterValue, neutron._get_network_by_uuid_or_name, self.client, network_name) self.client.find_network.assert_called_once_with( network_name, ignore_missing=False) @mock.patch.object(neutron, '_get_network_by_uuid_or_name', autospec=True) @mock.patch.object(neutron, '_get_port_by_uuid', autospec=True) class TestGetPhysnetsByPortUUID(base.TestCase): PORT_FIELDS = ['network_id'] NETWORK_FIELDS = ['provider:physical_network', 'segments'] def setUp(self): super(TestGetPhysnetsByPortUUID, self).setUp() self.client = mock.MagicMock() def test_get_physnets_by_port_uuid_single_segment(self, mock_gp, mock_gn): port_uuid = 'fake-port-uuid' network_uuid = 'fake-network-uuid' physnet = 'fake-physnet' mock_gp.return_value = stubs.FakeNeutronPort(network_id=network_uuid) mock_gn.return_value = stubs.FakeNeutronNetwork( **{'segments': [ {'provider:physical_network': physnet} ]}) result = neutron.get_physnets_by_port_uuid(self.client, port_uuid) mock_gp.assert_called_once_with(self.client, port_uuid) mock_gn.assert_called_once_with(self.client, network_uuid) self.assertEqual({physnet}, result) def test_get_physnets_by_port_uuid_no_segment_no_physnet( self, mock_gp, mock_gn): port_uuid = 'fake-port-uuid' network_uuid = 'fake-network-uuid' mock_gp.return_value = stubs.FakeNeutronPort(network_id=network_uuid) fake_network = stubs.FakeNeutronNetwork( **{'provider:physical_network': None}) mock_gn.return_value = fake_network result = neutron.get_physnets_by_port_uuid(self.client, port_uuid) mock_gp.assert_called_once_with(self.client, port_uuid) mock_gn.assert_called_once_with(self.client, network_uuid) self.assertEqual(set(), result) def test_get_physnets_by_port_uuid_no_segment( self, mock_gp, mock_gn): port_uuid = 'fake-port-uuid' network_uuid = 'fake-network-uuid' physnet = 'fake-physnet' mock_gp.return_value = stubs.FakeNeutronPort(network_id=network_uuid) fake_network = stubs.FakeNeutronNetwork( **{'provider:physical_network': physnet}) mock_gn.return_value = fake_network result = neutron.get_physnets_by_port_uuid(self.client, port_uuid) mock_gp.assert_called_once_with(self.client, port_uuid) mock_gn.assert_called_once_with(self.client, network_uuid) self.assertEqual({physnet}, result) def test_get_physnets_by_port_uuid_multiple_segments(self, mock_gp, mock_gn): port_uuid = 'fake-port-uuid' network_uuid = 'fake-network-uuid' physnet1 = 'fake-physnet-1' physnet2 = 'fake-physnet-2' mock_gp.return_value = stubs.FakeNeutronPort(network_id=network_uuid) mock_gn.return_value = stubs.FakeNeutronNetwork( **{'segments': [{'provider:physical_network': physnet1}, {'provider:physical_network': physnet2}]}) result = neutron.get_physnets_by_port_uuid(self.client, port_uuid) mock_gp.assert_called_once_with(self.client, port_uuid) mock_gn.assert_called_once_with(self.client, network_uuid) self.assertEqual({physnet1, physnet2}, result) def test_get_physnets_by_port_uuid_multiple_segments_no_physnet( self, mock_gp, mock_gn): port_uuid = 'fake-port-uuid' network_uuid = 'fake-network-uuid' mock_gp.return_value = stubs.FakeNeutronPort(network_id=network_uuid) mock_gn.return_value = stubs.FakeNeutronNetwork( **{'segments': [{'provider:physical_network': None}, {'provider:physical_network': None}]}) result = neutron.get_physnets_by_port_uuid(self.client, port_uuid) mock_gp.assert_called_once_with(self.client, port_uuid) mock_gn.assert_called_once_with(self.client, network_uuid) self.assertEqual(set(), result) def test_get_physnets_by_port_uuid_port_missing(self, mock_gp, mock_gn): port_uuid = 'fake-port-uuid' mock_gp.side_effect = exception.InvalidParameterValue('error') self.assertRaises(exception.InvalidParameterValue, neutron.get_physnets_by_port_uuid, self.client, port_uuid) mock_gp.assert_called_once_with(self.client, port_uuid) self.assertFalse(mock_gn.called) def test_get_physnets_by_port_uuid_port_failure(self, mock_gp, mock_gn): port_uuid = 'fake-port-uuid' mock_gp.side_effect = exception.NetworkError self.assertRaises(exception.NetworkError, neutron.get_physnets_by_port_uuid, self.client, port_uuid) mock_gp.assert_called_once_with(self.client, port_uuid) self.assertFalse(mock_gn.called) def test_get_physnets_by_port_uuid_network_missing( self, mock_gp, mock_gn): port_uuid = 'fake-port-uuid' network_uuid = 'fake-network-uuid' mock_gp.return_value = stubs.FakeNeutronPort(network_id=network_uuid) mock_gn.side_effect = exception.InvalidParameterValue('error') self.assertRaises(exception.InvalidParameterValue, neutron.get_physnets_by_port_uuid, self.client, port_uuid) mock_gp.assert_called_once_with(self.client, port_uuid) mock_gn.assert_called_once_with(self.client, network_uuid) def test_get_physnets_by_port_uuid_network_failure( self, mock_gp, mock_gn): port_uuid = 'fake-port-uuid' network_uuid = 'fake-network-uuid' mock_gp.return_value = stubs.FakeNeutronPort(network_id=network_uuid) mock_gn.side_effect = exception.NetworkError self.assertRaises(exception.NetworkError, neutron.get_physnets_by_port_uuid, self.client, port_uuid) mock_gp.assert_called_once_with(self.client, port_uuid) mock_gn.assert_called_once_with(self.client, network_uuid)