openstack/neutron
Code Issues Proposed changes

Merge "Implement external physical bridge mapping in linuxbridge"

Browse Source
This commit is contained in:
Jenkins 2015-09-05 17:04:45 +00:00 committed by Gerrit Code Review
commit ad2ea53388
5 changed files with 235 additions and 46 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
neutron/plugins/ml2/drivers/linuxbridge/agent/common/config.py
View File

@ -16,6 +16,7 @@ from oslo_config import cfg
from neutron.agent.common import config
DEFAULT_BRIDGE_MAPPINGS = []
DEFAULT_INTERFACE_MAPPINGS = []
DEFAULT_VXLAN_GROUP = '224.0.0.1'
@ -47,6 +48,9 @@ bridge_opts = [
cfg.ListOpt('physical_interface_mappings',
default=DEFAULT_INTERFACE_MAPPINGS,
help=_("List of <physical_network>:<physical_interface>")),
cfg.ListOpt('bridge_mappings',
default=DEFAULT_BRIDGE_MAPPINGS,
help=_("List of <physical_network>:<physical_bridge>")),
]
agent_opts = [

133
neutron/plugins/ml2/drivers/linuxbridge/agent/linuxbridge_neutron_agent.py
View File

@ -77,9 +77,11 @@ class NetworkSegment(object):
class LinuxBridgeManager(object):
def __init__(self, interface_mappings):
def __init__(self, bridge_mappings, interface_mappings):
self.bridge_mappings = bridge_mappings
self.interface_mappings = interface_mappings
self.validate_interface_mappings()
self.validate_bridge_mappings()
self.ip = ip_lib.IPWrapper()
# VXLAN related parameters:
self.local_ip = cfg.CONF.VXLAN.local_ip
@ -104,6 +106,14 @@ class LinuxBridgeManager(object):
{'intf': interface, 'net': physnet})
sys.exit(1)
def validate_bridge_mappings(self):
for physnet, bridge in self.bridge_mappings.items():
if not ip_lib.device_exists(bridge):
LOG.error(_LE("Bridge %(brq)s for physical network %(net)s"
" does not exist. Agent terminated!"),
{'brq': bridge, 'net': physnet})
sys.exit(1)
def interface_exists_on_bridge(self, bridge, interface):
directory = '/sys/class/net/%s/brif' % bridge
for filename in os.listdir(directory):
@ -111,6 +121,11 @@ class LinuxBridgeManager(object):
return True
return False
def get_existing_bridge_name(self, physical_network):
if not physical_network:
return None
return self.bridge_mappings.get(physical_network)
def get_bridge_name(self, network_id):
if not network_id:
LOG.warning(_LW("Invalid Network ID, will lead to incorrect "
@ -160,6 +175,11 @@ class LinuxBridgeManager(object):
for bridge in bridge_list:
if bridge.startswith(BRIDGE_NAME_PREFIX):
neutron_bridge_list.append(bridge)
# NOTE(nick-ma-z): Add pre-existing user-defined bridges
for bridge_name in self.bridge_mappings.values():
if bridge_name not in neutron_bridge_list:
neutron_bridge_list.append(bridge_name)
return neutron_bridge_list
def get_interfaces_on_bridge(self, bridge_name):
@ -197,13 +217,17 @@ class LinuxBridgeManager(object):
DEVICE_NAME_PLACEHOLDER, device_name)
return os.path.exists(bridge_port_path)
def ensure_vlan_bridge(self, network_id, physical_interface, vlan_id):
def ensure_vlan_bridge(self, network_id, phy_bridge_name,
physical_interface, vlan_id):
"""Create a vlan and bridge unless they already exist."""
interface = self.ensure_vlan(physical_interface, vlan_id)
bridge_name = self.get_bridge_name(network_id)
ips, gateway = self.get_interface_details(interface)
if self.ensure_bridge(bridge_name, interface, ips, gateway):
return interface
if phy_bridge_name:
return self.ensure_bridge(phy_bridge_name)
else:
bridge_name = self.get_bridge_name(network_id)
ips, gateway = self.get_interface_details(interface)
if self.ensure_bridge(bridge_name, interface, ips, gateway):
return interface
def ensure_vxlan_bridge(self, network_id, segmentation_id):
"""Create a vxlan and bridge unless they already exist."""
@ -225,16 +249,24 @@ class LinuxBridgeManager(object):
gateway = device.route.get_gateway(scope='global')
return ips, gateway
def ensure_flat_bridge(self, network_id, physical_interface):
def ensure_flat_bridge(self, network_id, phy_bridge_name,
physical_interface):
"""Create a non-vlan bridge unless it already exists."""
bridge_name = self.get_bridge_name(network_id)
ips, gateway = self.get_interface_details(physical_interface)
if self.ensure_bridge(bridge_name, physical_interface, ips, gateway):
return physical_interface
if phy_bridge_name:
return self.ensure_bridge(phy_bridge_name)
else:
bridge_name = self.get_bridge_name(network_id)
ips, gateway = self.get_interface_details(physical_interface)
if self.ensure_bridge(bridge_name, physical_interface, ips,
gateway):
return physical_interface
def ensure_local_bridge(self, network_id):
def ensure_local_bridge(self, network_id, phy_bridge_name):
"""Create a local bridge unless it already exists."""
bridge_name = self.get_bridge_name(network_id)
if phy_bridge_name:
bridge_name = phy_bridge_name
else:
bridge_name = self.get_bridge_name(network_id)
return self.ensure_bridge(bridge_name)
def ensure_vlan(self, physical_interface, vlan_id):
@ -389,15 +421,20 @@ class LinuxBridgeManager(object):
return
return self.ensure_vxlan_bridge(network_id, segmentation_id)
# NOTE(nick-ma-z): Obtain mappings of physical bridge and interfaces
physical_bridge = self.get_existing_bridge_name(physical_network)
physical_interface = self.interface_mappings.get(physical_network)
if not physical_interface:
LOG.error(_LE("No mapping for physical network %s"),
if not physical_bridge and not physical_interface:
LOG.error(_LE("No bridge or interface mappings"
" for physical network %s"),
physical_network)
return
if network_type == p_const.TYPE_FLAT:
return self.ensure_flat_bridge(network_id, physical_interface)
return self.ensure_flat_bridge(network_id, physical_bridge,
physical_interface)
elif network_type == p_const.TYPE_VLAN:
return self.ensure_vlan_bridge(network_id, physical_interface,
return self.ensure_vlan_bridge(network_id, physical_bridge,
physical_interface,
segmentation_id)
else:
LOG.error(_LE("Unknown network_type %(network_type)s for network "
@ -416,9 +453,13 @@ class LinuxBridgeManager(object):
"this host, skipped", tap_device_name)
return False
bridge_name = self.get_bridge_name(network_id)
if physical_network:
bridge_name = self.get_existing_bridge_name(physical_network)
else:
bridge_name = self.get_bridge_name(network_id)
if network_type == p_const.TYPE_LOCAL:
self.ensure_local_bridge(network_id)
self.ensure_local_bridge(network_id, bridge_name)
else:
phy_dev_name = self.ensure_physical_in_bridge(network_id,
network_type,
@ -495,6 +536,11 @@ class LinuxBridgeManager(object):
def remove_empty_bridges(self):
for network_id in list(self.network_map.keys()):
# NOTE(nick-ma-z): Don't remove pre-existing user-defined bridges
phy_net = self.network_map[network_id].physical_network
if phy_net and phy_net in self.bridge_mappings:
continue
bridge_name = self.get_bridge_name(network_id)
if not self.get_tap_devices_count(bridge_name):
self.delete_bridge(bridge_name)
@ -678,6 +724,19 @@ class LinuxBridgeRpcCallbacks(sg_rpc.SecurityGroupAgentRpcCallbackMixin,
def network_delete(self, context, **kwargs):
LOG.debug("network_delete received")
network_id = kwargs.get('network_id')
# NOTE(nick-ma-z): Don't remove pre-existing user-defined bridges
if network_id in self.agent.br_mgr.network_map:
phynet = self.agent.br_mgr.network_map[network_id].physical_network
if phynet and phynet in self.agent.br_mgr.bridge_mappings:
LOG.info(_LI("Physical network %s is defined in "
"bridge_mappings and cannot be deleted."),
network_id)
return
else:
LOG.error(_LE("Network %s is not available."), network_id)
return
bridge_name = self.agent.br_mgr.get_bridge_name(network_id)
LOG.debug("Delete %s", bridge_name)
self.agent.br_mgr.delete_bridge(bridge_name)
@ -773,10 +832,12 @@ class LinuxBridgeRpcCallbacks(sg_rpc.SecurityGroupAgentRpcCallbackMixin,
class LinuxBridgeNeutronAgentRPC(service.Service):
def __init__(self, interface_mappings, polling_interval,
def __init__(self, bridge_mappings, interface_mappings, polling_interval,
quitting_rpc_timeout):
"""Constructor.
:param bridge_mappings: dict mapping physical_networks to
physical_bridges.
:param interface_mappings: dict mapping physical_networks to
physical_interfaces.
:param polling_interval: interval (secs) to poll DB.
@ -785,13 +846,15 @@ class LinuxBridgeNeutronAgentRPC(service.Service):
"""
super(LinuxBridgeNeutronAgentRPC, self).__init__()
self.interface_mappings = interface_mappings
self.bridge_mappings = bridge_mappings
self.polling_interval = polling_interval
self.quitting_rpc_timeout = quitting_rpc_timeout
def start(self):
self.prevent_arp_spoofing = cfg.CONF.AGENT.prevent_arp_spoofing
self.setup_linux_bridge(self.interface_mappings)
configurations = {'interface_mappings': self.interface_mappings}
self.setup_linux_bridge(self.bridge_mappings, self.interface_mappings)
configurations = {'bridge_mappings': self.bridge_mappings,
'interface_mappings': self.interface_mappings}
if self.br_mgr.vxlan_mode != lconst.VXLAN_NONE:
configurations['tunneling_ip'] = self.br_mgr.local_ip
configurations['tunnel_types'] = [p_const.TYPE_VXLAN]
@ -869,11 +932,15 @@ class LinuxBridgeNeutronAgentRPC(service.Service):
self._report_state)
heartbeat.start(interval=report_interval)
def setup_linux_bridge(self, interface_mappings):
self.br_mgr = LinuxBridgeManager(interface_mappings)
def setup_linux_bridge(self, bridge_mappings, interface_mappings):
self.br_mgr = LinuxBridgeManager(bridge_mappings, interface_mappings)
def remove_port_binding(self, network_id, interface_id):
bridge_name = self.br_mgr.get_bridge_name(network_id)
def remove_port_binding(self, network_id, physical_network, interface_id):
if physical_network:
bridge_name = self.br_mgr.get_existing_bridge_name(
physical_network)
else:
bridge_name = self.br_mgr.get_bridge_name(network_id)
tap_device_name = self.br_mgr.get_tap_device_name(interface_id)
return self.br_mgr.remove_interface(bridge_name, tap_device_name)
@ -941,7 +1008,9 @@ class LinuxBridgeNeutronAgentRPC(service.Service):
self.agent_id,
cfg.CONF.host)
else:
physical_network = device_details['physical_network']
self.remove_port_binding(device_details['network_id'],
physical_network,
device_details['port_id'])
else:
LOG.info(_LI("Device %s not defined on plugin"), device)
@ -1073,9 +1142,19 @@ def main():
sys.exit(1)
LOG.info(_LI("Interface mappings: %s"), interface_mappings)
try:
bridge_mappings = n_utils.parse_mappings(
cfg.CONF.LINUX_BRIDGE.bridge_mappings)
except ValueError as e:
LOG.error(_LE("Parsing bridge_mappings failed: %s. "
"Agent terminated!"), e)
sys.exit(1)
LOG.info(_LI("Bridge mappings: %s"), bridge_mappings)
polling_interval = cfg.CONF.AGENT.polling_interval
quitting_rpc_timeout = cfg.CONF.AGENT.quitting_rpc_timeout
agent = LinuxBridgeNeutronAgentRPC(interface_mappings,
agent = LinuxBridgeNeutronAgentRPC(bridge_mappings,
interface_mappings,
polling_interval,
quitting_rpc_timeout)
LOG.info(_LI("Agent initialized successfully, now running... "))

4
neutron/plugins/ml2/drivers/linuxbridge/mech_driver/mech_linuxbridge.py
View File

@ -47,7 +47,9 @@ class LinuxbridgeMechanismDriver(mech_agent.SimpleAgentMechanismDriverBase):
p_constants.TYPE_VLAN])
def get_mappings(self, agent):
return agent['configurations'].get('interface_mappings', {})
mappings = dict(agent['configurations'].get('interface_mappings', {}),
**agent['configurations'].get('bridge_mappings', {}))
return mappings
def check_vlan_transparency(self, context):
"""Linuxbridge driver vlan transparency support."""

15
neutron/tests/functional/agent/test_l2_lb_agent.py
View File

@ -35,13 +35,22 @@ class LinuxBridgeAgentTests(test_ip_lib.IpLibTestFramework):
def test_validate_interface_mappings(self):
mappings = {'physnet1': 'int1', 'physnet2': 'int2'}
with testtools.ExpectedException(SystemExit):
lba.LinuxBridgeManager(mappings)
lba.LinuxBridgeManager({}, mappings)
self.manage_device(
self.generate_device_details()._replace(namespace=None,
name='int1'))
with testtools.ExpectedException(SystemExit):
lba.LinuxBridgeManager(mappings)
lba.LinuxBridgeManager({}, mappings)
self.manage_device(
self.generate_device_details()._replace(namespace=None,
name='int2'))
lba.LinuxBridgeManager(mappings)
lba.LinuxBridgeManager({}, mappings)
def test_validate_bridge_mappings(self):
mappings = {'physnet1': 'br-eth1'}
with testtools.ExpectedException(SystemExit):
lba.LinuxBridgeManager(mappings, {})
self.manage_device(
self.generate_device_details()._replace(namespace=None,
name='br-eth1'))
lba.LinuxBridgeManager(mappings, {})

125
neutron/tests/unit/plugins/ml2/drivers/linuxbridge/agent/test_linuxbridge_neutron_agent.py
View File

@ -31,6 +31,8 @@ from neutron.tests import base
LOCAL_IP = '192.168.0.33'
DEVICE_1 = 'tapabcdef01-12'
BRIDGE_MAPPINGS = {'physnet0': 'br-eth2'}
INTERFACE_MAPPINGS = {'physnet1': 'eth1'}
class FakeIpLinkCommand(object):
@ -47,14 +49,15 @@ class TestLinuxBridge(base.BaseTestCase):
def setUp(self):
super(TestLinuxBridge, self).setUp()
interface_mappings = {'physnet1': 'eth1'}
interface_mappings = INTERFACE_MAPPINGS
bridge_mappings = BRIDGE_MAPPINGS
with mock.patch.object(ip_lib.IPWrapper,
'get_device_by_ip', return_value=None),\
mock.patch.object(ip_lib, 'device_exists',
return_value=True):
self.linux_bridge = linuxbridge_neutron_agent.LinuxBridgeManager(
interface_mappings)
bridge_mappings, interface_mappings)
def test_ensure_physical_in_bridge_invalid(self):
result = self.linux_bridge.ensure_physical_in_bridge('network_id',
@ -107,7 +110,7 @@ class TestLinuxBridgeAgent(base.BaseTestCase):
with mock.patch.object(ip_lib.IPWrapper,
'get_device_by_ip', return_value=None):
self.agent = linuxbridge_neutron_agent.LinuxBridgeNeutronAgentRPC(
{}, 0, cfg.CONF.AGENT.quitting_rpc_timeout)
{}, {}, 0, cfg.CONF.AGENT.quitting_rpc_timeout)
with mock.patch.object(self.agent, "daemon_loop"):
self.agent.start()
@ -333,7 +336,9 @@ class TestLinuxBridgeAgent(base.BaseTestCase):
resync_needed = agent.treat_devices_added_updated(set(['tap1']))
self.assertFalse(resync_needed)
agent.remove_port_binding.assert_called_with('net123', 'port123')
agent.remove_port_binding.assert_called_with('net123',
'physnet1',
'port123')
self.assertFalse(agent.plugin_rpc.update_device_up.called)
def test_set_rpc_timeout(self):
@ -354,14 +359,15 @@ class TestLinuxBridgeAgent(base.BaseTestCase):
class TestLinuxBridgeManager(base.BaseTestCase):
def setUp(self):
super(TestLinuxBridgeManager, self).setUp()
self.interface_mappings = {'physnet1': 'eth1'}
self.interface_mappings = INTERFACE_MAPPINGS
self.bridge_mappings = BRIDGE_MAPPINGS
with mock.patch.object(ip_lib.IPWrapper,
'get_device_by_ip', return_value=None),\
mock.patch.object(ip_lib, 'device_exists',
return_value=True):
self.lbm = linuxbridge_neutron_agent.LinuxBridgeManager(
self.interface_mappings)
self.bridge_mappings, self.interface_mappings)
def test_interface_exists_on_bridge(self):
with mock.patch.object(os, 'listdir') as listdir_fn:
@ -373,6 +379,15 @@ class TestLinuxBridgeManager(base.BaseTestCase):
self.lbm.interface_exists_on_bridge("br-int", "abd")
)
def test_get_existing_bridge_name(self):
phy_net = 'physnet0'
self.assertEqual('br-eth2',
self.lbm.get_existing_bridge_name(phy_net))
phy_net = ''
self.assertEqual(None,
self.lbm.get_existing_bridge_name(phy_net))
def test_get_bridge_name(self):
nw_id = "123456789101112"
self.assertEqual(self.lbm.get_bridge_name(nw_id),
@ -416,10 +431,13 @@ class TestLinuxBridgeManager(base.BaseTestCase):
def test_get_all_neutron_bridges(self):
br_list = ["br-int", "brq1", "brq2", "br-ex"]
result = br_list[1:3]
result.append('br-eth2')
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])
result)
self.assertTrue(listdir_fn.called)
def test_get_interfaces_on_bridge(self):
@ -493,7 +511,7 @@ class TestLinuxBridgeManager(base.BaseTestCase):
list_fn.return_value = ipdict
with mock.patch.object(self.lbm, 'ensure_bridge') as ens:
self.assertEqual(
self.lbm.ensure_flat_bridge("123", "eth0"),
self.lbm.ensure_flat_bridge("123", None, "eth0"),
"eth0"
)
self.assertTrue(list_fn.called)
@ -501,6 +519,15 @@ class TestLinuxBridgeManager(base.BaseTestCase):
ens.assert_called_once_with("brq123", "eth0",
ipdict, gwdict)
def test_ensure_flat_bridge_with_existed_brq(self):
with mock.patch.object(self.lbm, 'ensure_bridge') as ens:
ens.return_value = "br-eth2"
self.assertEqual("br-eth2",
self.lbm.ensure_flat_bridge("123",
"br-eth2",
None))
ens.assert_called_with("br-eth2")
def test_ensure_vlan_bridge(self):
with mock.patch.object(self.lbm, 'ensure_vlan') as ens_vl_fn,\
mock.patch.object(self.lbm, 'ensure_bridge') as ens,\
@ -508,20 +535,44 @@ class TestLinuxBridgeManager(base.BaseTestCase):
'get_interface_details') as 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"),
self.assertEqual(self.lbm.ensure_vlan_bridge("123",
None,
"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"),
self.assertEqual(self.lbm.ensure_vlan_bridge("123",
None,
"eth0",
"1"),
"eth0.1")
ens.assert_called_with("brq123", "eth0.1", "ips", "gateway")
def test_ensure_vlan_bridge_with_existed_brq(self):
with mock.patch.object(self.lbm, 'ensure_vlan') as ens_vl_fn,\
mock.patch.object(self.lbm, 'ensure_bridge') as ens:
ens_vl_fn.return_value = None
ens.return_value = "br-eth2"
self.assertEqual("br-eth2",
self.lbm.ensure_vlan_bridge("123",
"br-eth2",
None,
None))
ens.assert_called_with("br-eth2")
def test_ensure_local_bridge(self):
with mock.patch.object(self.lbm, 'ensure_bridge') as ens_fn:
self.lbm.ensure_local_bridge("54321")
self.lbm.ensure_local_bridge("54321", None)
ens_fn.assert_called_once_with("brq54321")
def test_ensure_local_bridge_with_existed_brq(self):
with mock.patch.object(self.lbm, 'ensure_bridge') as ens_fn:
ens_fn.return_value = "br-eth2"
self.lbm.ensure_local_bridge("54321", 'br-eth2')
ens_fn.assert_called_once_with("br-eth2")
def test_ensure_vlan(self):
with mock.patch.object(ip_lib, 'device_exists') as de_fn:
de_fn.return_value = True
@ -661,6 +712,12 @@ class TestLinuxBridgeManager(base.BaseTestCase):
)
self.assertTrue(vlbr_fn.called)
def test_ensure_physical_in_bridge_with_existed_brq(self):
with mock.patch.object(linuxbridge_neutron_agent.LOG, 'error') as log:
self.lbm.ensure_physical_in_bridge("123", p_const.TYPE_FLAT,
"physnet9", "1")
self.assertEqual(1, log.call_count)
def test_add_tap_interface(self):
with mock.patch.object(ip_lib, "device_exists") as de_fn:
de_fn.return_value = False
@ -682,7 +739,7 @@ class TestLinuxBridgeManager(base.BaseTestCase):
p_const.TYPE_LOCAL,
"physnet1", None,
"tap1"))
en_fn.assert_called_with("123")
en_fn.assert_called_with("123", None)
get_br.return_value = False
bridge_device.addif.retun_value = True
@ -784,11 +841,12 @@ class TestLinuxBridgeManager(base.BaseTestCase):
self.assertFalse(updif_fn.called)
def test_delete_bridge_no_int_mappings(self):
bridge_mappings = {}
interface_mappings = {}
with mock.patch.object(ip_lib.IPWrapper,
'get_device_by_ip', return_value=None):
lbm = linuxbridge_neutron_agent.LinuxBridgeManager(
interface_mappings)
bridge_mappings, interface_mappings)
bridge_device = mock.Mock()
with mock.patch.object(ip_lib, "device_exists") as de_fn,\
@ -838,6 +896,23 @@ class TestLinuxBridgeManager(base.BaseTestCase):
self.lbm.remove_empty_bridges()
del_br_fn.assert_called_once_with('brqnet1')
def test_remove_empty_bridges_with_existed_brq(self):
phy_net = mock.Mock()
phy_net.physical_network = 'physnet0'
self.lbm.network_map = {'net1': mock.Mock(),
'net2': mock.Mock(),
'net3': phy_net}
def tap_count_side_effect(*args):
return 0
with mock.patch.object(self.lbm, "delete_bridge") as del_br_fn,\
mock.patch.object(self.lbm,
"get_tap_devices_count",
side_effect=tap_count_side_effect):
self.lbm.remove_empty_bridges()
self.assertEqual(2, del_br_fn.call_count)
def test_remove_interface(self):
bridge_device = mock.Mock()
with mock.patch.object(ip_lib, "device_exists") as de_fn,\
@ -975,8 +1050,10 @@ class TestLinuxBridgeRpcCallbacks(base.BaseTestCase):
'get_device_by_ip', return_value=None),\
mock.patch.object(ip_lib, 'device_exists',
return_value=True):
self.br_mgr = (linuxbridge_neutron_agent.
LinuxBridgeManager({'physnet1': 'eth1'}))
self.br_mgr = (
linuxbridge_neutron_agent.LinuxBridgeManager(
BRIDGE_MAPPINGS,
INTERFACE_MAPPINGS))
self.br_mgr.vxlan_mode = lconst.VXLAN_UCAST
segment = mock.Mock()
@ -991,6 +1068,11 @@ class TestLinuxBridgeRpcCallbacks(base.BaseTestCase):
)
def test_network_delete(self):
mock_net = mock.Mock()
mock_net.physical_network = None
self.lb_rpc.agent.br_mgr.network_map = {'123': mock_net}
with mock.patch.object(self.lb_rpc.agent.br_mgr,
"get_bridge_name") as get_br_fn,\
mock.patch.object(self.lb_rpc.agent.br_mgr,
@ -1000,6 +1082,19 @@ class TestLinuxBridgeRpcCallbacks(base.BaseTestCase):
get_br_fn.assert_called_with("123")
del_fn.assert_called_with("br0")
def test_network_delete_with_existed_brq(self):
mock_net = mock.Mock()
mock_net.physical_network = 'physnet0'
self.lb_rpc.agent.br_mgr.network_map = {'123': mock_net}
with mock.patch.object(linuxbridge_neutron_agent.LOG, 'info') as log,\
mock.patch.object(self.lb_rpc.agent.br_mgr,
"delete_bridge") as del_fn:
self.lb_rpc.network_delete("anycontext", network_id="123")
self.assertEqual(0, del_fn.call_count)
self.assertEqual(1, log.call_count)
def test_fdb_add(self):
fdb_entries = {'net_id':
{'ports':