ironic.tests.unit.drivers.modules.network.test_common

Source code for ironic.tests.unit.drivers.modules.network.test_common

#    Licensed under the Apache License, Version 2.0 (the "License"); you may
#    not use this file except in compliance with the License. You may obtain
#    a copy of the License at
#
#         http://www.apache.org/licenses/LICENSE-2.0
#
#    Unless required by applicable law or agreed to in writing, software
#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
#    License for the specific language governing permissions and limitations
#    under the License.

import mock
from neutronclient.common import exceptions as neutron_exceptions
from oslo_config import cfg
from oslo_utils import uuidutils

from ironic.common import exception
from ironic.common import neutron as neutron_common
from ironic.common import states
from ironic.common import utils as common_utils
from ironic.conductor import task_manager
from ironic.drivers.modules.network import common
from ironic.tests.unit.conductor import mgr_utils
from ironic.tests.unit.db import base as db_base
from ironic.tests.unit.objects import utils as obj_utils

CONF = cfg.CONF


[docs]class TestCommonFunctions(db_base.DbTestCase):
[docs] def setUp(self): super(TestCommonFunctions, self).setUp() self.config(enabled_drivers=['fake']) mgr_utils.mock_the_extension_manager() self.node = obj_utils.create_test_node(self.context, network_interface='neutron') self.port = obj_utils.create_test_port( self.context, node_id=self.node.id, address='52:54:00:cf:2d:32') self.vif_id = "fake_vif_id" self.client = mock.MagicMock()
def _objects_setup(self, set_physnets): pg1 = obj_utils.create_test_portgroup( self.context, node_id=self.node.id) pg1_ports = [] # This portgroup contains 2 ports, both of them without VIF. The ports # are assigned to physnet physnet1. physical_network = 'physnet1' if set_physnets else None for i in range(2): pg1_ports.append(obj_utils.create_test_port( self.context, node_id=self.node.id, address='52:54:00:cf:2d:0%d' % i, physical_network=physical_network, uuid=uuidutils.generate_uuid(), portgroup_id=pg1.id)) pg2 = obj_utils.create_test_portgroup( self.context, node_id=self.node.id, address='00:54:00:cf:2d:04', name='foo2', uuid=uuidutils.generate_uuid()) pg2_ports = [] # This portgroup contains 3 ports, one of them with 'some-vif' # attached, so the two free ones should be considered standalone. # The ports are assigned physnet physnet2. physical_network = 'physnet2' if set_physnets else None for i in range(2, 4): pg2_ports.append(obj_utils.create_test_port( self.context, node_id=self.node.id, address='52:54:00:cf:2d:0%d' % i, physical_network=physical_network, uuid=uuidutils.generate_uuid(), portgroup_id=pg2.id)) pg2_ports.append(obj_utils.create_test_port( self.context, node_id=self.node.id, address='52:54:00:cf:2d:04', physical_network=physical_network, extra={'vif_port_id': 'some-vif'}, uuid=uuidutils.generate_uuid(), portgroup_id=pg2.id)) # This portgroup has 'some-vif-2' attached to it and contains one port, # so neither portgroup nor port can be considered free. The ports are # assigned physnet3. physical_network = 'physnet3' if set_physnets else None pg3 = obj_utils.create_test_portgroup( self.context, node_id=self.node.id, address='00:54:00:cf:2d:05', name='foo3', uuid=uuidutils.generate_uuid(), internal_info={common.TENANT_VIF_KEY: 'some-vif-2'}) pg3_ports = [obj_utils.create_test_port( self.context, node_id=self.node.id, address='52:54:00:cf:2d:05', uuid=uuidutils.generate_uuid(), physical_network=physical_network, portgroup_id=pg3.id)] return pg1, pg1_ports, pg2, pg2_ports, pg3, pg3_ports
[docs] def test__get_free_portgroups_and_ports_no_port_physnets(self): self.node.network_interface = 'flat' self.node.save() pg1, pg1_ports, pg2, pg2_ports, pg3, pg3_ports = self._objects_setup( set_physnets=False) with task_manager.acquire(self.context, self.node.id) as task: free_port_like_objs = ( common._get_free_portgroups_and_ports(task, self.vif_id, {'anyphysnet'})) self.assertItemsEqual( [pg1.uuid, self.port.uuid] + [p.uuid for p in pg2_ports[:2]], [p.uuid for p in free_port_like_objs])
[docs] def test__get_free_portgroups_and_ports_no_physnets(self): self.node.network_interface = 'flat' self.node.save() pg1, pg1_ports, pg2, pg2_ports, pg3, pg3_ports = self._objects_setup( set_physnets=True) with task_manager.acquire(self.context, self.node.id) as task: free_port_like_objs = ( common._get_free_portgroups_and_ports(task, self.vif_id, set())) self.assertItemsEqual( [pg1.uuid, self.port.uuid] + [p.uuid for p in pg2_ports[:2]], [p.uuid for p in free_port_like_objs])
[docs] def test__get_free_portgroups_and_ports_no_matching_physnet(self): self.node.network_interface = 'flat' self.node.save() pg1, pg1_ports, pg2, pg2_ports, pg3, pg3_ports = self._objects_setup( set_physnets=True) with task_manager.acquire(self.context, self.node.id) as task: free_port_like_objs = ( common._get_free_portgroups_and_ports(task, self.vif_id, {'notaphysnet'})) self.assertItemsEqual( [self.port.uuid], [p.uuid for p in free_port_like_objs])
[docs] def test__get_free_portgroups_and_ports_physnet1(self): self.node.network_interface = 'flat' self.node.save() pg1, pg1_ports, pg2, pg2_ports, pg3, pg3_ports = self._objects_setup( set_physnets=True) with task_manager.acquire(self.context, self.node.id) as task: free_port_like_objs = ( common._get_free_portgroups_and_ports(task, self.vif_id, {'physnet1'})) self.assertItemsEqual( [pg1.uuid, self.port.uuid], [p.uuid for p in free_port_like_objs])
[docs] def test__get_free_portgroups_and_ports_physnet2(self): self.node.network_interface = 'flat' self.node.save() pg1, pg1_ports, pg2, pg2_ports, pg3, pg3_ports = self._objects_setup( set_physnets=True) with task_manager.acquire(self.context, self.node.id) as task: free_port_like_objs = ( common._get_free_portgroups_and_ports(task, self.vif_id, {'physnet2'})) self.assertItemsEqual( [self.port.uuid] + [p.uuid for p in pg2_ports[:2]], [p.uuid for p in free_port_like_objs])
[docs] def test__get_free_portgroups_and_ports_physnet3(self): self.node.network_interface = 'flat' self.node.save() pg1, pg1_ports, pg2, pg2_ports, pg3, pg3_ports = self._objects_setup( set_physnets=True) with task_manager.acquire(self.context, self.node.id) as task: free_port_like_objs = ( common._get_free_portgroups_and_ports(task, self.vif_id, {'physnet3'})) self.assertItemsEqual( [self.port.uuid], [p.uuid for p in free_port_like_objs])
[docs] def test__get_free_portgroups_and_ports_all_physnets(self): self.node.network_interface = 'flat' self.node.save() pg1, pg1_ports, pg2, pg2_ports, pg3, pg3_ports = self._objects_setup( set_physnets=True) physnets = {'physnet1', 'physnet2', 'physnet3'} with task_manager.acquire(self.context, self.node.id) as task: free_port_like_objs = ( common._get_free_portgroups_and_ports(task, self.vif_id, physnets)) self.assertItemsEqual( [pg1.uuid, self.port.uuid] + [p.uuid for p in pg2_ports[:2]], [p.uuid for p in free_port_like_objs])
[docs] @mock.patch.object(neutron_common, 'validate_port_info', autospec=True) def test__get_free_portgroups_and_ports_neutron_missed(self, vpi_mock): vpi_mock.return_value = False with task_manager.acquire(self.context, self.node.id) as task: free_port_like_objs = ( common._get_free_portgroups_and_ports(task, self.vif_id, {'anyphysnet'})) self.assertItemsEqual([], free_port_like_objs)
[docs] @mock.patch.object(neutron_common, 'validate_port_info', autospec=True) def test__get_free_portgroups_and_ports_neutron(self, vpi_mock): vpi_mock.return_value = True with task_manager.acquire(self.context, self.node.id) as task: free_port_like_objs = ( common._get_free_portgroups_and_ports(task, self.vif_id, {'anyphysnet'})) self.assertItemsEqual( [self.port.uuid], [p.uuid for p in free_port_like_objs])
[docs] @mock.patch.object(neutron_common, 'validate_port_info', autospec=True) def test__get_free_portgroups_and_ports_flat(self, vpi_mock): self.node.network_interface = 'flat' self.node.save() vpi_mock.return_value = True with task_manager.acquire(self.context, self.node.id) as task: free_port_like_objs = ( common._get_free_portgroups_and_ports(task, self.vif_id, {'anyphysnet'})) self.assertItemsEqual( [self.port.uuid], [p.uuid for p in free_port_like_objs])
[docs] @mock.patch.object(neutron_common, 'validate_port_info', autospec=True, return_value=True) def test_get_free_port_like_object_ports(self, vpi_mock): with task_manager.acquire(self.context, self.node.id) as task: res = common.get_free_port_like_object(task, self.vif_id, {'anyphysnet'}) self.assertEqual(self.port.uuid, res.uuid)
[docs] @mock.patch.object(neutron_common, 'validate_port_info', autospec=True, return_value=True) def test_get_free_port_like_object_ports_pxe_enabled_first(self, vpi_mock): self.port.pxe_enabled = False self.port.save() other_port = obj_utils.create_test_port( self.context, node_id=self.node.id, address='52:54:00:cf:2d:33', uuid=uuidutils.generate_uuid()) with task_manager.acquire(self.context, self.node.id) as task: res = common.get_free_port_like_object(task, self.vif_id, {'anyphysnet'}) self.assertEqual(other_port.uuid, res.uuid)
[docs] @mock.patch.object(neutron_common, 'validate_port_info', autospec=True, return_value=True) def test_get_free_port_like_object_ports_physnet_match_first(self, vpi_mock): self.port.pxe_enabled = False self.port.physical_network = 'physnet1' self.port.save() obj_utils.create_test_port( self.context, node_id=self.node.id, address='52:54:00:cf:2d:33', uuid=uuidutils.generate_uuid()) with task_manager.acquire(self.context, self.node.id) as task: res = common.get_free_port_like_object(task, self.vif_id, {'physnet1'}) self.assertEqual(self.port.uuid, res.uuid)
[docs] @mock.patch.object(neutron_common, 'validate_port_info', autospec=True, return_value=True) def test_get_free_port_like_object_ports_physnet_match_first2(self, vpi_mock): self.port.pxe_enabled = False self.port.physical_network = 'physnet1' self.port.save() pg = obj_utils.create_test_portgroup( self.context, node_id=self.node.id) obj_utils.create_test_port( self.context, node_id=self.node.id, address='52:54:00:cf:2d:01', uuid=uuidutils.generate_uuid(), portgroup_id=pg.id) with task_manager.acquire(self.context, self.node.id) as task: res = common.get_free_port_like_object(task, self.vif_id, {'physnet1'}) self.assertEqual(self.port.uuid, res.uuid)
[docs] @mock.patch.object(neutron_common, 'validate_port_info', autospec=True, return_value=True) def test_get_free_port_like_object_portgroup_first(self, vpi_mock): pg = obj_utils.create_test_portgroup( self.context, node_id=self.node.id) obj_utils.create_test_port( self.context, node_id=self.node.id, address='52:54:00:cf:2d:01', uuid=uuidutils.generate_uuid(), portgroup_id=pg.id) with task_manager.acquire(self.context, self.node.id) as task: res = common.get_free_port_like_object(task, self.vif_id, {'anyphysnet'}) self.assertEqual(pg.uuid, res.uuid)
[docs] @mock.patch.object(neutron_common, 'validate_port_info', autospec=True, return_value=True) def test_get_free_port_like_object_portgroup_physnet_match_first(self, vpi_mock): pg1 = obj_utils.create_test_portgroup( self.context, node_id=self.node.id) obj_utils.create_test_port( self.context, node_id=self.node.id, address='52:54:00:cf:2d:01', uuid=uuidutils.generate_uuid(), portgroup_id=pg1.id) pg2 = obj_utils.create_test_portgroup( self.context, node_id=self.node.id, uuid=uuidutils.generate_uuid(), name='pg2', address='52:54:00:cf:2d:01') obj_utils.create_test_port( self.context, node_id=self.node.id, address='52:54:00:cf:2d:02', uuid=uuidutils.generate_uuid(), portgroup_id=pg2.id, physical_network='physnet1') with task_manager.acquire(self.context, self.node.id) as task: res = common.get_free_port_like_object(task, self.vif_id, {'physnet1'}) self.assertEqual(pg2.uuid, res.uuid)
[docs] @mock.patch.object(neutron_common, 'validate_port_info', autospec=True, return_value=True) def test_get_free_port_like_object_ignores_empty_portgroup(self, vpi_mock): obj_utils.create_test_portgroup(self.context, node_id=self.node.id) with task_manager.acquire(self.context, self.node.id) as task: res = common.get_free_port_like_object(task, self.vif_id, {'anyphysnet'}) self.assertEqual(self.port.uuid, res.uuid)
[docs] @mock.patch.object(neutron_common, 'validate_port_info', autospec=True, return_value=True) def test_get_free_port_like_object_ignores_standalone_portgroup( self, vpi_mock): self.port.destroy() pg = obj_utils.create_test_portgroup( self.context, node_id=self.node.id) obj_utils.create_test_port( self.context, node_id=self.node.id, address='52:54:00:cf:2d:01', uuid=uuidutils.generate_uuid(), portgroup_id=pg.id, extra={'vif_port_id': 'some-vif'}) free_port = obj_utils.create_test_port( self.context, node_id=self.node.id, address='52:54:00:cf:2d:02', uuid=uuidutils.generate_uuid(), portgroup_id=pg.id) with task_manager.acquire(self.context, self.node.id) as task: res = common.get_free_port_like_object(task, self.vif_id, {'anyphysnet'}) self.assertEqual(free_port.uuid, res.uuid)
[docs] @mock.patch.object(neutron_common, 'validate_port_info', autospec=True, return_value=True) def test_get_free_port_like_object_vif_attached_to_portgroup( self, vpi_mock): pg = obj_utils.create_test_portgroup( self.context, node_id=self.node.id, internal_info={common.TENANT_VIF_KEY: self.vif_id}) obj_utils.create_test_port( self.context, node_id=self.node.id, address='52:54:00:cf:2d:01', uuid=uuidutils.generate_uuid(), portgroup_id=pg.id) with task_manager.acquire(self.context, self.node.id) as task: self.assertRaisesRegex( exception.VifAlreadyAttached, r"already attached to Ironic Portgroup", common.get_free_port_like_object, task, self.vif_id, {'anyphysnet'})
[docs] @mock.patch.object(neutron_common, 'validate_port_info', autospec=True, return_value=True) def test_get_free_port_like_object_vif_attached_to_portgroup_extra( self, vpi_mock): pg = obj_utils.create_test_portgroup( self.context, node_id=self.node.id, extra={'vif_port_id': self.vif_id}) obj_utils.create_test_port( self.context, node_id=self.node.id, address='52:54:00:cf:2d:01', uuid=uuidutils.generate_uuid(), portgroup_id=pg.id) with task_manager.acquire(self.context, self.node.id) as task: self.assertRaisesRegex( exception.VifAlreadyAttached, r"already attached to Ironic Portgroup", common.get_free_port_like_object, task, self.vif_id, {'anyphysnet'})
[docs] @mock.patch.object(neutron_common, 'validate_port_info', autospec=True, return_value=True) def test_get_free_port_like_object_vif_attached_to_port(self, vpi_mock): self.port.internal_info = {common.TENANT_VIF_KEY: self.vif_id} self.port.save() with task_manager.acquire(self.context, self.node.id) as task: self.assertRaisesRegex( exception.VifAlreadyAttached, r"already attached to Ironic Port\b", common.get_free_port_like_object, task, self.vif_id, {'anyphysnet'})
[docs] @mock.patch.object(neutron_common, 'validate_port_info', autospec=True, return_value=True) def test_get_free_port_like_object_vif_attached_to_port_extra( self, vpi_mock): self.port.extra = {'vif_port_id': self.vif_id} self.port.save() with task_manager.acquire(self.context, self.node.id) as task: self.assertRaisesRegex( exception.VifAlreadyAttached, r"already attached to Ironic Port\b", common.get_free_port_like_object, task, self.vif_id, {'anyphysnet'})
[docs] @mock.patch.object(neutron_common, 'validate_port_info', autospec=True, return_value=True) def test_get_free_port_like_object_nothing_free(self, vpi_mock): self.port.extra = {'vif_port_id': 'another-vif'} self.port.save() with task_manager.acquire(self.context, self.node.id) as task: self.assertRaises(exception.NoFreePhysicalPorts, common.get_free_port_like_object, task, self.vif_id, {'anyphysnet'})
[docs] @mock.patch.object(neutron_common, 'validate_port_info', autospec=True, return_value=True) def test_get_free_port_like_object_no_matching_physnets(self, vpi_mock): self.port.physical_network = 'physnet1' self.port.save() with task_manager.acquire(self.context, self.node.id) as task: self.assertRaises(exception.NoFreePhysicalPorts, common.get_free_port_like_object, task, self.vif_id, {'physnet2'})
[docs] @mock.patch.object(neutron_common, 'get_client', autospec=True) def test_plug_port_to_tenant_network_client(self, mock_gc): self.port.internal_info = {common.TENANT_VIF_KEY: self.vif_id} self.port.save() with task_manager.acquire(self.context, self.node.id) as task: common.plug_port_to_tenant_network(task, self.port, client=mock.MagicMock()) self.assertFalse(mock_gc.called)
[docs] @mock.patch.object(neutron_common, 'get_client', autospec=True) def test_plug_port_to_tenant_network_no_client(self, mock_gc): self.port.internal_info = {common.TENANT_VIF_KEY: self.vif_id} self.port.save() with task_manager.acquire(self.context, self.node.id) as task: common.plug_port_to_tenant_network(task, self.port) self.assertTrue(mock_gc.called)
[docs] @mock.patch.object(neutron_common, 'get_client', autospec=True) def test_plug_port_to_tenant_network_no_tenant_vif(self, mock_gc): nclient = mock.MagicMock() mock_gc.return_value = nclient self.port.extra = {} self.port.save() with task_manager.acquire(self.context, self.node.id) as task: self.assertRaisesRegex( exception.VifNotAttached, "not associated with port %s" % self.port.uuid, common.plug_port_to_tenant_network, task, self.port)
[docs]class TestVifPortIDMixin(db_base.DbTestCase):
[docs] def setUp(self): super(TestVifPortIDMixin, self).setUp() self.config(enabled_drivers=['fake']) mgr_utils.mock_the_extension_manager() self.interface = common.VIFPortIDMixin() self.node = obj_utils.create_test_node(self.context, network_interface='neutron') self.port = obj_utils.create_test_port( self.context, node_id=self.node.id, address='52:54:00:cf:2d:32', extra={'vif_port_id': uuidutils.generate_uuid(), 'client-id': 'fake1'})
[docs] def test__save_vif_to_port_like_obj_port(self): self.port.extra = {} self.port.save() vif_id = "fake_vif_id" self.interface._save_vif_to_port_like_obj(self.port, vif_id) self.port.refresh() self.assertIn(common.TENANT_VIF_KEY, self.port.internal_info) self.assertEqual(vif_id, self.port.internal_info[common.TENANT_VIF_KEY]) self.assertEqual({}, self.port.extra)
[docs] def test__save_vif_to_port_like_obj_portgroup(self): vif_id = "fake_vif_id" pg = obj_utils.create_test_portgroup( self.context, node_id=self.node.id) obj_utils.create_test_port( self.context, node_id=self.node.id, address='52:54:00:cf:2d:01', portgroup_id=pg.id, uuid=uuidutils.generate_uuid() ) self.interface._save_vif_to_port_like_obj(pg, vif_id) pg.refresh() self.assertIn(common.TENANT_VIF_KEY, pg.internal_info) self.assertEqual(vif_id, pg.internal_info[common.TENANT_VIF_KEY]) self.assertEqual({}, pg.extra)
[docs] def test__clear_vif_from_port_like_obj_in_extra_port(self): self.interface._clear_vif_from_port_like_obj(self.port) self.port.refresh() self.assertNotIn('vif_port_id', self.port.extra) self.assertNotIn(common.TENANT_VIF_KEY, self.port.internal_info)
[docs] def test__clear_vif_from_port_like_obj_in_internal_info_port(self): self.port.internal_info = { common.TENANT_VIF_KEY: self.port.extra['vif_port_id']} self.port.extra = {} self.port.save() self.interface._clear_vif_from_port_like_obj(self.port) self.port.refresh() self.assertNotIn('vif_port_id', self.port.extra) self.assertNotIn(common.TENANT_VIF_KEY, self.port.internal_info)
[docs] def test__clear_vif_from_port_like_obj_in_extra_portgroup(self): vif_id = uuidutils.generate_uuid() pg = obj_utils.create_test_portgroup( self.context, node_id=self.node.id, extra={'vif_port_id': vif_id}) obj_utils.create_test_port( self.context, node_id=self.node.id, address='52:54:00:cf:2d:01', portgroup_id=pg.id, uuid=uuidutils.generate_uuid() ) self.interface._clear_vif_from_port_like_obj(pg) pg.refresh() self.assertNotIn('vif_port_id', pg.extra) self.assertNotIn(common.TENANT_VIF_KEY, pg.internal_info)
[docs] def test__clear_vif_from_port_like_obj_in_internal_info_portgroup(self): vif_id = uuidutils.generate_uuid() pg = obj_utils.create_test_portgroup( self.context, node_id=self.node.id, internal_info={common.TENANT_VIF_KEY: vif_id}) obj_utils.create_test_port( self.context, node_id=self.node.id, address='52:54:00:cf:2d:01', portgroup_id=pg.id, uuid=uuidutils.generate_uuid() ) self.interface._clear_vif_from_port_like_obj(pg) pg.refresh() self.assertNotIn('vif_port_id', pg.extra) self.assertNotIn(common.TENANT_VIF_KEY, pg.internal_info)
[docs] def test__get_port_like_obj_by_vif_id_in_extra(self): vif_id = self.port.extra["vif_port_id"] with task_manager.acquire(self.context, self.node.id) as task: result = self.interface._get_port_like_obj_by_vif_id(task, vif_id) self.assertEqual(self.port.id, result.id)
[docs] def test__get_port_like_obj_by_vif_id_in_internal_info(self): vif_id = self.port.extra["vif_port_id"] self.port.internal_info = {common.TENANT_VIF_KEY: vif_id} self.port.extra = {} self.port.save() with task_manager.acquire(self.context, self.node.id) as task: result = self.interface._get_port_like_obj_by_vif_id(task, vif_id) self.assertEqual(self.port.id, result.id)
[docs] def test__get_port_like_obj_by_vif_id_not_attached(self): vif_id = self.port.extra["vif_port_id"] self.port.extra = {} self.port.save() with task_manager.acquire(self.context, self.node.id) as task: self.assertRaisesRegex(exception.VifNotAttached, "it is not attached to it.", self.interface._get_port_like_obj_by_vif_id, task, vif_id)
[docs] def test__get_vif_id_by_port_like_obj_in_extra(self): vif_id = self.port.extra["vif_port_id"] result = self.interface._get_vif_id_by_port_like_obj(self.port) self.assertEqual(vif_id, result)
[docs] def test__get_vif_id_by_port_like_obj_in_internal_info(self): vif_id = self.port.extra["vif_port_id"] self.port.internal_info = {common.TENANT_VIF_KEY: vif_id} self.port.extra = {} self.port.save() result = self.interface._get_vif_id_by_port_like_obj(self.port) self.assertEqual(vif_id, result)
[docs] def test__get_vif_id_by_port_like_obj_not_attached(self): self.port.extra = {} self.port.save() result = self.interface._get_vif_id_by_port_like_obj(self.port) self.assertIsNone(result)
[docs] def test_vif_list_extra(self): vif_id = uuidutils.generate_uuid() self.port.extra = {'vif_port_id': vif_id} self.port.save() pg_vif_id = uuidutils.generate_uuid() portgroup = obj_utils.create_test_portgroup( self.context, node_id=self.node.id, address='52:54:00:00:00:00', internal_info={common.TENANT_VIF_KEY: pg_vif_id}) obj_utils.create_test_port( self.context, node_id=self.node.id, portgroup_id=portgroup.id, address='52:54:00:cf:2d:01', uuid=uuidutils.generate_uuid()) with task_manager.acquire(self.context, self.node.id) as task: vifs = self.interface.vif_list(task) self.assertItemsEqual([{'id': pg_vif_id}, {'id': vif_id}], vifs)
[docs] def test_vif_list_internal(self): vif_id = uuidutils.generate_uuid() self.port.internal_info = {common.TENANT_VIF_KEY: vif_id} self.port.save() pg_vif_id = uuidutils.generate_uuid() portgroup = obj_utils.create_test_portgroup( self.context, node_id=self.node.id, address='52:54:00:00:00:00', internal_info={common.TENANT_VIF_KEY: pg_vif_id}) obj_utils.create_test_port( self.context, node_id=self.node.id, portgroup_id=portgroup.id, address='52:54:00:cf:2d:01', uuid=uuidutils.generate_uuid()) with task_manager.acquire(self.context, self.node.id) as task: vifs = self.interface.vif_list(task) self.assertItemsEqual([{'id': pg_vif_id}, {'id': vif_id}], vifs)
[docs] def test_vif_list_extra_and_internal_priority(self): vif_id = uuidutils.generate_uuid() vif_id2 = uuidutils.generate_uuid() self.port.extra = {'vif_port_id': vif_id2} self.port.internal_info = {common.TENANT_VIF_KEY: vif_id} self.port.save() with task_manager.acquire(self.context, self.node.id) as task: vifs = self.interface.vif_list(task) self.assertEqual([{'id': vif_id}], vifs)
[docs] def test_get_current_vif_extra_vif_port_id(self): extra = {'vif_port_id': 'foo'} self.port.extra = extra self.port.save() with task_manager.acquire(self.context, self.node.id) as task: vif = self.interface.get_current_vif(task, self.port) self.assertEqual('foo', vif)
[docs] def test_get_current_vif_internal_info_cleaning(self): internal_info = {'cleaning_vif_port_id': 'foo', 'tenant_vif_port_id': 'bar'} self.port.internal_info = internal_info self.port.save() with task_manager.acquire(self.context, self.node.id) as task: vif = self.interface.get_current_vif(task, self.port) self.assertEqual('foo', vif)
[docs] def test_get_current_vif_internal_info_provisioning(self): internal_info = {'provisioning_vif_port_id': 'foo', 'vif_port_id': 'bar'} self.port.internal_info = internal_info self.port.save() with task_manager.acquire(self.context, self.node.id) as task: vif = self.interface.get_current_vif(task, self.port) self.assertEqual('foo', vif)
[docs] def test_get_current_vif_internal_info_tenant_vif(self): internal_info = {'tenant_vif_port_id': 'bar'} self.port.internal_info = internal_info self.port.save() with task_manager.acquire(self.context, self.node.id) as task: vif = self.interface.get_current_vif(task, self.port) self.assertEqual('bar', vif)
[docs] def test_get_current_vif_none(self): internal_info = extra = {} self.port.internal_info = internal_info self.port.extra = extra self.port.save() with task_manager.acquire(self.context, self.node.id) as task: vif = self.interface.get_current_vif(task, self.port) self.assertIsNone(vif)
[docs]class TestNeutronVifPortIDMixin(db_base.DbTestCase):
[docs] def setUp(self): super(TestNeutronVifPortIDMixin, self).setUp() self.config(enabled_drivers=['fake']) mgr_utils.mock_the_extension_manager() self.interface = common.NeutronVIFPortIDMixin() self.node = obj_utils.create_test_node(self.context, network_interface='neutron') self.port = obj_utils.create_test_port( self.context, node_id=self.node.id, address='52:54:00:cf:2d:32', extra={'vif_port_id': uuidutils.generate_uuid(), 'client-id': 'fake1'}) self.neutron_port = {'id': '132f871f-eaec-4fed-9475-0d54465e0f00', 'mac_address': '52:54:00:cf:2d:32'}
[docs] @mock.patch.object(common.VIFPortIDMixin, '_save_vif_to_port_like_obj') @mock.patch.object(common, 'get_free_port_like_object', autospec=True) @mock.patch.object(neutron_common, 'get_client', autospec=True) @mock.patch.object(neutron_common, 'update_port_address', autospec=True) @mock.patch.object(neutron_common, 'get_physnets_by_port_uuid', autospec=True) def test_vif_attach(self, mock_gpbpi, mock_upa, mock_client, mock_gfp, mock_save): vif = {'id': "fake_vif_id"} mock_gfp.return_value = self.port with task_manager.acquire(self.context, self.node.id) as task: self.interface.vif_attach(task, vif) mock_client.assert_called_once_with() self.assertFalse(mock_gpbpi.called) mock_gfp.assert_called_once_with(task, 'fake_vif_id', set()) mock_client.return_value.show_port.assert_called_once_with( 'fake_vif_id') mock_upa.assert_called_once_with("fake_vif_id", self.port.address) mock_save.assert_called_once_with(self.port, "fake_vif_id")
[docs] @mock.patch.object(common.VIFPortIDMixin, '_save_vif_to_port_like_obj') @mock.patch.object(common, 'get_free_port_like_object', autospec=True) @mock.patch.object(neutron_common, 'get_client', autospec=True) @mock.patch.object(neutron_common, 'update_port_address', autospec=True) @mock.patch.object(neutron_common, 'get_physnets_by_port_uuid', autospec=True) def test_vif_attach_failure(self, mock_gpbpi, mock_upa, mock_client, mock_gfp, mock_save): vif = {'id': "fake_vif_id"} mock_gfp.side_effect = exception.NoFreePhysicalPorts(vif='fake-vif') with task_manager.acquire(self.context, self.node.id) as task: self.assertRaises(exception.NoFreePhysicalPorts, self.interface.vif_attach, task, vif) mock_gfp.assert_called_once_with(task, 'fake_vif_id', set()) self.assertFalse(mock_save.called)
[docs] @mock.patch.object(common.VIFPortIDMixin, '_save_vif_to_port_like_obj') @mock.patch.object(common, 'get_free_port_like_object', autospec=True) @mock.patch.object(neutron_common, 'get_client', autospec=True) @mock.patch.object(neutron_common, 'update_port_address', autospec=True) @mock.patch.object(neutron_common, 'get_physnets_by_port_uuid', autospec=True) def test_vif_attach_with_physnet(self, mock_gpbpi, mock_upa, mock_client, mock_gfp, mock_save): self.port.physical_network = 'physnet1' self.port.save() vif = {'id': "fake_vif_id"} mock_gpbpi.return_value = {'physnet1'} mock_gfp.return_value = self.port with task_manager.acquire(self.context, self.node.id) as task: self.interface.vif_attach(task, vif) mock_client.assert_called_once_with() mock_gpbpi.assert_called_once_with(mock_client.return_value, 'fake_vif_id') mock_gfp.assert_called_once_with(task, 'fake_vif_id', {'physnet1'}) mock_client.return_value.show_port.assert_called_once_with( 'fake_vif_id') mock_upa.assert_called_once_with("fake_vif_id", self.port.address) mock_save.assert_called_once_with(self.port, "fake_vif_id")
[docs] @mock.patch.object(common.VIFPortIDMixin, '_save_vif_to_port_like_obj') @mock.patch.object(common, 'plug_port_to_tenant_network', autospec=True) @mock.patch.object(common, 'get_free_port_like_object', autospec=True) @mock.patch.object(neutron_common, 'get_client', autospec=True) @mock.patch.object(neutron_common, 'update_port_address', autospec=True) @mock.patch.object(neutron_common, 'get_physnets_by_port_uuid', autospec=True) def test_vif_attach_active_node(self, mock_gpbpi, mock_upa, mock_client, mock_gfp, mock_plug, mock_save): self.node.provision_state = states.ACTIVE self.node.save() vif = {'id': "fake_vif_id"} mock_gfp.return_value = self.port with task_manager.acquire(self.context, self.node.id) as task: self.interface.vif_attach(task, vif) mock_client.assert_called_once_with() self.assertFalse(mock_gpbpi.called) mock_gfp.assert_called_once_with(task, 'fake_vif_id', set()) mock_client.return_value.show_port.assert_called_once_with( 'fake_vif_id') mock_upa.assert_called_once_with("fake_vif_id", self.port.address) mock_save.assert_called_once_with(self.port, "fake_vif_id") mock_plug.assert_called_once_with(task, self.port, mock.ANY)
[docs] @mock.patch.object(common.VIFPortIDMixin, '_save_vif_to_port_like_obj') @mock.patch.object(common, 'plug_port_to_tenant_network', autospec=True) @mock.patch.object(common, 'get_free_port_like_object', autospec=True) @mock.patch.object(neutron_common, 'get_client', autospec=True) @mock.patch.object(neutron_common, 'update_port_address', autospec=True) @mock.patch.object(neutron_common, 'get_physnets_by_port_uuid', autospec=True) def test_vif_attach_active_node_failure(self, mock_gpbpi, mock_upa, mock_client, mock_gfp, mock_plug, mock_save): self.node.provision_state = states.ACTIVE self.node.save() vif = {'id': "fake_vif_id"} mock_gfp.return_value = self.port mock_plug.side_effect = exception.NetworkError with task_manager.acquire(self.context, self.node.id) as task: self.assertRaises(exception.NetworkError, self.interface.vif_attach, task, vif) mock_client.assert_called_once_with() self.assertFalse(mock_gpbpi.called) mock_gfp.assert_called_once_with(task, 'fake_vif_id', set()) mock_client.return_value.show_port.assert_called_once_with( 'fake_vif_id') mock_upa.assert_called_once_with("fake_vif_id", self.port.address) mock_save.assert_called_once_with(self.port, "fake_vif_id") mock_plug.assert_called_once_with(task, self.port, mock.ANY)
[docs] @mock.patch.object(common.VIFPortIDMixin, '_save_vif_to_port_like_obj') @mock.patch.object(common, 'get_free_port_like_object', autospec=True) @mock.patch.object(neutron_common, 'get_client', autospec=True) @mock.patch.object(neutron_common, 'update_port_address') @mock.patch.object(neutron_common, 'get_physnets_by_port_uuid', autospec=True) def test_vif_attach_portgroup_no_address(self, mock_gpbpi, mock_upa, mock_client, mock_gfp, mock_save): pg = obj_utils.create_test_portgroup( self.context, node_id=self.node.id, address=None) mock_gfp.return_value = pg vif = {'id': "fake_vif_id"} with task_manager.acquire(self.context, self.node.id) as task: self.interface.vif_attach(task, vif) mock_client.assert_called_once_with() self.assertFalse(mock_gpbpi.called) mock_gfp.assert_called_once_with(task, 'fake_vif_id', set()) self.assertFalse(mock_client.return_value.show_port.called) self.assertFalse(mock_upa.called) mock_save.assert_called_once_with(pg, "fake_vif_id")
[docs] @mock.patch.object(common.VIFPortIDMixin, '_save_vif_to_port_like_obj') @mock.patch.object(neutron_common, 'get_client', autospec=True) @mock.patch.object(neutron_common, 'update_port_address') @mock.patch.object(neutron_common, 'get_physnets_by_port_uuid', autospec=True) def test_vif_attach_update_port_exception(self, mock_gpbpi, mock_upa, mock_client, mock_save): self.port.extra = {} self.port.physical_network = 'physnet1' self.port.save() vif = {'id': "fake_vif_id"} mock_gpbpi.return_value = {'physnet1'} mock_upa.side_effect = ( exception.FailedToUpdateMacOnPort(port_id='fake')) with task_manager.acquire(self.context, self.node.id) as task: self.assertRaisesRegex( exception.NetworkError, "can not update Neutron port", self.interface.vif_attach, task, vif) mock_client.assert_called_once_with() mock_gpbpi.assert_called_once_with(mock_client.return_value, 'fake_vif_id') mock_client.return_value.show_port.assert_called_once_with( 'fake_vif_id') self.assertFalse(mock_save.called)
[docs] @mock.patch.object(common.VIFPortIDMixin, '_save_vif_to_port_like_obj') @mock.patch.object(common, 'get_free_port_like_object', autospec=True) @mock.patch.object(neutron_common, 'get_client', autospec=True) @mock.patch.object(neutron_common, 'update_port_address') @mock.patch.object(neutron_common, 'get_physnets_by_port_uuid', autospec=True) def test_vif_attach_neutron_absent(self, mock_gpbpi, mock_upa, mock_client, mock_gfp, mock_save): self.port.physical_network = 'physnet1' self.port.save() vif = {'id': "fake_vif_id"} mock_gfp.return_value = self.port mock_client.return_value.show_port.side_effect = ( neutron_exceptions.NeutronClientException) mock_gpbpi.side_effect = exception.NetworkError with task_manager.acquire(self.context, self.node.id) as task: self.interface.vif_attach(task, vif) mock_client.assert_called_once_with() mock_gpbpi.assert_called_once_with(mock_client.return_value, 'fake_vif_id') mock_gfp.assert_called_once_with(task, 'fake_vif_id', set()) mock_client.return_value.show_port.assert_called_once_with( 'fake_vif_id') self.assertFalse(mock_upa.called) self.assertTrue(mock_save.called)
[docs] @mock.patch.object(common.VIFPortIDMixin, '_save_vif_to_port_like_obj') @mock.patch.object(common, 'get_free_port_like_object', autospec=True) @mock.patch.object(neutron_common, 'get_client') @mock.patch.object(neutron_common, 'update_port_address') @mock.patch.object(neutron_common, 'get_physnets_by_port_uuid', autospec=True) def test_vif_attach_portgroup_physnet_inconsistent(self, mock_gpbpi, mock_upa, mock_client, mock_gfp, mock_save): self.port.physical_network = 'physnet1' self.port.save() vif = {'id': "fake_vif_id"} mock_gpbpi.return_value = {'anyphysnet'} mock_gfp.side_effect = exception.PortgroupPhysnetInconsistent( portgroup='fake-portgroup-id', physical_networks='physnet1') with task_manager.acquire(self.context, self.node.id) as task: self.assertRaises( exception.PortgroupPhysnetInconsistent, self.interface.vif_attach, task, vif) mock_client.assert_called_once_with() mock_gpbpi.assert_called_once_with(mock_client.return_value, 'fake_vif_id') self.assertFalse(mock_upa.called) self.assertFalse(mock_save.called)
[docs] @mock.patch.object(common.VIFPortIDMixin, '_save_vif_to_port_like_obj') @mock.patch.object(common, 'get_free_port_like_object', autospec=True) @mock.patch.object(neutron_common, 'get_client') @mock.patch.object(neutron_common, 'update_port_address') @mock.patch.object(neutron_common, 'get_physnets_by_port_uuid', autospec=True) def test_vif_attach_multiple_segment_mappings(self, mock_gpbpi, mock_upa, mock_client, mock_gfp, mock_save): self.port.physical_network = 'physnet1' self.port.save() obj_utils.create_test_port( self.context, node_id=self.node.id, uuid=uuidutils.generate_uuid(), address='52:54:00:cf:2d:33', physical_network='physnet2') vif = {'id': "fake_vif_id"} mock_gpbpi.return_value = {'physnet1', 'physnet2'} with task_manager.acquire(self.context, self.node.id) as task: self.assertRaises( exception.VifInvalidForAttach, self.interface.vif_attach, task, vif) mock_client.assert_called_once_with() mock_gpbpi.assert_called_once_with(mock_client.return_value, 'fake_vif_id') self.assertFalse(mock_gfp.called) self.assertFalse(mock_upa.called) self.assertFalse(mock_save.called)
[docs] @mock.patch.object(common.VIFPortIDMixin, '_clear_vif_from_port_like_obj') @mock.patch.object(neutron_common, 'unbind_neutron_port', autospec=True) @mock.patch.object(common.VIFPortIDMixin, '_get_port_like_obj_by_vif_id') def test_vif_detach(self, mock_get, mock_unp, mock_clear): mock_get.return_value = self.port with task_manager.acquire(self.context, self.node.id) as task: self.interface.vif_detach(task, 'fake_vif_id') mock_get.assert_called_once_with(task, 'fake_vif_id') self.assertFalse(mock_unp.called) mock_clear.assert_called_once_with(self.port)
[docs] @mock.patch.object(common.VIFPortIDMixin, '_clear_vif_from_port_like_obj') @mock.patch.object(neutron_common, 'unbind_neutron_port', autospec=True) @mock.patch.object(common.VIFPortIDMixin, '_get_port_like_obj_by_vif_id') def test_vif_detach_portgroup(self, mock_get, mock_unp, mock_clear): pg = obj_utils.create_test_portgroup( self.context, node_id=self.node.id) mock_get.return_value = pg with task_manager.acquire(self.context, self.node.id) as task: self.interface.vif_detach(task, 'fake_vif_id') mock_get.assert_called_once_with(task, 'fake_vif_id') self.assertFalse(mock_unp.called) mock_clear.assert_called_once_with(pg)
[docs] @mock.patch.object(common.VIFPortIDMixin, '_clear_vif_from_port_like_obj') @mock.patch.object(neutron_common, 'unbind_neutron_port', autospec=True) @mock.patch.object(common.VIFPortIDMixin, '_get_port_like_obj_by_vif_id') def test_vif_detach_not_attached(self, mock_get, mock_unp, mock_clear): mock_get.side_effect = exception.VifNotAttached(vif='fake-vif', node='fake-node') with task_manager.acquire(self.context, self.node.id) as task: self.assertRaisesRegex( exception.VifNotAttached, "it is not attached to it.", self.interface.vif_detach, task, 'fake_vif_id') mock_get.assert_called_once_with(task, 'fake_vif_id') self.assertFalse(mock_unp.called) self.assertFalse(mock_clear.called)
[docs] @mock.patch.object(common.VIFPortIDMixin, '_clear_vif_from_port_like_obj') @mock.patch.object(neutron_common, 'unbind_neutron_port', autospec=True) @mock.patch.object(common.VIFPortIDMixin, '_get_port_like_obj_by_vif_id') def test_vif_detach_active_node(self, mock_get, mock_unp, mock_clear): self.node.provision_state = states.ACTIVE self.node.save() mock_get.return_value = self.port with task_manager.acquire(self.context, self.node.id) as task: self.interface.vif_detach(task, 'fake_vif_id') mock_get.assert_called_once_with(task, 'fake_vif_id') mock_clear.assert_called_once_with(self.port) mock_unp.assert_called_once_with('fake_vif_id')
[docs] @mock.patch.object(common.VIFPortIDMixin, '_clear_vif_from_port_like_obj') @mock.patch.object(neutron_common, 'unbind_neutron_port', autospec=True) @mock.patch.object(common.VIFPortIDMixin, '_get_port_like_obj_by_vif_id') def test_vif_detach_active_node_failure(self, mock_get, mock_unp, mock_clear): self.node.provision_state = states.ACTIVE self.node.save() mock_get.return_value = self.port mock_unp.side_effect = exception.NetworkError with task_manager.acquire(self.context, self.node.id) as task: self.assertRaises(exception.NetworkError, self.interface.vif_detach, task, 'fake_vif_id') mock_get.assert_called_once_with(task, 'fake_vif_id') mock_clear.assert_called_once_with(self.port) mock_unp.assert_called_once_with('fake_vif_id')
[docs] @mock.patch.object(common_utils, 'warn_about_deprecated_extra_vif_port_id', autospec=True) @mock.patch.object(neutron_common, 'update_port_address', autospec=True) def test_port_changed_address(self, mac_update_mock, mock_warn): new_address = '11:22:33:44:55:bb' self.port.address = new_address with task_manager.acquire(self.context, self.node.id) as task: self.interface.port_changed(task, self.port) mac_update_mock.assert_called_once_with( self.port.extra['vif_port_id'], new_address) self.assertFalse(mock_warn.called)
[docs] @mock.patch.object(neutron_common, 'update_port_address', autospec=True) def test_port_changed_address_VIF_MAC_update_fail(self, mac_update_mock): new_address = '11:22:33:44:55:bb' self.port.address = new_address mac_update_mock.side_effect = ( exception.FailedToUpdateMacOnPort(port_id=self.port.uuid)) with task_manager.acquire(self.context, self.node.id) as task: self.assertRaises(exception.FailedToUpdateMacOnPort, self.interface.port_changed, task, self.port) mac_update_mock.assert_called_once_with( self.port.extra['vif_port_id'], new_address)
[docs] @mock.patch.object(neutron_common, 'update_port_address', autospec=True) def test_port_changed_address_no_vif_id(self, mac_update_mock): self.port.extra = {} self.port.save() self.port.address = '11:22:33:44:55:bb' with task_manager.acquire(self.context, self.node.id) as task: self.interface.port_changed(task, self.port) self.assertFalse(mac_update_mock.called)
[docs] @mock.patch('ironic.dhcp.neutron.NeutronDHCPApi.update_port_dhcp_opts') def test_port_changed_client_id(self, dhcp_update_mock): expected_extra = {'vif_port_id': 'fake-id', 'client-id': 'fake2'} expected_dhcp_opts = [{'opt_name': '61', 'opt_value': 'fake2'}] self.port.extra = expected_extra with task_manager.acquire(self.context, self.node.id) as task: self.interface.port_changed(task, self.port) dhcp_update_mock.assert_called_once_with( 'fake-id', expected_dhcp_opts)
[docs] @mock.patch.object(common_utils, 'warn_about_deprecated_extra_vif_port_id', autospec=True) @mock.patch('ironic.dhcp.neutron.NeutronDHCPApi.update_port_dhcp_opts') def test_port_changed_vif(self, dhcp_update_mock, mock_warn): expected_extra = {'vif_port_id': 'new_ake-id', 'client-id': 'fake1'} self.port.extra = expected_extra with task_manager.acquire(self.context, self.node.id) as task: self.interface.port_changed(task, self.port) self.assertFalse(dhcp_update_mock.called) self.assertEqual(1, mock_warn.call_count)
[docs] @mock.patch.object(common_utils, 'warn_about_deprecated_extra_vif_port_id', autospec=True) @mock.patch('ironic.dhcp.neutron.NeutronDHCPApi.update_port_dhcp_opts') def test_port_changed_extra_add_new_key(self, dhcp_update_mock, mock_warn): self.port.extra = {'vif_port_id': 'fake-id'} self.port.save() expected_extra = self.port.extra expected_extra['foo'] = 'bar' self.port.extra = expected_extra with task_manager.acquire(self.context, self.node.id) as task: self.interface.port_changed(task, self.port) self.assertFalse(dhcp_update_mock.called) self.assertEqual(0, mock_warn.call_count)
[docs] @mock.patch.object(common_utils, 'warn_about_deprecated_extra_vif_port_id', autospec=True) @mock.patch('ironic.dhcp.neutron.NeutronDHCPApi.update_port_dhcp_opts') def test_port_changed_extra_no_deprecation_if_removing_vif( self, dhcp_update_mock, mock_warn): self.port.extra = {'vif_port_id': 'foo'} self.port.save() self.port.extra = {'foo': 'bar'} with task_manager.acquire(self.context, self.node.id) as task: self.interface.port_changed(task, self.port) self.assertFalse(dhcp_update_mock.called) self.assertEqual(0, mock_warn.call_count)
[docs] @mock.patch('ironic.dhcp.neutron.NeutronDHCPApi.update_port_dhcp_opts') def test_port_changed_client_id_fail(self, dhcp_update_mock): self.port.extra = {'vif_port_id': 'fake-id', 'client-id': 'fake2'} dhcp_update_mock.side_effect = ( exception.FailedToUpdateDHCPOptOnPort(port_id=self.port.uuid)) with task_manager.acquire(self.context, self.node.id) as task: self.assertRaises(exception.FailedToUpdateDHCPOptOnPort, self.interface.port_changed, task, self.port)
[docs] @mock.patch('ironic.dhcp.neutron.NeutronDHCPApi.update_port_dhcp_opts') def test_port_changed_client_id_no_vif_id(self, dhcp_update_mock): self.port.extra = {'client-id': 'fake1'} self.port.save() self.port.extra = {'client-id': 'fake2'} with task_manager.acquire(self.context, self.node.id) as task: self.interface.port_changed(task, self.port) self.assertFalse(dhcp_update_mock.called)
[docs] @mock.patch('ironic.dhcp.neutron.NeutronDHCPApi.update_port_dhcp_opts') def test_port_changed_message_format_failure(self, dhcp_update_mock): pg = obj_utils.create_test_portgroup( self.context, node_id=self.node.id, standalone_ports_supported=False) port = obj_utils.create_test_port(self.context, node_id=self.node.id, uuid=uuidutils.generate_uuid(), address="aa:bb:cc:dd:ee:01", extra={'vif_port_id': 'blah'}, pxe_enabled=False) port.portgroup_id = pg.id with task_manager.acquire(self.context, self.node.id) as task: self.assertRaisesRegex(exception.Conflict, "VIF blah is attached to the port", self.interface.port_changed, task, port)
def _test_port_changed(self, has_vif=False, in_portgroup=False, pxe_enabled=True, standalone_ports=True, expect_errors=False): pg = obj_utils.create_test_portgroup( self.context, node_id=self.node.id, standalone_ports_supported=standalone_ports) extra_vif = {'vif_port_id': uuidutils.generate_uuid()} if has_vif: extra = extra_vif opposite_extra = {} else: extra = {} opposite_extra = extra_vif opposite_pxe_enabled = not pxe_enabled pg_id = None if in_portgroup: pg_id = pg.id ports = [] # Update only portgroup id on existed port with different # combinations of pxe_enabled/vif_port_id p1 = obj_utils.create_test_port(self.context, node_id=self.node.id, uuid=uuidutils.generate_uuid(), address="aa:bb:cc:dd:ee:01", extra=extra, pxe_enabled=pxe_enabled) p1.portgroup_id = pg_id ports.append(p1) # Update portgroup_id/pxe_enabled/vif_port_id in one request p2 = obj_utils.create_test_port(self.context, node_id=self.node.id, uuid=uuidutils.generate_uuid(), address="aa:bb:cc:dd:ee:02", extra=opposite_extra, pxe_enabled=opposite_pxe_enabled) p2.extra = extra p2.pxe_enabled = pxe_enabled p2.portgroup_id = pg_id ports.append(p2) # Update portgroup_id and pxe_enabled p3 = obj_utils.create_test_port(self.context, node_id=self.node.id, uuid=uuidutils.generate_uuid(), address="aa:bb:cc:dd:ee:03", extra=extra, pxe_enabled=opposite_pxe_enabled) p3.pxe_enabled = pxe_enabled p3.portgroup_id = pg_id ports.append(p3) # Update portgroup_id and vif_port_id p4 = obj_utils.create_test_port(self.context, node_id=self.node.id, uuid=uuidutils.generate_uuid(), address="aa:bb:cc:dd:ee:04", pxe_enabled=pxe_enabled, extra=opposite_extra) p4.extra = extra p4.portgroup_id = pg_id ports.append(p4) for port in ports: with task_manager.acquire(self.context, self.node.id) as task: if not expect_errors: self.interface.port_changed(task, port) else: self.assertRaises(exception.Conflict, self.interface.port_changed, task, port)
[docs] def test_port_changed_novif_pxe_noportgroup(self): self._test_port_changed(has_vif=False, in_portgroup=False, pxe_enabled=True, expect_errors=False)
[docs] def test_port_changed_novif_nopxe_noportgroup(self): self._test_port_changed(has_vif=False, in_portgroup=False, pxe_enabled=False, expect_errors=False)
[docs] def test_port_changed_vif_pxe_noportgroup(self): self._test_port_changed(has_vif=True, in_portgroup=False, pxe_enabled=True, expect_errors=False)
[docs] def test_port_changed_vif_nopxe_noportgroup(self): self._test_port_changed(has_vif=True, in_portgroup=False, pxe_enabled=False, expect_errors=False)
[docs] def test_port_changed_novif_pxe_portgroup_standalone_ports(self): self._test_port_changed(has_vif=False, in_portgroup=True, pxe_enabled=True, standalone_ports=True, expect_errors=False)
[docs] def test_port_changed_novif_pxe_portgroup_nostandalone_ports(self): self._test_port_changed(has_vif=False, in_portgroup=True, pxe_enabled=True, standalone_ports=False, expect_errors=True)
[docs] def test_port_changed_novif_nopxe_portgroup_standalone_ports(self): self._test_port_changed(has_vif=False, in_portgroup=True, pxe_enabled=False, standalone_ports=True, expect_errors=False)
[docs] def test_port_changed_novif_nopxe_portgroup_nostandalone_ports(self): self._test_port_changed(has_vif=False, in_portgroup=True, pxe_enabled=False, standalone_ports=False, expect_errors=False)
[docs] def test_port_changed_vif_pxe_portgroup_standalone_ports(self): self._test_port_changed(has_vif=True, in_portgroup=True, pxe_enabled=True, standalone_ports=True, expect_errors=False)
[docs] def test_port_changed_vif_pxe_portgroup_nostandalone_ports(self): self._test_port_changed(has_vif=True, in_portgroup=True, pxe_enabled=True, standalone_ports=False, expect_errors=True)
[docs] def test_port_changed_vif_nopxe_portgroup_standalone_ports(self): self._test_port_changed(has_vif=True, in_portgroup=True, pxe_enabled=True, standalone_ports=True, expect_errors=False)
[docs] def test_port_changed_vif_nopxe_portgroup_nostandalone_ports(self): self._test_port_changed(has_vif=True, in_portgroup=True, pxe_enabled=False, standalone_ports=False, expect_errors=True)
[docs] @mock.patch.object(neutron_common, 'update_port_address', autospec=True) def test_update_portgroup_address(self, mac_update_mock): pg = obj_utils.create_test_portgroup( self.context, node_id=self.node.id, extra={'vif_port_id': 'fake-id'}) new_address = '11:22:33:44:55:bb' pg.address = new_address with task_manager.acquire(self.context, self.node.id) as task: self.interface.portgroup_changed(task, pg) mac_update_mock.assert_called_once_with('fake-id', new_address)
[docs] @mock.patch.object(neutron_common, 'update_port_address', autospec=True) def test_update_portgroup_remove_address(self, mac_update_mock): pg = obj_utils.create_test_portgroup( self.context, node_id=self.node.id, extra={'vif_port_id': 'fake-id'}) pg.address = None with task_manager.acquire(self.context, self.node.id) as task: self.interface.portgroup_changed(task, pg) self.assertFalse(mac_update_mock.called)
[docs] @mock.patch.object(common_utils, 'warn_about_deprecated_extra_vif_port_id', autospec=True) @mock.patch.object(neutron_common, 'update_port_address', autospec=True) def test_update_portgroup_vif(self, mac_update_mock, mock_warn): pg = obj_utils.create_test_portgroup( self.context, node_id=self.node.id) extra = {'vif_port_id': 'foo'} pg.extra = extra with task_manager.acquire(self.context, self.node.id) as task: self.interface.portgroup_changed(task, pg) self.assertFalse(mac_update_mock.called) self.assertEqual(1, mock_warn.call_count)
[docs] @mock.patch.object(common_utils, 'warn_about_deprecated_extra_vif_port_id', autospec=True) @mock.patch.object(neutron_common, 'update_port_address', autospec=True) def test_update_portgroup_vif_removal_no_deprecation(self, mac_update_mock, mock_warn): pg = obj_utils.create_test_portgroup( self.context, node_id=self.node.id, extra={'vif_port_id': 'foo'}) pg.extra = {} with task_manager.acquire(self.context, self.node.id) as task: self.interface.portgroup_changed(task, pg) self.assertFalse(mac_update_mock.called) self.assertFalse(mock_warn.called)
[docs] @mock.patch.object(common_utils, 'warn_about_deprecated_extra_vif_port_id', autospec=True) @mock.patch.object(neutron_common, 'update_port_address', autospec=True) def test_update_portgroup_extra_new_key(self, mac_update_mock, mock_warn): pg = obj_utils.create_test_portgroup( self.context, node_id=self.node.id, extra={'vif_port_id': 'vif-id'}) expected_extra = pg.extra expected_extra['foo'] = 'bar' pg.extra = expected_extra with task_manager.acquire(self.context, self.node.id) as task: self.interface.portgroup_changed(task, pg) self.assertFalse(mac_update_mock.called) self.assertFalse(mock_warn.called) self.assertEqual(expected_extra, pg.extra)
[docs] @mock.patch.object(neutron_common, 'update_port_address', autospec=True) def test_update_portgroup_address_fail(self, mac_update_mock): pg = obj_utils.create_test_portgroup( self.context, node_id=self.node.id, extra={'vif_port_id': 'fake-id'}) new_address = '11:22:33:44:55:bb' pg.address = new_address mac_update_mock.side_effect = ( exception.FailedToUpdateMacOnPort('boom')) with task_manager.acquire(self.context, self.node.id) as task: self.assertRaises(exception.FailedToUpdateMacOnPort, self.interface.portgroup_changed, task, pg) mac_update_mock.assert_called_once_with('fake-id', new_address)
[docs] @mock.patch.object(common_utils, 'warn_about_deprecated_extra_vif_port_id', autospec=True) @mock.patch.object(neutron_common, 'update_port_address', autospec=True) def test_update_portgroup_address_no_vif(self, mac_update_mock, mock_warn): pg = obj_utils.create_test_portgroup( self.context, node_id=self.node.id) new_address = '11:22:33:44:55:bb' pg.address = new_address with task_manager.acquire(self.context, self.node.id) as task: self.interface.portgroup_changed(task, pg) self.assertEqual(new_address, pg.address) self.assertFalse(mac_update_mock.called) self.assertFalse(mock_warn.called)
[docs] @mock.patch.object(neutron_common, 'update_port_address', autospec=True) def test_update_portgroup_nostandalone_ports_pxe_ports_exc( self, mac_update_mock): pg = obj_utils.create_test_portgroup( self.context, node_id=self.node.id) extra = {'vif_port_id': 'foo'} obj_utils.create_test_port( self.context, node_id=self.node.id, extra=extra, pxe_enabled=True, portgroup_id=pg.id, address="aa:bb:cc:dd:ee:01", uuid=uuidutils.generate_uuid()) pg.standalone_ports_supported = False with task_manager.acquire(self.context, self.node.id) as task: self.assertRaisesRegex(exception.Conflict, "VIF foo is attached to this port", self.interface.portgroup_changed, task, pg)
def _test_update_portgroup(self, has_vif=False, with_ports=False, pxe_enabled=True, standalone_ports=True, expect_errors=False): # NOTE(vsaienko) make sure that old values are opposite to new, # to guarantee that object.what_changes() returns true. old_standalone_ports_supported = not standalone_ports pg = obj_utils.create_test_portgroup( self.context, node_id=self.node.id, standalone_ports_supported=old_standalone_ports_supported) if with_ports: extra = {} if has_vif: extra = {'vif_port_id': uuidutils.generate_uuid()} obj_utils.create_test_port( self.context, node_id=self.node.id, extra=extra, pxe_enabled=pxe_enabled, portgroup_id=pg.id, address="aa:bb:cc:dd:ee:01", uuid=uuidutils.generate_uuid()) pg.standalone_ports_supported = standalone_ports with task_manager.acquire(self.context, self.node.id) as task: if not expect_errors: self.interface.portgroup_changed(task, pg) else: self.assertRaises(exception.Conflict, self.interface.portgroup_changed, task, pg)
[docs] def test_update_portgroup_standalone_ports_noports(self): self._test_update_portgroup(with_ports=False, standalone_ports=True, expect_errors=False)
[docs] def test_update_portgroup_standalone_ports_novif_pxe_ports(self): self._test_update_portgroup(with_ports=True, standalone_ports=True, has_vif=False, pxe_enabled=True, expect_errors=False)
[docs] def test_update_portgroup_nostandalone_ports_novif_pxe_ports(self): self._test_update_portgroup(with_ports=True, standalone_ports=False, has_vif=False, pxe_enabled=True, expect_errors=True)
[docs] def test_update_portgroup_nostandalone_ports_novif_nopxe_ports(self): self._test_update_portgroup(with_ports=True, standalone_ports=False, has_vif=False, pxe_enabled=False, expect_errors=False)
[docs] def test_update_portgroup_standalone_ports_novif_nopxe_ports(self): self._test_update_portgroup(with_ports=True, standalone_ports=True, has_vif=False, pxe_enabled=False, expect_errors=False)
[docs] def test_update_portgroup_standalone_ports_vif_pxe_ports(self): self._test_update_portgroup(with_ports=True, standalone_ports=True, has_vif=True, pxe_enabled=True, expect_errors=False)
[docs] def test_update_portgroup_nostandalone_ports_vif_pxe_ports(self): self._test_update_portgroup(with_ports=True, standalone_ports=False, has_vif=True, pxe_enabled=True, expect_errors=True)
[docs] def test_update_portgroup_standalone_ports_vif_nopxe_ports(self): self._test_update_portgroup(with_ports=True, standalone_ports=True, has_vif=True, pxe_enabled=False, expect_errors=False)
[docs] def test_update_portgroup_nostandalone_ports_vif_nopxe_ports(self): self._test_update_portgroup(with_ports=True, standalone_ports=False, has_vif=True, pxe_enabled=False, expect_errors=True)
Creative Commons Attribution 3.0 License

Except where otherwise noted, this document is licensed under Creative Commons Attribution 3.0 License. See all OpenStack Legal Documents.