want to use this feature.
Author: Chuck Short <zulcss@ubuntu.com>
Fowarded: not-needed
-diff -Naurp neutron-2014.1.b2.orig/neutron/tests/unit/linuxbridge/test_lb_neutron_agent.py neutron-2014.1.b2/neutron/tests/unit/linuxbridge/test_lb_neutron_agent.py
---- neutron-2014.1.b2.orig/neutron/tests/unit/linuxbridge/test_lb_neutron_agent.py 2014-01-23 10:13:25.000000000 -0500
-+++ neutron-2014.1.b2/neutron/tests/unit/linuxbridge/test_lb_neutron_agent.py 2014-02-03 08:53:04.409255073 -0500
-@@ -46,6 +46,7 @@ class FakeIpDevice(object):
+--- a/neutron/tests/unit/linuxbridge/test_lb_neutron_agent.py
++++ b/neutron/tests/unit/linuxbridge/test_lb_neutron_agent.py
+@@ -47,6 +47,7 @@ class FakeIpDevice(object):
class TestLinuxBridge(base.BaseTestCase):
def setUp(self):
+ self.skipTest("udev not consistently available in Ubuntu buildds")
super(TestLinuxBridge, self).setUp()
- self.addCleanup(cfg.CONF.reset)
interface_mappings = {'physnet1': 'eth1'}
-@@ -109,6 +110,7 @@ class TestLinuxBridgeAgent(base.BaseTest
- self.get_mac.return_value = '00:00:00:00:00:01'
+ root_helper = cfg.CONF.AGENT.root_helper
+@@ -95,6 +96,7 @@ class TestLinuxBridgeAgent(base.BaseTest
+ 'cc:dd:ee:ff:ab:cd brd ff:ff:ff:ff:ff:ff']
- def test_update_devices_failed(self):
-+ self.skipTest("udev not consistently available in Ubuntu buildds")
- agent = linuxbridge_neutron_agent.LinuxBridgeNeutronAgentRPC({},
- 0,
- None)
-@@ -130,6 +132,7 @@ class TestLinuxBridgeAgent(base.BaseTest
- self.assertEqual(3, log.call_count)
-
- def test_process_network_devices_failed(self):
+ def setUp(self):
+ self.skipTest("udev not consistently available in Ubuntu buildds")
- device_info = {'current': [1, 2, 3]}
- agent = linuxbridge_neutron_agent.LinuxBridgeNeutronAgentRPC({},
- 0,
-@@ -158,6 +161,7 @@ class TestLinuxBridgeAgent(base.BaseTest
+ super(TestLinuxBridgeAgent, self).setUp()
+ # disable setting up periodic state reporting
+ cfg.CONF.set_override('report_interval', 0, 'AGENT')
+@@ -163,6 +165,7 @@ class TestLinuxBridgeAgent(base.BaseTest
class TestLinuxBridgeManager(base.BaseTestCase):
def setUp(self):
super(TestLinuxBridgeManager, self).setUp()
self.interface_mappings = {'physnet1': 'eth1'}
self.root_helper = cfg.CONF.AGENT.root_helper
-@@ -667,6 +671,7 @@ class TestLinuxBridgeManager(base.BaseTe
-
- class TestLinuxBridgeRpcCallbacks(base.BaseTestCase):
- def setUp(self):
-+ self.skipTest("udev not consistently available in Ubuntu buildds")
- cfg.CONF.set_override('local_ip', LOCAL_IP, 'VXLAN')
- self.addCleanup(cfg.CONF.reset)
- super(TestLinuxBridgeRpcCallbacks, self).setUp()
-diff -Naurp neutron-2014.1.b2.orig/neutron/tests/unit/linuxbridge/test_lb_neutron_agent.py.orig neutron-2014.1.b2/neutron/tests/unit/linuxbridge/test_lb_neutron_agent.py.orig
---- neutron-2014.1.b2.orig/neutron/tests/unit/linuxbridge/test_lb_neutron_agent.py.orig 1969-12-31 19:00:00.000000000 -0500
-+++ neutron-2014.1.b2/neutron/tests/unit/linuxbridge/test_lb_neutron_agent.py.orig 2014-01-23 10:13:25.000000000 -0500
-@@ -0,0 +1,956 @@
-+# vim: tabstop=4 shiftwidth=4 softtabstop=4
-+
-+# 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
-+import testtools
-+
-+from neutron.agent.linux import ip_lib
-+from neutron.agent.linux import utils
-+from neutron.common import constants
-+from neutron.openstack.common.rpc import common as rpc_common
-+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'
-+
-+
-+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()
-+ self.addCleanup(cfg.CONF.reset)
-+ 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()
-+ cfg.CONF.set_override('rpc_backend',
-+ 'neutron.openstack.common.rpc.impl_fake')
-+ self.execute_p = mock.patch.object(ip_lib.IPWrapper, '_execute')
-+ self.execute = self.execute_p.start()
-+ self.addCleanup(self.execute_p.stop)
-+ 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.addCleanup(self.get_mac_p.stop)
-+ self.get_mac.return_value = '00:00:00:00:00:01'
-+
-+ def test_update_devices_failed(self):
-+ agent = linuxbridge_neutron_agent.LinuxBridgeNeutronAgentRPC({},
-+ 0,
-+ None)
-+ raise_exception = [0]
-+
-+ def info_mock(msg):
-+ if raise_exception[0] < 2:
-+ raise_exception[0] += 1
-+ else:
-+ raise RuntimeError()
-+ with mock.patch.object(agent.br_mgr,
-+ "update_devices") as update_devices:
-+ update_devices.side_effect = RuntimeError
-+ with mock.patch.object(linuxbridge_neutron_agent.LOG,
-+ 'info') as log:
-+ log.side_effect = info_mock
-+ with testtools.ExpectedException(RuntimeError):
-+ agent.daemon_loop()
-+ self.assertEqual(3, log.call_count)
-+
-+ def test_process_network_devices_failed(self):
-+ device_info = {'current': [1, 2, 3]}
-+ agent = linuxbridge_neutron_agent.LinuxBridgeNeutronAgentRPC({},
-+ 0,
-+ None)
-+ raise_exception = [0]
-+
-+ def info_mock(msg):
-+ if raise_exception[0] < 2:
-+ raise_exception[0] += 1
-+ else:
-+ raise RuntimeError()
-+
-+ with mock.patch.object(agent.br_mgr,
-+ "update_devices") as update_devices:
-+ update_devices.side_effect = device_info
-+ with contextlib.nested(
-+ mock.patch.object(linuxbridge_neutron_agent.LOG, 'info'),
-+ mock.patch.object(agent, 'process_network_devices')
-+ ) as (log, process_network_devices):
-+ log.side_effect = info_mock
-+ process_network_devices.side_effect = RuntimeError
-+ with testtools.ExpectedException(RuntimeError):
-+ agent.daemon_loop()
-+ self.assertEqual(3, log.call_count)
-+
-+
-+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_device_exists(self):
-+ with mock.patch.object(utils, 'execute') as execute_fn:
-+ self.assertTrue(self.lbm.device_exists("eth0"))
-+ execute_fn.side_effect = RuntimeError()
-+ self.assertFalse(self.lbm.device_exists("eth0"))
-+
-+ 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),
-+ "tap" + if_id[0:11])
-+ if_id = ""
-+ self.assertEqual(self.lbm.get_tap_device_name(if_id),
-+ "tap")
-+
-+ 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')
-+ ) as (exec_fn, listdir_fn):
-+ listdir_fn.return_value = ["qbr1"]
-+ self.assertEqual(self.lbm.get_interfaces_on_bridge("br0"),
-+ ["qbr1"])
-+
-+ 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/devices/virtual/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(self.lbm, '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")
-+ exec_fn.assert_called_twice()
-+ exec_fn.return_value = True
-+ self.assertIsNone(self.lbm.ensure_vlan("eth0", "1"))
-+ exec_fn.assert_called_once()
-+
-+ def test_ensure_vxlan(self):
-+ seg_id = "12345678"
-+ self.lbm.local_int = 'eth0'
-+ self.lbm.vxlan_mode = lconst.VXLAN_MCAST
-+ with mock.patch.object(self.lbm, '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_ensure_bridge(self):
-+ with contextlib.nested(
-+ mock.patch.object(self.lbm, 'device_exists'),
-+ 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(self.lbm, "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(self.lbm, "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(self.lbm, "delete_vxlan"),
-+ mock.patch.object(utils, "execute")
-+ ) as (de_fn, getif_fn, remif_fn, if_det_fn,
-+ updif_fn, del_vlan, 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.1", "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_vlan.assert_called_with("eth1.1")
-+ del_vxlan.assert_called_with("vxlan-1002")
-+
-+ 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(lbm, "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(self.lbm, "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(self.lbm, "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 test_update_devices(self):
-+ with mock.patch.object(self.lbm, "udev_get_tap_devices") as gt_fn:
-+ gt_fn.return_value = set(["dev1"])
-+ self.assertIsNone(self.lbm.update_devices(set(["dev1"])))
-+
-+ gt_fn.return_value = set(["dev1", "dev2"])
-+ self.assertEqual(self.lbm.update_devices(set(["dev2", "dev3"])),
-+ {"current": set(["dev1", "dev2"]),
-+ "added": set(["dev1"]),
-+ "removed": set(["dev3"])
-+ })
-+
-+ def _check_vxlan_support(self, kernel_version, vxlan_proxy_supported,
-+ fdb_append_supported, l2_population,
-+ expected_mode):
-+ def iproute_supported_side_effect(*args):
-+ if args[1] == 'proxy':
-+ return vxlan_proxy_supported
-+ elif args[1] == 'append':
-+ return fdb_append_supported
-+
-+ with contextlib.nested(
-+ mock.patch("platform.release", return_value=kernel_version),
-+ mock.patch.object(ip_lib, 'iproute_arg_supported',
-+ side_effect=iproute_supported_side_effect),
-+ ) as (kver_fn, ip_arg_fn):
-+ self.lbm.check_vxlan_support()
-+ self.assertEqual(self.lbm.vxlan_mode, expected_mode)
-+
-+ def test_vxlan_mode_ucast(self):
-+ self._check_vxlan_support(kernel_version='3.12',
-+ vxlan_proxy_supported=True,
-+ fdb_append_supported=True,
-+ l2_population=True,
-+ expected_mode=lconst.VXLAN_MCAST)
-+
-+ def test_vxlan_mode_mcast(self):
-+ self._check_vxlan_support(kernel_version='3.12',
-+ vxlan_proxy_supported=True,
-+ fdb_append_supported=False,
-+ l2_population=True,
-+ expected_mode=lconst.VXLAN_MCAST)
-+ self._check_vxlan_support(kernel_version='3.10',
-+ vxlan_proxy_supported=True,
-+ fdb_append_supported=True,
-+ l2_population=True,
-+ expected_mode=lconst.VXLAN_MCAST)
-+
-+ def test_vxlan_mode_unsupported(self):
-+ self._check_vxlan_support(kernel_version='3.7',
-+ vxlan_proxy_supported=True,
-+ fdb_append_supported=True,
-+ l2_population=False,
-+ expected_mode=lconst.VXLAN_NONE)
-+ self._check_vxlan_support(kernel_version='3.10',
-+ vxlan_proxy_supported=False,
-+ fdb_append_supported=False,
-+ l2_population=False,
-+ expected_mode=lconst.VXLAN_NONE)
-+ cfg.CONF.set_override('vxlan_group', '', 'VXLAN')
-+ self._check_vxlan_support(kernel_version='3.12',
-+ vxlan_proxy_supported=True,
-+ fdb_append_supported=True,
-+ l2_population=True,
-+ expected_mode=lconst.VXLAN_NONE)
-+
-+
-+class TestLinuxBridgeRpcCallbacks(base.BaseTestCase):
-+ def setUp(self):
-+ cfg.CONF.set_override('local_ip', LOCAL_IP, 'VXLAN')
-+ self.addCleanup(cfg.CONF.reset)
-+ super(TestLinuxBridgeRpcCallbacks, self).setUp()
-+
-+ self.u_execute_p = mock.patch('neutron.agent.linux.utils.execute')
-+ self.u_execute = self.u_execute_p.start()
-+ self.addCleanup(self.u_execute_p.stop)
-+
-+ 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_port_update(self):
-+ with contextlib.nested(
-+ mock.patch.object(self.lb_rpc.agent.br_mgr,
-+ "get_tap_device_name"),
-+ mock.patch.object(self.lb_rpc.agent.br_mgr,
-+ "udev_get_tap_devices"),
-+ mock.patch.object(self.lb_rpc.agent.br_mgr,
-+ "get_bridge_name"),
-+ mock.patch.object(self.lb_rpc.agent.br_mgr,
-+ "remove_interface"),
-+ mock.patch.object(self.lb_rpc.agent.br_mgr, "add_interface"),
-+ mock.patch.object(self.lb_rpc.agent,
-+ "plugin_rpc", create=True),
-+ mock.patch.object(self.lb_rpc.sg_agent,
-+ "refresh_firewall", create=True)
-+ ) as (get_tap_fn, udev_fn, getbr_fn, remif_fn,
-+ addif_fn, rpc_obj, reffw_fn):
-+ get_tap_fn.return_value = "tap123"
-+ udev_fn.return_value = ["tap123", "tap124"]
-+ port = {"admin_state_up": True,
-+ "id": "1234-5678",
-+ "network_id": "123-123"}
-+ self.lb_rpc.port_update("unused_context", port=port,
-+ vlan_id="1", physical_network="physnet1")
-+ self.assertFalse(reffw_fn.called)
-+ addif_fn.assert_called_with(port["network_id"], p_const.TYPE_VLAN,
-+ "physnet1", "1", port["id"])
-+
-+ self.lb_rpc.port_update("unused_context", port=port,
-+ network_type=p_const.TYPE_VLAN,
-+ segmentation_id="2",
-+ physical_network="physnet1")
-+ self.assertFalse(reffw_fn.called)
-+ addif_fn.assert_called_with(port["network_id"], p_const.TYPE_VLAN,
-+ "physnet1", "2", port["id"])
-+
-+ self.lb_rpc.port_update("unused_context", port=port,
-+ vlan_id=lconst.FLAT_VLAN_ID,
-+ physical_network="physnet1")
-+ self.assertFalse(reffw_fn.called)
-+ addif_fn.assert_called_with(port["network_id"], p_const.TYPE_FLAT,
-+ "physnet1", None, port["id"])
-+
-+ self.lb_rpc.port_update("unused_context", port=port,
-+ network_type=p_const.TYPE_FLAT,
-+ segmentation_id=None,
-+ physical_network="physnet1")
-+ self.assertFalse(reffw_fn.called)
-+ addif_fn.assert_called_with(port["network_id"], p_const.TYPE_FLAT,
-+ "physnet1", None, port["id"])
-+
-+ self.lb_rpc.port_update("unused_context", port=port,
-+ vlan_id=lconst.LOCAL_VLAN_ID,
-+ physical_network=None)
-+ self.assertFalse(reffw_fn.called)
-+ addif_fn.assert_called_with(port["network_id"], p_const.TYPE_LOCAL,
-+ None, None, port["id"])
-+
-+ self.lb_rpc.port_update("unused_context", port=port,
-+ network_type=p_const.TYPE_LOCAL,
-+ segmentation_id=None,
-+ physical_network=None)
-+ self.assertFalse(reffw_fn.called)
-+ addif_fn.assert_called_with(port["network_id"], p_const.TYPE_LOCAL,
-+ None, None, port["id"])
-+
-+ addif_fn.return_value = True
-+ self.lb_rpc.port_update("unused_context", port=port,
-+ network_type=p_const.TYPE_LOCAL,
-+ segmentation_id=None,
-+ physical_network=None)
-+ rpc_obj.update_device_up.assert_called_with(
-+ self.lb_rpc.context,
-+ "tap123",
-+ self.lb_rpc.agent.agent_id,
-+ cfg.CONF.host
-+ )
-+
-+ addif_fn.return_value = False
-+ self.lb_rpc.port_update("unused_context", port=port,
-+ network_type=p_const.TYPE_LOCAL,
-+ segmentation_id=None,
-+ physical_network=None)
-+ rpc_obj.update_device_down.assert_called_with(
-+ self.lb_rpc.context,
-+ "tap123",
-+ self.lb_rpc.agent.agent_id,
-+ cfg.CONF.host
-+ )
-+
-+ port["admin_state_up"] = False
-+ port["security_groups"] = True
-+ getbr_fn.return_value = "br0"
-+ self.lb_rpc.port_update("unused_context", port=port,
-+ vlan_id="1", physical_network="physnet1")
-+ self.assertTrue(reffw_fn.called)
-+ remif_fn.assert_called_with("br0", "tap123")
-+ rpc_obj.update_device_down.assert_called_with(
-+ self.lb_rpc.context,
-+ "tap123",
-+ self.lb_rpc.agent.agent_id,
-+ cfg.CONF.host
-+ )
-+
-+ def test_port_update_plugin_rpc_failed(self):
-+ with contextlib.nested(
-+ mock.patch.object(self.lb_rpc.agent.br_mgr,
-+ "get_tap_device_name"),
-+ mock.patch.object(self.lb_rpc.agent.br_mgr,
-+ "udev_get_tap_devices"),
-+ mock.patch.object(self.lb_rpc.agent.br_mgr,
-+ "get_bridge_name"),
-+ mock.patch.object(self.lb_rpc.agent.br_mgr,
-+ "remove_interface"),
-+ mock.patch.object(self.lb_rpc.agent.br_mgr, "add_interface"),
-+ mock.patch.object(self.lb_rpc.sg_agent,
-+ "refresh_firewall", create=True),
-+ mock.patch.object(self.lb_rpc.agent,
-+ "plugin_rpc", create=True),
-+ mock.patch.object(linuxbridge_neutron_agent.LOG, 'error'),
-+ ) as (get_tap_fn, udev_fn, _, _, _, _, plugin_rpc, log):
-+ get_tap_fn.return_value = "tap123"
-+ udev_fn.return_value = ["tap123", "tap124"]
-+ port = {"admin_state_up": True,
-+ "id": "1234-5678",
-+ "network_id": "123-123"}
-+ plugin_rpc.update_device_up.side_effect = rpc_common.Timeout
-+ self.lb_rpc.port_update(mock.Mock(), port=port)
-+ self.assertTrue(plugin_rpc.update_device_up.called)
-+ self.assertEqual(log.call_count, 1)
-+
-+ log.reset_mock()
-+ port["admin_state_up"] = False
-+ plugin_rpc.update_device_down.side_effect = rpc_common.Timeout
-+ self.lb_rpc.port_update(mock.Mock(), port=port)
-+ self.assertTrue(plugin_rpc.update_device_down.called)
-+ self.assertEqual(log.call_count, 1)
-+
-+ 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', 'add', 'port_ip', 'lladdr',
-+ 'port_mac', 'dev', 'vxlan-1', 'nud', 'permanent'],
-+ root_helper=self.root_helper,
-+ check_exit_code=False),
-+ mock.call(['bridge', 'fdb', 'add', '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', 'add', '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)
Code Review / openstack-build / neutron-build.git / blobdiff