# under the License.
import contextlib
+import os
import mock
from oslo.config import cfg
import testtools
from quantum.agent.linux import ip_lib
+from quantum.agent.linux import utils
from quantum.plugins.linuxbridge.agent import linuxbridge_quantum_agent
from quantum.plugins.linuxbridge.common import constants as lconst
from quantum.tests import base
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_quantum_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_all_quantum_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_quantum_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_bridge_for_tap_device(self):
+ with contextlib.nested(
+ mock.patch.object(self.lbm, "get_all_quantum_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.assertEqual(self.lbm.get_bridge_for_tap_device("tap4"),
+ None)
+
+ 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 mock.patch.object(self.lbm, 'ensure_vlan') as ens_vl_fn:
+ ens_vl_fn.return_value = "eth0.1"
+ with mock.patch.object(self.lbm, 'ensure_bridge') as ens:
+ self.assertEqual(
+ self.lbm.ensure_vlan_bridge("123", "eth0", "1"),
+ "eth0.1"
+ )
+ ens.assert_called_once_with("brq123", "eth0.1")
+
+ 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_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')
+ ) as (de_fn, exec_fn, upd_fn, ie_fn):
+ de_fn.return_value = False
+ exec_fn.return_value = False
+ self.assertIsNone(self.lbm.ensure_bridge("br0", None))
+ ie_fn.return_Value = False
+ self.lbm.ensure_bridge("br0", "eth0")
+ self.assertTrue(upd_fn.called)
+ 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")
+
+ def test_ensure_physical_in_bridge(self):
+ self.assertFalse(
+ self.lbm.ensure_physical_in_bridge("123", "phys", "1")
+ )
+ with mock.patch.object(self.lbm, "ensure_flat_bridge") as flbr_fn:
+ self.assertTrue(
+ self.lbm.ensure_physical_in_bridge("123", "physnet1", "-1")
+ )
+ 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", "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", "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", "physnet1",
+ "-2", "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", "physnet1",
+ "-2", "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", "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", "physnet-1", "1", "234")
+ add_tap.assert_called_with("123", "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(utils, "execute")
+ ) as (de_fn, getif_fn, remif_fn, if_det_fn,
+ updif_fn, del_vlan, 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"]
+ 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")
+
+ 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"])
+ })
+
+
+class TestLinuxBridgeRpcCallbacks(base.BaseTestCase):
+ def setUp(self):
+ super(TestLinuxBridgeRpcCallbacks, self).setUp()
+
+ class FakeLBAgent(object):
+ def __init__(self):
+ self.agent_id = 1
+ self.br_mgr = (linuxbridge_quantum_agent.
+ LinuxBridgeManager({'physnet1': 'eth1'},
+ cfg.CONF.AGENT.root_helper))
+
+ self.lb_rpc = linuxbridge_quantum_agent.LinuxBridgeRpcCallbacks(
+ object(),
+ FakeLBAgent()
+ )
+
+ 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"],
+ "physnet1", "1", port["id"])
+
+ 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
+ )
# License for the specific language governing permissions and limitations
# under the License.
+import contextlib
+import sys
+
import mock
from oslo.config import cfg
import testtools
+from quantum.agent.linux import ip_lib
+from quantum.agent.linux import ovs_lib
from quantum.plugins.openvswitch.agent import ovs_quantum_agent
from quantum.tests import base
'quantum.openstack.common.rpc.impl_fake')
cfg.CONF.set_override('report_interval', 0, 'AGENT')
kwargs = ovs_quantum_agent.create_agent_config_map(cfg.CONF)
- with mock.patch('quantum.plugins.openvswitch.agent.ovs_quantum_agent.'
- 'OVSQuantumAgent.setup_integration_br',
- return_value=mock.Mock()):
- with mock.patch('quantum.agent.linux.utils.get_interface_mac',
- return_value='00:00:00:00:00:01'):
- self.agent = ovs_quantum_agent.OVSQuantumAgent(**kwargs)
+
+ with contextlib.nested(
+ mock.patch('quantum.plugins.openvswitch.agent.ovs_quantum_agent.'
+ 'OVSQuantumAgent.setup_integration_br',
+ return_value=mock.Mock()),
+ mock.patch('quantum.agent.linux.utils.get_interface_mac',
+ return_value='00:00:00:00:00:01')):
+ self.agent = ovs_quantum_agent.OVSQuantumAgent(**kwargs)
+ self.agent.tun_br = mock.Mock()
self.agent.sg_agent = mock.Mock()
- def mock_port_bound(self, ofport=None):
+ def _mock_port_bound(self, ofport=None):
port = mock.Mock()
port.ofport = ofport
net_uuid = 'my-net-uuid'
self.assertEqual(delete_flows_func.called, ofport != -1)
def test_port_bound_deletes_flows_for_valid_ofport(self):
- self.mock_port_bound(ofport=1)
+ self._mock_port_bound(ofport=1)
def test_port_bound_ignores_flows_for_invalid_ofport(self):
- self.mock_port_bound(ofport=-1)
+ self._mock_port_bound(ofport=-1)
def test_port_dead(self):
with mock.patch.object(self.agent.int_br,
side_effect=Exception()):
self.assertTrue(self.agent.treat_devices_added([{}]))
- def mock_treat_devices_added(self, details, port, func_name):
+ def _mock_treat_devices_added(self, details, port, func_name):
"""
:param details: the details to return for the device
:param func_name: the function that should be called
:returns: whether the named function was called
"""
- with mock.patch.object(self.agent.plugin_rpc, 'get_device_details',
- return_value=details):
- with mock.patch.object(self.agent.int_br, 'get_vif_port_by_id',
- return_value=port):
- with mock.patch.object(self.agent, func_name) as func:
- self.assertFalse(self.agent.treat_devices_added([{}]))
+ with contextlib.nested(
+ mock.patch.object(self.agent.plugin_rpc, 'get_device_details',
+ return_value=details),
+ mock.patch.object(self.agent.int_br, 'get_vif_port_by_id',
+ return_value=port),
+ mock.patch.object(self.agent, func_name)
+ ) as (get_dev_fn, get_vif_func, func):
+ self.assertFalse(self.agent.treat_devices_added([{}]))
return func.called
def test_treat_devices_added_ignores_invalid_ofport(self):
port = mock.Mock()
port.ofport = -1
- self.assertFalse(self.mock_treat_devices_added(mock.MagicMock(), port,
- 'port_dead'))
+ self.assertFalse(self._mock_treat_devices_added(mock.MagicMock(), port,
+ 'port_dead'))
def test_treat_devices_added_marks_unknown_port_as_dead(self):
port = mock.Mock()
port.ofport = 1
- self.assertTrue(self.mock_treat_devices_added(mock.MagicMock(), port,
- 'port_dead'))
+ self.assertTrue(self._mock_treat_devices_added(mock.MagicMock(), port,
+ 'port_dead'))
def test_treat_devices_added_updates_known_port(self):
details = mock.MagicMock()
details.__contains__.side_effect = lambda x: True
- self.assertTrue(self.mock_treat_devices_added(details,
- mock.Mock(),
- 'treat_vif_port'))
+ self.assertTrue(self._mock_treat_devices_added(details,
+ mock.Mock(),
+ 'treat_vif_port'))
def test_treat_devices_removed_returns_true_for_missing_device(self):
with mock.patch.object(self.agent.plugin_rpc, 'update_device_down',
side_effect=Exception()):
self.assertTrue(self.agent.treat_devices_removed([{}]))
- def mock_treat_devices_removed(self, port_exists):
+ def _mock_treat_devices_removed(self, port_exists):
details = dict(exists=port_exists)
with mock.patch.object(self.agent.plugin_rpc, 'update_device_down',
return_value=details):
self.assertEqual(port_unbound.called, not port_exists)
def test_treat_devices_removed_unbinds_port(self):
- self.mock_treat_devices_removed(False)
+ self._mock_treat_devices_removed(True)
def test_treat_devices_removed_ignores_missing_port(self):
- self.mock_treat_devices_removed(False)
-
- def test_port_update(self):
- port = {'id': 1,
- 'network_id': 1,
- 'admin_state_up': True}
- with mock.patch.object(self.agent.int_br, 'get_vif_port_by_id',
- return_value='2'):
- with mock.patch.object(self.agent.plugin_rpc,
- 'update_device_up') as device_up:
- with mock.patch.object(self.agent, 'port_bound') as port_bound:
- self.agent.port_update(mock.Mock(), port=port)
- self.assertTrue(port_bound.called)
- self.assertTrue(device_up.called)
- with mock.patch.object(self.agent.plugin_rpc,
- 'update_device_down') as device_down:
- with mock.patch.object(self.agent, 'port_dead') as port_dead:
- port['admin_state_up'] = False
- self.agent.port_update(mock.Mock(), port=port)
- self.assertTrue(port_dead.called)
- self.assertTrue(device_down.called)
+ self._mock_treat_devices_removed(False)
def test_process_network_ports(self):
reply = {'current': set(['tap0']),
self.assertFalse(self.agent.process_network_ports(reply))
self.assertTrue(device_added.called)
self.assertTrue(device_removed.called)
+
+ def test_report_state(self):
+ with contextlib.nested(
+ mock.patch.object(self.agent.int_br, "get_vif_port_set"),
+ mock.patch.object(self.agent.state_rpc, "report_state")
+ ) as (get_vif_fn, report_st):
+ get_vif_fn.return_value = ["vif123", "vif234"]
+ self.agent._report_state()
+ self.assertTrue(get_vif_fn.called)
+ report_st.assert_called_with(self.agent.context,
+ self.agent.agent_state)
+ self.assertNotIn("start_flag", self.agent.agent_state)
+ self.assertEqual(
+ self.agent.agent_state["configurations"]["devices"], 2
+ )
+
+ def test_network_delete(self):
+ with mock.patch.object(self.agent, "reclaim_local_vlan") as recl_fn:
+ self.agent.network_delete("unused_context",
+ network_id="123")
+ self.assertFalse(recl_fn.called)
+
+ self.agent.local_vlan_map["123"] = "LVM object"
+ self.agent.network_delete("unused_context",
+ network_id="123")
+ recl_fn.assert_called_with("123", self.agent.local_vlan_map["123"])
+
+ def test_port_update(self):
+ with contextlib.nested(
+ mock.patch.object(self.agent.int_br, "get_vif_port_by_id"),
+ mock.patch.object(self.agent, "treat_vif_port"),
+ mock.patch.object(self.agent.plugin_rpc, "update_device_up"),
+ mock.patch.object(self.agent.plugin_rpc, "update_device_down")
+ ) as (getvif_fn, treatvif_fn, updup_fn, upddown_fn):
+ port = {"id": "123",
+ "network_id": "124",
+ "admin_state_up": False}
+ getvif_fn.return_value = "vif_port_obj"
+ self.agent.port_update("unused_context",
+ port=port,
+ network_type="vlan",
+ segmentation_id="1",
+ physical_network="physnet")
+ treatvif_fn.assert_called_with("vif_port_obj", "123",
+ "124", "vlan", "physnet",
+ "1", False)
+ upddown_fn.assert_called_with(self.agent.context,
+ "123", self.agent.agent_id)
+
+ port["admin_state_up"] = True
+ self.agent.port_update("unused_context",
+ port=port,
+ network_type="vlan",
+ segmentation_id="1",
+ physical_network="physnet")
+ updup_fn.assert_called_with(self.agent.context,
+ "123", self.agent.agent_id)
+
+ def test_setup_physical_bridges(self):
+ with contextlib.nested(
+ mock.patch.object(ip_lib, "device_exists"),
+ mock.patch.object(sys, "exit"),
+ mock.patch.object(ovs_lib.OVSBridge, "remove_all_flows"),
+ mock.patch.object(ovs_lib.OVSBridge, "add_flow"),
+ mock.patch.object(ovs_lib.OVSBridge, "add_port"),
+ mock.patch.object(ovs_lib.OVSBridge, "delete_port"),
+ mock.patch.object(self.agent.int_br, "add_port"),
+ mock.patch.object(self.agent.int_br, "delete_port"),
+ mock.patch.object(ip_lib.IPWrapper, "add_veth"),
+ mock.patch.object(ip_lib.IpLinkCommand, "delete"),
+ mock.patch.object(ip_lib.IpLinkCommand, "set_up")
+ ) as (devex_fn, sysexit_fn, remflows_fn, ovs_addfl_fn,
+ ovs_addport_fn, ovs_delport_fn, br_addport_fn,
+ br_delport_fn, addveth_fn, linkdel_fn, linkset_fn):
+ devex_fn.return_value = True
+ addveth_fn.return_value = (ip_lib.IPDevice("int-br-eth1"),
+ ip_lib.IPDevice("phy-br-eth1"))
+ ovs_addport_fn.return_value = "int_ofport"
+ br_addport_fn.return_value = "phys_veth"
+ self.agent.setup_physical_bridges({"physnet1": "br-eth"})
+ self.assertEqual(self.agent.int_ofports["physnet1"],
+ "phys_veth")
+ self.assertEqual(self.agent.phys_ofports["physnet1"],
+ "int_ofport")
+
+ def test_port_unbound(self):
+ with contextlib.nested(
+ mock.patch.object(self.agent.tun_br, "delete_flows"),
+ mock.patch.object(self.agent, "reclaim_local_vlan")
+ ) as (delfl_fn, reclvl_fn):
+ self.agent.enable_tunneling = True
+ lvm = mock.Mock()
+ lvm.network_type = "gre"
+ lvm.vif_ports = {"vif1": mock.Mock()}
+ self.agent.local_vlan_map["netuid12345"] = lvm
+ self.agent.port_unbound("vif1", "netuid12345")
+ self.assertTrue(delfl_fn.called)
+ self.assertTrue(reclvl_fn.called)
+ reclvl_fn.called = False
+
+ lvm.vif_ports = {}
+ self.agent.port_unbound("vif1", "netuid12345")
+ self.assertEqual(reclvl_fn.call_count, 2)
+
+ lvm.vif_ports = {"vif1": mock.Mock()}
+ self.agent.port_unbound("vif3", "netuid12345")
+ self.assertEqual(reclvl_fn.call_count, 2)
Code Review / openstack-build / neutron-build.git / commitdiff