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 os
  
  import mock
  from oslo.config import cfg
  
  from neutron.agent.linux import ip_lib
  from neutron.agent.linux import utils
  from neutron.common import constants
  from neutron.common import exceptions
  from neutron.plugins.common import constants as p_const
  from neutron.plugins.linuxbridge.agent import linuxbridge_neutron_agent
  from neutron.plugins.linuxbridge.common import constants as lconst
  from neutron.tests import base
  
  LOCAL_IP = '192.168.0.33'
  DEVICE_1 = 'tapabcdef01-12'
  
  
  class FakeIpLinkCommand(object):
      def set_up(self):
          pass
  
  
  class FakeIpDevice(object):
      def __init__(self):
          self.link = FakeIpLinkCommand()
  
  
  class TestLinuxBridge(base.BaseTestCase):
  
      def setUp(self):
          super(TestLinuxBridge, self).setUp()
          interface_mappings = {'physnet1': 'eth1'}
          root_helper = cfg.CONF.AGENT.root_helper
  
          self.linux_bridge = linuxbridge_neutron_agent.LinuxBridgeManager(
              interface_mappings, root_helper)
  
      def test_ensure_physical_in_bridge_invalid(self):
          result = self.linux_bridge.ensure_physical_in_bridge('network_id',
                                                               p_const.TYPE_VLAN,
                                                               'physnetx',
                                                               7)
          self.assertFalse(result)
  
      def test_ensure_physical_in_bridge_flat(self):
          with mock.patch.object(self.linux_bridge,
                                 'ensure_flat_bridge') as flat_bridge_func:
              self.linux_bridge.ensure_physical_in_bridge(
                  'network_id', p_const.TYPE_FLAT, 'physnet1', None)
          self.assertTrue(flat_bridge_func.called)
  
      def test_ensure_physical_in_bridge_vlan(self):
          with mock.patch.object(self.linux_bridge,
                                 'ensure_vlan_bridge') as vlan_bridge_func:
              self.linux_bridge.ensure_physical_in_bridge(
                  'network_id', p_const.TYPE_VLAN, 'physnet1', 7)
          self.assertTrue(vlan_bridge_func.called)
  
      def test_ensure_physical_in_bridge_vxlan(self):
          self.linux_bridge.vxlan_mode = lconst.VXLAN_UCAST
          with mock.patch.object(self.linux_bridge,
                                 'ensure_vxlan_bridge') as vxlan_bridge_func:
              self.linux_bridge.ensure_physical_in_bridge(
                  'network_id', 'vxlan', 'physnet1', 7)
          self.assertTrue(vxlan_bridge_func.called)
  
  
  class TestLinuxBridgeAgent(base.BaseTestCase):
  
      LINK_SAMPLE = [
          '1: lo: <LOOPBACK,UP,LOWER_UP> mtu 16436 qdisc noqueue \\'
          'state UNKNOWN \\'
          'link/loopback 00:00:00:00:00:00 brd 00:00:00:00:00:00',
          '2: eth77: <BROADCAST,MULTICAST,UP,LOWER_UP> mtu 1500 \\'
          'qdisc mq state UP qlen 1000\    link/ether \\'
          'cc:dd:ee:ff:ab:cd brd ff:ff:ff:ff:ff:ff']
  
      def setUp(self):
          super(TestLinuxBridgeAgent, self).setUp()
          # disable setting up periodic state reporting
          cfg.CONF.set_override('report_interval', 0, 'AGENT')
          cfg.CONF.set_default('firewall_driver',
                               'neutron.agent.firewall.NoopFirewallDriver',
                               group='SECURITYGROUP')
          self.execute_p = mock.patch.object(ip_lib.IPWrapper, '_execute')
          self.execute = self.execute_p.start()
          self.execute.return_value = '\n'.join(self.LINK_SAMPLE)
          self.get_mac_p = mock.patch('neutron.agent.linux.utils.'
                                      'get_interface_mac')
          self.get_mac = self.get_mac_p.start()
          self.get_mac.return_value = '00:00:00:00:00:01'
          self.agent = linuxbridge_neutron_agent.LinuxBridgeNeutronAgentRPC({},
                                                                            0,
                                                                            None)
  
      def test_treat_devices_removed_with_existed_device(self):
          agent = linuxbridge_neutron_agent.LinuxBridgeNeutronAgentRPC({},
                                                                       0,
                                                                       None)
          devices = [DEVICE_1]
          with contextlib.nested(
              mock.patch.object(agent.plugin_rpc, "update_device_down"),
              mock.patch.object(agent, "remove_devices_filter")
          ) as (fn_udd, fn_rdf):
              fn_udd.return_value = {'device': DEVICE_1,
                                     'exists': True}
              with mock.patch.object(linuxbridge_neutron_agent.LOG,
                                     'info') as log:
                  resync = agent.treat_devices_removed(devices)
                  self.assertEqual(2, log.call_count)
                  self.assertFalse(resync)
                  self.assertTrue(fn_udd.called)
                  self.assertTrue(fn_rdf.called)
  
      def test_treat_devices_removed_with_not_existed_device(self):
          agent = linuxbridge_neutron_agent.LinuxBridgeNeutronAgentRPC({},
                                                                       0,
                                                                       None)
          devices = [DEVICE_1]
          with contextlib.nested(
              mock.patch.object(agent.plugin_rpc, "update_device_down"),
              mock.patch.object(agent, "remove_devices_filter")
          ) as (fn_udd, fn_rdf):
              fn_udd.return_value = {'device': DEVICE_1,
                                     'exists': False}
              with mock.patch.object(linuxbridge_neutron_agent.LOG,
                                     'debug') as log:
                  resync = agent.treat_devices_removed(devices)
                  self.assertEqual(1, log.call_count)
                  self.assertFalse(resync)
                  self.assertTrue(fn_udd.called)
                  self.assertTrue(fn_rdf.called)
  
      def test_treat_devices_removed_failed(self):
          agent = linuxbridge_neutron_agent.LinuxBridgeNeutronAgentRPC({},
                                                                       0,
                                                                       None)
          devices = [DEVICE_1]
          with contextlib.nested(
              mock.patch.object(agent.plugin_rpc, "update_device_down"),
              mock.patch.object(agent, "remove_devices_filter")
          ) as (fn_udd, fn_rdf):
              fn_udd.side_effect = Exception()
              with mock.patch.object(linuxbridge_neutron_agent.LOG,
                                     'debug') as log:
                  resync = agent.treat_devices_removed(devices)
                  self.assertEqual(2, log.call_count)
                  self.assertTrue(resync)
                  self.assertTrue(fn_udd.called)
                  self.assertTrue(fn_rdf.called)
  
      def _test_scan_devices(self, previous, updated,
                             fake_current, expected, sync):
          self.agent.br_mgr = mock.Mock()
          self.agent.br_mgr.get_tap_devices.return_value = fake_current
  
          self.agent.updated_devices = updated
          results = self.agent.scan_devices(previous, sync)
          self.assertEqual(expected, results)
  
      def test_scan_devices_no_changes(self):
          previous = {'current': set([1, 2]),
                      'updated': set(),
                      'added': set(),
                      'removed': set()}
          fake_current = set([1, 2])
          updated = set()
          expected = {'current': set([1, 2]),
                      'updated': set(),
                      'added': set(),
                      'removed': set()}
  
          self._test_scan_devices(previous, updated, fake_current, expected,
                                  sync=False)
  
      def test_scan_devices_added_removed(self):
          previous = {'current': set([1, 2]),
                      'updated': set(),
                      'added': set(),
                      'removed': set()}
          fake_current = set([2, 3])
          updated = set()
          expected = {'current': set([2, 3]),
                      'updated': set(),
                      'added': set([3]),
                      'removed': set([1])}
  
          self._test_scan_devices(previous, updated, fake_current, expected,
                                  sync=False)
  
      def test_scan_devices_removed_retried_on_sync(self):
          previous = {'current': set([2, 3]),
                      'updated': set(),
                      'added': set(),
                      'removed': set([1])}
          fake_current = set([2, 3])
          updated = set()
          expected = {'current': set([2, 3]),
                      'updated': set(),
                      'added': set([2, 3]),
                      'removed': set([1])}
  
          self._test_scan_devices(previous, updated, fake_current, expected,
                                  sync=True)
  
      def test_scan_devices_vanished_removed_on_sync(self):
          previous = {'current': set([2, 3]),
                      'updated': set(),
                      'added': set(),
                      'removed': set([1])}
          # Device 2 disappeared.
          fake_current = set([3])
          updated = set()
          # Device 1 should be retried.
          expected = {'current': set([3]),
                      'updated': set(),
                      'added': set([3]),
                      'removed': set([1, 2])}
  
          self._test_scan_devices(previous, updated, fake_current, expected,
                                  sync=True)
  
      def test_scan_devices_updated(self):
          previous = {'current': set([1, 2]),
                      'updated': set(),
                      'added': set(),
                      'removed': set()}
          fake_current = set([1, 2])
          updated = set([1])
          expected = {'current': set([1, 2]),
                      'updated': set([1]),
                      'added': set(),
                      'removed': set()}
  
          self._test_scan_devices(previous, updated, fake_current, expected,
                                  sync=False)
  
      def test_scan_devices_updated_non_existing(self):
          previous = {'current': set([1, 2]),
                      'updated': set(),
                      'added': set(),
                      'removed': set()}
          fake_current = set([1, 2])
          updated = set([3])
          expected = {'current': set([1, 2]),
                      'updated': set(),
                      'added': set(),
                      'removed': set()}
  
          self._test_scan_devices(previous, updated, fake_current, expected,
                                  sync=False)
  
      def test_scan_devices_updated_on_sync(self):
          previous = {'current': set([1, 2]),
                      'updated': set([1]),
                      'added': set(),
                      'removed': set()}
          fake_current = set([1, 2])
          updated = set([2])
          expected = {'current': set([1, 2]),
                      'updated': set([1, 2]),
                      'added': set([1, 2]),
                      'removed': set()}
  
          self._test_scan_devices(previous, updated, fake_current, expected,
                                  sync=True)
  
      def test_process_network_devices(self):
          agent = self.agent
          device_info = {'current': set(),
                         'added': set(['tap3', 'tap4']),
                         'updated': set(['tap2', 'tap3']),
                         'removed': set(['tap1'])}
          agent.prepare_devices_filter = mock.Mock()
          agent.refresh_firewall = mock.Mock()
          agent.treat_devices_added_updated = mock.Mock(return_value=False)
          agent.treat_devices_removed = mock.Mock(return_value=False)
  
          agent.process_network_devices(device_info)
  
          agent.prepare_devices_filter.assert_called_with(set(['tap3', 'tap4']))
          self.assertTrue(agent.refresh_firewall.called)
          agent.treat_devices_added_updated.assert_called_with(set(['tap2',
                                                                    'tap3',
                                                                    'tap4']))
          agent.treat_devices_removed.assert_called_with(set(['tap1']))
  
      def test_treat_devices_added_updated_admin_state_up_true(self):
          agent = self.agent
          mock_details = {'device': 'dev123',
                          'port_id': 'port123',
                          'network_id': 'net123',
                          'admin_state_up': True,
                          'network_type': 'vlan',
                          'segmentation_id': 100,
                          'physical_network': 'physnet1'}
          agent.plugin_rpc = mock.Mock()
          agent.plugin_rpc.get_devices_details_list.return_value = [mock_details]
          agent.br_mgr = mock.Mock()
          agent.br_mgr.add_interface.return_value = True
          resync_needed = agent.treat_devices_added_updated(set(['tap1']))
  
          self.assertFalse(resync_needed)
          agent.br_mgr.add_interface.assert_called_with('net123', 'vlan',
                                                        'physnet1', 100,
                                                        'port123')
          self.assertTrue(agent.plugin_rpc.update_device_up.called)
  
      def test_treat_devices_added_updated_admin_state_up_false(self):
          agent = self.agent
          mock_details = {'device': 'dev123',
                          'port_id': 'port123',
                          'network_id': 'net123',
                          'admin_state_up': False,
                          'network_type': 'vlan',
                          'segmentation_id': 100,
                          'physical_network': 'physnet1'}
          agent.plugin_rpc = mock.Mock()
          agent.plugin_rpc.get_devices_details_list.return_value = [mock_details]
          agent.remove_port_binding = mock.Mock()
          resync_needed = agent.treat_devices_added_updated(set(['tap1']))
  
          self.assertFalse(resync_needed)
          agent.remove_port_binding.assert_called_with('net123', 'port123')
          self.assertFalse(agent.plugin_rpc.update_device_up.called)
  
  
  class TestLinuxBridgeManager(base.BaseTestCase):
      def setUp(self):
          super(TestLinuxBridgeManager, self).setUp()
          self.interface_mappings = {'physnet1': 'eth1'}
          self.root_helper = cfg.CONF.AGENT.root_helper
  
          self.lbm = linuxbridge_neutron_agent.LinuxBridgeManager(
              self.interface_mappings, self.root_helper)
  
      def test_interface_exists_on_bridge(self):
          with mock.patch.object(os, 'listdir') as listdir_fn:
              listdir_fn.return_value = ["abc"]
              self.assertTrue(
                  self.lbm.interface_exists_on_bridge("br-int", "abc")
              )
              self.assertFalse(
                  self.lbm.interface_exists_on_bridge("br-int", "abd")
              )
  
      def test_get_bridge_name(self):
          nw_id = "123456789101112"
          self.assertEqual(self.lbm.get_bridge_name(nw_id),
                           "brq" + nw_id[0:11])
          nw_id = ""
          self.assertEqual(self.lbm.get_bridge_name(nw_id),
                           "brq")
  
      def test_get_subinterface_name(self):
          self.assertEqual(self.lbm.get_subinterface_name("eth0", "0"),
                           "eth0.0")
          self.assertEqual(self.lbm.get_subinterface_name("eth0", ""),
                           "eth0.")
  
      def test_get_tap_device_name(self):
          if_id = "123456789101112"
          self.assertEqual(self.lbm.get_tap_device_name(if_id),
                           constants.TAP_DEVICE_PREFIX + if_id[0:11])
          if_id = ""
          self.assertEqual(self.lbm.get_tap_device_name(if_id),
                           constants.TAP_DEVICE_PREFIX)
  
      def test_get_vxlan_device_name(self):
          vn_id = constants.MAX_VXLAN_VNI
          self.assertEqual(self.lbm.get_vxlan_device_name(vn_id),
                           "vxlan-" + str(vn_id))
          self.assertIsNone(self.lbm.get_vxlan_device_name(vn_id + 1))
  
      def test_get_all_neutron_bridges(self):
          br_list = ["br-int", "brq1", "brq2", "br-ex"]
          with mock.patch.object(os, 'listdir') as listdir_fn:
              listdir_fn.return_value = br_list
              self.assertEqual(self.lbm.get_all_neutron_bridges(),
                               br_list[1:3])
              self.assertTrue(listdir_fn.called)
  
      def test_get_interfaces_on_bridge(self):
          with contextlib.nested(
              mock.patch.object(utils, 'execute'),
              mock.patch.object(os, 'listdir'),
              mock.patch.object(ip_lib, 'device_exists', return_value=True)
          ) as (exec_fn, listdir_fn, dev_exists_fn):
              listdir_fn.return_value = ["qbr1"]
              self.assertEqual(self.lbm.get_interfaces_on_bridge("br0"),
                               ["qbr1"])
  
      def test_get_interfaces_on_bridge_not_existing(self):
          with mock.patch.object(ip_lib, 'device_exists', return_value=False):
              self.assertEqual([], self.lbm.get_interfaces_on_bridge("br0"))
  
      def test_get_tap_devices_count(self):
          with mock.patch.object(os, 'listdir') as listdir_fn:
              listdir_fn.return_value = ['tap2101', 'eth0.100', 'vxlan-1000']
              self.assertEqual(self.lbm.get_tap_devices_count('br0'), 1)
              listdir_fn.side_effect = OSError()
              self.assertEqual(self.lbm.get_tap_devices_count('br0'), 0)
  
      def test_get_interface_by_ip(self):
          with contextlib.nested(
              mock.patch.object(ip_lib.IPWrapper, 'get_devices'),
              mock.patch.object(ip_lib.IpAddrCommand, 'list')
          ) as (get_dev_fn, ip_list_fn):
              device = mock.Mock()
              device.name = 'dev_name'
              get_dev_fn.return_value = [device]
              ip_list_fn.returnvalue = mock.Mock()
              self.assertEqual(self.lbm.get_interface_by_ip(LOCAL_IP),
                               'dev_name')
  
      def test_get_bridge_for_tap_device(self):
          with contextlib.nested(
              mock.patch.object(self.lbm, "get_all_neutron_bridges"),
              mock.patch.object(self.lbm, "get_interfaces_on_bridge")
          ) as (get_all_qbr_fn, get_if_fn):
              get_all_qbr_fn.return_value = ["br-int", "br-ex"]
              get_if_fn.return_value = ["tap1", "tap2", "tap3"]
              self.assertEqual(self.lbm.get_bridge_for_tap_device("tap1"),
                               "br-int")
              self.assertIsNone(self.lbm.get_bridge_for_tap_device("tap4"))
  
      def test_is_device_on_bridge(self):
          self.assertTrue(not self.lbm.is_device_on_bridge(""))
          with mock.patch.object(os.path, 'exists') as exists_fn:
              exists_fn.return_value = True
              self.assertTrue(self.lbm.is_device_on_bridge("tap1"))
              exists_fn.assert_called_with(
                  "/sys/class/net/tap1/brport"
              )
  
      def test_get_interface_details(self):
          with contextlib.nested(
              mock.patch.object(ip_lib.IpAddrCommand, 'list'),
              mock.patch.object(ip_lib.IpRouteCommand, 'get_gateway')
          ) as (list_fn, getgw_fn):
              gwdict = dict(gateway='1.1.1.1')
              getgw_fn.return_value = gwdict
              ipdict = dict(cidr='1.1.1.1/24',
                            broadcast='1.1.1.255',
                            scope='global',
                            ip_version=4,
                            dynamic=False)
              list_fn.return_value = ipdict
              ret = self.lbm.get_interface_details("eth0")
  
              self.assertTrue(list_fn.called)
              self.assertTrue(getgw_fn.called)
              self.assertEqual(ret, (ipdict, gwdict))
  
      def test_ensure_flat_bridge(self):
          with contextlib.nested(
              mock.patch.object(ip_lib.IpAddrCommand, 'list'),
              mock.patch.object(ip_lib.IpRouteCommand, 'get_gateway')
          ) as (list_fn, getgw_fn):
              gwdict = dict(gateway='1.1.1.1')
              getgw_fn.return_value = gwdict
              ipdict = dict(cidr='1.1.1.1/24',
                            broadcast='1.1.1.255',
                            scope='global',
                            ip_version=4,
                            dynamic=False)
              list_fn.return_value = ipdict
              with mock.patch.object(self.lbm, 'ensure_bridge') as ens:
                  self.assertEqual(
                      self.lbm.ensure_flat_bridge("123", "eth0"),
                      "eth0"
                  )
                  self.assertTrue(list_fn.called)
                  self.assertTrue(getgw_fn.called)
                  ens.assert_called_once_with("brq123", "eth0",
                                              ipdict, gwdict)
  
      def test_ensure_vlan_bridge(self):
          with contextlib.nested(
              mock.patch.object(self.lbm, 'ensure_vlan'),
              mock.patch.object(self.lbm, 'ensure_bridge'),
              mock.patch.object(self.lbm, 'get_interface_details'),
          ) as (ens_vl_fn, ens, get_int_det_fn):
              ens_vl_fn.return_value = "eth0.1"
              get_int_det_fn.return_value = (None, None)
              self.assertEqual(self.lbm.ensure_vlan_bridge("123", "eth0", "1"),
                               "eth0.1")
              ens.assert_called_with("brq123", "eth0.1", None, None)
  
              get_int_det_fn.return_value = ("ips", "gateway")
              self.assertEqual(self.lbm.ensure_vlan_bridge("123", "eth0", "1"),
                               "eth0.1")
              ens.assert_called_with("brq123", "eth0.1", "ips", "gateway")
  
      def test_ensure_local_bridge(self):
          with mock.patch.object(self.lbm, 'ensure_bridge') as ens_fn:
              self.lbm.ensure_local_bridge("54321")
              ens_fn.assert_called_once_with("brq54321")
  
      def test_ensure_vlan(self):
          with mock.patch.object(ip_lib, 'device_exists') as de_fn:
              de_fn.return_value = True
              self.assertEqual(self.lbm.ensure_vlan("eth0", "1"), "eth0.1")
              de_fn.return_value = False
              with mock.patch.object(utils, 'execute') as exec_fn:
                  exec_fn.return_value = False
                  self.assertEqual(self.lbm.ensure_vlan("eth0", "1"), "eth0.1")
                  # FIXME(kevinbenton): validate the params to the exec_fn calls
                  self.assertEqual(exec_fn.call_count, 2)
                  exec_fn.return_value = True
                  self.assertIsNone(self.lbm.ensure_vlan("eth0", "1"))
                  self.assertEqual(exec_fn.call_count, 3)
  
      def test_ensure_vxlan(self):
          seg_id = "12345678"
          self.lbm.local_int = 'eth0'
          self.lbm.vxlan_mode = lconst.VXLAN_MCAST
          with mock.patch.object(ip_lib, 'device_exists') as de_fn:
              de_fn.return_value = True
              self.assertEqual(self.lbm.ensure_vxlan(seg_id), "vxlan-" + seg_id)
              de_fn.return_value = False
              with mock.patch.object(self.lbm.ip,
                                     'add_vxlan') as add_vxlan_fn:
                  add_vxlan_fn.return_value = FakeIpDevice()
                  self.assertEqual(self.lbm.ensure_vxlan(seg_id),
                                   "vxlan-" + seg_id)
                  add_vxlan_fn.assert_called_with("vxlan-" + seg_id, seg_id,
                                                  group="224.0.0.1",
                                                  dev=self.lbm.local_int)
                  cfg.CONF.set_override('l2_population', 'True', 'VXLAN')
                  self.assertEqual(self.lbm.ensure_vxlan(seg_id),
                                   "vxlan-" + seg_id)
                  add_vxlan_fn.assert_called_with("vxlan-" + seg_id, seg_id,
                                                  group="224.0.0.1",
                                                  dev=self.lbm.local_int,
                                                  proxy=True)
  
      def test_update_interface_ip_details(self):
          gwdict = dict(gateway='1.1.1.1',
                        metric=50)
          ipdict = dict(cidr='1.1.1.1/24',
                        broadcast='1.1.1.255',
                        scope='global',
                        ip_version=4,
                        dynamic=False)
          with contextlib.nested(
              mock.patch.object(ip_lib.IpAddrCommand, 'add'),
              mock.patch.object(ip_lib.IpAddrCommand, 'delete')
          ) as (add_fn, del_fn):
              self.lbm.update_interface_ip_details("br0", "eth0",
                                                   [ipdict], None)
              self.assertTrue(add_fn.called)
              self.assertTrue(del_fn.called)
  
          with contextlib.nested(
              mock.patch.object(ip_lib.IpRouteCommand, 'add_gateway'),
              mock.patch.object(ip_lib.IpRouteCommand, 'delete_gateway')
          ) as (addgw_fn, delgw_fn):
              self.lbm.update_interface_ip_details("br0", "eth0",
                                                   None, gwdict)
              self.assertTrue(addgw_fn.called)
              self.assertTrue(delgw_fn.called)
  
      def test_bridge_exists_and_ensure_up(self):
          ip_lib_mock = mock.Mock()
          with mock.patch.object(ip_lib, 'IPDevice', return_value=ip_lib_mock):
              # device exists
              self.assertTrue(self.lbm._bridge_exists_and_ensure_up("br0"))
              self.assertTrue(ip_lib_mock.link.set_up.called)
              # device doesn't exists
              ip_lib_mock.link.set_up.side_effect = RuntimeError
              self.assertFalse(self.lbm._bridge_exists_and_ensure_up("br0"))
  
      def test_ensure_bridge(self):
          with contextlib.nested(
              mock.patch.object(self.lbm, '_bridge_exists_and_ensure_up'),
              mock.patch.object(utils, 'execute'),
              mock.patch.object(self.lbm, 'update_interface_ip_details'),
              mock.patch.object(self.lbm, 'interface_exists_on_bridge'),
              mock.patch.object(self.lbm, 'is_device_on_bridge'),
              mock.patch.object(self.lbm, 'get_bridge_for_tap_device'),
          ) as (de_fn, exec_fn, upd_fn, ie_fn, if_br_fn, get_if_br_fn):
              de_fn.return_value = False
              exec_fn.return_value = False
              self.assertEqual(self.lbm.ensure_bridge("br0", None), "br0")
              ie_fn.return_Value = False
              self.lbm.ensure_bridge("br0", "eth0")
              upd_fn.assert_called_with("br0", "eth0", None, None)
              ie_fn.assert_called_with("br0", "eth0")
  
              self.lbm.ensure_bridge("br0", "eth0", "ips", "gateway")
              upd_fn.assert_called_with("br0", "eth0", "ips", "gateway")
              ie_fn.assert_called_with("br0", "eth0")
  
              exec_fn.side_effect = Exception()
              de_fn.return_value = True
              self.lbm.ensure_bridge("br0", "eth0")
              ie_fn.assert_called_with("br0", "eth0")
  
              exec_fn.reset_mock()
              exec_fn.side_effect = None
              de_fn.return_value = True
              ie_fn.return_value = False
              get_if_br_fn.return_value = "br1"
              self.lbm.ensure_bridge("br0", "eth0")
              expected = [
                  mock.call(['brctl', 'delif', 'br1', 'eth0'],
                            root_helper=self.root_helper),
                  mock.call(['brctl', 'addif', 'br0', 'eth0'],
                            root_helper=self.root_helper),
              ]
              exec_fn.assert_has_calls(expected)
  
      def test_ensure_physical_in_bridge(self):
          self.assertFalse(
              self.lbm.ensure_physical_in_bridge("123", p_const.TYPE_VLAN,
                                                 "phys", "1")
          )
          with mock.patch.object(self.lbm, "ensure_flat_bridge") as flbr_fn:
              self.assertTrue(
                  self.lbm.ensure_physical_in_bridge("123", p_const.TYPE_FLAT,
                                                     "physnet1", None)
              )
              self.assertTrue(flbr_fn.called)
          with mock.patch.object(self.lbm, "ensure_vlan_bridge") as vlbr_fn:
              self.assertTrue(
                  self.lbm.ensure_physical_in_bridge("123", p_const.TYPE_VLAN,
                                                     "physnet1", "1")
              )
              self.assertTrue(vlbr_fn.called)
  
          with mock.patch.object(self.lbm, "ensure_vxlan_bridge") as vlbr_fn:
              self.lbm.vxlan_mode = lconst.VXLAN_MCAST
              self.assertTrue(
                  self.lbm.ensure_physical_in_bridge("123", p_const.TYPE_VXLAN,
                                                     "physnet1", "1")
              )
              self.assertTrue(vlbr_fn.called)
  
      def test_add_tap_interface(self):
          with mock.patch.object(ip_lib, "device_exists") as de_fn:
              de_fn.return_value = False
              self.assertFalse(
                  self.lbm.add_tap_interface("123", p_const.TYPE_VLAN,
                                             "physnet1", "1", "tap1")
              )
  
              de_fn.return_value = True
              with contextlib.nested(
                  mock.patch.object(self.lbm, "ensure_local_bridge"),
                  mock.patch.object(utils, "execute"),
                  mock.patch.object(self.lbm, "get_bridge_for_tap_device")
              ) as (en_fn, exec_fn, get_br):
                  exec_fn.return_value = False
                  get_br.return_value = True
                  self.assertTrue(self.lbm.add_tap_interface("123",
                                                             p_const.TYPE_LOCAL,
                                                             "physnet1", None,
                                                             "tap1"))
                  en_fn.assert_called_with("123")
  
                  get_br.return_value = False
                  exec_fn.return_value = True
                  self.assertFalse(self.lbm.add_tap_interface("123",
                                                              p_const.TYPE_LOCAL,
                                                              "physnet1", None,
                                                              "tap1"))
  
              with mock.patch.object(self.lbm,
                                     "ensure_physical_in_bridge") as ens_fn:
                  ens_fn.return_value = False
                  self.assertFalse(self.lbm.add_tap_interface("123",
                                                              p_const.TYPE_VLAN,
                                                              "physnet1", "1",
                                                              "tap1"))
  
      def test_add_interface(self):
          with mock.patch.object(self.lbm, "add_tap_interface") as add_tap:
              self.lbm.add_interface("123", p_const.TYPE_VLAN, "physnet-1",
                                     "1", "234")
              add_tap.assert_called_with("123", p_const.TYPE_VLAN, "physnet-1",
                                         "1", "tap234")
  
      def test_delete_vlan_bridge(self):
          with contextlib.nested(
              mock.patch.object(ip_lib, "device_exists"),
              mock.patch.object(self.lbm, "get_interfaces_on_bridge"),
              mock.patch.object(self.lbm, "remove_interface"),
              mock.patch.object(self.lbm, "get_interface_details"),
              mock.patch.object(self.lbm, "update_interface_ip_details"),
              mock.patch.object(self.lbm, "delete_vxlan"),
              mock.patch.object(utils, "execute")
          ) as (de_fn, getif_fn, remif_fn, if_det_fn,
                updif_fn, del_vxlan, exec_fn):
              de_fn.return_value = False
              self.lbm.delete_vlan_bridge("br0")
              self.assertFalse(getif_fn.called)
  
              de_fn.return_value = True
              getif_fn.return_value = ["eth0", "eth1", "vxlan-1002"]
              if_det_fn.return_value = ("ips", "gateway")
              exec_fn.return_value = False
              self.lbm.delete_vlan_bridge("br0")
              updif_fn.assert_called_with("eth1", "br0", "ips", "gateway")
              del_vxlan.assert_called_with("vxlan-1002")
  
      def test_delete_vlan_bridge_with_ip(self):
          with contextlib.nested(
              mock.patch.object(ip_lib, "device_exists"),
              mock.patch.object(self.lbm, "get_interfaces_on_bridge"),
              mock.patch.object(self.lbm, "remove_interface"),
              mock.patch.object(self.lbm, "get_interface_details"),
              mock.patch.object(self.lbm, "update_interface_ip_details"),
              mock.patch.object(self.lbm, "delete_vlan"),
              mock.patch.object(utils, "execute")
          ) as (de_fn, getif_fn, remif_fn, if_det_fn,
                updif_fn, del_vlan, exec_fn):
              de_fn.return_value = True
              getif_fn.return_value = ["eth0", "eth1.1"]
              if_det_fn.return_value = ("ips", "gateway")
              exec_fn.return_value = False
              self.lbm.delete_vlan_bridge("br0")
              updif_fn.assert_called_with("eth1.1", "br0", "ips", "gateway")
              self.assertFalse(del_vlan.called)
  
      def test_delete_vlan_bridge_no_ip(self):
          with contextlib.nested(
              mock.patch.object(ip_lib, "device_exists"),
              mock.patch.object(self.lbm, "get_interfaces_on_bridge"),
              mock.patch.object(self.lbm, "remove_interface"),
              mock.patch.object(self.lbm, "get_interface_details"),
              mock.patch.object(self.lbm, "update_interface_ip_details"),
              mock.patch.object(self.lbm, "delete_vlan"),
              mock.patch.object(utils, "execute")
          ) as (de_fn, getif_fn, remif_fn, if_det_fn,
                updif_fn, del_vlan, exec_fn):
              de_fn.return_value = True
              getif_fn.return_value = ["eth0", "eth1.1"]
              exec_fn.return_value = False
              if_det_fn.return_value = ([], None)
              self.lbm.delete_vlan_bridge("br0")
              del_vlan.assert_called_with("eth1.1")
              self.assertFalse(updif_fn.called)
  
      def test_delete_vxlan_bridge_no_int_mappings(self):
          interface_mappings = {}
          lbm = linuxbridge_neutron_agent.LinuxBridgeManager(
              interface_mappings, self.root_helper)
  
          with contextlib.nested(
              mock.patch.object(ip_lib, "device_exists"),
              mock.patch.object(lbm, "get_interfaces_on_bridge"),
              mock.patch.object(lbm, "remove_interface"),
              mock.patch.object(lbm, "delete_vxlan"),
              mock.patch.object(utils, "execute")
          ) as (de_fn, getif_fn, remif_fn, del_vxlan, exec_fn):
              de_fn.return_value = False
              lbm.delete_vlan_bridge("br0")
              self.assertFalse(getif_fn.called)
  
              de_fn.return_value = True
              getif_fn.return_value = ["vxlan-1002"]
              exec_fn.return_value = False
              lbm.delete_vlan_bridge("br0")
              del_vxlan.assert_called_with("vxlan-1002")
  
      def test_remove_empty_bridges(self):
          self.lbm.network_map = {'net1': mock.Mock(), 'net2': mock.Mock()}
  
          def tap_count_side_effect(*args):
              return 0 if args[0] == 'brqnet1' else 1
  
          with contextlib.nested(
              mock.patch.object(self.lbm, "delete_vlan_bridge"),
              mock.patch.object(self.lbm, "get_tap_devices_count",
                                side_effect=tap_count_side_effect),
          ) as (del_br_fn, count_tap_fn):
              self.lbm.remove_empty_bridges()
              del_br_fn.assert_called_once_with('brqnet1')
  
      def test_remove_interface(self):
          with contextlib.nested(
              mock.patch.object(ip_lib, "device_exists"),
              mock.patch.object(self.lbm, "is_device_on_bridge"),
              mock.patch.object(utils, "execute")
          ) as (de_fn, isdev_fn, exec_fn):
              de_fn.return_value = False
              self.assertFalse(self.lbm.remove_interface("br0", "eth0"))
              self.assertFalse(isdev_fn.called)
  
              de_fn.return_value = True
              isdev_fn.return_value = False
              self.assertTrue(self.lbm.remove_interface("br0", "eth0"))
  
              isdev_fn.return_value = True
              exec_fn.return_value = True
              self.assertFalse(self.lbm.remove_interface("br0", "eth0"))
  
              exec_fn.return_value = False
              self.assertTrue(self.lbm.remove_interface("br0", "eth0"))
  
      def test_delete_vlan(self):
          with contextlib.nested(
              mock.patch.object(ip_lib, "device_exists"),
              mock.patch.object(utils, "execute")
          ) as (de_fn, exec_fn):
              de_fn.return_value = False
              self.lbm.delete_vlan("eth1.1")
              self.assertFalse(exec_fn.called)
  
              de_fn.return_value = True
              exec_fn.return_value = False
              self.lbm.delete_vlan("eth1.1")
              self.assertTrue(exec_fn.called)
  
      def _check_vxlan_support(self, expected, vxlan_module_supported,
                               vxlan_ucast_supported, vxlan_mcast_supported):
          with contextlib.nested(
              mock.patch.object(self.lbm, 'vxlan_module_supported',
                                return_value=vxlan_module_supported),
              mock.patch.object(self.lbm, 'vxlan_ucast_supported',
                                return_value=vxlan_ucast_supported),
              mock.patch.object(self.lbm, 'vxlan_mcast_supported',
                                return_value=vxlan_mcast_supported)):
              if expected == lconst.VXLAN_NONE:
                  self.assertRaises(exceptions.VxlanNetworkUnsupported,
                                    self.lbm.check_vxlan_support)
                  self.assertEqual(expected, self.lbm.vxlan_mode)
              else:
                  self.lbm.check_vxlan_support()
                  self.assertEqual(expected, self.lbm.vxlan_mode)
  
      def test_check_vxlan_support(self):
          self._check_vxlan_support(expected=lconst.VXLAN_UCAST,
                                    vxlan_module_supported=True,
                                    vxlan_ucast_supported=True,
                                    vxlan_mcast_supported=True)
          self._check_vxlan_support(expected=lconst.VXLAN_MCAST,
                                    vxlan_module_supported=True,
                                    vxlan_ucast_supported=False,
                                    vxlan_mcast_supported=True)
  
          self._check_vxlan_support(expected=lconst.VXLAN_NONE,
                                    vxlan_module_supported=False,
                                    vxlan_ucast_supported=False,
                                    vxlan_mcast_supported=False)
          self._check_vxlan_support(expected=lconst.VXLAN_NONE,
                                    vxlan_module_supported=True,
                                    vxlan_ucast_supported=False,
                                    vxlan_mcast_supported=False)
  
      def _check_vxlan_module_supported(self, expected, execute_side_effect):
          with mock.patch.object(
                  utils, 'execute',
                  side_effect=execute_side_effect):
              self.assertEqual(expected, self.lbm.vxlan_module_supported())
  
      def test_vxlan_module_supported(self):
          self._check_vxlan_module_supported(
              expected=True,
              execute_side_effect=None)
          self._check_vxlan_module_supported(
              expected=False,
              execute_side_effect=RuntimeError())
  
      def _check_vxlan_ucast_supported(
              self, expected, l2_population, iproute_arg_supported, fdb_append):
          cfg.CONF.set_override('l2_population', l2_population, 'VXLAN')
          with contextlib.nested(
                  mock.patch.object(
                      ip_lib, 'device_exists', return_value=False),
                  mock.patch.object(self.lbm, 'delete_vxlan', return_value=None),
                  mock.patch.object(self.lbm, 'ensure_vxlan', return_value=None),
                  mock.patch.object(
                      utils, 'execute',
                      side_effect=None if fdb_append else RuntimeError()),
                  mock.patch.object(
                      ip_lib, 'iproute_arg_supported',
                      return_value=iproute_arg_supported)):
                  self.assertEqual(expected, self.lbm.vxlan_ucast_supported())
  
      def test_vxlan_ucast_supported(self):
          self._check_vxlan_ucast_supported(
              expected=False,
              l2_population=False, iproute_arg_supported=True, fdb_append=True)
          self._check_vxlan_ucast_supported(
              expected=False,
              l2_population=True, iproute_arg_supported=False, fdb_append=True)
          self._check_vxlan_ucast_supported(
              expected=False,
              l2_population=True, iproute_arg_supported=True, fdb_append=False)
          self._check_vxlan_ucast_supported(
              expected=True,
              l2_population=True, iproute_arg_supported=True, fdb_append=True)
  
      def _check_vxlan_mcast_supported(
              self, expected, vxlan_group, iproute_arg_supported):
          cfg.CONF.set_override('vxlan_group', vxlan_group, 'VXLAN')
          with mock.patch.object(
                  ip_lib, 'iproute_arg_supported',
                  return_value=iproute_arg_supported):
              self.assertEqual(expected, self.lbm.vxlan_mcast_supported())
  
      def test_vxlan_mcast_supported(self):
          self._check_vxlan_mcast_supported(
              expected=False,
              vxlan_group='',
              iproute_arg_supported=True)
          self._check_vxlan_mcast_supported(
              expected=False,
              vxlan_group='224.0.0.1',
              iproute_arg_supported=False)
          self._check_vxlan_mcast_supported(
              expected=True,
              vxlan_group='224.0.0.1',
              iproute_arg_supported=True)
  
  
  class TestLinuxBridgeRpcCallbacks(base.BaseTestCase):
      def setUp(self):
          cfg.CONF.set_override('local_ip', LOCAL_IP, 'VXLAN')
          super(TestLinuxBridgeRpcCallbacks, self).setUp()
  
          self.u_execute_p = mock.patch('neutron.agent.linux.utils.execute')
          self.u_execute = self.u_execute_p.start()
  
          class FakeLBAgent(object):
              def __init__(self):
                  self.agent_id = 1
                  self.br_mgr = (linuxbridge_neutron_agent.
                                 LinuxBridgeManager({'physnet1': 'eth1'},
                                                    cfg.CONF.AGENT.root_helper))
  
                  self.br_mgr.vxlan_mode = lconst.VXLAN_UCAST
                  segment = mock.Mock()
                  segment.network_type = 'vxlan'
                  segment.segmentation_id = 1
                  self.br_mgr.network_map['net_id'] = segment
  
          self.lb_rpc = linuxbridge_neutron_agent.LinuxBridgeRpcCallbacks(
              object(),
              FakeLBAgent()
          )
  
          self.root_helper = cfg.CONF.AGENT.root_helper
  
      def test_network_delete(self):
          with contextlib.nested(
              mock.patch.object(self.lb_rpc.agent.br_mgr, "get_bridge_name"),
              mock.patch.object(self.lb_rpc.agent.br_mgr, "delete_vlan_bridge")
          ) as (get_br_fn, del_fn):
              get_br_fn.return_value = "br0"
              self.lb_rpc.network_delete("anycontext", network_id="123")
              get_br_fn.assert_called_with("123")
              del_fn.assert_called_with("br0")
  
      def test_fdb_add(self):
          fdb_entries = {'net_id':
                         {'ports':
                          {'agent_ip': [constants.FLOODING_ENTRY,
                                        ['port_mac', 'port_ip']]},
                          'network_type': 'vxlan',
                          'segment_id': 1}}
  
          with mock.patch.object(utils, 'execute',
                                 return_value='') as execute_fn:
              self.lb_rpc.fdb_add(None, fdb_entries)
  
              expected = [
                  mock.call(['bridge', 'fdb', 'show', 'dev', 'vxlan-1'],
                            root_helper=self.root_helper),
                  mock.call(['bridge', 'fdb', 'add',
                             constants.FLOODING_ENTRY[0],
                             'dev', 'vxlan-1', 'dst', 'agent_ip'],
                            root_helper=self.root_helper,
                            check_exit_code=False),
                  mock.call(['ip', 'neigh', 'replace', 'port_ip', 'lladdr',
                             'port_mac', 'dev', 'vxlan-1', 'nud', 'permanent'],
                            root_helper=self.root_helper,
                            check_exit_code=False),
                  mock.call(['bridge', 'fdb', 'replace', 'port_mac', 'dev',
                             'vxlan-1', 'dst', 'agent_ip'],
                            root_helper=self.root_helper,
                            check_exit_code=False),
              ]
              execute_fn.assert_has_calls(expected)
  
      def test_fdb_ignore(self):
          fdb_entries = {'net_id':
                         {'ports':
                          {LOCAL_IP: [constants.FLOODING_ENTRY,
                                      ['port_mac', 'port_ip']]},
                          'network_type': 'vxlan',
                          'segment_id': 1}}
  
          with mock.patch.object(utils, 'execute',
                                 return_value='') as execute_fn:
              self.lb_rpc.fdb_add(None, fdb_entries)
              self.lb_rpc.fdb_remove(None, fdb_entries)
  
              self.assertFalse(execute_fn.called)
  
          fdb_entries = {'other_net_id':
                         {'ports':
                          {'192.168.0.67': [constants.FLOODING_ENTRY,
                                            ['port_mac', 'port_ip']]},
                          'network_type': 'vxlan',
                          'segment_id': 1}}
  
          with mock.patch.object(utils, 'execute',
                                 return_value='') as execute_fn:
              self.lb_rpc.fdb_add(None, fdb_entries)
              self.lb_rpc.fdb_remove(None, fdb_entries)
  
              self.assertFalse(execute_fn.called)
  
      def test_fdb_remove(self):
          fdb_entries = {'net_id':
                         {'ports':
                          {'agent_ip': [constants.FLOODING_ENTRY,
                                        ['port_mac', 'port_ip']]},
                          'network_type': 'vxlan',
                          'segment_id': 1}}
  
          with mock.patch.object(utils, 'execute',
                                 return_value='') as execute_fn:
              self.lb_rpc.fdb_remove(None, fdb_entries)
  
              expected = [
                  mock.call(['bridge', 'fdb', 'del',
                             constants.FLOODING_ENTRY[0],
                             'dev', 'vxlan-1', 'dst', 'agent_ip'],
                            root_helper=self.root_helper,
                            check_exit_code=False),
                  mock.call(['ip', 'neigh', 'del', 'port_ip', 'lladdr',
                             'port_mac', 'dev', 'vxlan-1'],
                            root_helper=self.root_helper,
                            check_exit_code=False),
                  mock.call(['bridge', 'fdb', 'del', 'port_mac',
                             'dev', 'vxlan-1', 'dst', 'agent_ip'],
                            root_helper=self.root_helper,
                            check_exit_code=False),
              ]
              execute_fn.assert_has_calls(expected)
  
      def test_fdb_update_chg_ip(self):
          fdb_entries = {'chg_ip':
                         {'net_id':
                          {'agent_ip':
                           {'before': [['port_mac', 'port_ip_1']],
                            'after': [['port_mac', 'port_ip_2']]}}}}
  
          with mock.patch.object(utils, 'execute',
                                 return_value='') as execute_fn:
              self.lb_rpc.fdb_update(None, fdb_entries)
  
              expected = [
                  mock.call(['ip', 'neigh', 'replace', 'port_ip_2', 'lladdr',
                             'port_mac', 'dev', 'vxlan-1', 'nud', 'permanent'],
                            root_helper=self.root_helper,
                            check_exit_code=False),
                  mock.call(['ip', 'neigh', 'del', 'port_ip_1', 'lladdr',
                             'port_mac', 'dev', 'vxlan-1'],
                            root_helper=self.root_helper,
                            check_exit_code=False)
              ]
              execute_fn.assert_has_calls(expected)
  