From: Eugene Nikanorov Date: Thu, 25 Apr 2013 04:19:58 +0000 (+0400) Subject: Add tests for LinuxBridge and OVS agents X-Git-Url: https://review.fuel-infra.org/gitweb?a=commitdiff_plain;h=d9c95e48becd7af36f71a821601389b8c97abe4b;p=openstack-build%2Fneutron-build.git Add tests for LinuxBridge and OVS agents Fixes bug 988067 Patch introduces some more unit tests for LinuxBridge and OVS agents. Change-Id: I95b8856e75b303520af654b08854c6bd0bc5d7ea --- diff --git a/quantum/tests/unit/linuxbridge/test_lb_quantum_agent.py b/quantum/tests/unit/linuxbridge/test_lb_quantum_agent.py index 92018e0f0..f2a6ebfb0 100644 --- a/quantum/tests/unit/linuxbridge/test_lb_quantum_agent.py +++ b/quantum/tests/unit/linuxbridge/test_lb_quantum_agent.py @@ -15,12 +15,14 @@ # 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 @@ -131,3 +133,393 @@ class TestLinuxBridgeAgent(base.BaseTestCase): 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 + ) diff --git a/quantum/tests/unit/openvswitch/test_ovs_quantum_agent.py b/quantum/tests/unit/openvswitch/test_ovs_quantum_agent.py index 88adadce4..ce6c146de 100644 --- a/quantum/tests/unit/openvswitch/test_ovs_quantum_agent.py +++ b/quantum/tests/unit/openvswitch/test_ovs_quantum_agent.py @@ -14,10 +14,15 @@ # 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 @@ -54,15 +59,18 @@ class TestOvsQuantumAgent(base.BaseTestCase): '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' @@ -72,10 +80,10 @@ class TestOvsQuantumAgent(base.BaseTestCase): 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, @@ -103,7 +111,7 @@ class TestOvsQuantumAgent(base.BaseTestCase): 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 @@ -111,39 +119,41 @@ class TestOvsQuantumAgent(base.BaseTestCase): :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): @@ -152,30 +162,10 @@ class TestOvsQuantumAgent(base.BaseTestCase): 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']), @@ -188,3 +178,110 @@ class TestOvsQuantumAgent(base.BaseTestCase): 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)