text: # Copyright (c) 2012 OpenStack Foundation.
  #
  #    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 contextlib
  import sys
  import time
  from unittest import mock
  
  import netaddr
  from neutron_lib.agent import constants as agent_consts
  from neutron_lib.api.definitions import portbindings
  from neutron_lib.api.definitions import provider_net
  from neutron_lib import constants as n_const
  from neutron_lib import rpc as n_rpc
  import os_vif
  from os_vif.objects import instance_info as vif_instance_object
  from oslo_config import cfg
  from oslo_log import log
  import oslo_messaging
  import testtools
  
  from neutron._i18n import _
  from neutron.agent.common import async_process
  from neutron.agent.common import ovs_lib
  from neutron.agent.common import polling
  from neutron.agent.common import utils
  from neutron.agent.linux import ip_lib
  from neutron.agent.linux import utils as linux_utils
  from neutron.api.rpc.callbacks import resources
  from neutron.objects.ports import Port
  from neutron.objects.ports import PortBinding
  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 import base
  from neutron.tests.unit.plugins.ml2.drivers.openvswitch.agent \
      import ovs_test_base
  from neutron.tests.unit.plugins.ml2.drivers.openvswitch.agent \
      import test_vlanmanager
  
  
  NOTIFIER = 'neutron.plugins.ml2.rpc.AgentNotifierApi'
  PULLAPI = 'neutron.api.rpc.handlers.resources_rpc.ResourcesPullRpcApi'
  OVS_LINUX_KERN_VERS_WITHOUT_VXLAN = "3.12.0"
  
  FAKE_MAC = '00:11:22:33:44:55'
  FAKE_IP1 = '10.0.0.1'
  FAKE_IP2 = '10.0.0.2'
  FAKE_IP6 = '2001:db8:42:42::10'
  
  TEST_PORT_ID1 = 'port-id-1'
  TEST_PORT_ID2 = 'port-id-2'
  TEST_PORT_ID3 = 'port-id-3'
  
  TEST_NETWORK_ID1 = 'net-id-1'
  TEST_NETWORK_ID2 = 'net-id-2'
  
  DEVICE_OWNER_COMPUTE = n_const.DEVICE_OWNER_COMPUTE_PREFIX + 'fake'
  
  
  class FakeVif(object):
      ofport = 99
      port_name = 'name'
      vif_mac = 'aa:bb:cc:11:22:33'
  
  
  class MockFixedIntervalLoopingCall(object):
      def __init__(self, f):
          self.f = f
  
      def start(self, interval=0):
          self.f()
  
  
  class ValidateTunnelTypes(ovs_test_base.OVSAgentConfigTestBase):
  
      def setUp(self):
          super(ValidateTunnelTypes, self).setUp()
          self.mock_validate_local_ip = mock.patch.object(
              self.mod_agent, 'validate_local_ip').start()
  
      def test_validate_tunnel_types_succeeds(self):
          cfg.CONF.set_override('local_ip', '10.10.10.10', group='OVS')
          cfg.CONF.set_override('tunnel_types', [n_const.TYPE_GRE],
                                group='AGENT')
          self.mod_agent.validate_tunnel_config(cfg.CONF.AGENT.tunnel_types,
                                                cfg.CONF.OVS.local_ip)
          self.mock_validate_local_ip.assert_called_once_with('10.10.10.10')
  
      def test_validate_tunnel_types_fails_for_invalid_tunnel_type(self):
          cfg.CONF.set_override('local_ip', '10.10.10.10', group='OVS')
          cfg.CONF.set_override('tunnel_types', ['foobar'], group='AGENT')
          with testtools.ExpectedException(SystemExit):
              self.mod_agent.validate_tunnel_config(cfg.CONF.AGENT.tunnel_types,
                                                    cfg.CONF.OVS.local_ip)
  
  
  class TestOvsNeutronAgent(object):
  
      def setUp(self):
          super(TestOvsNeutronAgent, self).setUp()
          self.useFixture(test_vlanmanager.LocalVlanManagerFixture())
          mock.patch(PULLAPI).start()
          notifier_p = mock.patch(NOTIFIER)
          notifier_cls = notifier_p.start()
          self.notifier = mock.Mock()
          notifier_cls.return_value = self.notifier
          systemd_patch = mock.patch('oslo_service.systemd.notify_once')
          self.systemd_notify = systemd_patch.start()
  
          cfg.CONF.set_default('firewall_driver',
                               'neutron.agent.firewall.NoopFirewallDriver',
                               group='SECURITYGROUP')
          cfg.CONF.set_default('quitting_rpc_timeout', 10, 'AGENT')
          cfg.CONF.set_default('local_ip', '127.0.0.1', 'OVS')
          cfg.CONF.set_default('host', 'host')
          mock.patch(
              'neutron.agent.ovsdb.native.helpers.enable_connection_uri').start()
          mock.patch(
              'neutron.agent.common.ovs_lib.OVSBridge.get_ports_attributes',
              return_value=[]).start()
  
          mock.patch('neutron.agent.common.ovs_lib.BaseOVS.config',
                     new_callable=mock.PropertyMock,
                     return_value={}).start()
          mock.patch('neutron.agent.ovsdb.impl_idl._connection').start()
          self.agent = self._make_agent()
          self.agent.sg_agent = mock.Mock()
  
      def _make_agent(self):
          with mock.patch.object(self.mod_agent.OVSNeutronAgent,
                                 'setup_integration_br'),\
                  mock.patch.object(self.mod_agent.OVSNeutronAgent,
                                    'setup_ancillary_bridges',
                                    return_value=[]),\
                  mock.patch('neutron.agent.linux.ip_lib.get_device_mac',
                             return_value='00:00:00:00:00:01'),\
                  mock.patch(
                      'neutron.agent.common.ovs_lib.BaseOVS.get_bridges'),\
                  mock.patch('oslo_service.loopingcall.FixedIntervalLoopingCall',
                             new=MockFixedIntervalLoopingCall),\
                  mock.patch(
                      'neutron.agent.common.ovs_lib.OVSBridge.' 'get_vif_ports',
                      return_value=[]),\
                  mock.patch('neutron.agent.rpc.PluginReportStateAPI.'
                             'has_alive_neutron_server'):
              ext_manager = mock.Mock()
              agent = self.mod_agent.OVSNeutronAgent(self._bridge_classes(),
                                                     ext_manager, cfg.CONF)
              agent.tun_br = self.br_tun_cls(br_name='br-tun')
              return agent
  
      def _mock_port_bound(self, ofport=None, new_local_vlan=None,
                           old_local_vlan=None, db_get_val=None):
          port = mock.Mock()
          port.ofport = ofport
          net_uuid = 'my-net-uuid'
          fixed_ips = [{'subnet_id': 'my-subnet-uuid',
                        'ip_address': '1.1.1.1'}]
          if old_local_vlan is not None:
              self.agent.vlan_manager.add(
                  net_uuid, old_local_vlan, None, None, None)
          with mock.patch.object(self.agent, 'int_br', autospec=True) as int_br:
              int_br.db_get_val.return_value = db_get_val
              int_br.set_db_attribute.return_value = True
              needs_binding = self.agent.port_bound(
                  port, net_uuid, 'local', None, None,
                  fixed_ips, DEVICE_OWNER_COMPUTE, False)
          if db_get_val is None:
              int_br.assert_not_called()
              self.assertFalse(needs_binding)
          else:
              vlan_mapping = {'net_uuid': net_uuid,
                              'network_type': 'local',
                              'physical_network': 'None'}
              int_br.set_db_attribute.assert_called_once_with(
                  "Port", mock.ANY, "other_config", vlan_mapping)
              self.assertTrue(needs_binding)
  
      def test_setup_physical_bridges_during_agent_initialization(self):
          with mock.patch.object(
              self.mod_agent.OVSNeutronAgent,
              'setup_physical_bridges') as setup_physical_bridges,\
                  mock.patch.object(
                      self.mod_agent.OVSNeutronAgent, 'setup_rpc') as setup_rpc:
              setup_rpc.side_effect = oslo_messaging.MessagingException(
                  "Test communication failure")
              try:
                  self._make_agent()
              except oslo_messaging.MessagingException:
                  pass
              setup_physical_bridges.assert_called_once_with(mock.ANY)
  
      def test_datapath_type_system(self):
          # verify kernel datapath is default
          expected = constants.OVS_DATAPATH_SYSTEM
          self.assertEqual(expected, self.agent.int_br.datapath_type)
  
      def test_datapath_type_netdev(self):
  
          with mock.patch.object(self.mod_agent.OVSNeutronAgent,
                                 'setup_integration_br'), \
              mock.patch.object(self.mod_agent.OVSNeutronAgent,
                             'setup_ancillary_bridges',
                             return_value=[]), \
              mock.patch('neutron.agent.linux.ip_lib.get_device_mac',
                      return_value='00:00:00:00:00:01'), \
              mock.patch(
                  'neutron.agent.common.ovs_lib.BaseOVS.get_bridges'), \
              mock.patch('oslo_service.loopingcall.FixedIntervalLoopingCall',
                         new=MockFixedIntervalLoopingCall), \
              mock.patch(
                  'neutron.agent.common.ovs_lib.OVSBridge.' 'get_vif_ports',
                  return_value=[]), \
              mock.patch('neutron.agent.common.ovs_lib.BaseOVS.config',
                         new_callable=mock.PropertyMock,
                         return_value={'datapath_types': ['netdev']}),\
              mock.patch('neutron.agent.rpc.PluginReportStateAPI.'
                         'has_alive_neutron_server'):
              # validate setting non default datapath
              expected = constants.OVS_DATAPATH_NETDEV
              cfg.CONF.set_override('datapath_type',
                                    expected,
                                    group='OVS')
              ext_manager = mock.Mock()
              self.agent = self.mod_agent.OVSNeutronAgent(self._bridge_classes(),
                                                          ext_manager, cfg.CONF)
              self.assertEqual(expected, self.agent.int_br.datapath_type)
  
      def test_agent_type_ovs(self):
          # verify agent_type is default
          expected = n_const.AGENT_TYPE_OVS
          self.assertEqual(expected,
                           self.agent.agent_state['agent_type'])
  
      def test_agent_available_local_vlans(self):
          expected = [n_const.MIN_VLAN_TAG,
                      n_const.MIN_VLAN_TAG + 1,
                      n_const.MAX_VLAN_TAG - 1,
                      n_const.MAX_VLAN_TAG]
          exception = [n_const.MIN_VLAN_TAG - 1,
                       n_const.MAX_VLAN_TAG + 1,
                       n_const.MAX_VLAN_TAG + 2]
          available_vlan = self.agent.available_local_vlans
          for tag in expected:
              self.assertIn(tag, available_vlan)
          for tag in exception:
              self.assertNotIn(tag, available_vlan)
  
      def _test_restore_local_vlan_maps(self, tag, segmentation_id='1'):
          port = mock.Mock()
          port.port_name = 'fake_port'
          net_uuid = 'fake_network_id'
          local_vlan_map = {'net_uuid': net_uuid,
                            'network_type': 'vlan',
                            'physical_network': 'fake_network'}
          if segmentation_id is not None:
              local_vlan_map['segmentation_id'] = segmentation_id
  
          # this is for the call inside get_vif_ports()
          get_interfaces = [{'name': port.port_name,
                             'ofport': '1',
                             'external_ids': {
                                 'iface-id': '1',
                                 'attached-mac': 'mac1'}},
                            {'name': 'invalid',
                             'ofport': ovs_lib.INVALID_OFPORT,
                             'external_ids': {
                                 'iface-id': '2',
                                 'attached-mac': 'mac2'}},
                            {'name': 'unassigned',
                             'ofport': ovs_lib.UNASSIGNED_OFPORT,
                             'external_ids': {
                                 'iface-id': '3',
                                 'attached-mac': 'mac3'}}]
          # this is for the call inside _restore_local_vlan_map()
          get_ports = [{'name': port.port_name,
                        'other_config': local_vlan_map,
                        'tag': tag}]
  
          with mock.patch.object(self.agent.int_br,
                                 'get_ports_attributes',
                                 side_effect=[get_interfaces, get_ports]) as gpa:
              self.agent._restore_local_vlan_map()
              expected_hints = {}
              if tag:
                  expected_hints[net_uuid] = tag
              self.assertEqual(expected_hints, self.agent._local_vlan_hints)
              # make sure invalid and unassigned ports were skipped
              gpa.assert_has_calls([
                  mock.call('Interface', columns=mock.ANY, if_exists=True),
                  mock.call('Port', columns=mock.ANY, ports=['fake_port'])
              ])
  
      def test_restore_local_vlan_map_with_device_has_tag(self):
          self._test_restore_local_vlan_maps(2)
  
      def test_restore_local_vlan_map_with_device_no_tag(self):
          self._test_restore_local_vlan_maps([])
  
      def test_restore_local_vlan_map_no_segmentation_id(self):
          self._test_restore_local_vlan_maps(2, segmentation_id=None)
  
      def test_restore_local_vlan_map_segmentation_id_compat(self):
          self._test_restore_local_vlan_maps(2, segmentation_id='None')
  
      def test_check_agent_configurations_for_dvr_raises(self):
          self.agent.enable_distributed_routing = True
          self.agent.enable_tunneling = True
          self.agent.l2_pop = False
          self.assertRaises(ValueError,
                            self.agent._check_agent_configurations)
  
      def test_check_agent_configurations_for_dvr(self):
          self.agent.enable_distributed_routing = True
          self.agent.enable_tunneling = True
          self.agent.l2_pop = True
          self.assertIsNone(self.agent._check_agent_configurations())
  
      def test_check_agent_configurations_for_dvr_with_vlan(self):
          self.agent.enable_distributed_routing = True
          self.agent.enable_tunneling = False
          self.agent.l2_pop = False
          self.assertIsNone(self.agent._check_agent_configurations())
  
      def test_port_bound_deletes_flows_for_valid_ofport(self):
          self._mock_port_bound(ofport=1, new_local_vlan=1, db_get_val={})
  
      def test_port_bound_ignores_flows_for_invalid_ofport(self):
          self._mock_port_bound(ofport=-1, new_local_vlan=1, db_get_val={})
  
      def test_port_bound_does_not_rewire_if_already_bound(self):
          self._mock_port_bound(
              ofport=-1, new_local_vlan=1, old_local_vlan=1, db_get_val={})
  
      def test_port_bound_not_found(self):
          self._mock_port_bound(ofport=1, new_local_vlan=1, db_get_val=None)
  
      def _test_port_dead(self, cur_tag=None):
          port = mock.Mock()
          port.ofport = 1
          with mock.patch.object(self.agent, 'int_br') as int_br:
              int_br.db_get_val.return_value = cur_tag
              self.agent.port_dead(port)
          if cur_tag is None or cur_tag == constants.DEAD_VLAN_TAG:
              self.assertFalse(int_br.set_db_attribute.called)
              self.assertFalse(int_br.drop_port.called)
          else:
              int_br.assert_has_calls([
                  mock.call.set_db_attribute("Port", mock.ANY, "tag",
                                             constants.DEAD_VLAN_TAG,
                                             log_errors=True),
                  mock.call.drop_port(in_port=port.ofport),
              ])
  
      def test_port_dead(self):
          self._test_port_dead()
  
      def test_port_dead_with_port_already_dead(self):
          self._test_port_dead(constants.DEAD_VLAN_TAG)
  
      def test_port_dead_with_valid_tag(self):
          self._test_port_dead(cur_tag=1)
  
      def mock_scan_ports(self, vif_port_set=None, registered_ports=None,
                          updated_ports=None, port_tags_dict=None, sync=False):
          if port_tags_dict is None:  # Because empty dicts evaluate as False.
              port_tags_dict = {}
          with mock.patch.object(self.agent.int_br,
                                 'get_vif_port_set',
                                 return_value=vif_port_set),\
                  mock.patch.object(self.agent.int_br,
                                    'get_port_tag_dict',
                                    return_value=port_tags_dict):
              return self.agent.scan_ports(registered_ports, sync, updated_ports)
  
      def test_scan_ports_returns_current_only_for_unchanged_ports(self):
          vif_port_set = set([1, 3])
          registered_ports = set([1, 3])
          expected = ovs_agent.PortInfo(current=vif_port_set)
          actual = self.mock_scan_ports(vif_port_set, registered_ports)
          self.assertEqual(expected, actual)
  
      def test_scan_ports_returns_port_changes(self):
          vif_port_set = set([1, 3])
          registered_ports = set([1, 2])
          expected = ovs_agent.PortInfo(current=vif_port_set, added=set([3]),
                                        removed=set([2]))
          actual = self.mock_scan_ports(vif_port_set, registered_ports)
          self.assertEqual(expected, actual)
  
      def test_scan_ports_returns_port_changes_with_sync(self):
          vif_port_set = set([1, 3])
          registered_ports = set([1, 2])
          expected = ovs_agent.PortInfo(current=vif_port_set, added=vif_port_set,
                                        removed=set([2]))
          actual = self.mock_scan_ports(vif_port_set, registered_ports,
                                        sync=True)
          self.assertEqual(expected, actual)
  
      def _test_scan_ports_with_updated_ports(self, updated_ports):
          vif_port_set = set([1, 3, 4])
          registered_ports = set([1, 2, 4])
          expected = ovs_agent.PortInfo(current=vif_port_set, added={3},
                                        removed={2}, updated={4})
          actual = self.mock_scan_ports(vif_port_set, registered_ports,
                                        updated_ports)
          self.assertEqual(expected, actual)
  
      def test_scan_ports_finds_known_updated_ports(self):
          self._test_scan_ports_with_updated_ports(set([4]))
  
      def test_scan_ports_ignores_unknown_updated_ports(self):
          # the port '5' was not seen on current ports. Hence it has either
          # never been wired or already removed and should be ignored
          self._test_scan_ports_with_updated_ports(set([4, 5]))
  
      def test_scan_ports_ignores_updated_port_if_removed(self):
          vif_port_set = set([1, 3])
          registered_ports = set([1, 2])
          updated_ports = set([1, 2])
          expected = ovs_agent.PortInfo(current=vif_port_set, added=set([3]),
                                        removed=set([2]), updated=set([1]))
          actual = self.mock_scan_ports(vif_port_set, registered_ports,
                                        updated_ports)
          self.assertEqual(expected, actual)
  
      def test_scan_ports_no_vif_changes_returns_updated_port_only(self):
          vif_port_set = set([1, 2, 3])
          registered_ports = set([1, 2, 3])
          updated_ports = set([2])
          expected = ovs_agent.PortInfo(current=vif_port_set, updated=set([2]))
          actual = self.mock_scan_ports(vif_port_set, registered_ports,
                                        updated_ports)
          self.assertEqual(expected, actual)
  
      def _test_process_ports_events(self, events, registered_ports,
                                     ancillary_ports, expected_ports,
                                     expected_ancillary, updated_ports=None,
                                     ):
          with mock.patch.object(self.agent, 'check_changed_vlans',
                                 return_value=set()):
              devices_not_ready_yet = set()
              failed_devices = {'added': set(), 'removed': set()}
              failed_ancillary_devices = {
                  'added': set(), 'removed': set()}
              actual = self.agent.process_ports_events(
                  events, registered_ports, ancillary_ports,
                  devices_not_ready_yet, failed_devices,
                  failed_ancillary_devices, updated_ports)
              self.assertEqual(
                  (expected_ports, expected_ancillary, devices_not_ready_yet),
                  actual)
  
      def test_process_ports_events_port_removed_and_added(self):
          port_id_one = 'f6f104bd-37c7-4f7b-9d70-53a6bb42728f'
          port_id_two = 'fbaf42ef-ab63-4cda-81d2-37ee55daac3a'
          events = {
              'removed':
                  [{'ofport': 1,
                    'external_ids': {'iface-id': port_id_one,
                                     'attached-mac': 'fa:16:3e:f6:1b:fb'},
                    'name': 'qvof6f104bd-37'},
                   {'ofport': 2,
                    'external_ids': {'iface-id': port_id_two,
                                     'attached-mac': 'fa:16:3e:a4:42:6e'},
                    'name': 'qvofbaf42ef-ab'}],
              'added':
                  [{'ofport': 3,
                    'external_ids': {'iface-id': port_id_one,
                                     'attached-mac': 'fa:16:3e:f6:1b:fb'},
                    'name': 'qvof6f104bd-37'},
                   {'ofport': 4,
                    'external_ids': {'iface-id': port_id_two,
                                     'attached-mac': 'fa:16:3e:a4:42:6e'},
                    'name': 'qvofbaf42ef-ab'}],
              'modified': []
          }
          registered_ports = {port_id_one, port_id_two}
          expected_ancillary = ovs_agent.PortInfo()
  
          # port was removed and then added
          expected_ports = ovs_agent.PortInfo(
              added={port_id_one, port_id_two},
              current={port_id_one, port_id_two},
              re_added={port_id_one, port_id_two}
          )
          with mock.patch.object(ovs_lib.BaseOVS, "port_exists",
                                 return_value=True):
              self._test_process_ports_events(events.copy(), registered_ports,
                                              set(), expected_ports,
                                              expected_ancillary)
  
          # port was added and then removed
          expected_ports = ovs_agent.PortInfo(removed={port_id_one, port_id_two})
          with mock.patch.object(ovs_lib.BaseOVS, "port_exists",
                                 return_value=False):
              self._test_process_ports_events(events.copy(), registered_ports,
                                              set(), expected_ports,
                                              expected_ancillary)
  
      def test_process_ports_events_returns_current_for_unchanged_ports(self):
          events = {'added': [], 'removed': [], 'modified': []}
          registered_ports = {1, 3}
          ancillary_ports = {2, 5}
          expected_ports = ovs_agent.PortInfo(current=registered_ports)
          expected_ancillary = ovs_agent.PortInfo(current=ancillary_ports)
          self._test_process_ports_events(events, registered_ports,
                                          ancillary_ports, expected_ports,
                                          expected_ancillary)
  
      def test_process_port_events_no_vif_changes_return_updated_port_only(self):
          events = {'added': [], 'removed': [], 'modified': []}
          registered_ports = {1, 2, 3}
          updated_ports = {2}
          expected_ports = ovs_agent.PortInfo(current=registered_ports,
                                              updated={2})
          expected_ancillary = ovs_agent.PortInfo()
          self._test_process_ports_events(events, registered_ports,
                                          set(), expected_ports,
                                          expected_ancillary, updated_ports)
  
      def test_process_port_events_ignores_removed_port_if_never_added(self):
          events = {'added': [], 'modified': [],
                    'removed': [{'name': 'port2', 'ofport': 2,
                                 'external_ids': {'attached-mac': 'test-mac'}}]}
          registered_ports = {1}
          expected_ports = ovs_agent.PortInfo(current=registered_ports)
          expected_ancillary = ovs_agent.PortInfo()
          devices_not_ready_yet = set()
          with mock.patch.object(self.agent.int_br, 'portid_from_external_ids',
                                 side_effect=[2]), \
              mock.patch.object(self.agent, 'check_changed_vlans',
                                return_value=set()):
              failed_devices = {'added': set(), 'removed': set()}
              failed_ancillary_devices = {
                  'added': set(), 'removed': set()}
              ports_not_ready_yet = set()
              actual = self.agent.process_ports_events(
                  events, registered_ports, set(), ports_not_ready_yet,
                  failed_devices, failed_ancillary_devices)
              self.assertEqual(
                  (expected_ports, expected_ancillary, devices_not_ready_yet),
                  actual)
  
      def test_process_port_events_port_not_ready_yet(self):
          events = {'added': [{'name': 'port5', 'ofport': [],
                    'external_ids': {'attached-mac': 'test-mac'}}],
                    'removed': [], 'modified': []}
          old_devices_not_ready = {'port4'}
          registered_ports = {1, 2, 3}
          expected_ports = ovs_agent.PortInfo(current={1, 2, 3, 4}, added={4},
                                              removed=set())
          self.agent.ancillary_brs = []
          expected_ancillary = ovs_agent.PortInfo()
          with mock.patch.object(self.agent.int_br, 'portid_from_external_ids',
                                 side_effect=[5, 4]), \
              mock.patch.object(self.agent, 'check_changed_vlans',
                                return_value=set()), \
              mock.patch.object(self.agent.int_br, 'get_ports_attributes',
                                return_value=[{'name': 'port4', 'ofport': 4,
                                               'external_ids': {
                                                   'attached-mac': 'mac4'}}]):
              expected_devices_not_ready = {'port5'}
              failed_devices = {'added': set(), 'removed': set()}
              failed_ancillary_devices = {
                  'added': set(), 'removed': set()}
              actual = self.agent.process_ports_events(
                  events, registered_ports, set(), old_devices_not_ready,
                  failed_devices, failed_ancillary_devices)
              self.assertEqual(
                  (expected_ports, expected_ancillary,
                   expected_devices_not_ready), actual)
  
      def _test_process_port_events_with_updated_ports(self, updated_ports):
          events = {'added': [{'name': 'port3', 'ofport': 3,
                              'external_ids': {'attached-mac': 'test-mac'}},
                              {'name': 'qg-port2', 'ofport': 6,
                               'external_ids': {'attached-mac': 'test-mac'}}],
                    'removed': [{'name': 'port2', 'ofport': 2,
                                 'external_ids': {'attached-mac': 'test-mac'}},
                                {'name': 'qg-port1', 'ofport': 5,
                                 'external_ids': {'attached-mac': 'test-mac'}}],
                    'modified': []}
          registered_ports = {1, 2, 4}
          ancillary_ports = {5, 8}
          expected_ports = ovs_agent.PortInfo(current={1, 3, 4}, added={3},
                                              removed={2}, updated=updated_ports)
          expected_ancillary = ovs_agent.PortInfo(current={6, 8}, added={6},
                                                  removed={5})
          ancillary_bridge = mock.Mock()
          ancillary_bridge.get_vif_port_set.return_value = {5, 6, 8}
          self.agent.ancillary_brs = [ancillary_bridge]
          with mock.patch.object(self.agent.int_br, 'portid_from_external_ids',
                                side_effect=[3, 6, 2, 5]), \
              mock.patch.object(self.agent, 'check_changed_vlans',
                                return_value=set()):
  
              devices_not_ready_yet = set()
              failed_devices = {'added': set(), 'removed': set()}
              failed_ancillary_devices = {
                  'added': set(), 'removed': set()}
              actual = self.agent.process_ports_events(
                  events, registered_ports, ancillary_ports,
                  devices_not_ready_yet, failed_devices,
                  failed_ancillary_devices, updated_ports)
              self.assertEqual(
                  (expected_ports, expected_ancillary, devices_not_ready_yet),
                  actual)
  
      def test_process_port_events_returns_port_changes(self):
          self._test_process_port_events_with_updated_ports(set())
  
      def test_process_port_events_finds_known_updated_ports(self):
          self._test_process_port_events_with_updated_ports({4})
  
      def test_process_port_events_ignores_unknown_updated_ports(self):
          # the port '10' was not seen on current ports. Hence it has either
          # never been wired or already removed and should be ignored
          self._test_process_port_events_with_updated_ports({4, 10})
  
      def test_process_port_events_ignores_updated_port_if_removed(self):
          self._test_process_port_events_with_updated_ports({4, 5})
  
      def test_update_ports_returns_changed_vlan(self):
          br = self.br_int_cls('br-int')
          mac = "ca:fe:de:ad:be:ef"
          port = ovs_lib.VifPort(1, 1, 1, mac, br)
          self.agent.vlan_manager.add(
              '1', 1, '1', None, 1, {port.vif_id: port})
          vif_port_set = set([1, 3])
          registered_ports = set([1, 2])
          port_tags_dict = {1: []}
          expected = ovs_agent.PortInfo(added={3}, current=vif_port_set,
                                        removed={2}, updated={1})
          with mock.patch.object(self.agent, 'tun_br', autospec=True), \
                  mock.patch.object(self.agent.plugin_rpc,
                                    'update_device_list') as upd_l:
              actual = self.mock_scan_ports(
                  vif_port_set, registered_ports, port_tags_dict=port_tags_dict)
          self.assertEqual(expected, actual)
          upd_l.assert_called_once_with(mock.ANY, [], set([1]),
                                        self.agent.agent_id,
                                        self.agent.conf.host)
  
      def test_update_retries_map_and_remove_devs_not_to_retry(self):
          failed_devices_retries_map = {
              'device_not_to_retry': constants.MAX_DEVICE_RETRIES,
              'device_to_retry': 2,
              'ancillary_not_to_retry': constants.MAX_DEVICE_RETRIES,
              'ancillary_to_retry': 1}
          failed_devices = {
              'added': set(['device_not_to_retry']),
              'removed': set(['device_to_retry', 'new_device'])}
          failed_ancillary_devices = {'added': set(['ancillary_to_retry']),
                                      'removed': set(['ancillary_not_to_retry'])}
          expected_failed_devices_retries_map = {
              'device_to_retry': 3, 'new_device': 1, 'ancillary_to_retry': 2}
          (new_failed_devices_retries_map, devices_not_to_retry,
           ancillary_devices_not_t_retry) = self.agent._get_devices_not_to_retry(
              failed_devices, failed_ancillary_devices,
              failed_devices_retries_map)
          self.agent._remove_devices_not_to_retry(
              failed_devices, failed_ancillary_devices, devices_not_to_retry,
              ancillary_devices_not_t_retry)
          self.assertIn('device_to_retry', failed_devices['removed'])
          self.assertNotIn('device_not_to_retry', failed_devices['added'])
          self.assertEqual(
              expected_failed_devices_retries_map,
              new_failed_devices_retries_map)
  
      def test_add_port_tag_info(self):
          lvm = mock.Mock()
          lvm.vlan = 1
          self.agent.vlan_manager.mapping["net1"] = lvm
          ovs_db_list = [{'name': 'tap1',
                          'tag': [],
                          'other_config': {'segmentation_id': '1'}},
                         {'name': 'tap2',
                          'tag': [],
                          'other_config': {}},
                         {'name': 'tap3',
                          'tag': [],
                          'other_config': None}]
          vif_port1 = mock.Mock()
          vif_port1.port_name = 'tap1'
          vif_port2 = mock.Mock()
          vif_port2.port_name = 'tap2'
          vif_port3 = mock.Mock()
          vif_port3.port_name = 'tap3'
          port_details = [
              {'network_id': 'net1', 'vif_port': vif_port1},
              {'network_id': 'net1', 'vif_port': vif_port2},
              {'network_id': 'net1', 'vif_port': vif_port3}]
          with mock.patch.object(self.agent, 'int_br') as int_br:
              int_br.get_ports_attributes.return_value = ovs_db_list
              self.agent._add_port_tag_info(port_details)
              set_db_attribute_calls = \
                  [mock.call.set_db_attribute("Port", "tap1",
                      "other_config", {"segmentation_id": "1", "tag": "1"}),
                   mock.call.set_db_attribute("Port", "tap2",
                      "other_config", {"tag": "1"}),
                   mock.call.set_db_attribute("Port", "tap3",
                      "other_config", {"tag": "1"})]
              int_br.assert_has_calls(set_db_attribute_calls, any_order=True)
  
      def test_add_port_tag_info_with_tagged_ports(self):
          lvm = mock.Mock()
          lvm.vlan = 1
          self.agent.vlan_manager.mapping["net1"] = lvm
          ovs_db_list1 = [{'name': 'tap1',
                           'tag': 1,
                           'other_config': {'segmentation_id': '1', 'tag': '1'}}]
          ovs_db_list2 = [{'name': 'tap2',
                           'tag': 2,
                           'other_config': {'segmentation_id': '1', 'tag': '1'}},
                          {'name': 'tap3',
                           'tag': 1,
                           'other_config': {'segmentation_id': '2', 'tag': '2'}}]
          vif_port1 = mock.Mock()
          vif_port1.port_name = 'tap1'
          vif_port2 = mock.Mock()
          vif_port2.port_name = 'tap2'
          vif_port2.ofport = 7
          vif_port3 = mock.Mock()
          vif_port3.port_name = 'tap3'
          vif_port3.ofport = 8
          port_details1 = [{'network_id': 'net1', 'vif_port': vif_port1}]
          port_details2 = [{'network_id': 'net1', 'vif_port': vif_port2},
                           {'network_id': 'net1', 'vif_port': vif_port3}]
          with mock.patch.object(self.agent, 'int_br') as int_br:
              int_br.get_ports_attributes.return_value = ovs_db_list1
              self.agent._add_port_tag_info(port_details1)
              int_br.set_db_attribute.assert_not_called()
              # Reset mock to check port with changed tag
              int_br.reset_mock()
              int_br.get_ports_attributes.return_value = ovs_db_list2
              self.agent._add_port_tag_info(port_details2)
              expected_calls = \
                  [mock.call.set_db_attribute("Port", "tap2",
                      "other_config", {'segmentation_id': '1', 'tag': '1'}),
                   mock.call.uninstall_flows(in_port=7),
                   mock.call.set_db_attribute("Port", "tap3",
                       "other_config", {'segmentation_id': '2', 'tag': '1'}),
                   mock.call.uninstall_flows(in_port=8)]
              int_br.assert_has_calls(expected_calls)
  
      def test_bind_devices(self):
          devices_up = ['tap1']
          devices_down = ['tap2']
          self.agent.vlan_manager.mapping["net1"] = mock.Mock()
          ovs_db_list = [{'name': 'tap1', 'tag': []},
                         {'name': 'tap2', 'tag': []}]
          vif_port1 = mock.Mock()
          vif_port1.port_name = 'tap1'
          vif_port2 = mock.Mock()
          vif_port2.port_name = 'tap2'
          port_details = [
              {'network_id': 'net1', 'vif_port': vif_port1,
               'device': devices_up[0],
               'device_owner': 'network:dhcp',
               'admin_state_up': True},
              {'network_id': 'net1', 'vif_port': vif_port2,
               'device': devices_down[0],
               'device_owner': 'network:dhcp',
               'admin_state_up': False}]
          with mock.patch.object(
              self.agent.plugin_rpc, 'update_device_list',
              return_value={'devices_up': devices_up,
                            'devices_down': devices_down,
                            'failed_devices_up': [],
                            'failed_devices_down': []}) as update_devices, \
                  mock.patch.object(self.agent,
                                    'int_br') as int_br:
              int_br.get_ports_attributes.return_value = ovs_db_list
              self.agent._bind_devices(port_details)
              update_devices.assert_called_once_with(mock.ANY, devices_up,
                                                     devices_down,
                                                     mock.ANY, mock.ANY,
                                                     refresh_tunnels=True)
  
      def _test_bind_devices_sets_refresh_tunnels(self, tun_ofports, expected):
          self.agent.iter_num = 3
          self.agent.prevent_arp_spoofing = False
          self.agent.vlan_manager.add('fake_network', 1,
                                      n_const.TYPE_VXLAN, None, 1)
          ovs_db_list = [{'name': 'fake_device', 'tag': []}]
          self.agent.vlan_manager.get('fake_network').tun_ofports = tun_ofports
          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': []}) as update_devices, \
                  mock.patch.object(self.agent,
                                    'int_br') as int_br:
              int_br.get_ports_attributes.return_value = ovs_db_list
              self.agent._bind_devices(need_binding_ports)
              update_devices.assert_called_once_with(mock.ANY, ['fake_device'],
                                                     [], mock.ANY, mock.ANY,
                                                     refresh_tunnels=expected)
  
      def test_bind_devices_sets_refresh_tunnels_if_tunnels_missing(self):
          self._test_bind_devices_sets_refresh_tunnels([], True)
  
      def test_bind_devices_does_not_set_refresh_tunnels_if_tunnels_exist(self):
          self._test_bind_devices_sets_refresh_tunnels([1, 2, 3], False)
  
      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.vlan_manager.add('fake_network', 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.get_ports_attributes.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.
  
          :param details: the details to return for the device
          :param port: the port that get_vif_port_by_id should return
          :param func_name: the function that should be called
          :returns: whether the named function was called
          """
          with mock.patch.object(self.agent.plugin_rpc,
                                 'get_devices_details_list_and_failed_devices',
                                 return_value={'devices': [details],
                                               'failed_devices': []}),\
                  mock.patch.object(self.agent.int_br,
                                    'get_vifs_by_ids',
                                    return_value={details['device']: port}),\
                  mock.patch.object(self.agent.plugin_rpc, 'update_device_list',
                                    return_value={'devices_up': [],
                                                  'devices_down': details,
                                                  'failed_devices_up': [],
                                                  'failed_devices_down': []}),\
                  mock.patch.object(self.agent.int_br,
                      'get_port_tag_dict',
                      return_value={}),\
                  mock.patch.object(self.agent, func_name) as func:
              skip_devs, _, need_bound_devices, _ = (
                  self.agent.treat_devices_added_or_updated([], False, set()))
              # The function should not raise
              self.assertFalse(skip_devs)
              return func.called
  
      def test_treat_devices_added_updated_no_active_binding(self):
          details = {'device': 'id',
                     n_const.NO_ACTIVE_BINDING: True}
          port = mock.Mock()
          with mock.patch.object(self.agent.plugin_rpc,
                                 'get_devices_details_list_and_failed_devices',
                                 return_value={'devices': [details],
                                               'failed_devices': []}),\
                  mock.patch.object(self.agent.int_br,
                                    'get_vifs_by_ids',
                                    return_value={details['device']: port}),\
                  mock.patch.object(self.agent, 'port_dead') as func:
              skip_devs, binding_no_activated_devices, _, _ = (
                  self.agent.treat_devices_added_or_updated([], False, set()))
              self.assertFalse(skip_devs)
              self.assertTrue(func.called)
              self.assertIn('id', binding_no_activated_devices)
  
      def test_treat_devices_added_updated_ignores_invalid_ofport(self):
          port = mock.Mock()
          port.ofport = -1
          self.assertFalse(self._mock_treat_devices_added_updated(
              mock.MagicMock(), port, 'port_dead'))
  
      def test_treat_devices_added_updated_marks_unknown_port_as_dead(self):
          port = mock.Mock()
          port.ofport = 1
          self.assertTrue(self._mock_treat_devices_added_updated(
              mock.MagicMock(), port, 'port_dead'))
  
      def test_treat_devices_added_does_not_process_missing_port(self):
          with mock.patch.object(
              self.agent.plugin_rpc,
              'get_devices_details_list_and_failed_devices') as get_dev_fn,\
                  mock.patch.object(self.agent.int_br,
                                    'get_vif_port_by_id',
                                    return_value=None):
              get_dev_fn.assert_not_called()
  
      def test_treat_devices_added_updated_updates_known_port(self):
          details = mock.MagicMock()
          details.__contains__.side_effect = lambda x: True
          self.assertTrue(self._mock_treat_devices_added_updated(
              details, mock.Mock(), 'treat_vif_port'))
  
      def test_treat_devices_added_updated_sends_vif_port_into_extension_manager(
              self, *args):
          details = mock.MagicMock()
          details.__contains__.side_effect = lambda x: True
          port = mock.MagicMock()
  
          def fake_handle_port(context, port):
              self.assertIn('vif_port', port)
  
          with mock.patch.object(self.agent.plugin_rpc,
                                 'get_devices_details_list_and_failed_devices',
                                 return_value={'devices': [details],
                                               'failed_devices': []}),\
              mock.patch.object(self.agent.ext_manager,
                                'handle_port', new=fake_handle_port),\
              mock.patch.object(self.agent.int_br,
                                'get_vifs_by_ids',
                                return_value={details['device']: port}),\
              mock.patch.object(self.agent, 'treat_vif_port',
                                return_value=False):
  
              self.agent.treat_devices_added_or_updated([], False, set())
  
      def test_treat_devices_added_updated_skips_if_port_not_found(self):
          dev_mock = mock.MagicMock()
          dev_mock.__getitem__.return_value = 'the_skipped_one'
          with mock.patch.object(self.agent.plugin_rpc,
                                 'get_devices_details_list_and_failed_devices',
                                 return_value={'devices': [dev_mock],
                                               'failed_devices': []}),\
                  mock.patch.object(self.agent.int_br,
                      'get_port_tag_dict',
                      return_value={}),\
                  mock.patch.object(self.agent.int_br,
                                    'get_vifs_by_ids',
                                    return_value={}),\
                  mock.patch.object(self.agent.ext_manager,
                                    "delete_port") as ext_mgr_delete_port,\
                  mock.patch.object(self.agent,
                                    'treat_vif_port') as treat_vif_port:
              skip_devs = self.agent.treat_devices_added_or_updated(
                  [], False, set())
              # The function should return False for resync and no device
              # processed
              self.assertEqual((['the_skipped_one'], set(), [], set()),
                               skip_devs)
              ext_mgr_delete_port.assert_called_once_with(
                  self.agent.context, {'port_id': 'the_skipped_one'})
              treat_vif_port.assert_not_called()
  
      def test_treat_devices_added_failed_devices(self):
          dev_mock = 'the_failed_one'
          with mock.patch.object(self.agent.plugin_rpc,
                                 'get_devices_details_list_and_failed_devices',
                                 return_value={'devices': [],
                                               'failed_devices': [dev_mock]}),\
                  mock.patch.object(self.agent.int_br,
                                    'get_vifs_by_ids',
                                    return_value={}),\
                  mock.patch.object(self.agent,
                                    'treat_vif_port') as treat_vif_port:
              failed_devices = {'added': set(), 'removed': set()}
              (_, _, _, failed_devices['added']) = (
                  self.agent.treat_devices_added_or_updated([], False, set()))
              # The function should return False for resync and no device
              # processed
              self.assertEqual(set([dev_mock]), failed_devices.get('added'))
              treat_vif_port.assert_not_called()
  
      def test_treat_devices_added_updated_put_port_down(self):
          fake_details_dict = {'admin_state_up': False,
                               'port_id': 'xxx',
                               'device': 'xxx',
                               'network_id': 'yyy',
                               'physical_network': 'foo',
                               'segmentation_id': 'bar',
                               'network_type': 'baz',
                               'fixed_ips': [{'subnet_id': 'my-subnet-uuid',
                                              'ip_address': '1.1.1.1'}],
                               'device_owner': DEVICE_OWNER_COMPUTE
                               }
  
          with mock.patch.object(self.agent.plugin_rpc,
                                 'get_devices_details_list_and_failed_devices',
                                 return_value={'devices': [fake_details_dict],
                                               'failed_devices': []}),\
                  mock.patch.object(self.agent.int_br,
                                    'get_vifs_by_ids',
                                    return_value={'xxx': mock.MagicMock()}),\
                  mock.patch.object(self.agent.int_br, 'get_port_tag_dict',
                                    return_value={}),\
                  mock.patch.object(self.agent,
                                    'treat_vif_port') as treat_vif_port:
              skip_devs, _, need_bound_devices, _ = (
                  self.agent.treat_devices_added_or_updated([], False, set()))
              # The function should return False for resync
              self.assertFalse(skip_devs)
              self.assertTrue(treat_vif_port.called)
  
      def _mock_treat_devices_removed(self, port_exists):
          details = dict(exists=port_exists)
          with mock.patch.object(self.agent.plugin_rpc,
                                 'update_device_list',
                                 return_value={'devices_up': [],
                                               'devices_down': details,
                                               'failed_devices_up': [],
                                               'failed_devices_down': []}):
              with mock.patch.object(self.agent, 'port_unbound') as port_unbound:
                  with mock.patch.object(self.agent.int_br,
                                         'get_vif_port_by_id',
                                         return_value=None):
                      self.assertFalse(self.agent.treat_devices_removed([{}]))
          self.assertTrue(port_unbound.called)
  
      def test_treat_devices_removed_unbinds_port(self):
          self._mock_treat_devices_removed(True)
  
      def test_treat_devices_removed_ignores_missing_port(self):
          self._mock_treat_devices_removed(False)
  
      def test_treat_devices_removed_failed_devices(self):
          dev_mock = 'the_failed_one'
          with mock.patch.object(self.agent.plugin_rpc,
                                 'update_device_list',
                                 return_value={'devices_up': [],
                                               'devices_down': [],
                                               'failed_devices_up': [],
                                               'failed_devices_down': [
                                                   dev_mock]}):
              with mock.patch.object(self.agent.int_br,
                                     'get_vif_port_by_id',
                                     return_value=None):
                  failed_devices = {'added': set(), 'removed': set()}
                  failed_devices['removed'] = \
                      self.agent.treat_devices_removed([{}])
                  self.assertEqual(set([dev_mock]),
                                   failed_devices.get('removed'))
  
      def test_treat_devices_removed_ext_delete_port(self):
          port_id = 'fake-id'
  
          m_delete = mock.patch.object(self.agent.ext_manager, 'delete_port')
          m_rpc = mock.patch.object(self.agent.plugin_rpc, 'update_device_list',
                                    return_value={'devices_up': [],
                                                  'devices_down': [],
                                                  'failed_devices_up': [],
                                                  'failed_devices_down': []})
          m_unbound = mock.patch.object(self.agent, 'port_unbound')
          with m_delete as delete, m_rpc, m_unbound:
              with mock.patch.object(self.agent.int_br,
                                     'get_vif_port_by_id',
                                     return_value=None):
                  self.agent.treat_devices_removed([port_id])
                  delete.assert_called_with(mock.ANY, {'port_id': port_id})
  
      def test_treat_vif_port_shut_down_port(self):
          details = mock.MagicMock()
          vif_port = type('vif_port', (object,), {
              "vif_id": "12",
              "iface-id": "407a79e0-e0be-4b7d-92a6-513b2161011b",
              "vif_mac": "fa:16:3e:68:46:7b",
              "port_name": "qr-407a79e0-e0",
              "ofport": -1,
              "bridge_name": "br-int"})
          with mock.patch.object(
                  self.agent.plugin_rpc, 'update_device_down'
          ) as update_device_down, mock.patch.object(
              self.agent, "port_dead"
          ) as port_dead:
              port_needs_binding = self.agent.treat_vif_port(
                  vif_port, details['port_id'],
                  details['network_id'],
                  details['network_type'],
                  details['physical_network'],
                  details['segmentation_id'],
                  False,
                  details['fixed_ips'],
                  details['device_owner'], False)
          self.assertFalse(port_needs_binding)
          port_dead.assert_called_once_with(vif_port)
          update_device_down.assert_called_once_with(
              self.agent.context, details['port_id'], self.agent.agent_id,
              self.agent.conf.host)
  
      def test_bind_port_with_missing_network(self):
          vif_port = mock.Mock()
          vif_port.name.return_value = 'port'
          self.agent._bind_devices([{'network_id': 'non-existent',
                                     'vif_port': vif_port}])
  
      def _test_process_network_ports(self, port_info, skipped_devices=None,
                                      binding_no_activated_devices=None):
          failed_devices = {'added': set(), 'removed': set()}
          skipped_devices = skipped_devices or []
          binding_no_activated_devices = binding_no_activated_devices or set()
          added_devices = port_info.get('added', set())
          with mock.patch.object(self.agent.sg_agent,
                                 "setup_port_filters") as setup_port_filters,\
                  mock.patch.object(
                      self.agent, "treat_devices_added_or_updated",
                      return_value=(
                          skipped_devices, binding_no_activated_devices, [],
                          failed_devices['added'])) as device_added_updated,\
                  mock.patch.object(self.agent.int_br, "get_ports_attributes",
                                    return_value=[]),\
                  mock.patch.object(self.agent,
                                    "treat_devices_removed",
                                    return_value=(
                                        failed_devices[
                                            'removed'])) as device_removed,\
                  mock.patch.object(self.agent,
                                    "treat_devices_skipped",
                                    return_value=(
                                        skipped_devices)) as device_skipped:
              self.assertEqual(
                  failed_devices,
                  self.agent.process_network_ports(port_info, False))
              setup_port_filters.assert_called_once_with(
                  (added_devices - set(skipped_devices) -
                      binding_no_activated_devices),
                  port_info.get('updated', set()))
              devices_added_updated = (added_devices |
                                       port_info.get('updated', set()))
              if devices_added_updated:
                  device_added_updated.assert_called_once_with(
                      devices_added_updated, False, set())
              if port_info.get('removed', set()):
                  device_removed.assert_called_once_with(port_info['removed'])
              if skipped_devices:
                  device_skipped.assert_called_once_with(set(skipped_devices))
  
      def test_process_network_ports(self):
          self._test_process_network_ports(
              {'current': set(['tap0']),
               'removed': set(['eth0']),
               'added': set(['eth1'])})
  
      def test_process_network_port_with_updated_ports(self):
          self._test_process_network_ports(
              {'current': set(['tap0', 'tap1']),
               'updated': set(['tap1', 'eth1']),
               'removed': set(['eth0']),
               'added': set(['eth1'])})
  
      def test_process_network_port_with_skipped_ports(self):
          port_info = {'current': set(['tap0', 'tap1']),
                       'removed': set(['eth0']),
                       'added': set(['eth1', 'eth2'])}
          self._test_process_network_ports(port_info, skipped_devices=['eth1'])
  
      def test_process_network_port_with_binding_no_activated_devices(self):
          port_info = {'current': set(['tap0', 'tap1']),
                       'removed': set(['eth0']),
                       'added': set(['eth1', 'eth2', 'eth3'])}
          self._test_process_network_ports(
              port_info, skipped_devices=['eth1'],
              binding_no_activated_devices=set(['eth3']))
  
      def test_process_network_port_with_empty_port(self):
          self._test_process_network_ports({})
  
      @mock.patch.object(linux_utils, 'execute', return_value=False)
      def test_hybrid_plug_flag_based_on_firewall(self, *args):
          cfg.CONF.set_default(
              'firewall_driver',
              'neutron.agent.firewall.NoopFirewallDriver',
              group='SECURITYGROUP')
          agt = self._make_agent()
          self.assertFalse(agt.agent_state['configurations']['ovs_hybrid_plug'])
          cfg.CONF.set_default(
              'firewall_driver',
              'neutron.agent.linux.openvswitch_firewall.OVSFirewallDriver',
              group='SECURITYGROUP')
          with mock.patch('neutron.agent.linux.openvswitch_firewall.'
                          'OVSFirewallDriver.initialize_bridge'):
              agt = self._make_agent()
          self.assertFalse(agt.agent_state['configurations']['ovs_hybrid_plug'])
          cfg.CONF.set_default(
              'firewall_driver',
              'neutron.agent.linux.iptables_firewall.'
              'OVSHybridIptablesFirewallDriver',
              group='SECURITYGROUP')
          with mock.patch('neutron.agent.linux.ip_conntrack.'
                          'IpConntrackManager._populate_initial_zone_map'):
              agt = self._make_agent()
          self.assertTrue(agt.agent_state['configurations']['ovs_hybrid_plug'])
  
      def test_report_state(self):
          with mock.patch.object(self.agent.state_rpc,
                                 "report_state") as report_st:
              self.agent.int_br_device_count = 5
              self.systemd_notify.assert_not_called()
              self.agent._report_state()
              report_st.assert_called_with(self.agent.context,
                                           self.agent.agent_state, True)
              self.systemd_notify.assert_called_once_with()
              self.systemd_notify.reset_mock()
              # agent keeps sending "start_flag" while iter 0 not completed
              self.assertIn("start_flag", self.agent.agent_state)
              self.assertEqual(
                  self.agent.agent_state["configurations"]["devices"],
                  self.agent.int_br_device_count
              )
              self.agent._report_state()
              report_st.assert_called_with(self.agent.context,
                                           self.agent.agent_state, True)
              self.systemd_notify.assert_not_called()
  
      def test_report_state_fail(self):
          with mock.patch.object(self.agent.state_rpc,
                                 "report_state") as report_st:
              report_st.side_effect = Exception()
              self.agent._report_state()
              report_st.assert_called_with(self.agent.context,
                                           self.agent.agent_state, True)
              self.agent._report_state()
              report_st.assert_called_with(self.agent.context,
                                           self.agent.agent_state, True)
              self.systemd_notify.assert_not_called()
  
      def test_report_state_revived(self):
          with mock.patch.object(self.agent.state_rpc,
                                 "report_state") as report_st:
              report_st.return_value = agent_consts.AGENT_REVIVED
              self.agent._report_state()
              self.assertTrue(self.agent.fullsync)
  
      def test_port_update(self):
          port_arg = {"id": TEST_PORT_ID1}
          with mock.patch.object(self.agent.plugin_rpc.remote_resource_cache,
                                 "get_resource_by_id") as mocked_resource:
              port = Port()
              port['mac_address'] = netaddr.EUI(FAKE_MAC)
              port['device_id'] = '0'
              port_bind = PortBinding()
              port_bind['host'] = 'host'
              port_bind['vnic_type'] = 'normal'
              port.bindings = [port_bind]
              mocked_resource.return_value = port
              self.agent.port_update("unused_context",
                                     port=port_arg,
                                     network_type="vlan",
                                     segmentation_id="1",
                                     physical_network="physnet")
              self.assertEqual(set([TEST_PORT_ID1]), self.agent.updated_ports)
              self.assertEqual([], self.agent.updated_smartnic_ports)
  
      def test_port_update_smartnic(self):
          cfg.CONF.set_default('baremetal_smartnic', True, group='AGENT')
          port_arg = {"id": TEST_PORT_ID1}
          with mock.patch.object(self.agent.plugin_rpc.remote_resource_cache,
                                 "get_resource_by_id") as mocked_resource:
              port = Port()
              port['id'] = 'd850ed99-5f46-47bc-8c06-86d9d519c46a'
              port['mac_address'] = netaddr.EUI(FAKE_MAC)
              port['device_id'] = '0'
              bindings_data = PortBinding()
              bindings_data['host'] = 'host'
              bindings_data['vnic_type'] = portbindings.VNIC_SMARTNIC
              bindings_data['vif_type'] = portbindings.VIF_TYPE_OVS
              bindings_data['profile'] = {
                  'local_link_information': [{'port_id': 'rep_port'}]}
              port.bindings = [bindings_data]
              mocked_resource.return_value = port
              self.agent.port_update("unused_context",
                                     port=port_arg)
              expected_smartnic_port_data = {
                  'mac': port['mac_address'],
                  'vm_uuid': port['device_id'],
                  'vif_name': 'rep_port',
                  'iface_id': port['id'],
                  'vif_type': bindings_data['vif_type']
              }
              self.assertEqual({TEST_PORT_ID1}, self.agent.updated_ports)
              self.assertEqual([expected_smartnic_port_data],
                               self.agent.updated_smartnic_ports)
  
      def test_port_update_unbound_smartnic_port(self):
          cfg.CONF.set_default('baremetal_smartnic', True, group='AGENT')
          port_arg = {"id": 'd850ed99-5f46-47bc-8c06-86d9d519c46a'}
          vif_name = "rep0-0"
          vif_id = port_arg["id"]
          vif_mac = FAKE_MAC
          self.agent.current_smartnic_ports_map = {
              vif_id: {
                  'vif_mac': vif_mac,
                  'vif_name': vif_name}}
  
          with mock.patch.object(self.agent.plugin_rpc.remote_resource_cache,
                                 "get_resource_by_id") as mocked_resource:
              port = Port()
              port['id'] = port_arg["id"]
              port['mac_address'] = netaddr.EUI(FAKE_MAC)
              port['device_id'] = '0'
              bindings_data = PortBinding()
              bindings_data['host'] = ''
              bindings_data['vnic_type'] = portbindings.VNIC_SMARTNIC
              bindings_data['vif_type'] = portbindings.VIF_TYPE_UNBOUND
              bindings_data['profile'] = {}
              port.bindings = [bindings_data]
              mocked_resource.return_value = port
              self.agent.port_update("unused_context",
                                     port=port_arg)
              expected_smartnic_port_data = [{
                  'mac': port['mac_address'],
                  'vm_uuid': '',
                  'vif_name': 'rep0-0',
                  'iface_id': port['id'],
                  'vif_type': portbindings.VIF_TYPE_UNBOUND
              }]
              mocked_resource.assert_called_with(resources.PORT, port['id'])
              self.assertEqual({port['id']}, self.agent.updated_ports)
              self.assertEqual(expected_smartnic_port_data,
                               self.agent.updated_smartnic_ports)
  
      def test_port_update_unbound_smartnic_port_not_belong_to_the_agent(self):
          cfg.CONF.set_default('baremetal_smartnic', True, group='AGENT')
          port_arg = {"id": 'd850ed99-5f46-47bc-8c06-86d9d519c46a'}
          vif_name = "rep0-0"
          vif_id = 'd850ed99-5f46-47bc-8c06-86d9d519c46b'
          vif_mac = FAKE_MAC
          self.agent.current_smartnic_ports_map = {
              vif_id: {
                  'vif_mac': vif_mac,
                  'vif_name': vif_name}}
  
          with mock.patch.object(self.agent.plugin_rpc.remote_resource_cache,
                                 "get_resource_by_id") as mocked_resource:
              port = Port()
              port['id'] = port_arg["id"]
              port['mac_address'] = netaddr.EUI(FAKE_MAC)
              port['device_id'] = '0'
              bindings_data = PortBinding()
              bindings_data['host'] = ''
              bindings_data['vnic_type'] = portbindings.VNIC_SMARTNIC
              bindings_data['vif_type'] = portbindings.VIF_TYPE_UNBOUND
              bindings_data['profile'] = {}
              port.bindings = [bindings_data]
              mocked_resource.return_value = port
              self.agent.port_update("unused_context",
                                     port=port_arg)
              mocked_resource.assert_called_with(resources.PORT, port['id'])
              self.assertEqual({port['id']}, self.agent.updated_ports)
              self.assertEqual([], self.agent.updated_smartnic_ports)
  
      def test_port_delete_after_update(self):
          """Make sure a port is not marked for delete and update."""
          port = {'id': TEST_PORT_ID1}
  
          self.agent.port_update(context=None, port=port)
          self.agent.port_delete(context=None, port_id=port['id'])
          self.assertEqual(set(), self.agent.updated_ports)
          self.assertEqual(set([port['id']]), self.agent.deleted_ports)
  
      def test_process_deleted_ports_cleans_network_ports(self):
          self.agent._update_port_network(TEST_PORT_ID1, TEST_NETWORK_ID1)
          self.agent.port_delete(context=None, port_id=TEST_PORT_ID1)
          self.agent.sg_agent = mock.Mock()
          self.agent.int_br = mock.Mock()
  
          @contextlib.contextmanager
          def bridge_deferred(*args, **kwargs):
              yield
  
          self.agent.int_br.deferred = mock.Mock(side_effect=bridge_deferred)
          self.agent.process_deleted_ports(port_info={})
          self.assertEqual(set(), self.agent.network_ports[TEST_NETWORK_ID1])
  
      def test_network_update(self):
          """Network update marks port for update. """
          network = {'id': TEST_NETWORK_ID1}
          port = {'id': TEST_PORT_ID1, 'network_id': network['id']}
  
          self.agent._update_port_network(port['id'], port['network_id'])
          with mock.patch.object(self.agent.plugin_rpc, 'get_network_details'), \
                  mock.patch.object(self.agent,
                                    '_update_network_segmentation_id'):
              self.agent.network_update(context=None, network=network)
          self.assertEqual(set([port['id']]), self.agent.updated_ports)
  
      def test_network_update_outoforder(self):
          """Network update arrives later than port_delete.
  
          But the main agent loop still didn't process the ports,
          so we ensure the port is not marked for update.
          """
          network = {'id': TEST_NETWORK_ID1}
          port = {'id': TEST_PORT_ID1, 'network_id': network['id']}
  
          self.agent._update_port_network(port['id'], port['network_id'])
          self.agent.port_delete(context=None, port_id=port['id'])
          with mock.patch.object(self.agent.plugin_rpc, 'get_network_details'), \
                  mock.patch.object(self.agent,
                                    '_update_network_segmentation_id'):
              self.agent.network_update(context=None, network=network)
          self.assertEqual(set(), self.agent.updated_ports)
  
      def test_update_port_network(self):
          """Ensure ports are associated and moved across networks correctly."""
          self.agent._update_port_network(TEST_PORT_ID1, TEST_NETWORK_ID1)
          self.agent._update_port_network(TEST_PORT_ID2, TEST_NETWORK_ID1)
          self.agent._update_port_network(TEST_PORT_ID3, TEST_NETWORK_ID2)
          self.agent._update_port_network(TEST_PORT_ID1, TEST_NETWORK_ID2)
  
          self.assertEqual(set([TEST_PORT_ID2]),
                           self.agent.network_ports[TEST_NETWORK_ID1])
          self.assertEqual(set([TEST_PORT_ID1, TEST_PORT_ID3]),
                           self.agent.network_ports[TEST_NETWORK_ID2])
  
      def test_port_delete(self):
          vif = FakeVif()
          with mock.patch.object(self.agent, 'int_br') as int_br:
              int_br.get_vif_by_port_id.return_value = vif.port_name
              int_br.get_vif_port_by_id.return_value = vif
              self.agent.port_delete("unused_context",
                                     port_id='id')
              self.agent.process_deleted_ports(port_info={})
              # the main things we care about are that it gets put in the
              # dead vlan and gets blocked
              int_br.set_db_attribute.assert_any_call(
                  'Port', vif.port_name, 'tag', constants.DEAD_VLAN_TAG,
                  log_errors=False)
              int_br.drop_port.assert_called_once_with(in_port=vif.ofport)
  
      def test_port_delete_removed_port(self):
          with mock.patch.object(self.agent, 'int_br') as int_br:
              self.agent.port_delete("unused_context",
                                     port_id='id')
              # if it was removed from the bridge, we shouldn't be processing it
              self.agent.process_deleted_ports(port_info={'removed': {'id', }})
              int_br.set_db_attribute.assert_not_called()
              int_br.drop_port.assert_not_called()
  
      def test_binding_deactivate_not_for_host(self):
          self.agent.binding_deactivate('unused_context', port_id='id',
                                        host='other_host')
          self.assertEqual(set(), self.agent.deactivated_bindings)
  
      def test_binding_deactivate(self):
          vif = FakeVif()
          with mock.patch.object(self.agent, 'int_br') as int_br:
              int_br.get_vif_port_by_id.return_value = vif
              self.agent.binding_deactivate('unused_context', port_id='id',
                                            host='host')
              self.assertEqual(set(['id']), self.agent.deactivated_bindings)
              self.agent.process_deactivated_bindings(port_info={})
              int_br.get_vif_port_by_id.assert_called_once_with('id')
              int_br.delete_port.assert_called_once_with(vif.port_name)
              self.assertEqual(set(), self.agent.deactivated_bindings)
  
      def test_binding_deactivate_removed_port(self):
          with mock.patch.object(self.agent, 'int_br') as int_br:
              self.agent.binding_deactivate('unused_context', port_id='id',
                                            host='host')
              self.assertEqual(set(['id']), self.agent.deactivated_bindings)
              self.agent.process_deactivated_bindings(
                  port_info={'removed': {'id', }})
              int_br.get_vif_port_by_id.assert_not_called()
              int_br.delete_port.assert_not_called()
              self.assertEqual(set(), self.agent.deactivated_bindings)
  
      def test_binding_activate(self):
          self.agent.binding_activate('context', port_id='id', host='host')
          self.assertIn('id', self.agent.activated_bindings)
  
      def test_binding_activate_not_for_host(self):
          self.agent.binding_activate('context', port_id='id', host='other-host')
          self.assertEqual(set(), self.agent.activated_bindings)
  
      def test_process_activated_bindings(self):
          port_info = {}
          port_info['added'] = set(['added_port_id'])
          port_info['current'] = set(['activated_port_id'])
          self.agent.process_activated_bindings(port_info,
                                                set(['activated_port_id']))
          self.assertIn('added_port_id', port_info['added'])
          self.assertIn('activated_port_id', port_info['added'])
  
      def test_process_activated_bindings_activated_port_not_present(self):
          port_info = {}
          port_info['added'] = set(['added_port_id'])
          port_info['current'] = set()
          self.agent.process_activated_bindings(port_info,
                                                set(['activated_port_id']))
          self.assertIn('added_port_id', port_info['added'])
          self.assertNotIn('activated_port_id', port_info['added'])
  
      def _test_setup_physical_bridges(self, port_exists=False,
                                       dvr_enabled=False):
          self.agent.enable_distributed_routing = dvr_enabled
          with mock.patch.object(ip_lib.IPDevice, "exists") as devex_fn,\
                  mock.patch.object(sys, "exit"),\
                  mock.patch.object(self.agent, 'br_phys_cls') as phys_br_cls,\
                  mock.patch.object(self.agent, 'int_br') as int_br,\
                  mock.patch.object(self.agent, '_check_bridge_datapath_id'),\
                  mock.patch.object(ovs_lib.BaseOVS, 'get_bridges'):
              devex_fn.return_value = True
              parent = mock.MagicMock()
              phys_br = phys_br_cls()
              parent.attach_mock(phys_br_cls, 'phys_br_cls')
              parent.attach_mock(phys_br, 'phys_br')
              parent.attach_mock(int_br, 'int_br')
              if port_exists:
                  phys_br.get_port_ofport.return_value = "phy_ofport"
                  int_br.get_port_ofport.return_value = "int_ofport"
              else:
                  phys_br.add_patch_port.return_value = "phy_ofport"
                  int_br.add_patch_port.return_value = "int_ofport"
              phys_br.port_exists.return_value = port_exists
              int_br.port_exists.return_value = port_exists
              self.agent.setup_physical_bridges({"physnet1": "br-eth"})
              expected_calls = [
                  mock.call.phys_br_cls('br-eth'),
                  mock.call.phys_br.create(),
                  mock.call.phys_br.set_secure_mode(),
                  mock.call.phys_br.setup_controllers(mock.ANY),
                  mock.call.phys_br.setup_default_table(),
                  mock.call.int_br.db_get_val('Interface', 'int-br-eth',
                                              'type', log_errors=False),
                  # Have to use __getattr__ here to avoid mock._Call.__eq__
                  # method being called
                  mock.call.int_br.db_get_val().__getattr__('__eq__')('veth'),
                  mock.call.int_br.port_exists('int-br-eth'),
              ]
              if port_exists:
                  expected_calls += [
                      mock.call.int_br.get_port_ofport('int-br-eth'),
                  ]
              else:
                  expected_calls += [
                      mock.call.int_br.add_patch_port(
                          'int-br-eth', constants.NONEXISTENT_PEER),
                  ]
              expected_calls += [
                  mock.call.phys_br.port_exists('phy-br-eth'),
              ]
              if port_exists:
                  expected_calls += [
                      mock.call.phys_br.get_port_ofport('phy-br-eth'),
                  ]
              else:
                  expected_calls += [
                      mock.call.phys_br.add_patch_port(
                          'phy-br-eth', constants.NONEXISTENT_PEER),
                  ]
              expected_calls += [
                      mock.call.int_br.drop_port(in_port='int_ofport')
              ]
              if not dvr_enabled:
                  expected_calls += [
                          mock.call.phys_br.drop_port(in_port='phy_ofport')
                  ]
              expected_calls += [
                  mock.call.int_br.set_db_attribute('Interface', 'int-br-eth',
                                                    'options',
                                                    {'peer': 'phy-br-eth'}),
                  mock.call.phys_br.set_db_attribute('Interface', 'phy-br-eth',
                                                     'options',
                                                     {'peer': 'int-br-eth'}),
              ]
              parent.assert_has_calls(expected_calls)
              self.assertEqual("int_ofport",
                               self.agent.int_ofports["physnet1"])
              self.assertEqual("phy_ofport",
                               self.agent.phys_ofports["physnet1"])
  
      def test_setup_physical_bridges(self):
          self._test_setup_physical_bridges()
  
      def test_setup_physical_bridges_port_exists(self):
          self._test_setup_physical_bridges(port_exists=True)
  
      def test_setup_physical_bridges_dvr_enabled(self):
          self._test_setup_physical_bridges(dvr_enabled=True)
  
      def test_setup_physical_bridges_using_veth_interconnection(self):
          self.agent.use_veth_interconnection = True
          with mock.patch.object(ip_lib.IPDevice, "exists") as devex_fn,\
                  mock.patch.object(sys, "exit"),\
                  mock.patch.object(utils, "execute") as utilsexec_fn,\
                  mock.patch.object(self.agent, 'br_phys_cls') as phys_br_cls,\
                  mock.patch.object(self.agent, 'int_br') as int_br,\
                  mock.patch.object(self.agent, '_check_bridge_datapath_id'),\
                  mock.patch.object(ip_lib.IPWrapper, "add_veth") as addveth_fn,\
                  mock.patch.object(ip_lib.IpLinkCommand,
                                    "delete") as linkdel_fn,\
                  mock.patch.object(ip_lib.IpLinkCommand, "set_up"),\
                  mock.patch.object(ip_lib.IpLinkCommand, "set_mtu"),\
                  mock.patch.object(ovs_lib.BaseOVS, "get_bridges") as get_br_fn:
              devex_fn.return_value = True
              parent = mock.MagicMock()
              parent.attach_mock(utilsexec_fn, 'utils_execute')
              parent.attach_mock(linkdel_fn, 'link_delete')
              parent.attach_mock(addveth_fn, 'add_veth')
              addveth_fn.return_value = (ip_lib.IPDevice("int-br-eth1"),
                                         ip_lib.IPDevice("phy-br-eth1"))
              phys_br = phys_br_cls()
              phys_br.add_port.return_value = "phys_veth_ofport"
              int_br.add_port.return_value = "int_veth_ofport"
              get_br_fn.return_value = ["br-eth"]
              self.agent.setup_physical_bridges({"physnet1": "br-eth"})
              expected_calls = [mock.call.link_delete(),
                                mock.call.utils_execute(['udevadm',
                                                         'settle',
                                                         '--timeout=10']),
                                mock.call.add_veth('int-br-eth',
                                                   'phy-br-eth')]
              parent.assert_has_calls(expected_calls, any_order=False)
              self.assertEqual("int_veth_ofport",
                               self.agent.int_ofports["physnet1"])
              self.assertEqual("phys_veth_ofport",
                               self.agent.phys_ofports["physnet1"])
              int_br.add_port.assert_called_with("int-br-eth")
              phys_br.add_port.assert_called_with("phy-br-eth")
  
      def _test_setup_physical_bridges_change_from_veth_to_patch_conf(
              self, port_exists=False):
          with mock.patch.object(sys, "exit"),\
                  mock.patch.object(self.agent, 'br_phys_cls') as phys_br_cls,\
                  mock.patch.object(self.agent, 'int_br') as int_br,\
                  mock.patch.object(self.agent.int_br, 'db_get_val',
                                    return_value='veth'), \
                  mock.patch.object(self.agent, '_check_bridge_datapath_id'), \
                  mock.patch.object(ovs_lib.BaseOVS, 'get_bridges'):
              phys_br = phys_br_cls()
              parent = mock.MagicMock()
              parent.attach_mock(phys_br_cls, 'phys_br_cls')
              parent.attach_mock(phys_br, 'phys_br')
              parent.attach_mock(int_br, 'int_br')
              if port_exists:
                  phys_br.get_port_ofport.return_value = "phy_ofport"
                  int_br.get_port_ofport.return_value = "int_ofport"
              else:
                  phys_br.add_patch_port.return_value = "phy_ofport"
                  int_br.add_patch_port.return_value = "int_ofport"
              phys_br.port_exists.return_value = port_exists
              int_br.port_exists.return_value = port_exists
              self.agent.setup_physical_bridges({"physnet1": "br-eth"})
              expected_calls = [
                  mock.call.phys_br_cls('br-eth'),
                  mock.call.phys_br.create(),
                  mock.call.phys_br.set_secure_mode(),
                  mock.call.phys_br.setup_controllers(mock.ANY),
                  mock.call.phys_br.setup_default_table(),
                  mock.call.int_br.delete_port('int-br-eth'),
                  mock.call.phys_br.delete_port('phy-br-eth'),
                  mock.call.int_br.port_exists('int-br-eth'),
              ]
              if port_exists:
                  expected_calls += [
                      mock.call.int_br.get_port_ofport('int-br-eth'),
                  ]
              else:
                  expected_calls += [
                      mock.call.int_br.add_patch_port(
                          'int-br-eth', constants.NONEXISTENT_PEER),
                  ]
              expected_calls += [
                  mock.call.phys_br.port_exists('phy-br-eth'),
              ]
              if port_exists:
                  expected_calls += [
                      mock.call.phys_br.get_port_ofport('phy-br-eth'),
                  ]
              else:
                  expected_calls += [
                      mock.call.phys_br.add_patch_port(
                          'phy-br-eth', constants.NONEXISTENT_PEER),
                  ]
              expected_calls += [
                  mock.call.int_br.drop_port(in_port='int_ofport'),
                  mock.call.phys_br.drop_port(in_port='phy_ofport'),
                  mock.call.int_br.set_db_attribute('Interface', 'int-br-eth',
                                                    'options',
                                                    {'peer': 'phy-br-eth'}),
                  mock.call.phys_br.set_db_attribute('Interface', 'phy-br-eth',
                                                     'options',
                                                     {'peer': 'int-br-eth'}),
              ]
              parent.assert_has_calls(expected_calls)
              self.assertEqual("int_ofport",
                               self.agent.int_ofports["physnet1"])
              self.assertEqual("phy_ofport",
                               self.agent.phys_ofports["physnet1"])
  
      def test_setup_physical_bridges_change_from_veth_to_patch_conf(self):
          self._test_setup_physical_bridges_change_from_veth_to_patch_conf()
  
      def test_setup_physical_bridges_change_from_veth_to_patch_conf_port_exists(
              self):
          self._test_setup_physical_bridges_change_from_veth_to_patch_conf(
              port_exists=True)
  
      def test_setup_tunnel_br(self):
          self.tun_br = mock.Mock()
          with mock.patch.object(self.agent.int_br,
                                 "add_patch_port",
                                 return_value=1) as int_patch_port,\
                  mock.patch.object(self.agent.tun_br,
                                    "add_patch_port",
                                    return_value=1) as tun_patch_port,\
                  mock.patch.object(self.agent.tun_br, 'bridge_exists',
                                    return_value=False),\
                  mock.patch.object(self.agent.tun_br, 'create') as create_tun,\
                  mock.patch.object(self.agent.tun_br,
                                    'setup_controllers') as setup_controllers,\
                  mock.patch.object(self.agent.tun_br, 'port_exists',
                                    return_value=False),\
                  mock.patch.object(self.agent.int_br, 'port_exists',
                                    return_value=False),\
                  mock.patch.object(sys, "exit"):
              self.agent.setup_tunnel_br(None)
              self.agent.setup_tunnel_br()
              self.assertTrue(create_tun.called)
              self.assertTrue(setup_controllers.called)
              self.assertTrue(int_patch_port.called)
              self.assertTrue(tun_patch_port.called)
  
      def test_setup_tunnel_br_ports_exits_drop_flows(self):
          cfg.CONF.set_override('drop_flows_on_start', True, 'AGENT')
          with mock.patch.object(self.agent.tun_br, 'port_exists',
                                 return_value=True),\
                  mock.patch.object(self.agent, 'tun_br'),\
                  mock.patch.object(self.agent.int_br, 'port_exists',
                                    return_value=True),\
                  mock.patch.object(self.agent.tun_br, 'setup_controllers'),\
                  mock.patch.object(self.agent, 'patch_tun_ofport', new=2),\
                  mock.patch.object(self.agent, 'patch_int_ofport', new=2),\
                  mock.patch.object(self.agent.tun_br,
                                    'uninstall_flows') as delete,\
                  mock.patch.object(self.agent.int_br,
                                    "add_patch_port") as int_patch_port,\
                  mock.patch.object(self.agent.tun_br,
                                    "add_patch_port") as tun_patch_port,\
                  mock.patch.object(sys, "exit"):
              self.agent.setup_tunnel_br(None)
              self.agent.setup_tunnel_br()
              int_patch_port.assert_not_called()
              tun_patch_port.assert_not_called()
              self.assertTrue(delete.called)
  
      def test_setup_tunnel_port(self):
          self.agent.tun_br = mock.Mock()
          self.agent.l2_pop = False
          self.agent.udp_vxlan_port = 8472
          self.agent.tun_br_ofports['vxlan'] = {}
          self.agent.local_ip = '2.3.4.5'
          with mock.patch.object(self.agent.tun_br,
                                 "add_tunnel_port",
                                 return_value='6') as add_tun_port_fn,\
                  mock.patch.object(self.agent.tun_br, "add_flow"):
              self.agent._setup_tunnel_port(self.agent.tun_br, 'portname',
                                            '1.2.3.4', 'vxlan')
              self.assertTrue(add_tun_port_fn.called)
  
      def test_port_unbound(self):
          with mock.patch.object(self.agent, "reclaim_local_vlan") as reclvl_fn:
              self.agent.enable_tunneling = True
              lvm = mock.Mock()
              lvm.network_type = "gre"
              lvm.vif_ports = {"vif1": mock.Mock()}
              self.agent.vlan_manager.mapping["netuid12345"] = lvm
              self.agent.port_unbound("vif1", "netuid12345")
              self.assertTrue(reclvl_fn.called)
  
              lvm.vif_ports = {}
              self.agent.port_unbound("vif1", "netuid12345")
              self.assertEqual(2, reclvl_fn.call_count)
  
              lvm.vif_ports = {"vif1": mock.Mock()}
              self.agent.port_unbound("vif3", "netuid12345")
              self.assertEqual(2, reclvl_fn.call_count)
  
      def _prepare_l2_pop_ofports(self):
          lvm1 = mock.Mock()
          lvm1.network_type = 'gre'
          lvm1.vlan = 'vlan1'
          lvm1.segmentation_id = 'seg1'
          lvm1.tun_ofports = set(['1'])
          lvm2 = mock.Mock()
          lvm2.network_type = 'gre'
          lvm2.vlan = 'vlan2'
          lvm2.segmentation_id = 'seg2'
          lvm2.tun_ofports = set(['1', '2'])
          self.agent.vlan_manager.mapping = {'net1': lvm1, 'net2': lvm2}
          self.agent.tun_br_ofports = {'gre':
                                       {'1.1.1.1': '1', '2.2.2.2': '2'}}
          self.agent.arp_responder_enabled = True
  
      def test_fdb_ignore_network(self):
          self._prepare_l2_pop_ofports()
          fdb_entry = {'net3': {}}
          with mock.patch.object(self.agent.tun_br, 'add_flow') as add_flow_fn,\
                  mock.patch.object(self.agent.tun_br,
                                    'uninstall_flows') as del_flow_fn,\
                  mock.patch.object(self.agent,
                                    '_setup_tunnel_port') as add_tun_fn,\
                  mock.patch.object(self.agent,
                                    'cleanup_tunnel_port') as clean_tun_fn:
              self.agent.fdb_add(None, fdb_entry)
              add_flow_fn.assert_not_called()
              add_tun_fn.assert_not_called()
              self.agent.fdb_remove(None, fdb_entry)
              del_flow_fn.assert_not_called()
              clean_tun_fn.assert_not_called()
  
      def test_fdb_ignore_self(self):
          self._prepare_l2_pop_ofports()
          self.agent.local_ip = 'agent_ip'
          fdb_entry = {'net2':
                       {'network_type': 'gre',
                        'segment_id': 'tun2',
                        'ports':
                        {'agent_ip':
                         [l2pop_rpc.PortInfo(FAKE_MAC, FAKE_IP1),
                          n_const.FLOODING_ENTRY]}}}
          with mock.patch.object(self.agent, 'tun_br', autospec=True) as tun_br:
              self.agent.fdb_add(None, fdb_entry)
              tun_br.add_port.assert_not_called()
  
              self.agent.fdb_remove(None, fdb_entry)
              tun_br.delete_port.assert_not_called()
  
      def test_fdb_add_flows(self):
          self._prepare_l2_pop_ofports()
          fdb_entry = {'net1':
                       {'network_type': 'gre',
                        'segment_id': 'tun1',
                        'ports':
                        {'2.2.2.2':
                         [l2pop_rpc.PortInfo(FAKE_MAC, FAKE_IP1),
                          n_const.FLOODING_ENTRY]}}}
  
          with mock.patch.object(self.agent, 'tun_br', autospec=True) as tun_br,\
                  mock.patch.object(self.agent,
                                    '_setup_tunnel_port',
                                    autospec=True) as add_tun_fn:
              self.agent.fdb_add(None, fdb_entry)
              add_tun_fn.assert_not_called()
              expected_calls = [
                  mock.call.install_arp_responder('vlan1', FAKE_IP1, FAKE_MAC),
                  mock.call.install_unicast_to_tun('vlan1', 'seg1', '2',
                                                   FAKE_MAC),
                  mock.call.install_flood_to_tun('vlan1', 'seg1',
                                                 set(['1', '2'])),
              ]
              tun_br.assert_has_calls(expected_calls)
  
      def test_fdb_del_flows(self):
          self._prepare_l2_pop_ofports()
          fdb_entry = {'net2':
                       {'network_type': 'gre',
                        'segment_id': 'tun2',
                        'ports':
                        {'2.2.2.2':
                         [l2pop_rpc.PortInfo(FAKE_MAC, FAKE_IP1),
                          n_const.FLOODING_ENTRY]}}}
          with mock.patch.object(self.agent, 'tun_br', autospec=True) as br_tun:
              self.agent.fdb_remove(None, fdb_entry)
              expected_calls = [
                  mock.call.delete_arp_responder('vlan2', FAKE_IP1),
                  mock.call.delete_unicast_to_tun('vlan2', FAKE_MAC),
                  mock.call.install_flood_to_tun('vlan2', 'seg2', set(['1'])),
                  mock.call.delete_port('gre-02020202'),
                  mock.call.cleanup_tunnel_port('2'),
              ]
              br_tun.assert_has_calls(expected_calls)
  
      def test_fdb_add_port(self):
          self._prepare_l2_pop_ofports()
          fdb_entry = {'net1':
                       {'network_type': 'gre',
                        'segment_id': 'tun1',
                        'ports': {'1.1.1.1': [l2pop_rpc.PortInfo(FAKE_MAC,
                                                                 FAKE_IP1)]}}}
          with mock.patch.object(self.agent, 'tun_br', autospec=True) as tun_br,\
                  mock.patch.object(self.agent,
                                    '_setup_tunnel_port') as add_tun_fn:
              self.agent.fdb_add(None, fdb_entry)
              add_tun_fn.assert_not_called()
              fdb_entry['net1']['ports']['10.10.10.10'] = [
                  l2pop_rpc.PortInfo(FAKE_MAC, FAKE_IP1)]
              self.agent.fdb_add(None, fdb_entry)
              add_tun_fn.assert_called_with(
                  tun_br, 'gre-0a0a0a0a', '10.10.10.10', 'gre')
  
      def test_fdb_del_port(self):
          self._prepare_l2_pop_ofports()
          fdb_entry = {'net2':
                       {'network_type': 'gre',
                        'segment_id': 'tun2',
                        'ports': {'2.2.2.2': [n_const.FLOODING_ENTRY]}}}
          with mock.patch.object(self.agent, 'tun_br', autospec=True) as tun_br:
              self.agent.fdb_remove(None, fdb_entry)
              tun_br.delete_port.assert_called_once_with('gre-02020202')
  
      def test_fdb_update_chg_ip(self):
          self._prepare_l2_pop_ofports()
          fdb_entries = {'chg_ip':
                         {'net1':
                          {'agent_ip':
                           {'before': [l2pop_rpc.PortInfo(FAKE_MAC, FAKE_IP1)],
                            'after': [l2pop_rpc.PortInfo(FAKE_MAC, FAKE_IP2)]}}}}
          with mock.patch.object(self.agent, 'tun_br', autospec=True) as tun_br:
              self.agent.fdb_update(None, fdb_entries)
              tun_br.assert_has_calls([
                  mock.call.install_arp_responder('vlan1', FAKE_IP2, FAKE_MAC),
                  mock.call.delete_arp_responder('vlan1', FAKE_IP1)
              ])
  
      def test_del_fdb_flow_idempotency(self):
          lvm = mock.Mock()
          lvm.network_type = 'gre'
          lvm.vlan = 'vlan1'
          lvm.segmentation_id = 'seg1'
          lvm.tun_ofports = set(['1', '2'])
          with mock.patch.object(self.agent.tun_br, 'mod_flow') as mod_flow_fn,\
                  mock.patch.object(self.agent.tun_br,
                                    'uninstall_flows') as uninstall_flows_fn:
              self.agent.del_fdb_flow(self.agent.tun_br, n_const.FLOODING_ENTRY,
                                      '1.1.1.1', lvm, '3')
              mod_flow_fn.assert_not_called()
              uninstall_flows_fn.assert_not_called()
  
      def test_recl_lv_port_to_preserve(self):
          self._prepare_l2_pop_ofports()
          self.agent.l2_pop = True
          self.agent.enable_tunneling = True
          with mock.patch.object(self.agent, 'tun_br', autospec=True) as tun_br:
              self.agent.reclaim_local_vlan('net1')
              tun_br.cleanup_tunnel_port.assert_not_called()
  
      def test_recl_lv_port_to_remove(self):
          self._prepare_l2_pop_ofports()
          self.agent.l2_pop = True
          self.agent.enable_tunneling = True
          with mock.patch.object(self.agent, 'tun_br', autospec=True) as tun_br:
              self.agent.reclaim_local_vlan('net2')
              tun_br.delete_port.assert_called_once_with('gre-02020202')
  
      def _test_ext_br_recreated(self, setup_bridges_side_effect):
          bridge_mappings = {'physnet0': 'br-ex0',
                             'physnet1': 'br-ex1'}
          ex_br_mocks = [mock.Mock(br_name='br-ex0'),
                         mock.Mock(br_name='br-ex1')]
          phys_bridges = {'physnet0': ex_br_mocks[0],
                          'physnet1': ex_br_mocks[1]},
          bridges_added = ['br-ex0']
          expected_added_bridges = (
              bridges_added if setup_bridges_side_effect else [])
          with mock.patch.object(self.agent, 'check_ovs_status',
                                 return_value=constants.OVS_NORMAL), \
                  mock.patch.object(self.agent, '_agent_has_updates',
                                    side_effect=TypeError('loop exit')), \
                  mock.patch.dict(self.agent.bridge_mappings, bridge_mappings,
                                  clear=True), \
                  mock.patch.dict(self.agent.phys_brs, phys_bridges,
                                  clear=True), \
                  mock.patch.object(self.agent, 'setup_physical_bridges') as \
                  setup_physical_bridges, \
                  mock.patch.object(self.agent.ovs.ovsdb, 'idl_monitor') as \
                  mock_idl_monitor:
              mock_idl_monitor.bridges_added = bridges_added
              setup_physical_bridges.side_effect = setup_bridges_side_effect
              try:
                  self.agent.rpc_loop(polling_manager=mock.Mock())
              except TypeError:
                  pass
          # Setup bridges should be called once even if it will raise Runtime
          # Error because there is raised TypeError in _agent_has_updates to stop
          # agent after first loop iteration
          setup_physical_bridges.assert_called_once_with({'physnet0': 'br-ex0'})
          self.assertEqual(expected_added_bridges, self.agent.added_bridges)
  
      def test_ext_br_recreated(self):
          self._test_ext_br_recreated(setup_bridges_side_effect=None)
  
      def test_ext_br_recreated_fail_setup_physical_bridge(self):
          self._test_ext_br_recreated(setup_bridges_side_effect=RuntimeError)
  
      def test_daemon_loop_uses_polling_manager(self):
          ex_br_mock = mock.Mock(br_name="br-ex0")
          with mock.patch.object(polling, 'get_polling_manager') as \
                  mock_get_pm, \
                  mock.patch.object(self.agent, 'rpc_loop') as mock_loop, \
                  mock.patch.dict(self.agent.phys_brs, {'physnet0': ex_br_mock},
                                  clear=True), \
                  mock.patch.object(self.agent.ovs.ovsdb, 'idl_monitor') as \
                  mock_idl_monitor:
              self.agent.daemon_loop()
          mock_get_pm.assert_called_with(True,
                                         constants.DEFAULT_OVSDBMON_RESPAWN)
          mock_loop.assert_called_once_with(polling_manager=mock.ANY)
          mock_idl_monitor.start_bridge_monitor.assert_called()
  
      def test_setup_tunnel_port_invalid_ofport(self):
          remote_ip = '1.2.3.4'
          with mock.patch.object(
              self.agent.tun_br,
              'add_tunnel_port',
              return_value=ovs_lib.INVALID_OFPORT) as add_tunnel_port_fn,\
                  mock.patch.object(self.mod_agent.LOG, 'error') as log_error_fn:
              self.agent.local_ip = '1.2.3.4'
              ofport = self.agent._setup_tunnel_port(
                  self.agent.tun_br, 'gre-1', remote_ip, n_const.TYPE_GRE)
              add_tunnel_port_fn.assert_called_once_with(
                  'gre-1', remote_ip, self.agent.local_ip, n_const.TYPE_GRE,
                  self.agent.vxlan_udp_port, self.agent.dont_fragment,
                  self.agent.tunnel_csum, self.agent.tos)
              log_error_fn.assert_called_once_with(
                  _("Failed to set-up %(type)s tunnel port to %(ip)s"),
                  {'type': n_const.TYPE_GRE, 'ip': remote_ip})
              self.assertEqual(0, ofport)
  
      def test_setup_tunnel_port_invalid_address_mismatch(self):
          remote_ip = '2001:db8::2'
          with mock.patch.object(self.mod_agent.LOG, 'error') as log_error_fn:
              self.agent.local_ip = '1.2.3.4'
              ofport = self.agent._setup_tunnel_port(
                  self.agent.tun_br, 'gre-1', remote_ip, n_const.TYPE_GRE)
              log_error_fn.assert_called_once_with(
                  _("IP version mismatch, cannot create tunnel: "
                    "local_ip=%(lip)s remote_ip=%(rip)s"),
                  {'lip': self.agent.local_ip, 'rip': remote_ip})
              self.assertEqual(0, ofport)
  
      def test_setup_tunnel_port_invalid_netaddr_exception(self):
          remote_ip = '2001:db8::2'
          with mock.patch.object(self.mod_agent.LOG, 'error') as log_error_fn:
              self.agent.local_ip = '1.2.3.4.5'
              ofport = self.agent._setup_tunnel_port(
                  self.agent.tun_br, 'gre-1', remote_ip, n_const.TYPE_GRE)
              log_error_fn.assert_called_once_with(
                  _("Invalid local or remote IP, cannot create tunnel: "
                    "local_ip=%(lip)s remote_ip=%(rip)s"),
                  {'lip': self.agent.local_ip, 'rip': remote_ip})
              self.assertEqual(0, ofport)
  
      def test_setup_tunnel_port_error_negative_df_disabled(self):
          remote_ip = '1.2.3.4'
          with mock.patch.object(
              self.agent.tun_br,
              'add_tunnel_port',
              return_value=ovs_lib.INVALID_OFPORT) as add_tunnel_port_fn,\
                  mock.patch.object(self.mod_agent.LOG, 'error') as log_error_fn:
              self.agent.dont_fragment = False
              self.agent.tunnel_csum = False
              self.agent.local_ip = '2.3.4.5'
              ofport = self.agent._setup_tunnel_port(
                  self.agent.tun_br, 'gre-1', remote_ip, n_const.TYPE_GRE)
              add_tunnel_port_fn.assert_called_once_with(
                  'gre-1', remote_ip, self.agent.local_ip, n_const.TYPE_GRE,
                  self.agent.vxlan_udp_port, self.agent.dont_fragment,
                  self.agent.tunnel_csum, self.agent.tos)
              log_error_fn.assert_called_once_with(
                  _("Failed to set-up %(type)s tunnel port to %(ip)s"),
                  {'type': n_const.TYPE_GRE, 'ip': remote_ip})
              self.assertEqual(0, ofport)
  
      def test_setup_tunnel_port_error_negative_tunnel_csum(self):
          remote_ip = '1.2.3.4'
          with mock.patch.object(
              self.agent.tun_br,
              'add_tunnel_port',
              return_value=ovs_lib.INVALID_OFPORT) as add_tunnel_port_fn,\
                  mock.patch.object(self.mod_agent.LOG, 'error') as log_error_fn:
              self.agent.dont_fragment = True
              self.agent.tunnel_csum = True
              self.agent.local_ip = '2.3.4.5'
              ofport = self.agent._setup_tunnel_port(
                  self.agent.tun_br, 'gre-1', remote_ip, n_const.TYPE_GRE)
              add_tunnel_port_fn.assert_called_once_with(
                  'gre-1', remote_ip, self.agent.local_ip, n_const.TYPE_GRE,
                  self.agent.vxlan_udp_port, self.agent.dont_fragment,
                  self.agent.tunnel_csum, self.agent.tos)
              log_error_fn.assert_called_once_with(
                  _("Failed to set-up %(type)s tunnel port to %(ip)s"),
                  {'type': n_const.TYPE_GRE, 'ip': remote_ip})
              self.assertEqual(0, ofport)
  
      def test_setup_tunnel_port_error_negative_tos_inherit(self):
          remote_ip = '1.2.3.4'
          with mock.patch.object(
              self.agent.tun_br,
              'add_tunnel_port',
              return_value=ovs_lib.INVALID_OFPORT) as add_tunnel_port_fn,\
                  mock.patch.object(self.mod_agent.LOG, 'error') as log_error_fn:
              self.agent.tos = 'inherit'
              self.agent.local_ip = '2.3.4.5'
              ofport = self.agent._setup_tunnel_port(
                  self.agent.tun_br, 'gre-1', remote_ip, n_const.TYPE_GRE)
              add_tunnel_port_fn.assert_called_once_with(
                  'gre-1', remote_ip, self.agent.local_ip, n_const.TYPE_GRE,
                  self.agent.vxlan_udp_port, self.agent.dont_fragment,
                  self.agent.tunnel_csum, self.agent.tos)
              log_error_fn.assert_called_once_with(
                  _("Failed to set-up %(type)s tunnel port to %(ip)s"),
                  {'type': n_const.TYPE_GRE, 'ip': remote_ip})
              self.assertEqual(0, ofport)
  
      def test_tunnel_sync_with_ml2_plugin(self):
          fake_tunnel_details = {'tunnels': [{'ip_address': '100.101.31.15'}]}
          with mock.patch.object(self.agent.plugin_rpc,
                                 'tunnel_sync',
                                 return_value=fake_tunnel_details),\
                  mock.patch.object(
                      self.agent,
                      '_setup_tunnel_port') as _setup_tunnel_port_fn,\
                  mock.patch.object(self.agent,
                                    'cleanup_stale_flows') as cleanup:
              self.agent.tunnel_types = ['vxlan']
              self.agent.tunnel_sync()
              expected_calls = [mock.call(self.agent.tun_br, 'vxlan-64651f0f',
                                          '100.101.31.15', 'vxlan')]
              _setup_tunnel_port_fn.assert_has_calls(expected_calls)
              cleanup.assert_not_called()
  
      def test_tunnel_sync_invalid_ip_address(self):
          fake_tunnel_details = {'tunnels': [{'ip_address': '300.300.300.300'},
                                             {'ip_address': '100.100.100.100'}]}
          with mock.patch.object(self.agent.plugin_rpc,
                                 'tunnel_sync',
                                 return_value=fake_tunnel_details),\
                  mock.patch.object(
                      self.agent,
                      '_setup_tunnel_port') as _setup_tunnel_port_fn,\
                  mock.patch.object(self.agent,
                                    'cleanup_stale_flows') as cleanup:
              self.agent.tunnel_types = ['vxlan']
              self.agent.tunnel_sync()
              _setup_tunnel_port_fn.assert_called_once_with(self.agent.tun_br,
                                                            'vxlan-64646464',
                                                            '100.100.100.100',
                                                            'vxlan')
              cleanup.assert_not_called()
  
      def test_tunnel_sync_setup_tunnel_flood_flow_once(self):
          fake_tunnel_details = {'tunnels': [{'ip_address': '200.200.200.200'},
                                             {'ip_address': '100.100.100.100'}]}
          with mock.patch.object(self.agent.plugin_rpc,
                                 'tunnel_sync',
                                 return_value=fake_tunnel_details),\
                  mock.patch.object(
                      self.agent,
                      '_setup_tunnel_port') as _setup_tunnel_port_fn,\
                  mock.patch.object(
                      self.agent,
                      '_setup_tunnel_flood_flow') as _setup_tunnel_flood_flow:
              self.agent.tunnel_types = ['vxlan']
              self.agent.tunnel_sync()
              expected_calls = [mock.call(self.agent.tun_br, 'vxlan-c8c8c8c8',
                                          '200.200.200.200', 'vxlan'),
                                mock.call(self.agent.tun_br, 'vxlan-64646464',
                                          '100.100.100.100', 'vxlan')]
              _setup_tunnel_port_fn.assert_has_calls(expected_calls)
              _setup_tunnel_flood_flow.assert_called_once_with(self.agent.tun_br,
                                                               'vxlan')
  
      def test_tunnel_update(self):
          kwargs = {'tunnel_ip': '10.10.10.10',
                    'tunnel_type': 'gre'}
          self.agent._setup_tunnel_port = mock.Mock()
          self.agent.enable_tunneling = True
          self.agent.tunnel_types = ['gre']
          self.agent.l2_pop = False
          self.agent.tunnel_update(context=None, **kwargs)
          expected_calls = [
              mock.call(self.agent.tun_br, 'gre-0a0a0a0a', '10.10.10.10', 'gre')]
          self.agent._setup_tunnel_port.assert_has_calls(expected_calls)
  
      def test_tunnel_delete(self):
          kwargs = {'tunnel_ip': '10.10.10.10',
                    'tunnel_type': 'gre'}
          self.agent.enable_tunneling = True
          self.agent.tunnel_types = ['gre']
          self.agent.tun_br_ofports = {'gre': {'10.10.10.10': '1'}}
          with mock.patch.object(
              self.agent, 'cleanup_tunnel_port'
          ) as clean_tun_fn:
              self.agent.tunnel_delete(context=None, **kwargs)
              self.assertTrue(clean_tun_fn.called)
  
      def test_reset_tunnel_ofports(self):
          tunnel_handles = self.agent.tun_br_ofports
          self.agent.tun_br_ofports = {'gre': {'10.10.10.10': '1'}}
          self.agent._reset_tunnel_ofports()
          self.assertEqual(self.agent.tun_br_ofports, tunnel_handles)
  
      def _test_ovs_status(self, *args):
          reply2 = {'current': set(['tap0']),
                    'added': set(['tap2']),
                    'removed': set([])}
  
          reply3 = {'current': set(['tap2']),
                    'added': set([]),
                    'removed': set(['tap0'])}
  
          reply_ancillary = {'current': set([]),
                             'added': set([]),
                             'removed': set([])}
  
          self.agent.enable_tunneling = True
  
          with mock.patch.object(async_process.AsyncProcess, "_spawn"),\
                  mock.patch.object(async_process.AsyncProcess, "start"),\
                  mock.patch.object(async_process.AsyncProcess,
                                    "is_active", return_value=True),\
                  mock.patch.object(async_process.AsyncProcess, "stop"),\
                  mock.patch.object(log.KeywordArgumentAdapter,
                                    'exception') as log_exception,\
                  mock.patch.object(self.mod_agent.OVSNeutronAgent,
                                    'process_ports_events') as process_p_events,\
                  mock.patch.object(
                      self.mod_agent.OVSNeutronAgent,
                      'process_network_ports') as process_network_ports,\
                  mock.patch.object(self.mod_agent.OVSNeutronAgent,
                                    'check_ovs_status') as check_ovs_status,\
                  mock.patch.object(self.mod_agent.OVSNeutronAgent,
                                    'setup_integration_br') as setup_int_br,\
                  mock.patch.object(self.mod_agent.OVSNeutronAgent,
                                    'setup_physical_bridges') as setup_phys_br,\
                  mock.patch.object(time, 'sleep'),\
                  mock.patch.object(
                      self.mod_agent.OVSNeutronAgent,
                      'update_stale_ofport_rules') as update_stale, \
                  mock.patch.object(self.mod_agent.OVSNeutronAgent,
                                    'cleanup_stale_flows') as cleanup, \
                  mock.patch.object(self.mod_agent.OVSNeutronAgent,
                                    'setup_tunnel_br') as setup_tunnel_br,\
                  mock.patch.object(
                      self.mod_agent.OVSNeutronAgent,
                      'setup_tunnel_br_flows') as setup_tunnel_br_flows,\
                  mock.patch.object(
                      self.mod_agent.OVSNeutronAgent,
                      '_reset_tunnel_ofports') as reset_tunnel_ofports, \
                  mock.patch.object(self.agent.ovs.ovsdb, 'idl_monitor'),\
                  mock.patch.object(self.agent.state_rpc,
                                    'report_state') as report_st:
              log_exception.side_effect = Exception(
                  'Fake exception to get out of the loop')
              devices_not_ready = set()
              process_p_events.side_effect = [(reply2, reply_ancillary,
                                               devices_not_ready),
                                              (reply3, reply_ancillary,
                                               devices_not_ready)]
              failed_devices = {'added': set(), 'removed': set()}
              failed_ancillary_devices = {'added': set(), 'removed': set()}
              process_network_ports.side_effect = [
                  failed_devices,
                  Exception('Fake exception to get out of the loop')]
              check_ovs_status.side_effect = args
  
              if self.agent.enable_tunneling:
                  self.agent.agent_state.pop("start_flag")
  
              try:
                  self.agent.daemon_loop()
              except Exception:
                  pass
  
              process_p_events.assert_has_calls([
                  mock.call({'removed': [], 'added': [], 'modified': []},
                            set(), set(), set(),
                            failed_devices, failed_ancillary_devices,
                            set()),
                  mock.call({'removed': [], 'added': [], 'modified': []},
                            set(['tap0']), set(),
                            set(), failed_devices, failed_ancillary_devices,
                            set())
              ])
  
              process_network_ports.assert_has_calls([
                  mock.call(reply2, False),
                  mock.call(reply3, True)
              ])
              cleanup.assert_called_once_with()
              self.assertTrue(update_stale.called)
              # Verify the OVS restart we triggered in the loop
              # re-setup the bridges
              setup_int_br.assert_has_calls([mock.call()])
              setup_phys_br.assert_has_calls([mock.call({})])
              # Ensure that tunnel handles are reset and bridge
              # and flows reconfigured.
              self.assertTrue(reset_tunnel_ofports.called)
              self.assertTrue(setup_tunnel_br_flows.called)
              self.assertTrue(setup_tunnel_br.called)
              if self.agent.enable_tunneling:
                  self.agent.agent_state['start_flag'] = True
                  report_st.assert_called_once_with(
                      self.agent.context, self.agent.agent_state, True)
  
      def test_ovs_status(self):
          self._test_ovs_status(constants.OVS_NORMAL,
                                constants.OVS_DEAD,
                                constants.OVS_RESTARTED)
          # OVS will not DEAD in some exception, like DBConnectionError.
          self._test_ovs_status(constants.OVS_NORMAL,
                                constants.OVS_RESTARTED)
  
      def test_rpc_loop_fail_to_process_network_ports_keep_flows(self):
          with mock.patch.object(async_process.AsyncProcess, "_spawn"),\
                  mock.patch.object(async_process.AsyncProcess, "start"),\
                  mock.patch.object(async_process.AsyncProcess,
                                    "is_active", return_value=True),\
                  mock.patch.object(async_process.AsyncProcess, "stop"),\
                  mock.patch.object(
                      self.mod_agent.OVSNeutronAgent,
                      'process_network_ports') as process_network_ports,\
                  mock.patch.object(self.mod_agent.OVSNeutronAgent,
                                    'check_ovs_status') as check_ovs_status,\
                  mock.patch.object(time, 'sleep'),\
                  mock.patch.object(
                      self.mod_agent.OVSNeutronAgent,
                      'update_stale_ofport_rules') as update_stale, \
                  mock.patch.object(self.mod_agent.OVSNeutronAgent,
                                    'cleanup_stale_flows') as cleanup,\
                  mock.patch.object(
                      self.mod_agent.OVSNeutronAgent,
                      '_check_and_handle_signal') as check_and_handle_signal, \
                  mock.patch.object(self.agent.ovs.ovsdb, 'idl_monitor'):
              process_network_ports.side_effect = Exception("Trigger resync")
              check_ovs_status.return_value = constants.OVS_NORMAL
              check_and_handle_signal.side_effect = [True, False]
              self.agent.daemon_loop()
              self.assertTrue(update_stale.called)
              cleanup.assert_not_called()
  
      def test_set_rpc_timeout(self):
          with mock.patch.object(n_rpc.BackingOffClient,
                                 'set_max_timeout') as smt:
              self.agent._handle_sigterm(None, None)
              for rpc_client in (self.agent.plugin_rpc.client,
                                 self.agent.sg_plugin_rpc.client,
                                 self.agent.dvr_plugin_rpc.client,
                                 self.agent.state_rpc.client):
                  smt.assert_called_with(10)
  
      def test_set_rpc_timeout_no_value(self):
          self.agent.quitting_rpc_timeout = None
          with mock.patch.object(self.agent, 'set_rpc_timeout') as mock_set_rpc:
              self.agent._handle_sigterm(None, None)
          mock_set_rpc.assert_not_called()
  
      def test_arp_spoofing_network_port(self):
          int_br = mock.create_autospec(self.agent.int_br)
          self.agent.setup_arp_spoofing_protection(
              int_br, FakeVif(),
              {'device_owner': n_const.DEVICE_OWNER_ROUTER_INTF})
          self.assertTrue(int_br.delete_arp_spoofing_protection.called)
          int_br.install_arp_spoofing_protection.assert_not_called()
  
      def test_arp_spoofing_port_security_disabled(self):
          int_br = mock.create_autospec(self.agent.int_br)
          self.agent.setup_arp_spoofing_protection(
              int_br, FakeVif(), {'port_security_enabled': False})
          self.assertTrue(int_br.delete_arp_spoofing_protection.called)
          int_br.install_arp_spoofing_protection.assert_not_called()
  
      def test_arp_spoofing_basic_rule_setup(self):
          vif = FakeVif()
          fake_details = {'fixed_ips': [], 'device_owner': 'nobody'}
          self.agent.prevent_arp_spoofing = True
          int_br = mock.create_autospec(self.agent.int_br)
          self.agent.setup_arp_spoofing_protection(int_br, vif, fake_details)
          int_br.delete_arp_spoofing_allow_rules.assert_has_calls(
              [mock.call(port=vif.ofport)])
          int_br.install_arp_spoofing_protection.assert_has_calls(
              [mock.call(ip_addresses=set(), port=vif.ofport)])
  
      def test_arp_spoofing_basic_rule_setup_fixed_ipv6(self):
          vif = FakeVif()
          fake_details = {'fixed_ips': [{'ip_address': 'fdf8:f53b:82e4::1'}],
                          'device_owner': 'nobody'}
          self.agent.prevent_arp_spoofing = True
          br = mock.create_autospec(self.agent.int_br)
          self.agent.setup_arp_spoofing_protection(br, vif, fake_details)
          br.delete_arp_spoofing_allow_rules.assert_has_calls(
              [mock.call(port=vif.ofport)])
          self.assertTrue(br.install_icmpv6_na_spoofing_protection.called)
  
      def test_arp_spoofing_fixed_and_allowed_addresses(self):
          vif = FakeVif()
          fake_details = {
              'device_owner': 'nobody',
              'fixed_ips': [{'ip_address': '192.168.44.100'},
                            {'ip_address': '192.168.44.101'}],
              'allowed_address_pairs': [{'ip_address': '192.168.44.102/32'},
                                        {'ip_address': '192.168.44.103/32'}]
          }
          self.agent.prevent_arp_spoofing = True
          int_br = mock.create_autospec(self.agent.int_br)
          self.agent.setup_arp_spoofing_protection(int_br, vif, fake_details)
          # make sure all addresses are allowed
          addresses = {'192.168.44.100', '192.168.44.101', '192.168.44.102/32',
                       '192.168.44.103/32'}
          int_br.install_arp_spoofing_protection.assert_has_calls(
              [mock.call(port=vif.ofport, ip_addresses=addresses)])
  
      def test_arp_spoofing_fixed_and_allowed_addresses_ipv6(self):
          vif = FakeVif()
          fake_details = {
              'device_owner': 'nobody',
              'fixed_ips': [{'ip_address': '2001:db8::1'},
                            {'ip_address': '2001:db8::2'}],
              'allowed_address_pairs': [{'ip_address': '2001:db8::200',
                                         'mac_address': 'aa:22:33:44:55:66'}]
          }
          self.agent.prevent_arp_spoofing = True
          int_br = mock.create_autospec(self.agent.int_br)
          self.agent.setup_arp_spoofing_protection(int_br, vif, fake_details)
          # make sure all addresses are allowed including ipv6 LLAs
          addresses = {'2001:db8::1', '2001:db8::2', '2001:db8::200',
                       'fe80::a822:33ff:fe44:5566', 'fe80::a8bb:ccff:fe11:2233'}
          int_br.install_icmpv6_na_spoofing_protection.assert_has_calls(
              [mock.call(port=vif.ofport, ip_addresses=addresses)])
  
      def test__get_ofport_moves(self):
          previous = {'port1': 1, 'port2': 2}
          current = {'port1': 5, 'port2': 2}
          # we expect it to tell us port1 moved
          expected = ['port1']
          self.assertEqual(expected,
                           self.agent._get_ofport_moves(current, previous))
  
      def test_update_stale_ofport_rules_clears_old(self):
          self.agent.prevent_arp_spoofing = True
          self.agent.vifname_to_ofport_map = {'port1': 1, 'port2': 2}
          self.agent.int_br = mock.Mock()
          # simulate port1 was removed
          newmap = {'port2': 2}
          self.agent.int_br.get_vif_port_to_ofport_map.return_value = newmap
          self.agent.update_stale_ofport_rules()
          # rules matching port 1 should have been deleted
          self.agent.int_br.delete_arp_spoofing_protection.assert_has_calls(
              [mock.call(port=1)])
          # make sure the state was updated with the new map
          self.assertEqual(newmap, self.agent.vifname_to_ofport_map)
  
      def test_update_stale_ofport_rules_treats_moved(self):
          self.agent.prevent_arp_spoofing = True
          self.agent.vifname_to_ofport_map = {'port1': 1, 'port2': 2}
          self.agent.treat_devices_added_or_updated = mock.Mock()
          self.agent.int_br = mock.Mock()
          # simulate port1 was moved
          newmap = {'port2': 2, 'port1': 90}
          self.agent.int_br.get_vif_port_to_ofport_map.return_value = newmap
          ofport_changed_ports = self.agent.update_stale_ofport_rules()
          self.assertEqual(['port1'], ofport_changed_ports)
  
      def test_update_stale_ofport_rules_removes_drop_flow(self):
          self.agent.prevent_arp_spoofing = False
          self.agent.vifname_to_ofport_map = {'port1': 1, 'port2': 2}
          self.agent.int_br = mock.Mock()
          # simulate port1 was removed
          newmap = {'port2': 2}
          self.agent.int_br.get_vif_port_to_ofport_map.return_value = newmap
          self.agent.update_stale_ofport_rules()
          # drop flow rule matching port 1 should have been deleted
          ofport_changed_ports = self.agent.update_stale_ofport_rules()
          expected = [
              mock.call(in_port=1)
          ]
          self.agent.int_br.uninstall_flows.assert_has_calls(expected)
          self.assertEqual(newmap, self.agent.vifname_to_ofport_map)
          self.agent.int_br.delete_arp_spoofing_protection.assert_not_called()
          self.assertEqual([], ofport_changed_ports)
  
      def test__setup_tunnel_port_while_new_mapping_is_added(self):
          """Test setup_tunnel_port while adding a new mapping
  
          Test that _setup_tunnel_port doesn't fail if new vlan mapping is
          added in a different coroutine while iterating over existing mappings.
          See bug 1449944 for more info.
          """
  
          def add_new_vlan_mapping(*args, **kwargs):
              self.agent.vlan_manager.add('bar', 1, 2, 3, 4)
          bridge = mock.Mock()
          tunnel_type = 'vxlan'
          self.agent.tun_br_ofports = {tunnel_type: dict()}
          self.agent.l2_pop = False
          self.agent.vlan_manager.add('foo', 4, tunnel_type, 2, 1)
          self.agent.local_ip = '2.3.4.5'
          bridge.install_flood_to_tun.side_effect = add_new_vlan_mapping
          self.agent._setup_tunnel_port(bridge, 1, '1.2.3.4',
                                        tunnel_type=tunnel_type)
          self.agent._setup_tunnel_flood_flow(bridge, tunnel_type)
          self.assertIn('bar', self.agent.vlan_manager)
  
      def test_setup_entry_for_arp_reply_ignores_ipv6_addresses(self):
          self.agent.arp_responder_enabled = True
          ip = '2001:db8::1'
          br = mock.Mock()
          self.agent.setup_entry_for_arp_reply(
              br, 'add', mock.Mock(), mock.Mock(), ip)
          br.install_arp_responder.assert_not_called()
  
      def test_configurations_has_rp_bandwidth(self):
          self.assertIn(n_const.RP_BANDWIDTHS,
                        self.agent.agent_state['configurations'])
  
      def test_configurations_has_rp_default_inventory(self):
          self.assertIn(n_const.RP_INVENTORY_DEFAULTS,
                        self.agent.agent_state['configurations'])
          rp_inv_defaults = \
              self.agent.agent_state['configurations'][
                  n_const.RP_INVENTORY_DEFAULTS]
          self.assertListEqual(
              sorted(['reserved', 'min_unit', 'allocation_ratio', 'step_size']),
              sorted(list(rp_inv_defaults)))
          self.assertEqual(1.0, rp_inv_defaults['allocation_ratio'])
          self.assertEqual(1, rp_inv_defaults['min_unit'])
          self.assertEqual(1, rp_inv_defaults['step_size'])
          self.assertEqual(0, rp_inv_defaults['reserved'])
  
      def test__validate_rp_bandwidth_bridges(self):
          cfg.CONF.set_override('bridge_mappings', [], 'OVS')
          cfg.CONF.set_override(n_const.RP_BANDWIDTHS,
                                ['no_such_br_in_bridge_mappings:1:1'],
                                'OVS')
          self.assertRaises(ValueError, self._make_agent)
  
      def test__check_bridge_datapath_id(self):
          datapath_id = u'0000622486fa3f42'
          datapath_ids_set = set()
          for i in range(5):
              dpid = format((i << 48) + int(datapath_id, 16), '0x').zfill(16)
              bridge = mock.Mock()
              bridge.br_name = 'bridge_%s' % i
              bridge.get_datapath_id = mock.Mock(return_value=datapath_id)
              self.agent._check_bridge_datapath_id(bridge, datapath_ids_set)
              self.assertEqual(i + 1, len(datapath_ids_set))
              self.assertIn(dpid, datapath_ids_set)
              if i == 0:
                  bridge.set_datapath_id.assert_not_called()
              else:
                  bridge.set_datapath_id.assert_called_once_with(dpid)
  
      def test__update_network_segmentation_id(self):
          network = {'id': 'my-net-uuid',
                     provider_net.SEGMENTATION_ID: 1005,
                     provider_net.PHYSICAL_NETWORK: 'provider_net',
                     provider_net.NETWORK_TYPE: n_const.TYPE_VLAN}
          self.agent.vlan_manager.add('my-net-uuid', 5, n_const.TYPE_VLAN,
                                      'provider_net', 1004, None)
          mock_phys_br = mock.Mock()
          self.agent.phys_brs['provider_net'] = mock_phys_br
          self.agent.phys_ofports['provider_net'] = 'phy_ofport'
          self.agent.int_ofports['provider_net'] = 'int_ofport'
  
          with mock.patch.object(self.agent.int_br, 'reclaim_local_vlan') \
                  as mock_reclaim_local_vlan, \
                  mock.patch.object(self.agent.int_br, 'provision_local_vlan') \
                  as mock_provision_local_vlan:
              self.agent._update_network_segmentation_id(network)
              mock_reclaim_local_vlan.assert_called_once_with(
                  port='int_ofport', segmentation_id=1004)
              mock_provision_local_vlan.assert_called_once_with(
                  port='int_ofport', lvid=5, segmentation_id=1005)
          mock_phys_br.reclaim_local_vlan.assert_called_once_with(
              port='phy_ofport', lvid=5)
  
      def test__update_network_segmentation_id_not_vlan(self):
          network = {provider_net.NETWORK_TYPE: 'not_vlan'}
          with mock.patch.object(self.agent.vlan_manager, 'get') as mock_get:
              self.agent._update_network_segmentation_id(network)
              mock_get.assert_not_called()
  
      def test__update_network_segmentation_id_vlan_not_found(self):
          network = {'id': 'my-net-uuid',
                     provider_net.SEGMENTATION_ID: 1005,
                     provider_net.NETWORK_TYPE: n_const.TYPE_VLAN,
                     provider_net.PHYSICAL_NETWORK: 'default_network'}
          with mock.patch.object(self.agent.vlan_manager,
                                 'update_segmentation_id') as mock_update_segid:
              self.agent._update_network_segmentation_id(network)
              mock_update_segid.assert_not_called()
  
      def test__update_network_segmentation_id_segmentation_id_not_updated(self):
          network = {'id': 'my-net-uuid',
                     provider_net.SEGMENTATION_ID: 1005,
                     provider_net.NETWORK_TYPE: n_const.TYPE_VLAN,
                     provider_net.PHYSICAL_NETWORK: 'default_network'}
          self.agent.vlan_manager.add('my-net-uuid', 5, n_const.TYPE_VLAN,
                                      'provider_net', 1005, None)
          with mock.patch.object(self.agent.vlan_manager,
                                 'update_segmentation_id') as mock_update_segid:
              self.agent._update_network_segmentation_id(network)
              mock_update_segid.assert_not_called()
  
      def test__update_network_segmentation_id_multisegments(self):
          network = {'id': 'my-net-uuid',
                     'segments': [mock.ANY]}
          with mock.patch.object(self.agent.vlan_manager,
                                 'get') as mock_get:
              self.agent._update_network_segmentation_id(network)
              mock_get.assert_not_called()
  
      def _test_treat_smartnic_port(self, vif_type):
          vm_uuid = "407a79e0-e0be-4b7d-92a6-513b2161011b"
          iface_id = "407a79e0-e0be-4b7d-92a6-513b2161011c"
          rep_port = 'rep0-0'
          mac = FAKE_MAC
          smartnic_data = {
              'mac': mac,
              'vm_uuid': vm_uuid,
              'vif_name': rep_port,
              'iface_id': iface_id,
              'vif_type': vif_type}
  
          cfg.CONF.set_default('baremetal_smartnic', True, group='AGENT')
          agent = self._make_agent()
          instance_info = vif_instance_object.InstanceInfo(uuid=vm_uuid)
          vif = agent._get_vif_object(iface_id, rep_port, mac)
          with mock.patch.object(os_vif, 'plug') as plug_mock, \
                  mock.patch.object(os_vif, 'unplug') as unplug_mock, \
                  mock.patch('os_vif.objects.instance_info.InstanceInfo',
                             return_value=instance_info), \
                  mock.patch.object(agent, '_get_vif_object',
                                    return_value=vif):
  
              agent.treat_smartnic_port(smartnic_data)
  
              if vif_type == portbindings.VIF_TYPE_OVS:
                  plug_mock.assert_called_once_with(vif, instance_info)
              else:
                  unplug_mock.assert_called_once_with(vif, instance_info)
  
      def test_treat_smartnic_port_add(self):
          self._test_treat_smartnic_port('ovs')
  
      def test_treat_smartnic_port_remove(self):
          self._test_treat_smartnic_port('unbound')
  
      def test_process_smartnic_ports_remove(self):
          port_id = "407a79e0-e0be-4b7d-92a6-513b2161011a"
          rep_port = 'rep0-0'
          mac = FAKE_MAC
          ovs_port = mock.Mock()
          ovs_port.vif_mac = mac
          ovs_port.port_name = rep_port
          ovs_port.vif_id = port_id
          ports_int_br = [ovs_port]
          expected_smartnic_ports_processed_list = [
              {'iface_id': port_id,
               'vif_name': rep_port,
               'mac': mac,
               'vif_type': portbindings.VIF_TYPE_UNBOUND,
               'vm_uuid': ''}]
          expected_current_smartnic_ports_map = {
              port_id: {
                  'vif_mac': mac,
                  'vif_name': rep_port}}
          with mock.patch.object(self.agent.plugin_rpc,
                                 "get_ports_by_vnic_type_and_host",
                                 return_value=[]),\
                  mock.patch.object(self.agent.int_br,
                                    "get_vif_ports",
                                    return_value=ports_int_br):
              self.agent.process_smartnic_ports()
              self.assertEqual(expected_smartnic_ports_processed_list,
                               self.agent.updated_smartnic_ports)
              self.assertEqual(expected_current_smartnic_ports_map,
                               self.agent.current_smartnic_ports_map)
  
      def test_process_smartnic_ports(self):
          port_id = "407a79e0-e0be-4b7d-92a6-513b2161011a"
          rep_port = 'rep0-0'
          mac = FAKE_MAC
          ovs_port = mock.Mock()
          ovs_port.vif_mac = mac
          ovs_port.port_name = rep_port
          ovs_port.vif_id = port_id
          ports_int_br = [ovs_port]
  
          PORT_TO_PROCESS = {
              'binding:profile': {'local_link_information': [
                  {'hostname': 'host1', 'port_id': rep_port}]},
              'mac_address': FAKE_MAC,
              'device_id': "407a79e0-e0be-4b7d-92a6-513b2161011e",
              'id': "407a79e0-e0be-4b7d-92a6-513b2161011c",
              'binding:vif_type': portbindings.VIF_TYPE_OVS
          }
          expected_smartnic_ports_processed_list = [
              {'iface_id': port_id,
               'vif_name': rep_port,
               'mac': mac,
               'vif_type': portbindings.VIF_TYPE_UNBOUND,
               'vm_uuid': ''},
              {'iface_id': "407a79e0-e0be-4b7d-92a6-513b2161011c",
               'vif_name': rep_port,
               'mac': mac,
               'vif_type': portbindings.VIF_TYPE_OVS,
               'vm_uuid': "407a79e0-e0be-4b7d-92a6-513b2161011e"}]
          expected_current_smartnic_ports_map = {
              port_id: {
                  'vif_mac': mac,
                  'vif_name': rep_port}}
          with mock.patch.object(self.agent.plugin_rpc,
                                 "get_ports_by_vnic_type_and_host",
                                 return_value=[PORT_TO_PROCESS]),\
              mock.patch.object(self.agent.int_br,
                                "get_vif_ports",
                                return_value=ports_int_br):
              self.agent.process_smartnic_ports()
              self.assertEqual(expected_smartnic_ports_processed_list,
                               self.agent.updated_smartnic_ports)
              self.assertEqual(expected_current_smartnic_ports_map,
                               self.agent.current_smartnic_ports_map)
  
      def test_add_bound_port_to_updated_smartnic_ports(self):
          mac = FAKE_MAC
          vm_uuid = "407a79e0-e0be-4b7d-92a6-513b2161011b"
          rep_port = 'rep0-0'
          iface_id = "407a79e0-e0be-4b7d-92a6-513b2161011c"
          self.agent._add_port_to_updated_smartnic_ports(
              mac,
              rep_port,
              iface_id,
              portbindings.VIF_TYPE_OVS,
              vm_uuid,)
  
          smartnic_data = {
              'mac': mac,
              'vm_uuid': vm_uuid,
              'vif_name': rep_port,
              'iface_id': iface_id,
              'vif_type': portbindings.VIF_TYPE_OVS}
          self.assertEqual([smartnic_data],
                           self.agent.updated_smartnic_ports)
  
      def test_add_unbound_port_to_updated_smartnic_ports(self):
          vif_mac = FAKE_MAC
          vif_name = 'rep0-0'
          vif_id = "407a79e0-e0be-4b7d-92a6-513b2161011a"
          self.agent._add_port_to_updated_smartnic_ports(
              vif_mac,
              vif_name,
              vif_id,
              portbindings.VIF_TYPE_UNBOUND)
          smartnic_data = {
              'mac': vif_mac,
              'vm_uuid': '',
              'vif_name': vif_name,
              'iface_id': vif_id,
              'vif_type': portbindings.VIF_TYPE_UNBOUND}
          self.assertEqual([smartnic_data],
                           self.agent.updated_smartnic_ports)
  
      def test_create_smartnic_port_map_entry_data(self):
          mac = FAKE_MAC
          rep_port = 'rep0-0'
          expected_return_value = {"vif_mac": mac,
                                   "vif_name": rep_port}
          int_br_smartnic_port_map =\
              self.agent.create_smartnic_port_map_entry_data(mac, rep_port)
          self.assertEqual(int_br_smartnic_port_map, expected_return_value)
  
  
  class TestOvsNeutronAgentOSKen(TestOvsNeutronAgent,
                                 ovs_test_base.OVSOSKenTestBase):
      def test_cleanup_stale_flows(self):
          uint64_max = (1 << 64) - 1
          with mock.patch.object(self.agent.int_br,
                                 'dump_flows') as dump_flows,\
                  mock.patch.object(self.agent.int_br,
                                    'uninstall_flows') as uninstall_flows:
              self.agent.int_br.set_agent_uuid_stamp(1234)
              fake_flows = [
                  # mock os_ken.ofproto.ofproto_v1_3_parser.OFPFlowStats
                  mock.Mock(cookie=1234, table_id=0),
                  mock.Mock(cookie=17185, table_id=2),
                  mock.Mock(cookie=9029, table_id=2),
                  mock.Mock(cookie=1234, table_id=3),
              ]
              dump_flows.return_value = fake_flows
              self.agent.iter_num = 3
              self.agent.cleanup_stale_flows()
  
              dump_flows_expected = [
                  mock.call(tid) for tid in constants.INT_BR_ALL_TABLES]
              dump_flows.assert_has_calls(dump_flows_expected)
  
              expected = [mock.call(cookie=17185,
                                    cookie_mask=uint64_max),
                          mock.call(cookie=9029,
                                    cookie_mask=uint64_max)]
              uninstall_flows.assert_has_calls(expected, any_order=True)
              self.assertEqual(len(constants.INT_BR_ALL_TABLES) * len(expected),
                               len(uninstall_flows.mock_calls))
  
  
  class AncillaryBridgesTest(object):
  
      def setUp(self):
          super(AncillaryBridgesTest, self).setUp()
          conn_patcher = mock.patch(
              'neutron.agent.ovsdb.impl_idl._connection')
          conn_patcher.start()
          self.addCleanup(conn_patcher.stop)
          mock.patch(PULLAPI).start()
          notifier_p = mock.patch(NOTIFIER)
          notifier_cls = notifier_p.start()
          self.notifier = mock.Mock()
          notifier_cls.return_value = self.notifier
          cfg.CONF.set_default('firewall_driver',
                               'neutron.agent.firewall.NoopFirewallDriver',
                               group='SECURITYGROUP')
          cfg.CONF.set_override('report_interval', 0, 'AGENT')
          mock.patch('neutron.agent.common.ovs_lib.BaseOVS.config',
                     new_callable=mock.PropertyMock,
                     return_value={}).start()
  
      def _test_ancillary_bridges(self, bridges, ancillary):
          device_ids = ancillary[:]
  
          def pullup_side_effect(*args, **kwargs):
              # Check that the device_id exists, if it does return it
              # if it does not return None
              try:
                  device_ids.remove(args[0])
                  return args[0]
              except Exception:
                  return None
  
          with mock.patch.object(self.mod_agent.OVSNeutronAgent,
                                 'setup_integration_br'),\
                  mock.patch('neutron.agent.linux.ip_lib.get_device_mac',
                             return_value='00:00:00:00:00:01'),\
                  mock.patch('neutron.agent.common.ovs_lib.BaseOVS.get_bridges',
                             return_value=bridges),\
                  mock.patch('neutron.agent.common.ovs_lib.BaseOVS.'
                             'get_bridge_external_bridge_id',
                             side_effect=pullup_side_effect),\
                  mock.patch(
                      'neutron.agent.common.ovs_lib.OVSBridge.'
                      'get_ports_attributes',
                      return_value=[]),\
                  mock.patch(
                      'neutron.agent.common.ovs_lib.OVSBridge.' 'get_vif_ports',
                      return_value=[]),\
                  mock.patch('neutron.agent.rpc.PluginReportStateAPI.'
                             'has_alive_neutron_server'):
              ext_manager = mock.Mock()
              self.agent = self.mod_agent.OVSNeutronAgent(self._bridge_classes(),
                                                          ext_manager, cfg.CONF)
              self.assertEqual(len(ancillary), len(self.agent.ancillary_brs))
              if ancillary:
                  bridges = [br.br_name for br in self.agent.ancillary_brs]
                  for br in ancillary:
                      self.assertIn(br, bridges)
  
      def test_ancillary_bridges_single(self):
          bridges = ['br-int', 'br-ex']
          self._test_ancillary_bridges(bridges, ['br-ex'])
  
      def test_ancillary_bridges_none(self):
          bridges = ['br-int']
          self._test_ancillary_bridges(bridges, [])
  
      def test_ancillary_bridges_multiple(self):
          bridges = ['br-int', 'br-ex1', 'br-ex2']
          self._test_ancillary_bridges(bridges, ['br-ex1', 'br-ex2'])
  
      def mock_scan_ancillary_ports(self, vif_port_set=None,
                                    registered_ports=None, sync=False):
          bridges = ['br-int', 'br-ex']
          ancillary = ['br-ex']
  
          with mock.patch.object(self.mod_agent.OVSNeutronAgent,
                                 'setup_integration_br'), \
                  mock.patch.object(self.mod_agent.OVSNeutronAgent,
                                    '_restore_local_vlan_map'), \
                  mock.patch('neutron.agent.common.ovs_lib.BaseOVS.get_bridges',
                             return_value=bridges), \
                  mock.patch('neutron.agent.common.ovs_lib.BaseOVS.'
                             'get_bridge_external_bridge_id',
                             side_effect=ancillary), \
                  mock.patch('neutron.agent.common.ovs_lib.OVSBridge.'
                             'get_vif_port_set',
                             return_value=vif_port_set),\
                  mock.patch('neutron.agent.rpc.PluginReportStateAPI.'
                             'has_alive_neutron_server'):
              ext_manager = mock.Mock()
              self.agent = self.mod_agent.OVSNeutronAgent(self._bridge_classes(),
                                                          ext_manager, cfg.CONF)
              return self.agent.scan_ancillary_ports(registered_ports, sync)
  
      def test_scan_ancillary_ports_returns_cur_only_for_unchanged_ports(self):
          vif_port_set = set([1, 2])
          registered_ports = set([1, 2])
          expected = ovs_agent.PortInfo(current=vif_port_set)
          actual = self.mock_scan_ancillary_ports(vif_port_set, registered_ports)
          self.assertEqual(expected, actual)
  
      def test_scan_ancillary_ports_returns_port_changes(self):
          vif_port_set = set([1, 3])
          registered_ports = set([1, 2])
          expected = ovs_agent.PortInfo(current=vif_port_set, added={3},
                                        removed={2})
          actual = self.mock_scan_ancillary_ports(vif_port_set, registered_ports)
          self.assertEqual(expected, actual)
  
      def test_scan_ancillary_ports_returns_port_changes_with_sync(self):
          vif_port_set = set([1, 3])
          registered_ports = set([1, 2])
          expected = ovs_agent.PortInfo(current=vif_port_set, added=vif_port_set,
                                        removed={2})
          actual = self.mock_scan_ancillary_ports(vif_port_set, registered_ports,
                                                  sync=True)
          self.assertEqual(expected, actual)
  
  
  class AncillaryBridgesTestOSKen(AncillaryBridgesTest,
                                ovs_test_base.OVSOSKenTestBase):
      pass
  
  
  class TestOvsDvrNeutronAgent(object):
  
      def setUp(self):
          super(TestOvsDvrNeutronAgent, self).setUp()
          mock.patch(PULLAPI).start()
          notifier_p = mock.patch(NOTIFIER)
          notifier_cls = notifier_p.start()
          self.notifier = mock.Mock()
          notifier_cls.return_value = self.notifier
          cfg.CONF.set_default('firewall_driver',
                               'neutron.agent.firewall.NoopFirewallDriver',
                               group='SECURITYGROUP')
  
          mock.patch('neutron.agent.common.ovs_lib.BaseOVS.config',
                     new_callable=mock.PropertyMock,
                     return_value={}).start()
          mock.patch('neutron.agent.ovsdb.impl_idl._connection').start()
          with mock.patch.object(self.mod_agent.OVSNeutronAgent,
                                 'setup_integration_br'),\
                  mock.patch.object(self.mod_agent.OVSNeutronAgent,
                                    'setup_ancillary_bridges',
                                    return_value=[]),\
                  mock.patch('neutron.agent.linux.ip_lib.get_device_mac',
                             return_value='00:00:00:00:00:01'),\
                  mock.patch(
                      'neutron.agent.common.ovs_lib.BaseOVS.get_bridges'),\
                  mock.patch('oslo_service.loopingcall.'
                             'FixedIntervalLoopingCall',
                             new=MockFixedIntervalLoopingCall),\
                  mock.patch(
                      'neutron.agent.common.ovs_lib.OVSBridge.'
                      'get_ports_attributes',
                      return_value=[]),\
                  mock.patch(
                      'neutron.agent.common.ovs_lib.OVSBridge.' 'get_vif_ports',
                      return_value=[]),\
                  mock.patch('neutron.agent.rpc.PluginReportStateAPI.'
                             'has_alive_neutron_server'):
              ext_manager = mock.Mock()
              self.agent = self.mod_agent.OVSNeutronAgent(self._bridge_classes(),
                                                         ext_manager, cfg.CONF)
              self.agent.tun_br = self.br_tun_cls(br_name='br-tun')
          self.agent.sg_agent = mock.Mock()
  
      def _setup_for_dvr_test(self):
          self._port = mock.Mock()
          self._port.ofport = 10
          self._port.vif_id = "1234-5678-90"
          self._physical_network = 'physeth1'
          self._old_local_vlan = None
          self._segmentation_id = 2001
          self.agent.enable_distributed_routing = True
          self.agent.enable_tunneling = True
          self.agent.patch_tun_ofport = 1
          self.agent.patch_int_ofport = 2
          self.agent.dvr_agent.local_ports = {}
          self.agent.vlan_manager = self.useFixture(
              test_vlanmanager.LocalVlanManagerFixture()).manager
          self.agent.dvr_agent.enable_distributed_routing = True
          self.agent.dvr_agent.enable_tunneling = True
          self.agent.dvr_agent.patch_tun_ofport = 1
          self.agent.dvr_agent.patch_int_ofport = 2
          self.agent.dvr_agent.tun_br = mock.Mock()
          self.agent.dvr_agent.phys_brs[self._physical_network] = mock.Mock()
          self.agent.dvr_agent.bridge_mappings = {self._physical_network:
                                                  'br-eth1'}
          self.agent.dvr_agent.int_ofports[self._physical_network] = 30
          self.agent.dvr_agent.phys_ofports[self._physical_network] = 40
          self.agent.dvr_agent.local_dvr_map = {}
          self.agent.dvr_agent.registered_dvr_macs = set()
          self.agent.dvr_agent.dvr_mac_address = 'aa:22:33:44:55:66'
          self._net_uuid = 'my-net-uuid'
          self._fixed_ips = [{'subnet_id': 'my-subnet-uuid',
                              'ip_address': '1.1.1.1'}]
          self._compute_port = mock.Mock()
          self._compute_port.ofport = 20
          self._compute_port.vif_id = "1234-5678-91"
          self._compute_fixed_ips = [{'subnet_id': 'my-subnet-uuid',
                                      'ip_address': '1.1.1.3'}]
  
      @staticmethod
      def _expected_port_bound(port, lvid, is_dvr=True,
                               network_type=n_const.TYPE_VXLAN):
          resp = [
              mock.call.db_get_val('Port', port.port_name, 'other_config'),
              mock.call.set_db_attribute('Port', port.port_name, 'other_config',
                                         mock.ANY),
          ]
          if is_dvr:
              resp = [
                  mock.call.get_vifs_by_ids([]),
                  mock.call.install_dvr_dst_mac_for_arp(
                      network_type,
                      dvr_mac=port.dvr_mac,
                      gateway_mac=port.vif_mac,
                      rtr_port=port.ofport,
                      vlan_tag=lvid)
              ] + resp
          return resp
  
      @staticmethod
      def _expected_port_unbound(port, lvid, is_dvr=True,
                                 network_type=n_const.TYPE_VXLAN):
          resp = []
          if is_dvr:
              resp = [
                  mock.call.delete_dvr_dst_mac_for_arp(
                      network_type=network_type,
                      dvr_mac=port.dvr_mac,
                      gateway_mac=port.vif_mac,
                      rtr_port=port.ofport,
                      vlan_tag=lvid)
              ]
          return resp
  
      def _expected_install_dvr_process(self, lvid, port, ip_version,
                                        gateway_ip):
          if ip_version == n_const.IP_VERSION_4:
              ipvx_calls = [
                  mock.call.install_dvr_process_ipv4(
                      vlan_tag=lvid,
                      gateway_ip=gateway_ip),
              ]
          else:
              ipvx_calls = [
                  mock.call.install_dvr_process_ipv6(
                      vlan_tag=lvid,
                      gateway_mac=port.vif_mac),
              ]
          return ipvx_calls + [
              mock.call.install_dvr_process(
                  vlan_tag=lvid,
                  dvr_mac_address=self.agent.dvr_agent.dvr_mac_address,
                  vif_mac=port.vif_mac,
              ),
          ]
  
      def _test_port_bound_for_dvr_on_physical_network(
              self, device_owner, network_type, ip_version=n_const.IP_VERSION_4):
          self._setup_for_dvr_test()
          if ip_version == n_const.IP_VERSION_4:
              gateway_ip = '1.1.1.10'
              cidr = '1.1.1.0/24'
          else:
              gateway_ip = '2001:100::1'
              cidr = '2001:100::0/64'
          self._port.vif_mac = 'aa:bb:cc:11:22:33'
          self._port.dvr_mac = self.agent.dvr_agent.dvr_mac_address
          gateway_mac = 'aa:bb:cc:66:66:66'
          self._compute_port.vif_mac = '77:88:99:00:11:22'
          physical_network = self._physical_network
          segmentation_id = self._segmentation_id
          if network_type == n_const.TYPE_FLAT:
              segmentation_id = None
          int_br = mock.create_autospec(self.agent.int_br)
          tun_br = mock.create_autospec(self.agent.tun_br)
          phys_br = mock.create_autospec(self.br_phys_cls('br-phys'))
          int_br.set_db_attribute.return_value = True
          int_br.db_get_val.return_value = {}
          with mock.patch.object(self.agent.dvr_agent.plugin_rpc,
                                 'get_subnet_for_dvr',
                                 return_value={'gateway_ip': gateway_ip,
                                               'cidr': cidr,
                                               'ip_version': ip_version,
                                               'gateway_mac': gateway_mac}),\
                  mock.patch.object(self.agent.dvr_agent.plugin_rpc,
                                    'get_ports_on_host_by_subnet',
                                    return_value=[]),\
                  mock.patch.object(self.agent.dvr_agent.int_br,
                                    'get_vif_port_by_id',
                                    return_value=self._port),\
                  mock.patch.object(self.agent, 'int_br', new=int_br),\
                  mock.patch.object(self.agent, 'tun_br', new=tun_br),\
                  mock.patch.dict(self.agent.phys_brs,
                                  {physical_network: phys_br}),\
                  mock.patch.object(self.agent.dvr_agent, 'int_br', new=int_br),\
                  mock.patch.object(self.agent.dvr_agent, 'tun_br', new=tun_br),\
                  mock.patch.dict(self.agent.dvr_agent.phys_brs,
                                  {physical_network: phys_br}):
              self.agent.port_bound(
                  self._port, self._net_uuid, network_type,
                  physical_network, segmentation_id, self._fixed_ips,
                  n_const.DEVICE_OWNER_DVR_INTERFACE, False)
              phy_ofp = self.agent.dvr_agent.phys_ofports[physical_network]
              int_ofp = self.agent.dvr_agent.int_ofports[physical_network]
              lvid = self.agent.vlan_manager.get(self._net_uuid).vlan
              expected_on_phys_br = [
                  mock.call.provision_local_vlan(
                      port=phy_ofp,
                      lvid=lvid,
                      segmentation_id=segmentation_id,
                      distributed=True,
                  ),
              ] + self._expected_install_dvr_process(
                  port=self._port,
                  lvid=lvid,
                  ip_version=ip_version,
                  gateway_ip=self._fixed_ips[0]['ip_address'])
              expected_on_int_br = [
                  mock.call.provision_local_vlan(
                      port=int_ofp,
                      lvid=lvid,
                      segmentation_id=segmentation_id,
                  ),
              ] + self._expected_port_bound(self._port, lvid,
                                            network_type=network_type)
              int_br.assert_has_calls(expected_on_int_br)
              tun_br.assert_not_called()
              phys_br.assert_has_calls(expected_on_phys_br)
              int_br.reset_mock()
              tun_br.reset_mock()
              phys_br.reset_mock()
              self.agent.port_bound(self._compute_port, self._net_uuid,
                                    network_type, physical_network,
                                    segmentation_id,
                                    self._compute_fixed_ips,
                                    device_owner, False)
              expected_on_int_br = [
                  mock.call.install_dvr_to_src_mac(
                      network_type=network_type,
                      gateway_mac=gateway_mac,
                      dst_mac=self._compute_port.vif_mac,
                      dst_port=self._compute_port.ofport,
                      vlan_tag=segmentation_id,
                  ),
              ] + self._expected_port_bound(self._compute_port, lvid,
                                            is_dvr=False,
                                            network_type=network_type)
              int_br.assert_has_calls(expected_on_int_br)
              tun_br.assert_not_called()
              phys_br.assert_not_called()
  
      def _test_port_bound_for_dvr_on_vxlan_network(
              self, device_owner, ip_version=n_const.IP_VERSION_4):
          self._setup_for_dvr_test()
          if ip_version == n_const.IP_VERSION_4:
              gateway_ip = '1.1.1.1'
              cidr = '1.1.1.0/24'
          else:
              gateway_ip = '2001:100::1'
              cidr = '2001:100::0/64'
          network_type = n_const.TYPE_VXLAN
          self._port.vif_mac = gateway_mac = 'aa:bb:cc:11:22:33'
          self._port.dvr_mac = self.agent.dvr_agent.dvr_mac_address
          self._compute_port.vif_mac = '77:88:99:00:11:22'
          physical_network = self._physical_network
          segmentation_id = self._segmentation_id
          int_br = mock.create_autospec(self.agent.int_br)
          tun_br = mock.create_autospec(self.agent.tun_br)
          phys_br = mock.create_autospec(self.br_phys_cls('br-phys'))
          int_br.set_db_attribute.return_value = True
          int_br.db_get_val.return_value = {}
          with mock.patch.object(self.agent.dvr_agent.plugin_rpc,
                                 'get_subnet_for_dvr',
                                 return_value={'gateway_ip': gateway_ip,
                                               'cidr': cidr,
                                               'ip_version': ip_version,
                                               'gateway_mac': gateway_mac}),\
                  mock.patch.object(self.agent.dvr_agent.plugin_rpc,
                                    'get_ports_on_host_by_subnet',
                                    return_value=[]),\
                  mock.patch.object(self.agent.dvr_agent.int_br,
                                    'get_vif_port_by_id',
                                    return_value=self._port),\
                  mock.patch.object(self.agent, 'int_br', new=int_br),\
                  mock.patch.object(self.agent, 'tun_br', new=tun_br),\
                  mock.patch.dict(self.agent.phys_brs,
                                  {physical_network: phys_br}),\
                  mock.patch.object(self.agent.dvr_agent, 'int_br', new=int_br),\
                  mock.patch.object(self.agent.dvr_agent, 'tun_br', new=tun_br),\
                  mock.patch.dict(self.agent.dvr_agent.phys_brs,
                                  {physical_network: phys_br}):
              self.agent.port_bound(
                  self._port, self._net_uuid, network_type,
                  physical_network, segmentation_id, self._fixed_ips,
                  n_const.DEVICE_OWNER_DVR_INTERFACE, False)
              lvid = self.agent.vlan_manager.get(self._net_uuid).vlan
              expected_on_int_br = self._expected_port_bound(
                  self._port, lvid)
              expected_on_tun_br = [
                  mock.call.provision_local_vlan(
                      network_type=network_type,
                      segmentation_id=segmentation_id,
                      lvid=lvid,
                      distributed=True),
              ] + self._expected_install_dvr_process(
                  port=self._port,
                  lvid=lvid,
                  ip_version=ip_version,
                  gateway_ip=gateway_ip)
              int_br.assert_has_calls(expected_on_int_br)
              tun_br.assert_has_calls(expected_on_tun_br)
              phys_br.assert_not_called()
              int_br.reset_mock()
              tun_br.reset_mock()
              phys_br.reset_mock()
              self.agent.port_bound(self._compute_port, self._net_uuid,
                                    network_type, physical_network,
                                    segmentation_id,
                                    self._compute_fixed_ips,
                                    device_owner, False)
              expected_on_int_br = [
                  mock.call.install_dvr_to_src_mac(
                      network_type=network_type,
                      gateway_mac=gateway_mac,
                      dst_mac=self._compute_port.vif_mac,
                      dst_port=self._compute_port.ofport,
                      vlan_tag=lvid,
                  ),
              ] + self._expected_port_bound(self._compute_port, lvid, False,
                                            network_type)
              int_br.assert_has_calls(expected_on_int_br)
              tun_br.assert_not_called()
              phys_br.assert_not_called()
  
      def test_port_bound_for_dvr_with_compute_ports(self):
          self._test_port_bound_for_dvr_on_physical_network(
              device_owner=DEVICE_OWNER_COMPUTE,
              network_type=n_const.TYPE_VLAN)
          self._test_port_bound_for_dvr_on_physical_network(
              device_owner=DEVICE_OWNER_COMPUTE,
              network_type=n_const.TYPE_VLAN,
              ip_version=n_const.IP_VERSION_6)
          self._test_port_bound_for_dvr_on_physical_network(
              device_owner=DEVICE_OWNER_COMPUTE,
              network_type=n_const.TYPE_FLAT)
          self._test_port_bound_for_dvr_on_physical_network(
              device_owner=DEVICE_OWNER_COMPUTE,
              network_type=n_const.TYPE_FLAT,
              ip_version=n_const.IP_VERSION_6)
          self._test_port_bound_for_dvr_on_vxlan_network(
              device_owner=DEVICE_OWNER_COMPUTE)
          self._test_port_bound_for_dvr_on_vxlan_network(
              device_owner=DEVICE_OWNER_COMPUTE,
              ip_version=n_const.IP_VERSION_6)
  
      def test_port_bound_for_dvr_with_dhcp_ports(self):
          self._test_port_bound_for_dvr_on_physical_network(
              device_owner=n_const.DEVICE_OWNER_DHCP,
              network_type=n_const.TYPE_VLAN)
          self._test_port_bound_for_dvr_on_physical_network(
              device_owner=n_const.DEVICE_OWNER_DHCP,
              network_type=n_const.TYPE_VLAN,
              ip_version=n_const.IP_VERSION_6)
          self._test_port_bound_for_dvr_on_physical_network(
              device_owner=n_const.DEVICE_OWNER_DHCP,
              network_type=n_const.TYPE_FLAT)
          self._test_port_bound_for_dvr_on_physical_network(
              device_owner=n_const.DEVICE_OWNER_DHCP,
              network_type=n_const.TYPE_FLAT,
              ip_version=n_const.IP_VERSION_6)
          self._test_port_bound_for_dvr_on_vxlan_network(
              device_owner=n_const.DEVICE_OWNER_DHCP)
          self._test_port_bound_for_dvr_on_vxlan_network(
              device_owner=n_const.DEVICE_OWNER_DHCP,
              ip_version=n_const.IP_VERSION_6)
  
      def test_port_bound_for_dvr_with_csnat_ports(self):
          self._setup_for_dvr_test()
          int_br, tun_br = self._port_bound_for_dvr_with_csnat_ports()
          lvid = self.agent.vlan_manager.get(self._net_uuid).vlan
          expected_on_int_br = [
              mock.call.install_dvr_to_src_mac(
                  network_type='vxlan',
                  gateway_mac='aa:bb:cc:11:22:33',
                  dst_mac=self._port.vif_mac,
                  dst_port=self._port.ofport,
                  vlan_tag=lvid,
              ),
          ] + self._expected_port_bound(self._port, lvid, is_dvr=False)
          int_br.assert_has_calls(expected_on_int_br)
          expected_on_tun_br = [
              mock.call.provision_local_vlan(
                  network_type='vxlan',
                  lvid=lvid,
                  segmentation_id=None,
                  distributed=True,
              ),
          ]
          tun_br.assert_has_calls(expected_on_tun_br)
  
      def test_port_bound_for_dvr_with_csnat_port_without_passing_fixed_ip(self):
          self._setup_for_dvr_test()
          int_br = mock.create_autospec(self.agent.int_br)
          tun_br = mock.create_autospec(self.agent.tun_br)
          int_br.set_db_attribute.return_value = True
          int_br.db_get_val.return_value = {}
          with mock.patch.object(self.agent.dvr_agent.plugin_rpc,
                                 'get_subnet_for_dvr') as mock_getsub,\
                  mock.patch.object(self.agent.dvr_agent.plugin_rpc,
                                    'get_ports_on_host_by_subnet',
                                    return_value=[]),\
                  mock.patch.object(self.agent.dvr_agent.int_br,
                                    'get_vif_port_by_id',
                                    return_value=self._port),\
                  mock.patch.object(self.agent, 'int_br', new=int_br),\
                  mock.patch.object(self.agent, 'tun_br', new=tun_br),\
                  mock.patch.object(self.agent.dvr_agent, 'int_br', new=int_br),\
                  mock.patch.object(self.agent.dvr_agent, 'tun_br', new=tun_br):
              self.agent.port_bound(
                  self._port, self._net_uuid, 'vxlan',
                  None, None, self._fixed_ips,
                  n_const.DEVICE_OWNER_ROUTER_SNAT,
                  False)
              mock_getsub.assert_called_with(
                  self.agent.context, mock.ANY, fixed_ips=None)
  
      def test_port_bound_for_dvr_with_csnat_ports_ofport_change(self):
          self._setup_for_dvr_test()
          self._port_bound_for_dvr_with_csnat_ports()
          # simulate a replug
          self._port.ofport = 12
          int_br, tun_br = self._port_bound_for_dvr_with_csnat_ports()
          lvid = self.agent.vlan_manager.get(self._net_uuid).vlan
          expected_on_int_br = [
              mock.call.delete_dvr_to_src_mac(
                  network_type='vxlan',
                  dst_mac=self._port.vif_mac,
                  vlan_tag=lvid,
              ),
              mock.call.install_dvr_to_src_mac(
                  network_type='vxlan',
                  gateway_mac='aa:bb:cc:11:22:33',
                  dst_mac=self._port.vif_mac,
                  dst_port=self._port.ofport,
                  vlan_tag=lvid,
              ),
          ] + self._expected_port_bound(self._port, lvid, is_dvr=False)
          int_br.assert_has_calls(expected_on_int_br)
          # a local vlan was already provisioned so there should be no new
          # calls to tunbr
          tun_br.assert_not_called()
          # make sure ofport was updated
          self.assertEqual(12,
              self.agent.dvr_agent.local_ports[self._port.vif_id].ofport)
  
      def _port_bound_for_dvr_with_csnat_ports(self):
          int_br = mock.create_autospec(self.agent.int_br)
          tun_br = mock.create_autospec(self.agent.tun_br)
          int_br.set_db_attribute.return_value = True
          int_br.db_get_val.return_value = {}
          with mock.patch.object(self.agent.dvr_agent.plugin_rpc,
                                 'get_subnet_for_dvr',
                                 return_value={'gateway_ip': '1.1.1.1',
                                 'cidr': '1.1.1.0/24',
                                 'ip_version': n_const.IP_VERSION_4,
                                 'gateway_mac': 'aa:bb:cc:11:22:33'}),\
                  mock.patch.object(self.agent.dvr_agent.plugin_rpc,
                                    'get_ports_on_host_by_subnet',
                                    return_value=[]),\
                  mock.patch.object(self.agent.dvr_agent.int_br,
                                    'get_vif_port_by_id',
                                    return_value=self._port),\
                  mock.patch.object(self.agent, 'int_br', new=int_br),\
                  mock.patch.object(self.agent, 'tun_br', new=tun_br),\
                  mock.patch.object(self.agent.dvr_agent, 'int_br', new=int_br),\
                  mock.patch.object(self.agent.dvr_agent, 'tun_br', new=tun_br):
              self.agent.port_bound(
                  self._port, self._net_uuid, 'vxlan',
                  None, None, self._fixed_ips,
                  n_const.DEVICE_OWNER_ROUTER_SNAT,
                  False)
          return int_br, tun_br
  
      def test_port_bound_for_dvr_with_csnat_ports_without_subnet(self):
          self._setup_for_dvr_test()
          int_br = mock.create_autospec(self.agent.int_br)
          tun_br = mock.create_autospec(self.agent.tun_br)
  
          # get_subnet_for_dvr RPC returns {} on error
          with mock.patch.object(self.agent.dvr_agent.plugin_rpc,
                                 'get_subnet_for_dvr',
                                 return_value={}),\
                  mock.patch.object(self.agent, 'int_br', new=int_br),\
                  mock.patch.object(self.agent, 'tun_br', new=tun_br),\
                  mock.patch.object(self.agent.dvr_agent, 'int_br', new=int_br),\
                  mock.patch.object(self.agent.dvr_agent, 'tun_br', new=tun_br):
              self.agent.port_bound(
                  self._port, self._net_uuid, 'vxlan',
                  None, None, self._fixed_ips,
                  n_const.DEVICE_OWNER_ROUTER_SNAT,
                  False)
              int_br.install_dvr_to_src_mac.assert_not_called()
  
      def test_treat_devices_removed_for_dvr_interface(self):
          self._test_treat_devices_removed_for_dvr_interface()
          self._test_treat_devices_removed_for_dvr_interface(
              ip_version=n_const.IP_VERSION_6)
          self._test_treat_devices_removed_for_dvr_interface(network_type='vlan')
          self._test_treat_devices_removed_for_dvr_interface(
              ip_version=n_const.IP_VERSION_6, network_type='vlan')
  
      def _test_treat_devices_removed_for_dvr_interface(
              self, ip_version=n_const.IP_VERSION_4, network_type='vxlan'):
          self._setup_for_dvr_test()
          if ip_version == n_const.IP_VERSION_4:
              gateway_ip = '1.1.1.1'
              cidr = '1.1.1.0/24'
          else:
              gateway_ip = '2001:100::1'
              cidr = '2001:100::0/64'
          self._port.dvr_mac = self.agent.dvr_agent.dvr_mac_address
          gateway_mac = 'aa:bb:cc:11:22:33'
          int_br = mock.create_autospec(self.agent.int_br)
          tun_br = mock.create_autospec(self.agent.tun_br)
          int_br.set_db_attribute.return_value = True
          int_br.db_get_val.return_value = {}
          with mock.patch.object(self.agent.dvr_agent.plugin_rpc,
                                 'get_subnet_for_dvr',
                                 return_value={'gateway_ip': gateway_ip,
                                 'cidr': cidr,
                                 'ip_version': ip_version,
                                 'gateway_mac': gateway_mac}),\
                  mock.patch.object(self.agent.dvr_agent.plugin_rpc,
                                    'get_ports_on_host_by_subnet',
                                    return_value=[]),\
                  mock.patch.object(self.agent, 'int_br', new=int_br),\
                  mock.patch.object(self.agent, 'tun_br', new=tun_br),\
                  mock.patch.object(self.agent.dvr_agent, 'int_br', new=int_br),\
                  mock.patch.object(self.agent.dvr_agent, 'tun_br', new=tun_br),\
                  mock.patch.object(self.agent.dvr_agent.int_br,
                                    'get_vif_port_by_id',
                                    return_value=self._port):
              if network_type == 'vlan':
                  self.agent.port_bound(self._port, self._net_uuid,
                                        network_type, self._physical_network,
                                        self._segmentation_id,
                                        self._compute_fixed_ips,
                                        n_const.DEVICE_OWNER_DVR_INTERFACE,
                                        False)
              else:
                  self.agent.port_bound(
                      self._port, self._net_uuid, 'vxlan',
                      None, None, self._fixed_ips,
                      n_const.DEVICE_OWNER_DVR_INTERFACE,
                      False)
                  lvid = self.agent.vlan_manager.get(self._net_uuid).vlan
                  int_br.assert_has_calls(
                      self._expected_port_bound(self._port, lvid))
                  expected_on_tun_br = [
                      mock.call.provision_local_vlan(network_type='vxlan',
                          lvid=lvid, segmentation_id=None, distributed=True),
                  ] + self._expected_install_dvr_process(
                      port=self._port,
                      lvid=lvid,
                      ip_version=ip_version,
                      gateway_ip=gateway_ip)
                  tun_br.assert_has_calls(expected_on_tun_br)
  
          int_br.reset_mock()
          tun_br.reset_mock()
          phys_br = mock.create_autospec(self.br_phys_cls('br-phys'))
          with mock.patch.object(self.agent, 'reclaim_local_vlan'),\
                  mock.patch.object(self.agent.plugin_rpc, 'update_device_list',
                                    return_value={
                                        'devices_up': [],
                                        'devices_down': [self._port.vif_id],
                                        'failed_devices_up': [],
                                        'failed_devices_down': []}),\
                  mock.patch.object(self.agent, 'int_br', new=int_br),\
                  mock.patch.object(self.agent, 'tun_br', new=tun_br),\
                  mock.patch.dict(self.agent.phys_brs,
                                  {self._physical_network: phys_br}),\
                  mock.patch.object(self.agent.dvr_agent, 'int_br', new=int_br),\
                  mock.patch.object(self.agent.dvr_agent, 'tun_br', new=tun_br),\
                  mock.patch.dict(self.agent.dvr_agent.phys_brs,
                                  {self._physical_network: phys_br}),\
                  mock.patch.object(self.agent.int_br,
                                    'get_vif_port_by_id',
                                    return_value=None):
              failed_devices = {'added': set(), 'removed': set()}
              failed_devices['removed'] = self.agent.treat_devices_removed(
                  [self._port.vif_id])
              lvid = self.agent.vlan_manager.get(self._net_uuid).vlan
              if ip_version == n_const.IP_VERSION_4:
                  expected = [
                      mock.call.delete_dvr_process_ipv4(
                          vlan_tag=lvid,
                          gateway_ip=gateway_ip),
                  ]
              else:
                  expected = [
                      mock.call.delete_dvr_process_ipv6(
                          vlan_tag=lvid,
                          gateway_mac=gateway_mac),
                  ]
              expected.extend([
                  mock.call.delete_dvr_process(
                      vlan_tag=lvid,
                      vif_mac=self._port.vif_mac),
              ])
              if network_type == 'vlan':
                  expected_unbound_dvr = self._expected_port_unbound(self._port,
                      self._segmentation_id, network_type=network_type)
                  int_br.assert_has_calls(expected_unbound_dvr)
                  tun_br.assert_not_called()
                  phys_br.assert_has_calls(expected)
                  self.assertEqual({}, self.agent.dvr_agent.local_ports)
              else:
                  expected_unbound_dvr = self._expected_port_unbound(self._port,
                      lvid, network_type=network_type)
                  int_br.assert_has_calls(expected_unbound_dvr)
                  tun_br.assert_has_calls(expected)
                  phys_br.assert_not_called()
  
      def _test_treat_devices_removed_for_dvr(self, device_owner,
                                              ip_version=n_const.IP_VERSION_4):
          self._setup_for_dvr_test()
          if ip_version == n_const.IP_VERSION_4:
              gateway_ip = '1.1.1.1'
              cidr = '1.1.1.0/24'
          else:
              gateway_ip = '2001:100::1'
              cidr = '2001:100::0/64'
          self._port.dvr_mac = self.agent.dvr_agent.dvr_mac_address
          gateway_mac = 'aa:bb:cc:11:22:33'
          int_br = mock.create_autospec(self.agent.int_br)
          tun_br = mock.create_autospec(self.agent.tun_br)
          int_br.set_db_attribute.return_value = True
          int_br.db_get_val.return_value = {}
          with mock.patch.object(self.agent.dvr_agent.plugin_rpc,
                                 'get_subnet_for_dvr',
                                 return_value={'gateway_ip': gateway_ip,
                                 'cidr': cidr,
                                 'ip_version': ip_version,
                                 'gateway_mac': gateway_mac}),\
                  mock.patch.object(self.agent.dvr_agent.plugin_rpc,
                                    'get_ports_on_host_by_subnet',
                                    return_value=[]),\
                  mock.patch.object(self.agent.dvr_agent.int_br,
                                    'get_vif_port_by_id',
                                    return_value=self._port),\
                  mock.patch.object(self.agent, 'int_br', new=int_br),\
                  mock.patch.object(self.agent, 'tun_br', new=tun_br),\
                  mock.patch.object(self.agent.dvr_agent, 'int_br', new=int_br),\
                  mock.patch.object(self.agent.dvr_agent, 'tun_br', new=tun_br):
              self.agent.port_bound(
                  self._port, self._net_uuid, 'vxlan',
                  None, None, self._fixed_ips,
                  n_const.DEVICE_OWNER_DVR_INTERFACE,
                  False)
              lvid = self.agent.vlan_manager.get(self._net_uuid).vlan
              int_br.assert_has_calls(
                  self._expected_port_bound(self._port, lvid))
              expected_on_tun_br = [
                  mock.call.provision_local_vlan(
                      network_type='vxlan',
                      segmentation_id=None,
                      lvid=lvid,
                      distributed=True),
              ] + self._expected_install_dvr_process(
                  port=self._port,
                  lvid=lvid,
                  ip_version=ip_version,
                  gateway_ip=gateway_ip)
              tun_br.assert_has_calls(expected_on_tun_br)
              int_br.reset_mock()
              tun_br.reset_mock()
              self.agent.port_bound(self._compute_port,
                                    self._net_uuid, 'vxlan',
                                    None, None,
                                    self._compute_fixed_ips,
                                    device_owner, False)
              int_br.assert_has_calls(
                  [
                      mock.call.install_dvr_to_src_mac(
                          network_type='vxlan',
                          gateway_mac='aa:bb:cc:11:22:33',
                          dst_mac=self._compute_port.vif_mac,
                          dst_port=self._compute_port.ofport,
                          vlan_tag=lvid,
                      ),
                  ] + self._expected_port_bound(self._compute_port, lvid, False))
              tun_br.assert_not_called()
  
          int_br.reset_mock()
          tun_br.reset_mock()
          with mock.patch.object(self.agent, 'reclaim_local_vlan'),\
                  mock.patch.object(self.agent.plugin_rpc, 'update_device_list',
                                    return_value={
                                        'devices_up': [],
                                        'devices_down': [
                                            self._compute_port.vif_id],
                                        'failed_devices_up': [],
                                        'failed_devices_down': []}),\
                  mock.patch.object(self.agent, 'int_br', new=int_br),\
                  mock.patch.object(self.agent, 'tun_br', new=tun_br),\
                  mock.patch.object(self.agent.dvr_agent, 'int_br', new=int_br),\
                  mock.patch.object(self.agent.dvr_agent, 'tun_br', new=tun_br):
              failed_devices = {'added': set(), 'removed': set()}
              failed_devices['removed'] = self.agent.treat_devices_removed(
                  [self._compute_port.vif_id])
              int_br.assert_has_calls([
                  mock.call.delete_dvr_to_src_mac(
                      network_type='vxlan',
                      vlan_tag=lvid,
                      dst_mac=self._compute_port.vif_mac,
                  ),
              ])
              tun_br.assert_not_called()
  
      def test_treat_devices_removed_for_dvr_with_compute_ports(self):
          self._test_treat_devices_removed_for_dvr(
              device_owner=DEVICE_OWNER_COMPUTE)
          self._test_treat_devices_removed_for_dvr(
              device_owner=DEVICE_OWNER_COMPUTE, ip_version=n_const.IP_VERSION_6)
  
      def test_treat_devices_removed_for_dvr_with_dhcp_ports(self):
          self._test_treat_devices_removed_for_dvr(
              device_owner=n_const.DEVICE_OWNER_DHCP)
          self._test_treat_devices_removed_for_dvr(
              device_owner=n_const.DEVICE_OWNER_DHCP,
              ip_version=n_const.IP_VERSION_6)
  
      def test_treat_devices_removed_for_dvr_csnat_port(self):
          self._setup_for_dvr_test()
          gateway_mac = 'aa:bb:cc:11:22:33'
          int_br = mock.create_autospec(self.agent.int_br)
          tun_br = mock.create_autospec(self.agent.tun_br)
          int_br.set_db_attribute.return_value = True
          int_br.db_get_val.return_value = {}
          with mock.patch.object(self.agent.dvr_agent.plugin_rpc,
                                 'get_subnet_for_dvr',
                                 return_value={'gateway_ip': '1.1.1.1',
                                 'cidr': '1.1.1.0/24',
                                 'ip_version': n_const.IP_VERSION_4,
                                 'gateway_mac': gateway_mac}),\
                  mock.patch.object(self.agent.dvr_agent.plugin_rpc,
                                    'get_ports_on_host_by_subnet',
                                    return_value=[]),\
                  mock.patch.object(self.agent.dvr_agent.int_br,
                                    'get_vif_port_by_id',
                                    return_value=self._port),\
                  mock.patch.object(self.agent, 'int_br', new=int_br),\
                  mock.patch.object(self.agent, 'tun_br', new=tun_br),\
                  mock.patch.object(self.agent.dvr_agent, 'int_br', new=int_br),\
                  mock.patch.object(self.agent.dvr_agent, 'tun_br', new=tun_br):
              self.agent.port_bound(
                  self._port, self._net_uuid, 'vxlan',
                  None, None, self._fixed_ips,
                  n_const.DEVICE_OWNER_ROUTER_SNAT,
                  False)
              lvid = self.agent.vlan_manager.get(self._net_uuid).vlan
              expected_on_int_br = [
                  mock.call.install_dvr_to_src_mac(
                      network_type='vxlan',
                      gateway_mac=gateway_mac,
                      dst_mac=self._port.vif_mac,
                      dst_port=self._port.ofport,
                      vlan_tag=lvid,
                  ),
              ] + self._expected_port_bound(self._port, lvid, is_dvr=False)
              int_br.assert_has_calls(expected_on_int_br)
              expected_on_tun_br = [
                  mock.call.provision_local_vlan(
                      network_type='vxlan',
                      lvid=lvid,
                      segmentation_id=None,
                      distributed=True,
                  ),
              ]
              tun_br.assert_has_calls(expected_on_tun_br)
  
          int_br.reset_mock()
          tun_br.reset_mock()
          with mock.patch.object(self.agent, 'reclaim_local_vlan'),\
                  mock.patch.object(self.agent.plugin_rpc, 'update_device_list',
                                    return_value={
                                        'devices_up': [],
                                        'devices_down': [self._port.vif_id],
                                        'failed_devices_up': [],
                                        'failed_devices_down': []}),\
                  mock.patch.object(self.agent, 'int_br', new=int_br),\
                  mock.patch.object(self.agent, 'tun_br', new=tun_br),\
                  mock.patch.object(self.agent.dvr_agent, 'int_br', new=int_br),\
                  mock.patch.object(self.agent.dvr_agent, 'tun_br', new=tun_br),\
                  mock.patch.object(self.agent.int_br, 'get_vif_port_by_id',
                                    return_value=None):
              failed_devices = {'added': set(), 'removed': set()}
              failed_devices['removed'] = self.agent.treat_devices_removed(
                  [self._port.vif_id])
              expected_on_int_br = [
                  mock.call.delete_dvr_to_src_mac(
                      network_type='vxlan',
                      dst_mac=self._port.vif_mac,
                      vlan_tag=lvid,
                  ),
              ]
              int_br.assert_has_calls(expected_on_int_br)
              tun_br.assert_not_called()
  
      def test_setup_dvr_flows_on_int_br(self):
          self._setup_for_dvr_test()
          int_br = mock.create_autospec(self.agent.int_br)
          tun_br = mock.create_autospec(self.agent.tun_br)
          with mock.patch.object(self.agent, 'int_br', new=int_br),\
                  mock.patch.object(self.agent, 'tun_br', new=tun_br),\
                  mock.patch.object(self.agent.dvr_agent, 'int_br', new=int_br),\
                  mock.patch.object(self.agent.dvr_agent, 'tun_br', new=tun_br),\
                  mock.patch.object(self.agent.dvr_agent.plugin_rpc,
                                    'get_dvr_mac_address_list',
                                    return_value=[{'host': 'cn1',
                                    'mac_address': 'aa-bb-cc-dd-ee-ff'},
                                    {'host': 'cn2',
                                    'mac_address': '11-22-33-44-55-66'}]):
              self.agent.dvr_agent.setup_dvr_flows_on_integ_br()
              self.assertTrue(self.agent.dvr_agent.in_distributed_mode())
              physical_networks = list(
                  self.agent.dvr_agent.bridge_mappings.keys())
              ioport = self.agent.dvr_agent.int_ofports[physical_networks[0]]
              expected_on_int_br = [
                  # setup_dvr_flows_on_integ_br
                  mock.call.setup_canary_table(),
                  mock.call.install_drop(table_id=constants.DVR_TO_SRC_MAC,
                                         priority=1),
                  mock.call.install_drop(
                      table_id=constants.DVR_TO_SRC_MAC_PHYSICAL,
                      priority=1),
                  mock.call.install_drop(table_id=constants.LOCAL_SWITCHING,
                                         priority=2,
                                         in_port=ioport),
              ]
              int_br.assert_has_calls(expected_on_int_br)
              tun_br.assert_not_called()
  
      def test_get_dvr_mac_address(self):
          self._setup_for_dvr_test()
          self.agent.dvr_agent.dvr_mac_address = None
          with mock.patch.object(self.agent.dvr_agent.plugin_rpc,
                                 'get_dvr_mac_address_by_host',
                                 return_value={'host': 'cn1',
                                    'mac_address': 'aa-22-33-44-55-66'}):
              self.agent.dvr_agent.get_dvr_mac_address()
              self.assertEqual('aa:22:33:44:55:66',
                               self.agent.dvr_agent.dvr_mac_address)
              self.assertTrue(self.agent.dvr_agent.in_distributed_mode())
  
      def test_get_dvr_mac_address_exception(self):
          self._setup_for_dvr_test()
          self.agent.dvr_agent.dvr_mac_address = None
          int_br = mock.create_autospec(self.agent.int_br)
          with mock.patch.object(self.agent.dvr_agent.plugin_rpc,
                                 'get_dvr_mac_address_by_host',
                                 side_effect=oslo_messaging.RemoteError),\
                  mock.patch.object(self.agent, 'int_br', new=int_br),\
                  mock.patch.object(self.agent.dvr_agent, 'int_br', new=int_br):
              with testtools.ExpectedException(SystemExit):
                  self.agent.dvr_agent.get_dvr_mac_address()
                  self.assertIsNone(self.agent.dvr_agent.dvr_mac_address)
                  self.assertFalse(self.agent.dvr_agent.in_distributed_mode())
  
      def test_get_dvr_mac_address_retried(self):
          valid_entry = {'host': 'cn1', 'mac_address': 'aa-22-33-44-55-66'}
          raise_timeout = oslo_messaging.MessagingTimeout()
          # Raise a timeout the first 2 times it calls get_dvr_mac_address()
          self._setup_for_dvr_test()
          self.agent.dvr_agent.dvr_mac_address = None
          with mock.patch.object(self.agent.dvr_agent.plugin_rpc,
                                 'get_dvr_mac_address_by_host',
                                 side_effect=(raise_timeout, raise_timeout,
                                              valid_entry)):
              self.agent.dvr_agent.get_dvr_mac_address()
              self.assertEqual('aa:22:33:44:55:66',
                               self.agent.dvr_agent.dvr_mac_address)
              self.assertTrue(self.agent.dvr_agent.in_distributed_mode())
              self.assertEqual(self.agent.dvr_agent.plugin_rpc.
                               get_dvr_mac_address_by_host.call_count, 3)
  
      def test_get_dvr_mac_address_retried_max(self):
          raise_timeout = oslo_messaging.MessagingTimeout()
          # Raise a timeout every time until we give up, currently 5 tries
          self._setup_for_dvr_test()
          self.agent.dvr_agent.dvr_mac_address = None
          int_br = mock.create_autospec(self.agent.int_br)
          with mock.patch.object(self.agent.dvr_agent.plugin_rpc,
                                 'get_dvr_mac_address_by_host',
                                 side_effect=raise_timeout),\
                  mock.patch.object(utils, "execute"),\
                  mock.patch.object(self.agent, 'int_br', new=int_br),\
                  mock.patch.object(self.agent.dvr_agent, 'int_br', new=int_br):
              with testtools.ExpectedException(SystemExit):
                  self.agent.dvr_agent.get_dvr_mac_address()
                  self.assertIsNone(self.agent.dvr_agent.dvr_mac_address)
                  self.assertFalse(self.agent.dvr_agent.in_distributed_mode())
                  self.assertEqual(self.agent.dvr_agent.plugin_rpc.
                                   get_dvr_mac_address_by_host.call_count, 5)
  
      def test_dvr_mac_address_update(self):
          self._setup_for_dvr_test()
          newhost = 'cn2'
          newmac = 'aa:bb:cc:dd:ee:ff'
          int_br = mock.create_autospec(self.agent.int_br)
          tun_br = mock.create_autospec(self.agent.tun_br)
          phys_br = mock.create_autospec(self.br_phys_cls('br-phys'))
          physical_network = 'physeth1'
          with mock.patch.object(self.agent, 'int_br', new=int_br),\
                  mock.patch.object(self.agent, 'tun_br', new=tun_br),\
                  mock.patch.dict(self.agent.phys_brs,
                                  {physical_network: phys_br}),\
                  mock.patch.object(self.agent.dvr_agent, 'int_br', new=int_br),\
                  mock.patch.object(self.agent.dvr_agent, 'tun_br', new=tun_br),\
                  mock.patch.dict(self.agent.dvr_agent.phys_brs,
                                  {physical_network: phys_br}):
              self.agent.dvr_agent.\
                  dvr_mac_address_update(
                      dvr_macs=[{'host': newhost,
                                 'mac_address': newmac}])
              expected_on_int_br = [
                  mock.call.add_dvr_mac_physical(
                      mac=newmac,
                      port=self.agent.int_ofports[physical_network]),
                  mock.call.add_dvr_mac_tun(
                      mac=newmac,
                      port=self.agent.patch_tun_ofport),
              ]
              expected_on_tun_br = [
                  mock.call.add_dvr_mac_tun(
                      mac=newmac,
                      port=self.agent.patch_int_ofport),
              ]
              expected_on_phys_br = [
                  mock.call.add_dvr_mac_physical(
                      mac=newmac,
                      port=self.agent.phys_ofports[physical_network]),
              ]
              int_br.assert_has_calls(expected_on_int_br)
              tun_br.assert_has_calls(expected_on_tun_br)
              phys_br.assert_has_calls(expected_on_phys_br)
          int_br.reset_mock()
          tun_br.reset_mock()
          phys_br.reset_mock()
          with mock.patch.object(self.agent, 'int_br', new=int_br),\
                  mock.patch.object(self.agent, 'tun_br', new=tun_br),\
                  mock.patch.dict(self.agent.phys_brs,
                                  {physical_network: phys_br}),\
                  mock.patch.object(self.agent.dvr_agent, 'int_br', new=int_br),\
                  mock.patch.object(self.agent.dvr_agent, 'tun_br', new=tun_br),\
                  mock.patch.dict(self.agent.dvr_agent.phys_brs,
                                  {physical_network: phys_br}):
              self.agent.dvr_agent.dvr_mac_address_update(dvr_macs=[])
              expected_on_int_br = [
                  mock.call.remove_dvr_mac_vlan(
                      mac=newmac),
                  mock.call.remove_dvr_mac_tun(
                      mac=newmac,
                      port=self.agent.patch_tun_ofport),
              ]
              expected_on_tun_br = [
                  mock.call.remove_dvr_mac_tun(
                      mac=newmac),
              ]
              expected_on_phys_br = [
                  mock.call.remove_dvr_mac_vlan(
                      mac=newmac),
              ]
              int_br.assert_has_calls(expected_on_int_br)
              tun_br.assert_has_calls(expected_on_tun_br)
              phys_br.assert_has_calls(expected_on_phys_br)
  
      def test_ovs_restart(self):
          self._setup_for_dvr_test()
          reset_methods = (
              'reset_ovs_parameters', 'reset_dvr_parameters',
              'setup_dvr_flows_on_integ_br', 'setup_dvr_flows_on_tun_br',
              'setup_dvr_flows_on_phys_br', 'setup_dvr_mac_flows_on_all_brs')
          reset_mocks = [mock.patch.object(self.agent.dvr_agent, method).start()
                         for method in reset_methods]
          tun_br = mock.create_autospec(self.agent.tun_br)
          with mock.patch.object(self.agent,
                                 'check_ovs_status',
                                 return_value=constants.OVS_RESTARTED),\
                  mock.patch.object(self.agent,
                                    '_agent_has_updates',
                                    side_effect=TypeError('loop exit')),\
                  mock.patch.object(self.agent, 'tun_br', new=tun_br),\
                  mock.patch.object(self.agent, 'setup_physical_bridges'),\
                  mock.patch.object(self.agent, 'setup_integration_br'),\
                  mock.patch.object(self.agent, 'setup_tunnel_br'),\
                  mock.patch.object(self.agent, 'state_rpc'), \
                  mock.patch.object(self.agent.ovs.ovsdb, 'idl_monitor'):
              try:
                  self.agent.rpc_loop(polling_manager=mock.Mock())
              except TypeError:
                  pass
          self.assertTrue(all([x.called for x in reset_mocks]))
  
      def test_rpc_loop_survives_error_in_check_canary_table(self):
          with mock.patch.object(self.agent.int_br,
                                 'check_canary_table',
                                 side_effect=TypeError('borked')),\
                  mock.patch.object(self.agent, '_check_and_handle_signal',
                                    side_effect=[True, False]):
              self.agent.rpc_loop(polling_manager=mock.Mock())
  
      def _test_scan_ports_failure(self, scan_method_name):
          with mock.patch.object(self.agent,
                                 'check_ovs_status',
                                 return_value=constants.OVS_RESTARTED),\
                  mock.patch.object(self.agent, scan_method_name,
                                 side_effect=TypeError('broken')),\
                  mock.patch.object(self.agent, '_agent_has_updates',
                                    return_value=True),\
                  mock.patch.object(self.agent, '_check_and_handle_signal',
                                    side_effect=[True, False]),\
                  mock.patch.object(self.agent, 'setup_physical_bridges'),\
                  mock.patch.object(self.agent, 'setup_integration_br'),\
                  mock.patch.object(self.agent, 'state_rpc'), \
                  mock.patch.object(self.agent.ovs.ovsdb, 'idl_monitor'):
              # block RPC calls and bridge calls
              self.agent.rpc_loop(polling_manager=mock.Mock())
  
      def test_scan_ports_failure(self):
          self._test_scan_ports_failure('scan_ports')
  
      def test_scan_ancillary_ports_failure(self):
          with mock.patch.object(self.agent, 'scan_ports'):
              with mock.patch.object(self.agent, 'update_stale_ofport_rules'):
                  self.agent.ancillary_brs = mock.Mock()
                  self._test_scan_ports_failure('scan_ancillary_ports')
  
      def test_ext_br_recreated(self):
          self._setup_for_dvr_test()
          reset_methods = (
              'reset_ovs_parameters', 'reset_dvr_parameters',
              'setup_dvr_flows_on_integ_br', 'setup_dvr_flows_on_tun_br',
              'setup_dvr_flows_on_phys_br', 'setup_dvr_mac_flows_on_all_brs')
          for method in reset_methods:
              mock.patch.object(self.agent.dvr_agent, method).start()
          bridge_mappings = {'physnet0': 'br-ex0',
                             'physnet1': 'br-ex1'}
          ex_br_mocks = [mock.Mock(br_name='br-ex0'),
                         mock.Mock(br_name='br-ex1')]
          phys_bridges = {'physnet0': ex_br_mocks[0],
                          'physnet1': ex_br_mocks[1]},
          bridges_added = ['br-ex0']
          with mock.patch.object(self.agent, 'check_ovs_status',
                                 return_value=constants.OVS_NORMAL), \
                  mock.patch.object(self.agent, '_agent_has_updates',
                                    side_effect=TypeError('loop exit')), \
                  mock.patch.dict(self.agent.bridge_mappings, bridge_mappings,
                                  clear=True), \
                  mock.patch.dict(self.agent.phys_brs, phys_bridges,
                                  clear=True), \
                  mock.patch.object(self.agent, 'setup_physical_bridges') as \
                  setup_physical_bridges, \
                  mock.patch.object(self.agent.ovs.ovsdb, 'idl_monitor') as \
                  mock_idl_monitor:
              mock_idl_monitor.bridges_added = bridges_added
              try:
                  self.agent.rpc_loop(polling_manager=mock.Mock())
              except TypeError:
                  pass
          # Setup bridges should be called once even if it will raise Runtime
          # Error because TypeError is raised in _agent_has_updates to stop
          # agent after first loop iteration
          setup_physical_bridges.assert_called_once_with({'physnet0': 'br-ex0'})
          # Ensure dvr_agent methods were called correctly
          self.agent.dvr_agent.reset_ovs_parameters.assert_called_once_with(
              self.agent.int_br, self.agent.tun_br, self.agent.phys_brs,
              self.agent.patch_int_ofport, self.agent.patch_tun_ofport)
          self.agent.dvr_agent.reset_dvr_parameters.assert_called_once_with()
          (self.agent.dvr_agent.setup_dvr_flows_on_phys_br.
           assert_called_once_with({'physnet0': 'br-ex0'}))
          (self.agent.dvr_agent.setup_dvr_flows_on_integ_br.
              assert_called_once_with())
          (self.agent.dvr_agent.setup_dvr_flows_on_tun_br.
              assert_called_once_with())
          (self.agent.dvr_agent.setup_dvr_mac_flows_on_all_brs.
              assert_called_once_with())
  
  
  class TestOvsDvrNeutronAgentOSKen(TestOvsDvrNeutronAgent,
                                  ovs_test_base.OVSOSKenTestBase):
      pass
  
  
  class TestValidateTunnelLocalIP(base.BaseTestCase):
      def test_validate_local_ip_with_valid_ip(self):
          mock_get_device_by_ip = mock.patch.object(
              ip_lib.IPWrapper, 'get_device_by_ip').start()
          ovs_agent.validate_local_ip(FAKE_IP1)
          mock_get_device_by_ip.assert_called_once_with(FAKE_IP1)
  
      def test_validate_local_ip_with_valid_ipv6(self):
          mock_get_device_by_ip = mock.patch.object(
              ip_lib.IPWrapper, 'get_device_by_ip').start()
          ovs_agent.validate_local_ip(FAKE_IP6)
          mock_get_device_by_ip.assert_called_once_with(FAKE_IP6)
  
      def test_validate_local_ip_with_none_ip(self):
          with testtools.ExpectedException(SystemExit):
              ovs_agent.validate_local_ip(None)
  
      def test_validate_local_ip_with_invalid_ip(self):
          mock_get_device_by_ip = mock.patch.object(
              ip_lib.IPWrapper, 'get_device_by_ip').start()
          mock_get_device_by_ip.return_value = None
          with testtools.ExpectedException(SystemExit):
              ovs_agent.validate_local_ip(FAKE_IP1)
          mock_get_device_by_ip.assert_called_once_with(FAKE_IP1)
  
      def test_validate_local_ip_with_invalid_ipv6(self):
          mock_get_device_by_ip = mock.patch.object(
              ip_lib.IPWrapper, 'get_device_by_ip').start()
          mock_get_device_by_ip.return_value = None
          with testtools.ExpectedException(SystemExit):
              ovs_agent.validate_local_ip(FAKE_IP6)
          mock_get_device_by_ip.assert_called_once_with(FAKE_IP6)
  
  
  class TestOvsAgentTunnelName(base.BaseTestCase):
      def test_get_tunnel_hash_invalid_address(self):
          hashlen = n_const.DEVICE_NAME_MAX_LEN
          self.assertIsNone(
              ovs_agent.OVSNeutronAgent.get_tunnel_hash('a.b.c.d', hashlen))
  
      def test_get_tunnel_name_vxlan(self):
          self.assertEqual(
              'vxlan-7f000002',
              ovs_agent.OVSNeutronAgent.get_tunnel_name(
                  'vxlan', '127.0.0.1', '127.0.0.2'))
  
      def test_get_tunnel_name_gre(self):
          self.assertEqual(
              'gre-7f000002',
              ovs_agent.OVSNeutronAgent.get_tunnel_name(
                  'gre', '127.0.0.1', '127.0.0.2'))
  
      def test_get_tunnel_name_vxlan_ipv6(self):
          self.assertEqual(
              'vxlan-pehtjzksi',
              ovs_agent.OVSNeutronAgent.get_tunnel_name(
                  'vxlan', '2001:db8::1', '2001:db8::2'))
  
      def test_get_tunnel_name_gre_ipv6(self):
          self.assertEqual(
              'gre-pehtjzksiqr',
              ovs_agent.OVSNeutronAgent.get_tunnel_name(
                  'gre', '2001:db8::1', '2001:db8::2'))
  