Add tests for LinuxBridge and OVS agents
Fixes bug 988067 Patch introduces some more unit tests for LinuxBridge and OVS agents. Change-Id: I95b8856e75b303520af654b08854c6bd0bc5d7ea
This commit is contained in:
Eugene Nikanorov
@@ -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
|
||||
)
|
||||
|
||||
@@ -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)
|
||||
|
||||
Reference in New Issue
Block a user