device = port_detail['device']
# Do not bind a port if it's already bound
cur_tag = tags_by_name.get(port.port_name)
+ if cur_tag != lvm.vlan:
+ self.int_br.delete_flows(in_port=port.ofport)
+ if self.prevent_arp_spoofing:
+ self.setup_arp_spoofing_protection(self.int_br,
+ port, port_detail)
if cur_tag != lvm.vlan:
self.int_br.set_db_attribute(
"Port", port.port_name, "tag", lvm.vlan)
- if port.ofport != -1:
- # NOTE(yamamoto): Remove possible drop_port flow
- # installed by port_dead.
- self.int_br.delete_flows(in_port=port.ofport)
# update plugin about port status
# FIXME(salv-orlando): Failures while updating device status
# ofport-based rules, so make arp_spoofing protection a conditional
# until something else uses ofport
if not self.prevent_arp_spoofing:
- return
+ return []
previous = self.vifname_to_ofport_map
current = self.int_br.get_vif_port_to_ofport_map()
# if any ofport numbers have changed, re-process the devices as
# added ports so any rules based on ofport numbers are updated.
moved_ports = self._get_ofport_moves(current, previous)
- if moved_ports:
- self.treat_devices_added_or_updated(moved_ports,
- ovs_restarted=False)
# delete any stale rules based on removed ofports
ofports_deleted = set(previous.values()) - set(current.values())
# store map for next iteration
self.vifname_to_ofport_map = current
+ return moved_ports
@staticmethod
def _get_ofport_moves(current, previous):
details['fixed_ips'],
details['device_owner'],
ovs_restarted)
- if self.prevent_arp_spoofing:
- self.setup_arp_spoofing_protection(self.int_br,
- port, details)
if need_binding:
details['vif_port'] = port
need_binding_devices.append(details)
reg_ports = (set() if ovs_restarted else ports)
port_info = self.scan_ports(reg_ports, updated_ports_copy)
self.process_deleted_ports(port_info)
- self.update_stale_ofport_rules()
+ ofport_changed_ports = self.update_stale_ofport_rules()
+ if ofport_changed_ports:
+ port_info.setdefault('updated', set()).update(
+ ofport_changed_ports)
LOG.debug("Agent rpc_loop - iteration:%(iter_num)d - "
"port information retrieved. "
"Elapsed:%(elapsed).3f",
from neutron.plugins.common import constants as p_const
from neutron.plugins.ml2.drivers.l2pop import rpc as l2pop_rpc
from neutron.plugins.ml2.drivers.openvswitch.agent.common import constants
+from neutron.plugins.ml2.drivers.openvswitch.agent import ovs_neutron_agent \
+ as ovs_agent
from neutron.tests.unit.plugins.ml2.drivers.openvswitch.agent \
import ovs_test_base
devices_down,
mock.ANY, mock.ANY)
+ def _test_arp_spoofing(self, enable_prevent_arp_spoofing):
+ self.agent.prevent_arp_spoofing = enable_prevent_arp_spoofing
+
+ ovs_db_list = [{'name': 'fake_device', 'tag': []}]
+ self.agent.local_vlan_map = {
+ 'fake_network': ovs_agent.LocalVLANMapping(1, None, None, 1)}
+ vif_port = mock.Mock()
+ vif_port.port_name = 'fake_device'
+ vif_port.ofport = 1
+ need_binding_ports = [{'network_id': 'fake_network',
+ 'vif_port': vif_port,
+ 'device': 'fake_device',
+ 'admin_state_up': True}]
+ with mock.patch.object(
+ self.agent.plugin_rpc, 'update_device_list',
+ return_value={'devices_up': [],
+ 'devices_down': [],
+ 'failed_devices_up': [],
+ 'failed_devices_down': []}), \
+ mock.patch.object(self.agent,
+ 'int_br') as int_br, \
+ mock.patch.object(
+ self.agent,
+ 'setup_arp_spoofing_protection') as setup_arp:
+ int_br.db_list.return_value = ovs_db_list
+ self.agent._bind_devices(need_binding_ports)
+ self.assertEqual(enable_prevent_arp_spoofing, setup_arp.called)
+
+ def test_setup_arp_spoofing_protection_enable(self):
+ self._test_arp_spoofing(True)
+
+ def test_setup_arp_spoofing_protection_disabled(self):
+ self._test_arp_spoofing(False)
+
def _mock_treat_devices_added_updated(self, details, port, func_name):
"""Mock treat devices added or updated.
self.agent._handle_sigterm(None, None)
self.assertFalse(mock_set_rpc.called)
- def test_arp_spoofing_disabled(self):
- self.agent.prevent_arp_spoofing = False
- # all of this is required just to get to the part of
- # treat_devices_added_or_updated that checks the prevent_arp_spoofing
- # flag
- self.agent.int_br = mock.create_autospec(self.agent.int_br)
- self.agent.treat_vif_port = mock.Mock()
- self.agent.get_vif_port_by_id = mock.Mock(return_value=FakeVif())
- self.agent.plugin_rpc = mock.Mock()
- plist = [{a: a for a in ('port_id', 'network_id', 'network_type',
- 'physical_network', 'segmentation_id',
- 'admin_state_up', 'fixed_ips', 'device',
- 'device_owner')}]
- (self.agent.plugin_rpc.get_devices_details_list_and_failed_devices.
- return_value) = {'devices': plist, 'failed_devices': []}
- self.agent.plugin_rpc.update_device_list.return_value = {
- 'devices_up': plist, 'devices_down': [], 'failed_devices_up': [],
- 'failed_devices_down': []}
- self.agent.setup_arp_spoofing_protection = mock.Mock()
- self.agent.treat_devices_added_or_updated([], False)
- self.assertFalse(self.agent.setup_arp_spoofing_protection.called)
-
def test_arp_spoofing_port_security_disabled(self):
int_br = mock.create_autospec(self.agent.int_br)
self.agent.setup_arp_spoofing_protection(
# simulate port1 was moved
newmap = {'port2': 2, 'port1': 90}
self.agent.int_br.get_vif_port_to_ofport_map.return_value = newmap
- self.agent.update_stale_ofport_rules()
- self.agent.treat_devices_added_or_updated.assert_called_with(
- ['port1'], ovs_restarted=False)
+ ofport_changed_ports = self.agent.update_stale_ofport_rules()
+ self.assertEqual(['port1'], ofport_changed_ports)
def test__setup_tunnel_port_while_new_mapping_is_added(self):
"""
log_exception.side_effect = Exception(
'Fake exception to get out of the loop')
scan_ports.side_effect = [reply2, reply3]
+ update_stale.return_value = []
process_network_ports.side_effect = [
False, Exception('Fake exception to get out of the loop')]
])
process_network_ports.assert_has_calls([
mock.call({'current': set(['tap0']),
- 'removed': set([]),
- 'added': set(['tap2'])}, False),
+ 'removed': set([]),
+ 'added': set(['tap2'])}, False),
mock.call({'current': set(['tap2']),
- 'removed': set(['tap0']),
- 'added': set([])}, False)
+ 'removed': set(['tap0']),
+ 'added': set([])}, False)
])
self.assertTrue(update_stale.called)
self._verify_mock_calls()
Code Review / openstack-build / neutron-build.git / commitdiff