Merge "Replace mox in unit tests with mock"
This commit is contained in:
commit
a361a2ab60
47
neutron/tests/tools.py
Normal file
47
neutron/tests/tools.py
Normal file
@ -0,0 +1,47 @@
|
||||
# Copyright (c) 2013 NEC Corporation
|
||||
# All Rights Reserved.
|
||||
#
|
||||
# Licensed under the Apache License, Version 2.0 (the "License"); you may
|
||||
# not use this file except in compliance with the License. You may obtain
|
||||
# a copy of the License at
|
||||
#
|
||||
# http://www.apache.org/licenses/LICENSE-2.0
|
||||
#
|
||||
# Unless required by applicable law or agreed to in writing, software
|
||||
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
|
||||
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
|
||||
# License for the specific language governing permissions and limitations
|
||||
# under the License.
|
||||
#
|
||||
# @author: Akihiro Motoki, NEC Corporation
|
||||
|
||||
|
||||
"""setup_mock_calls and verify_mock_calls are convenient methods
|
||||
to setup a sequence of mock calls.
|
||||
|
||||
expected_calls_and_values is a list of (expected_call, return_value):
|
||||
|
||||
expected_calls_and_values = [
|
||||
(mock.call(["ovs-vsctl", self.TO, '--', "--may-exist", "add-port",
|
||||
self.BR_NAME, pname], root_helper=self.root_helper),
|
||||
None),
|
||||
(mock.call(["ovs-vsctl", self.TO, "set", "Interface",
|
||||
pname, "type=gre"], root_helper=self.root_helper),
|
||||
None),
|
||||
....
|
||||
]
|
||||
|
||||
* expected_call should be mock.call(expected_arg, ....)
|
||||
* return_value is passed to side_effect of a mocked call.
|
||||
A return value or an exception can be specified.
|
||||
"""
|
||||
|
||||
|
||||
def setup_mock_calls(mocked_call, expected_calls_and_values):
|
||||
return_values = [call[1] for call in expected_calls_and_values]
|
||||
mocked_call.side_effect = return_values
|
||||
|
||||
|
||||
def verify_mock_calls(mocked_call, expected_calls_and_values):
|
||||
expected_calls = [call[0] for call in expected_calls_and_values]
|
||||
mocked_call.assert_has_calls(expected_calls)
|
@ -18,9 +18,7 @@
|
||||
import os
|
||||
|
||||
import mock
|
||||
import mox
|
||||
from oslo.config import cfg
|
||||
import stubout
|
||||
import testtools
|
||||
|
||||
from neutron import context
|
||||
@ -78,11 +76,10 @@ class MetaNeutronPluginV2Test(base.BaseTestCase):
|
||||
self.context = context.get_admin_context()
|
||||
|
||||
db.configure_db()
|
||||
self.addCleanup(db.clear_db)
|
||||
|
||||
setup_metaplugin_conf()
|
||||
|
||||
self.mox = mox.Mox()
|
||||
self.stubs = stubout.StubOutForTesting()
|
||||
self.client_cls_p = mock.patch('neutronclient.v2_0.client.Client')
|
||||
client_cls = self.client_cls_p.start()
|
||||
self.client_inst = mock.Mock()
|
||||
@ -304,11 +301,3 @@ class MetaNeutronPluginV2Test(base.BaseTestCase):
|
||||
self.fail("AttributeError Error is not raised")
|
||||
|
||||
self.fail("No Error is not raised")
|
||||
|
||||
def tearDown(self):
|
||||
self.mox.UnsetStubs()
|
||||
self.stubs.UnsetAll()
|
||||
self.stubs.SmartUnsetAll()
|
||||
self.mox.VerifyAll()
|
||||
db.clear_db()
|
||||
super(MetaNeutronPluginV2Test, self).tearDown()
|
||||
|
@ -18,10 +18,9 @@
|
||||
import random
|
||||
import string
|
||||
|
||||
import mox
|
||||
import mock
|
||||
import netaddr
|
||||
|
||||
from neutron import context
|
||||
from neutron.openstack.common import uuidutils
|
||||
from neutron.plugins.nec.common import ofc_client as ofc
|
||||
from neutron.plugins.nec.db import api as ndb
|
||||
@ -50,10 +49,10 @@ class PFCDriverTestBase(base.BaseTestCase):
|
||||
|
||||
def setUp(self):
|
||||
super(PFCDriverTestBase, self).setUp()
|
||||
self.mox = mox.Mox()
|
||||
self.driver = drivers.get_driver(self.driver)(TestConfig)
|
||||
self.mox.StubOutWithMock(ofc.OFCClient, 'do_request')
|
||||
self.addCleanup(self.mox.UnsetStubs)
|
||||
self.do_request = mock.patch.object(ofc.OFCClient,
|
||||
'do_request').start()
|
||||
self.addCleanup(mock.patch.stopall)
|
||||
|
||||
def get_ofc_item_random_params(self):
|
||||
"""create random parameters for ofc_item test."""
|
||||
@ -85,14 +84,12 @@ class PFCDriverTestBase(base.BaseTestCase):
|
||||
body['description'] = ofc_description
|
||||
if post_id:
|
||||
body['id'] = ofc_t
|
||||
ofc.OFCClient.do_request("POST", path, body=body)
|
||||
self.do_request.return_value = None
|
||||
else:
|
||||
ofc.OFCClient.do_request("POST", path, body=body).\
|
||||
AndReturn({'id': ofc_t})
|
||||
self.mox.ReplayAll()
|
||||
self.do_request.return_value = {'id': ofc_t}
|
||||
|
||||
ret = self.driver.create_tenant(description, t)
|
||||
self.mox.VerifyAll()
|
||||
self.do_request.assert_called_once_with("POST", path, body=body)
|
||||
self.assertEqual(ret, tenant_path)
|
||||
|
||||
def testa_create_tenant(self):
|
||||
@ -104,11 +101,9 @@ class PFCDriverTestBase(base.BaseTestCase):
|
||||
t, n, p = self.get_ofc_item_random_params()
|
||||
|
||||
path = "/tenants/%s" % _ofc(t)
|
||||
ofc.OFCClient.do_request("DELETE", path)
|
||||
self.mox.ReplayAll()
|
||||
|
||||
self.driver.delete_tenant(path)
|
||||
self.mox.VerifyAll()
|
||||
self.do_request.assert_called_once_with("DELETE", path)
|
||||
|
||||
def testd_create_network(self):
|
||||
t, n, p = self.get_ofc_item_random_params()
|
||||
@ -119,12 +114,10 @@ class PFCDriverTestBase(base.BaseTestCase):
|
||||
post_path = "%s/networks" % tenant_path
|
||||
body = {'description': ofc_description}
|
||||
network = {'id': _ofc(n)}
|
||||
ofc.OFCClient.do_request("POST", post_path, body=body).\
|
||||
AndReturn(network)
|
||||
self.mox.ReplayAll()
|
||||
self.do_request.return_value = network
|
||||
|
||||
ret = self.driver.create_network(tenant_path, description, n)
|
||||
self.mox.VerifyAll()
|
||||
self.do_request.assert_called_once_with("POST", post_path, body=body)
|
||||
net_path = "/tenants/%s/networks/%s" % (_ofc(t), _ofc(n))
|
||||
self.assertEqual(ret, net_path)
|
||||
|
||||
@ -132,11 +125,9 @@ class PFCDriverTestBase(base.BaseTestCase):
|
||||
t, n, p = self.get_ofc_item_random_params()
|
||||
|
||||
net_path = "/tenants/%s/networks/%s" % (_ofc(t), _ofc(n))
|
||||
ofc.OFCClient.do_request("DELETE", net_path)
|
||||
self.mox.ReplayAll()
|
||||
|
||||
self.driver.delete_network(net_path)
|
||||
self.mox.VerifyAll()
|
||||
self.do_request.assert_called_once_with("DELETE", net_path)
|
||||
|
||||
def testg_create_port(self):
|
||||
t, n, p = self.get_ofc_item_random_params()
|
||||
@ -149,11 +140,10 @@ class PFCDriverTestBase(base.BaseTestCase):
|
||||
'port': str(p.port_no),
|
||||
'vid': str(p.vlan_id)}
|
||||
port = {'id': _ofc(p.id)}
|
||||
ofc.OFCClient.do_request("POST", post_path, body=body).AndReturn(port)
|
||||
self.mox.ReplayAll()
|
||||
self.do_request.return_value = port
|
||||
|
||||
ret = self.driver.create_port(net_path, p, p.id)
|
||||
self.mox.VerifyAll()
|
||||
self.do_request.assert_called_once_with("POST", post_path, body=body)
|
||||
self.assertEqual(ret, port_path)
|
||||
|
||||
def testh_delete_port(self):
|
||||
@ -161,11 +151,9 @@ class PFCDriverTestBase(base.BaseTestCase):
|
||||
|
||||
port_path = "/tenants/%s/networks/%s/ports/%s" % (_ofc(t), _ofc(n),
|
||||
_ofc(p.id))
|
||||
ofc.OFCClient.do_request("DELETE", port_path)
|
||||
self.mox.ReplayAll()
|
||||
|
||||
self.driver.delete_port(port_path)
|
||||
self.mox.VerifyAll()
|
||||
self.do_request.assert_called_once_with("DELETE", port_path)
|
||||
|
||||
def test_filter_supported(self):
|
||||
self.assertFalse(self.driver.filter_supported())
|
||||
@ -180,23 +168,16 @@ class PFCV3DriverTest(PFCDriverTestBase):
|
||||
|
||||
def testa_create_tenant(self):
|
||||
t, n, p = self.get_ofc_item_random_params()
|
||||
self.mox.ReplayAll()
|
||||
|
||||
ret = self.driver.create_tenant('dummy_desc', t)
|
||||
self.mox.VerifyAll()
|
||||
|
||||
self.assertEqual(0, self.do_request.call_count)
|
||||
ofc_t_path = "/tenants/" + self._generate_ofc_tenant_id(t)
|
||||
self.assertEqual(ofc_t_path, ret)
|
||||
|
||||
def testc_delete_tenant(self):
|
||||
t, n, p = self.get_ofc_item_random_params()
|
||||
|
||||
path = "/tenants/%s" % _ofc(t)
|
||||
# There is no API call.
|
||||
self.mox.ReplayAll()
|
||||
|
||||
self.driver.delete_tenant(path)
|
||||
self.mox.VerifyAll()
|
||||
self.assertEqual(0, self.do_request.call_count)
|
||||
|
||||
|
||||
class PFCV4DriverTest(PFCDriverTestBase):
|
||||
@ -214,12 +195,10 @@ class PFCV5DriverTest(PFCDriverTestBase):
|
||||
tenant_path = "/tenants/%s" % _ofc(t)
|
||||
post_path = "%s/routers" % tenant_path
|
||||
router = {'id': _ofc(r)}
|
||||
ofc.OFCClient.do_request("POST", post_path,
|
||||
body=None).AndReturn(router)
|
||||
self.mox.ReplayAll()
|
||||
self.do_request.return_value = router
|
||||
|
||||
ret = self.driver.create_router(tenant_path, description, r)
|
||||
self.mox.VerifyAll()
|
||||
self.do_request.assert_called_once_with("POST", post_path, body=None)
|
||||
router_path = "/tenants/%s/routers/%s" % (_ofc(t), _ofc(r))
|
||||
self.assertEqual(ret, router_path)
|
||||
|
||||
@ -228,11 +207,9 @@ class PFCV5DriverTest(PFCDriverTestBase):
|
||||
r = uuidutils.generate_uuid()
|
||||
|
||||
router_path = "/tenants/%s/routers/%s" % (_ofc(t), _ofc(r))
|
||||
ofc.OFCClient.do_request("DELETE", router_path)
|
||||
self.mox.ReplayAll()
|
||||
|
||||
self.driver.delete_router(router_path)
|
||||
self.mox.VerifyAll()
|
||||
self.do_request.assert_called_once_with("DELETE", router_path)
|
||||
|
||||
def test_add_router_interface(self):
|
||||
t = uuidutils.generate_uuid()
|
||||
@ -249,13 +226,12 @@ class PFCV5DriverTest(PFCDriverTestBase):
|
||||
'ip_address': ip_address,
|
||||
'mac_address': mac_address}
|
||||
inf = {'id': _ofc(p)}
|
||||
ofc.OFCClient.do_request("POST", infs_path,
|
||||
body=body).AndReturn(inf)
|
||||
self.mox.ReplayAll()
|
||||
self.do_request.return_value = inf
|
||||
|
||||
ret = self.driver.add_router_interface(router_path, net_path,
|
||||
ip_address, mac_address)
|
||||
self.mox.VerifyAll()
|
||||
self.do_request.assert_called_once_with("POST", infs_path, body=body)
|
||||
|
||||
inf_path = "%s/interfaces/%s" % (router_path, _ofc(p))
|
||||
self.assertEqual(ret, inf_path)
|
||||
|
||||
@ -269,22 +245,16 @@ class PFCV5DriverTest(PFCDriverTestBase):
|
||||
ip_address = '10.1.1.1/24'
|
||||
mac_address = '11:22:33:44:55:66'
|
||||
|
||||
body = {'ip_address': ip_address,
|
||||
'mac_address': mac_address}
|
||||
ofc.OFCClient.do_request("PUT", inf_path, body=body)
|
||||
|
||||
body = {'ip_address': ip_address}
|
||||
ofc.OFCClient.do_request("PUT", inf_path, body=body)
|
||||
|
||||
body = {'mac_address': mac_address}
|
||||
ofc.OFCClient.do_request("PUT", inf_path, body=body)
|
||||
|
||||
self.mox.ReplayAll()
|
||||
|
||||
self.driver.update_router_interface(inf_path, ip_address, mac_address)
|
||||
self.driver.update_router_interface(inf_path, ip_address=ip_address)
|
||||
self.driver.update_router_interface(inf_path, mac_address=mac_address)
|
||||
self.mox.VerifyAll()
|
||||
|
||||
self.do_request.assert_has_calls([
|
||||
mock.call("PUT", inf_path, body={'ip_address': ip_address,
|
||||
'mac_address': mac_address}),
|
||||
mock.call("PUT", inf_path, body={'ip_address': ip_address}),
|
||||
mock.call("PUT", inf_path, body={'mac_address': mac_address}),
|
||||
])
|
||||
|
||||
def test_delete_router_interface(self):
|
||||
t = uuidutils.generate_uuid()
|
||||
@ -293,11 +263,9 @@ class PFCV5DriverTest(PFCDriverTestBase):
|
||||
|
||||
router_path = "/tenants/%s/routers/%s" % (_ofc(t), _ofc(r))
|
||||
inf_path = "%s/interfaces/%s" % (router_path, _ofc(p))
|
||||
ofc.OFCClient.do_request("DELETE", inf_path)
|
||||
self.mox.ReplayAll()
|
||||
|
||||
self.driver.delete_router_interface(inf_path)
|
||||
self.mox.VerifyAll()
|
||||
self.do_request.assert_called_once_with("DELETE", inf_path)
|
||||
|
||||
def _get_route_id(self, dest, nexthop):
|
||||
dest = netaddr.IPNetwork(dest)
|
||||
@ -313,13 +281,11 @@ class PFCV5DriverTest(PFCDriverTestBase):
|
||||
nexthop = '192.168.100.10'
|
||||
body = {'destination': dest, 'nexthop': nexthop}
|
||||
route_id = self._get_route_id(dest, nexthop)
|
||||
ofc.OFCClient.do_request("POST", routes_path,
|
||||
body=body).AndReturn({'id': route_id})
|
||||
self.mox.ReplayAll()
|
||||
self.do_request.return_value = {'id': route_id}
|
||||
|
||||
ret = self.driver.add_router_route(router_path, '10.1.1.0/24',
|
||||
'192.168.100.10')
|
||||
self.mox.VerifyAll()
|
||||
self.do_request.assert_called_once_with("POST", routes_path, body=body)
|
||||
route_path = routes_path + '/' + route_id
|
||||
self.assertEqual(ret, route_path)
|
||||
|
||||
@ -332,11 +298,9 @@ class PFCV5DriverTest(PFCDriverTestBase):
|
||||
|
||||
route_id = self._get_route_id('10.1.1.0/24', '192.168.100.10')
|
||||
route_path = routes_path + '/' + route_id
|
||||
ofc.OFCClient.do_request("DELETE", route_path)
|
||||
self.mox.ReplayAll()
|
||||
|
||||
self.driver.delete_router_route(route_path)
|
||||
self.mox.VerifyAll()
|
||||
self.do_request.assert_called_once_with("DELETE", route_path)
|
||||
|
||||
def test_list_router_routes(self):
|
||||
t = uuidutils.generate_uuid()
|
||||
@ -350,11 +314,10 @@ class PFCV5DriverTest(PFCDriverTestBase):
|
||||
data = {'routes': [{'id': self._get_route_id(route[0], route[1]),
|
||||
'destination': route[0], 'nexthop': route[1]}
|
||||
for route in routes]}
|
||||
ofc.OFCClient.do_request("GET", routes_path).AndReturn(data)
|
||||
self.mox.ReplayAll()
|
||||
self.do_request.return_value = data
|
||||
|
||||
ret = self.driver.list_router_routes(router_path)
|
||||
self.mox.VerifyAll()
|
||||
self.do_request.assert_called_once_with("GET", routes_path)
|
||||
|
||||
expected = [{'id': (routes_path + "/" +
|
||||
self._get_route_id(route[0], route[1])),
|
||||
@ -412,12 +375,12 @@ class PFCIdConvertTest(base.BaseTestCase):
|
||||
|
||||
def setUp(self):
|
||||
super(PFCIdConvertTest, self).setUp()
|
||||
self.mox = mox.Mox()
|
||||
self.driver = drivers.get_driver(self.driver)(TestConfig)
|
||||
self.ctx = self.mox.CreateMock(context.Context)
|
||||
self.ctx = mock.Mock()
|
||||
self.ctx.session = "session"
|
||||
self.mox.StubOutWithMock(ndb, 'get_ofc_id_lookup_both')
|
||||
self.addCleanup(self.mox.UnsetStubs)
|
||||
self.get_ofc_id_lookup_both = mock.patch.object(
|
||||
ndb, 'get_ofc_id_lookup_both').start()
|
||||
self.addCleanup(mock.patch.stopall)
|
||||
|
||||
def generate_random_ids(self, count=1):
|
||||
if count == 1:
|
||||
@ -437,26 +400,24 @@ class PFCIdConvertTest(base.BaseTestCase):
|
||||
|
||||
def test_convert_network_id(self):
|
||||
t_id, ofc_t_id, ofc_n_id = self.generate_random_ids(3)
|
||||
ndb.get_ofc_id_lookup_both(
|
||||
self.ctx.session, 'ofc_tenant', t_id).AndReturn(ofc_t_id)
|
||||
self.mox.ReplayAll()
|
||||
self.get_ofc_id_lookup_both.return_value = ofc_t_id
|
||||
|
||||
ret = self.driver.convert_ofc_network_id(self.ctx, ofc_n_id, t_id)
|
||||
self.assertEqual(ret, ('/tenants/%(tenant)s/networks/%(network)s' %
|
||||
{'tenant': ofc_t_id, 'network': ofc_n_id}))
|
||||
self.mox.VerifyAll()
|
||||
self.get_ofc_id_lookup_both.assert_called_once_with(
|
||||
self.ctx.session, 'ofc_tenant', t_id)
|
||||
|
||||
def test_convert_network_id_with_new_tenant_id(self):
|
||||
t_id, ofc_t_id, ofc_n_id = self.generate_random_ids(3)
|
||||
ofc_t_path = '/tenants/%s' % ofc_t_id
|
||||
ndb.get_ofc_id_lookup_both(
|
||||
self.ctx.session, 'ofc_tenant', t_id).AndReturn(ofc_t_path)
|
||||
self.mox.ReplayAll()
|
||||
self.get_ofc_id_lookup_both.return_value = ofc_t_path
|
||||
|
||||
ret = self.driver.convert_ofc_network_id(self.ctx, ofc_n_id, t_id)
|
||||
self.assertEqual(ret, ('/tenants/%(tenant)s/networks/%(network)s' %
|
||||
{'tenant': ofc_t_id, 'network': ofc_n_id}))
|
||||
self.mox.VerifyAll()
|
||||
self.get_ofc_id_lookup_both.assert_called_once_with(
|
||||
self.ctx.session, 'ofc_tenant', t_id)
|
||||
|
||||
def test_convert_network_id_noconv(self):
|
||||
t_id = 'dummy'
|
||||
@ -470,34 +431,32 @@ class PFCIdConvertTest(base.BaseTestCase):
|
||||
t_id, n_id = self.generate_random_ids(2)
|
||||
ofc_t_id, ofc_n_id, ofc_p_id = self.generate_random_ids(3)
|
||||
|
||||
ndb.get_ofc_id_lookup_both(
|
||||
self.ctx.session, 'ofc_network', n_id).AndReturn(ofc_n_id)
|
||||
ndb.get_ofc_id_lookup_both(
|
||||
self.ctx.session, 'ofc_tenant', t_id).AndReturn(ofc_t_id)
|
||||
self.mox.ReplayAll()
|
||||
self.get_ofc_id_lookup_both.side_effect = [ofc_n_id, ofc_t_id]
|
||||
|
||||
ret = self.driver.convert_ofc_port_id(self.ctx, ofc_p_id, t_id, n_id)
|
||||
exp = ('/tenants/%(tenant)s/networks/%(network)s/ports/%(port)s' %
|
||||
{'tenant': ofc_t_id, 'network': ofc_n_id, 'port': ofc_p_id})
|
||||
self.assertEqual(ret, exp)
|
||||
self.mox.VerifyAll()
|
||||
self.get_ofc_id_lookup_both.assert_has_calls([
|
||||
mock.call(self.ctx.session, 'ofc_network', n_id),
|
||||
mock.call(self.ctx.session, 'ofc_tenant', t_id),
|
||||
])
|
||||
|
||||
def test_convert_port_id_with_new_tenant_id(self):
|
||||
t_id, n_id = self.generate_random_ids(2)
|
||||
ofc_t_id, ofc_n_id, ofc_p_id = self.generate_random_ids(3)
|
||||
|
||||
ofc_t_path = '/tenants/%s' % ofc_t_id
|
||||
ndb.get_ofc_id_lookup_both(
|
||||
self.ctx.session, 'ofc_network', n_id).AndReturn(ofc_n_id)
|
||||
ndb.get_ofc_id_lookup_both(
|
||||
self.ctx.session, 'ofc_tenant', t_id).AndReturn(ofc_t_path)
|
||||
self.mox.ReplayAll()
|
||||
self.get_ofc_id_lookup_both.side_effect = [ofc_n_id, ofc_t_path]
|
||||
|
||||
ret = self.driver.convert_ofc_port_id(self.ctx, ofc_p_id, t_id, n_id)
|
||||
exp = ('/tenants/%(tenant)s/networks/%(network)s/ports/%(port)s' %
|
||||
{'tenant': ofc_t_id, 'network': ofc_n_id, 'port': ofc_p_id})
|
||||
self.assertEqual(ret, exp)
|
||||
self.mox.VerifyAll()
|
||||
self.get_ofc_id_lookup_both.assert_has_calls([
|
||||
mock.call(self.ctx.session, 'ofc_network', n_id),
|
||||
mock.call(self.ctx.session, 'ofc_tenant', t_id),
|
||||
])
|
||||
|
||||
def test_convert_port_id_with_new_network_id(self):
|
||||
t_id, n_id = self.generate_random_ids(2)
|
||||
@ -505,15 +464,14 @@ class PFCIdConvertTest(base.BaseTestCase):
|
||||
|
||||
ofc_n_path = ('/tenants/%(tenant)s/networks/%(network)s' %
|
||||
{'tenant': ofc_t_id, 'network': ofc_n_id})
|
||||
ndb.get_ofc_id_lookup_both(
|
||||
self.ctx.session, 'ofc_network', n_id).AndReturn(ofc_n_path)
|
||||
self.mox.ReplayAll()
|
||||
self.get_ofc_id_lookup_both.return_value = ofc_n_path
|
||||
|
||||
ret = self.driver.convert_ofc_port_id(self.ctx, ofc_p_id, t_id, n_id)
|
||||
exp = ('/tenants/%(tenant)s/networks/%(network)s/ports/%(port)s' %
|
||||
{'tenant': ofc_t_id, 'network': ofc_n_id, 'port': ofc_p_id})
|
||||
self.assertEqual(ret, exp)
|
||||
self.mox.VerifyAll()
|
||||
self.get_ofc_id_lookup_both.assert_called_once_with(
|
||||
self.ctx.session, 'ofc_network', n_id)
|
||||
|
||||
def test_convert_port_id_noconv(self):
|
||||
t_id = n_id = 'dummy'
|
||||
|
@ -17,9 +17,8 @@
|
||||
|
||||
import random
|
||||
|
||||
import mox
|
||||
import mock
|
||||
|
||||
from neutron import context
|
||||
from neutron.openstack.common import uuidutils
|
||||
from neutron.plugins.nec.common import ofc_client
|
||||
from neutron.plugins.nec.db import api as ndb
|
||||
@ -40,10 +39,10 @@ class TremaDriverTestBase(base.BaseTestCase):
|
||||
|
||||
def setUp(self):
|
||||
super(TremaDriverTestBase, self).setUp()
|
||||
self.mox = mox.Mox()
|
||||
self.driver = drivers.get_driver(self.driver_name)(TestConfig)
|
||||
self.mox.StubOutWithMock(ofc_client.OFCClient, 'do_request')
|
||||
self.addCleanup(self.mox.UnsetStubs)
|
||||
self.do_request = mock.patch.object(ofc_client.OFCClient,
|
||||
'do_request').start()
|
||||
self.addCleanup(mock.patch.stopall)
|
||||
|
||||
def get_ofc_item_random_params(self):
|
||||
"""create random parameters for ofc_item test."""
|
||||
@ -61,50 +60,39 @@ class TremaDriverNetworkTestBase(TremaDriverTestBase):
|
||||
|
||||
def test_create_tenant(self):
|
||||
t, n, p = self.get_ofc_item_random_params()
|
||||
# There is no API call.
|
||||
self.mox.ReplayAll()
|
||||
ret = self.driver.create_tenant('dummy_desc', t)
|
||||
self.mox.VerifyAll()
|
||||
ofc_t_path = "/tenants/%s" % t
|
||||
self.assertEqual(ofc_t_path, ret)
|
||||
# There is no API call.
|
||||
self.assertEqual(0, self.do_request.call_count)
|
||||
|
||||
def test_update_tenant(self):
|
||||
t, n, p = self.get_ofc_item_random_params()
|
||||
path = "/tenants/%s" % t
|
||||
# There is no API call.
|
||||
self.mox.ReplayAll()
|
||||
self.driver.update_tenant(path, 'dummy_desc')
|
||||
self.mox.VerifyAll()
|
||||
# There is no API call.
|
||||
self.assertEqual(0, self.do_request.call_count)
|
||||
|
||||
def testc_delete_tenant(self):
|
||||
t, n, p = self.get_ofc_item_random_params()
|
||||
path = "/tenants/%s" % t
|
||||
# There is no API call.
|
||||
self.mox.ReplayAll()
|
||||
self.driver.delete_tenant(path)
|
||||
self.mox.VerifyAll()
|
||||
# There is no API call.
|
||||
self.assertEqual(0, self.do_request.call_count)
|
||||
|
||||
def testa_create_network(self):
|
||||
t, n, p = self.get_ofc_item_random_params()
|
||||
description = "desc of %s" % n
|
||||
|
||||
body = {'id': n, 'description': description}
|
||||
ofc_client.OFCClient.do_request("POST", "/networks", body=body)
|
||||
self.mox.ReplayAll()
|
||||
|
||||
ret = self.driver.create_network(t, description, n)
|
||||
self.mox.VerifyAll()
|
||||
self.do_request.assert_called_once_with("POST", "/networks", body=body)
|
||||
self.assertEqual(ret, '/networks/%s' % n)
|
||||
|
||||
def testc_delete_network(self):
|
||||
t, n, p = self.get_ofc_item_random_params()
|
||||
|
||||
net_path = "/networks/%s" % n
|
||||
ofc_client.OFCClient.do_request("DELETE", net_path)
|
||||
self.mox.ReplayAll()
|
||||
|
||||
self.driver.delete_network(net_path)
|
||||
self.mox.VerifyAll()
|
||||
self.do_request.assert_called_once_with("DELETE", net_path)
|
||||
|
||||
|
||||
class TremaPortBaseDriverTest(TremaDriverNetworkTestBase):
|
||||
@ -116,29 +104,21 @@ class TremaPortBaseDriverTest(TremaDriverNetworkTestBase):
|
||||
|
||||
def testd_create_port(self):
|
||||
_t, n, p = self.get_ofc_item_random_params()
|
||||
|
||||
net_path = "/networks/%s" % n
|
||||
body = {'id': p.id,
|
||||
'datapath_id': p.datapath_id,
|
||||
'port': str(p.port_no),
|
||||
'vid': str(p.vlan_id)}
|
||||
ofc_client.OFCClient.do_request("POST",
|
||||
"/networks/%s/ports" % n, body=body)
|
||||
self.mox.ReplayAll()
|
||||
|
||||
ret = self.driver.create_port(net_path, p, p.id)
|
||||
self.mox.VerifyAll()
|
||||
self.do_request.assert_called_once_with(
|
||||
"POST", "/networks/%s/ports" % n, body=body)
|
||||
self.assertEqual(ret, '/networks/%s/ports/%s' % (n, p.id))
|
||||
|
||||
def testd_delete_port(self):
|
||||
t, n, p = self.get_ofc_item_random_params()
|
||||
|
||||
p_path = "/networks/%s/ports/%s" % (n, p.id)
|
||||
ofc_client.OFCClient.do_request("DELETE", p_path)
|
||||
self.mox.ReplayAll()
|
||||
|
||||
self.driver.delete_port(p_path)
|
||||
self.mox.VerifyAll()
|
||||
self.do_request.assert_called_once_with("DELETE", p_path)
|
||||
|
||||
|
||||
class TremaPortMACBaseDriverTest(TremaDriverNetworkTestBase):
|
||||
@ -158,16 +138,16 @@ class TremaPortMACBaseDriverTest(TremaDriverNetworkTestBase):
|
||||
'datapath_id': p.datapath_id,
|
||||
'port': str(p.port_no),
|
||||
'vid': str(p.vlan_id)}
|
||||
ofc_client.OFCClient.do_request("POST", path_1, body=body_1)
|
||||
path_2 = "/networks/%s/ports/%s/attachments" % (n, dummy_port)
|
||||
body_2 = {'id': p.id, 'mac': p.mac}
|
||||
ofc_client.OFCClient.do_request("POST", path_2, body=body_2)
|
||||
path_3 = "/networks/%s/ports/%s" % (n, dummy_port)
|
||||
ofc_client.OFCClient.do_request("DELETE", path_3)
|
||||
self.mox.ReplayAll()
|
||||
|
||||
ret = self.driver.create_port(net_path, p, p.id)
|
||||
self.mox.VerifyAll()
|
||||
|
||||
self.do_request.assert_has_calls([
|
||||
mock.call("POST", path_1, body=body_1),
|
||||
mock.call("POST", path_2, body=body_2),
|
||||
mock.call("DELETE", path_3)
|
||||
])
|
||||
port_path = "/networks/%s/ports/%s/attachments/%s" % (n, dummy_port,
|
||||
p.id)
|
||||
self.assertEqual(ret, port_path)
|
||||
@ -175,13 +155,9 @@ class TremaPortMACBaseDriverTest(TremaDriverNetworkTestBase):
|
||||
def testd_delete_port(self):
|
||||
t, n, p = self.get_ofc_item_random_params()
|
||||
dummy_port = "dummy-%s" % p.id
|
||||
|
||||
path = "/networks/%s/ports/%s/attachments/%s" % (n, dummy_port, p.id)
|
||||
ofc_client.OFCClient.do_request("DELETE", path)
|
||||
self.mox.ReplayAll()
|
||||
|
||||
self.driver.delete_port(path)
|
||||
self.mox.VerifyAll()
|
||||
self.do_request.assert_called_once_with("DELETE", path)
|
||||
|
||||
|
||||
class TremaMACBaseDriverTest(TremaDriverNetworkTestBase):
|
||||
@ -193,26 +169,18 @@ class TremaMACBaseDriverTest(TremaDriverNetworkTestBase):
|
||||
|
||||
def testd_create_port(self):
|
||||
t, n, p = self.get_ofc_item_random_params()
|
||||
|
||||
net_path = "/networks/%s" % n
|
||||
path = "/networks/%s/attachments" % n
|
||||
body = {'id': p.id, 'mac': p.mac}
|
||||
ofc_client.OFCClient.do_request("POST", path, body=body)
|
||||
self.mox.ReplayAll()
|
||||
|
||||
ret = self.driver.create_port(net_path, p, p.id)
|
||||
self.mox.VerifyAll()
|
||||
self.do_request.assert_called_once_with("POST", path, body=body)
|
||||
self.assertEqual(ret, '/networks/%s/attachments/%s' % (n, p.id))
|
||||
|
||||
def testd_delete_port(self):
|
||||
t, n, p = self.get_ofc_item_random_params()
|
||||
|
||||
path = "/networks/%s/attachments/%s" % (n, p.id)
|
||||
ofc_client.OFCClient.do_request("DELETE", path)
|
||||
self.mox.ReplayAll()
|
||||
|
||||
self.driver.delete_port(path)
|
||||
self.mox.VerifyAll()
|
||||
self.do_request.assert_called_once_with("DELETE", path)
|
||||
|
||||
|
||||
class TremaFilterDriverTest(TremaDriverTestBase):
|
||||
@ -280,11 +248,8 @@ class TremaFilterDriverTest(TremaDriverTestBase):
|
||||
if non_ofp_wildcards:
|
||||
body['wildcards'] = ','.join(non_ofp_wildcards)
|
||||
|
||||
ofc_client.OFCClient.do_request("POST", "/filters", body=body)
|
||||
self.mox.ReplayAll()
|
||||
|
||||
ret = self.driver.create_filter(net_path, f, p, f['id'])
|
||||
self.mox.VerifyAll()
|
||||
self.do_request.assert_called_once_with("POST", "/filters", body=body)
|
||||
self.assertEqual(ret, '/filters/%s' % f['id'])
|
||||
|
||||
def test_create_filter_accept(self):
|
||||
@ -368,13 +333,9 @@ class TremaFilterDriverTest(TremaDriverTestBase):
|
||||
|
||||
def testb_delete_filter(self):
|
||||
t, n, p = self.get_ofc_item_random_params()
|
||||
|
||||
f_path = "/filters/%s" % uuidutils.generate_uuid()
|
||||
ofc_client.OFCClient.do_request("DELETE", f_path)
|
||||
self.mox.ReplayAll()
|
||||
|
||||
self.driver.delete_filter(f_path)
|
||||
self.mox.VerifyAll()
|
||||
self.do_request.assert_called_once_with("DELETE", f_path)
|
||||
|
||||
|
||||
def generate_random_ids(count=1):
|
||||
@ -390,9 +351,7 @@ class TremaIdConvertTest(base.BaseTestCase):
|
||||
def setUp(self):
|
||||
super(TremaIdConvertTest, self).setUp()
|
||||
self.driver = drivers.get_driver(self.driver_name)(TestConfig)
|
||||
self.mox = mox.Mox()
|
||||
self.ctx = self.mox.CreateMock(context.Context)
|
||||
self.addCleanup(self.mox.UnsetStubs)
|
||||
self.ctx = mock.Mock()
|
||||
|
||||
def test_convert_tenant_id(self):
|
||||
ofc_t_id = generate_random_ids(1)
|
||||
@ -430,40 +389,38 @@ class TremaIdConvertTest(base.BaseTestCase):
|
||||
class TremaIdConvertTestBase(base.BaseTestCase):
|
||||
def setUp(self):
|
||||
super(TremaIdConvertTestBase, self).setUp()
|
||||
self.mox = mox.Mox()
|
||||
self.driver = drivers.get_driver(self.driver_name)(TestConfig)
|
||||
self.ctx = self.mox.CreateMock(context.Context)
|
||||
self.ctx = mock.Mock()
|
||||
self.ctx.session = "session"
|
||||
self.mox.StubOutWithMock(ndb, 'get_ofc_id_lookup_both')
|
||||
self.addCleanup(self.mox.UnsetStubs)
|
||||
self.get_ofc_id_lookup_both = mock.patch.object(
|
||||
ndb, 'get_ofc_id_lookup_both').start()
|
||||
self.addCleanup(mock.patch.stopall)
|
||||
|
||||
def _test_convert_port_id(self, port_path_template):
|
||||
t_id, n_id = generate_random_ids(2)
|
||||
ofc_n_id, ofc_p_id = generate_random_ids(2)
|
||||
|
||||
ndb.get_ofc_id_lookup_both(
|
||||
self.ctx.session, 'ofc_network', n_id).AndReturn(ofc_n_id)
|
||||
self.mox.ReplayAll()
|
||||
self.get_ofc_id_lookup_both.return_value = ofc_n_id
|
||||
|
||||
ret = self.driver.convert_ofc_port_id(self.ctx, ofc_p_id, t_id, n_id)
|
||||
exp = port_path_template % {'network': ofc_n_id, 'port': ofc_p_id}
|
||||
self.assertEqual(ret, exp)
|
||||
self.mox.VerifyAll()
|
||||
self.get_ofc_id_lookup_both.assert_called_once_with(
|
||||
self.ctx.session, 'ofc_network', n_id)
|
||||
|
||||
def _test_convert_port_id_with_new_network_id(self, port_path_template):
|
||||
t_id, n_id = generate_random_ids(2)
|
||||
ofc_n_id, ofc_p_id = generate_random_ids(2)
|
||||
|
||||
ofc_n_path = '/networks/%s' % ofc_n_id
|
||||
ndb.get_ofc_id_lookup_both(
|
||||
self.ctx.session, 'ofc_network', n_id).AndReturn(ofc_n_path)
|
||||
self.mox.ReplayAll()
|
||||
self.get_ofc_id_lookup_both.return_value = ofc_n_path
|
||||
|
||||
ret = self.driver.convert_ofc_port_id(self.ctx, ofc_p_id, t_id, n_id)
|
||||
exp = port_path_template % {'network': ofc_n_id, 'port': ofc_p_id}
|
||||
|
||||
self.assertEqual(ret, exp)
|
||||
self.mox.VerifyAll()
|
||||
self.get_ofc_id_lookup_both.assert_called_once_with(
|
||||
self.ctx.session, 'ofc_network', n_id)
|
||||
|
||||
def _test_convert_port_id_noconv(self, port_path_template):
|
||||
t_id = n_id = 'dummy'
|
||||
|
@ -16,7 +16,6 @@
|
||||
# @author: Dan Wendlandt, Nicira, Inc.
|
||||
|
||||
import mock
|
||||
import mox
|
||||
import testtools
|
||||
|
||||
from neutron.agent.linux import ovs_lib
|
||||
@ -24,6 +23,7 @@ from neutron.agent.linux import utils
|
||||
from neutron.openstack.common import jsonutils
|
||||
from neutron.openstack.common import uuidutils
|
||||
from neutron.tests import base
|
||||
from neutron.tests import tools
|
||||
|
||||
|
||||
class TestBaseOVS(base.BaseTestCase):
|
||||
@ -111,15 +111,13 @@ class OVS_Lib_Test(base.BaseTestCase):
|
||||
self.BR_NAME = "br-int"
|
||||
self.TO = "--timeout=2"
|
||||
|
||||
self.mox = mox.Mox()
|
||||
self.root_helper = 'sudo'
|
||||
self.br = ovs_lib.OVSBridge(self.BR_NAME, self.root_helper)
|
||||
self.mox.StubOutWithMock(utils, "execute")
|
||||
self.addCleanup(self.mox.UnsetStubs)
|
||||
self.execute = mock.patch.object(utils, "execute").start()
|
||||
self.addCleanup(mock.patch.stopall)
|
||||
|
||||
def test_vifport(self):
|
||||
"""Create and stringify vif port, confirm no exceptions."""
|
||||
self.mox.ReplayAll()
|
||||
|
||||
pname = "vif1.0"
|
||||
ofport = 5
|
||||
@ -137,78 +135,35 @@ class OVS_Lib_Test(base.BaseTestCase):
|
||||
# test __str__
|
||||
str(port)
|
||||
|
||||
self.mox.VerifyAll()
|
||||
|
||||
def test_create(self):
|
||||
self.br.add_bridge(self.BR_NAME)
|
||||
self.mox.ReplayAll()
|
||||
|
||||
self.br.create()
|
||||
self.mox.VerifyAll()
|
||||
|
||||
def test_destroy(self):
|
||||
self.br.delete_bridge(self.BR_NAME)
|
||||
self.mox.ReplayAll()
|
||||
|
||||
self.br.destroy()
|
||||
self.mox.VerifyAll()
|
||||
|
||||
def test_reset_bridge(self):
|
||||
self.br.destroy()
|
||||
self.br.create()
|
||||
self.mox.ReplayAll()
|
||||
|
||||
self.br.reset_bridge()
|
||||
self.mox.VerifyAll()
|
||||
|
||||
def test_delete_port(self):
|
||||
pname = "tap5"
|
||||
utils.execute(["ovs-vsctl", self.TO, "--", "--if-exists",
|
||||
self.br.delete_port(pname)
|
||||
self.execute.assert_called_once_with(
|
||||
["ovs-vsctl", self.TO, "--", "--if-exists",
|
||||
"del-port", self.BR_NAME, pname],
|
||||
root_helper=self.root_helper)
|
||||
|
||||
self.mox.ReplayAll()
|
||||
self.br.delete_port(pname)
|
||||
self.mox.VerifyAll()
|
||||
|
||||
def test_add_flow(self):
|
||||
ofport = "99"
|
||||
vid = 4000
|
||||
lsw_id = 18
|
||||
cidr = '192.168.1.0/24'
|
||||
utils.execute(["ovs-ofctl", "add-flow", self.BR_NAME,
|
||||
"hard_timeout=0,idle_timeout=0,"
|
||||
"priority=2,dl_src=ca:fe:de:ad:be:ef"
|
||||
",actions=strip_vlan,output:0"],
|
||||
root_helper=self.root_helper)
|
||||
utils.execute(["ovs-ofctl", "add-flow", self.BR_NAME,
|
||||
"hard_timeout=0,idle_timeout=0,"
|
||||
"priority=1,actions=normal"],
|
||||
root_helper=self.root_helper)
|
||||
utils.execute(["ovs-ofctl", "add-flow", self.BR_NAME,
|
||||
"hard_timeout=0,idle_timeout=0,"
|
||||
"priority=2,actions=drop"],
|
||||
root_helper=self.root_helper)
|
||||
utils.execute(["ovs-ofctl", "add-flow", self.BR_NAME,
|
||||
"hard_timeout=0,idle_timeout=0,"
|
||||
"priority=2,in_port=%s,actions=drop" % ofport],
|
||||
root_helper=self.root_helper)
|
||||
utils.execute(["ovs-ofctl", "add-flow", self.BR_NAME,
|
||||
"hard_timeout=0,idle_timeout=0,"
|
||||
"priority=4,in_port=%s,dl_vlan=%s,"
|
||||
"actions=strip_vlan,set_tunnel:%s,normal"
|
||||
% (ofport, vid, lsw_id)],
|
||||
root_helper=self.root_helper)
|
||||
utils.execute(["ovs-ofctl", "add-flow", self.BR_NAME,
|
||||
"hard_timeout=0,idle_timeout=0,"
|
||||
"priority=3,tun_id=%s,actions="
|
||||
"mod_vlan_vid:%s,output:%s"
|
||||
% (lsw_id, vid, ofport)], root_helper=self.root_helper)
|
||||
utils.execute(["ovs-ofctl", "add-flow", self.BR_NAME,
|
||||
"hard_timeout=0,idle_timeout=0,"
|
||||
"priority=4,arp,nw_src=%s,actions=drop" % cidr],
|
||||
root_helper=self.root_helper)
|
||||
self.mox.ReplayAll()
|
||||
|
||||
self.br.add_flow(priority=2, dl_src="ca:fe:de:ad:be:ef",
|
||||
actions="strip_vlan,output:0")
|
||||
@ -223,71 +178,98 @@ class OVS_Lib_Test(base.BaseTestCase):
|
||||
actions="mod_vlan_vid:%s,output:%s" %
|
||||
(vid, ofport))
|
||||
self.br.add_flow(priority=4, proto='arp', nw_src=cidr, actions='drop')
|
||||
self.mox.VerifyAll()
|
||||
expected_calls = [
|
||||
mock.call(["ovs-ofctl", "add-flow", self.BR_NAME,
|
||||
"hard_timeout=0,idle_timeout=0,"
|
||||
"priority=2,dl_src=ca:fe:de:ad:be:ef"
|
||||
",actions=strip_vlan,output:0"],
|
||||
process_input=None, root_helper=self.root_helper),
|
||||
mock.call(["ovs-ofctl", "add-flow", self.BR_NAME,
|
||||
"hard_timeout=0,idle_timeout=0,"
|
||||
"priority=1,actions=normal"],
|
||||
process_input=None, root_helper=self.root_helper),
|
||||
mock.call(["ovs-ofctl", "add-flow", self.BR_NAME,
|
||||
"hard_timeout=0,idle_timeout=0,"
|
||||
"priority=2,actions=drop"],
|
||||
process_input=None, root_helper=self.root_helper),
|
||||
mock.call(["ovs-ofctl", "add-flow", self.BR_NAME,
|
||||
"hard_timeout=0,idle_timeout=0,"
|
||||
"priority=2,in_port=%s,actions=drop" % ofport],
|
||||
process_input=None, root_helper=self.root_helper),
|
||||
mock.call(["ovs-ofctl", "add-flow", self.BR_NAME,
|
||||
"hard_timeout=0,idle_timeout=0,"
|
||||
"priority=4,in_port=%s,dl_vlan=%s,"
|
||||
"actions=strip_vlan,set_tunnel:%s,normal"
|
||||
% (ofport, vid, lsw_id)],
|
||||
process_input=None, root_helper=self.root_helper),
|
||||
mock.call(["ovs-ofctl", "add-flow", self.BR_NAME,
|
||||
"hard_timeout=0,idle_timeout=0,"
|
||||
"priority=3,tun_id=%s,actions="
|
||||
"mod_vlan_vid:%s,output:%s"
|
||||
% (lsw_id, vid, ofport)],
|
||||
process_input=None, root_helper=self.root_helper),
|
||||
mock.call(["ovs-ofctl", "add-flow", self.BR_NAME,
|
||||
"hard_timeout=0,idle_timeout=0,"
|
||||
"priority=4,arp,nw_src=%s,actions=drop" % cidr],
|
||||
process_input=None, root_helper=self.root_helper),
|
||||
]
|
||||
self.execute.assert_has_calls(expected_calls)
|
||||
|
||||
def test_get_port_ofport(self):
|
||||
pname = "tap99"
|
||||
ofport = "6"
|
||||
utils.execute(["ovs-vsctl", self.TO, "get",
|
||||
"Interface", pname, "ofport"],
|
||||
root_helper=self.root_helper).AndReturn(ofport)
|
||||
self.mox.ReplayAll()
|
||||
|
||||
self.execute.return_value = ofport
|
||||
self.assertEqual(self.br.get_port_ofport(pname), ofport)
|
||||
self.mox.VerifyAll()
|
||||
self.execute.assert_called_once_with(
|
||||
["ovs-vsctl", self.TO, "get", "Interface", pname, "ofport"],
|
||||
root_helper=self.root_helper)
|
||||
|
||||
def test_get_datapath_id(self):
|
||||
datapath_id = '"0000b67f4fbcc149"'
|
||||
utils.execute(["ovs-vsctl", self.TO, "get",
|
||||
"Bridge", self.BR_NAME, "datapath_id"],
|
||||
root_helper=self.root_helper).AndReturn(datapath_id)
|
||||
self.mox.ReplayAll()
|
||||
|
||||
self.execute.return_value = datapath_id
|
||||
self.assertEqual(self.br.get_datapath_id(), datapath_id.strip('"'))
|
||||
self.mox.VerifyAll()
|
||||
self.execute.assert_called_once_with(
|
||||
["ovs-vsctl", self.TO, "get",
|
||||
"Bridge", self.BR_NAME, "datapath_id"],
|
||||
root_helper=self.root_helper)
|
||||
|
||||
def test_count_flows(self):
|
||||
utils.execute(["ovs-ofctl", "dump-flows", self.BR_NAME],
|
||||
root_helper=self.root_helper,
|
||||
process_input=None).AndReturn('ignore\nflow-1\n')
|
||||
self.mox.ReplayAll()
|
||||
|
||||
self.execute.return_value = 'ignore\nflow-1\n'
|
||||
# counts the number of flows as total lines of output - 2
|
||||
self.assertEqual(self.br.count_flows(), 1)
|
||||
self.mox.VerifyAll()
|
||||
self.execute.assert_called_once_with(
|
||||
["ovs-ofctl", "dump-flows", self.BR_NAME],
|
||||
root_helper=self.root_helper,
|
||||
process_input=None)
|
||||
|
||||
def test_delete_flow(self):
|
||||
ofport = "5"
|
||||
lsw_id = 40
|
||||
vid = 39
|
||||
utils.execute(["ovs-ofctl", "del-flows", self.BR_NAME,
|
||||
"in_port=" + ofport], root_helper=self.root_helper)
|
||||
utils.execute(["ovs-ofctl", "del-flows", self.BR_NAME,
|
||||
"tun_id=%s" % lsw_id], root_helper=self.root_helper)
|
||||
utils.execute(["ovs-ofctl", "del-flows", self.BR_NAME,
|
||||
"dl_vlan=%s" % vid], root_helper=self.root_helper)
|
||||
self.mox.ReplayAll()
|
||||
|
||||
self.br.delete_flows(in_port=ofport)
|
||||
self.br.delete_flows(tun_id=lsw_id)
|
||||
self.br.delete_flows(dl_vlan=vid)
|
||||
self.mox.VerifyAll()
|
||||
expected_calls = [
|
||||
mock.call(["ovs-ofctl", "del-flows", self.BR_NAME,
|
||||
"in_port=" + ofport],
|
||||
process_input=None, root_helper=self.root_helper),
|
||||
mock.call(["ovs-ofctl", "del-flows", self.BR_NAME,
|
||||
"tun_id=%s" % lsw_id],
|
||||
process_input=None, root_helper=self.root_helper),
|
||||
mock.call(["ovs-ofctl", "del-flows", self.BR_NAME,
|
||||
"dl_vlan=%s" % vid],
|
||||
process_input=None, root_helper=self.root_helper),
|
||||
]
|
||||
self.execute.assert_has_calls(expected_calls)
|
||||
|
||||
def test_defer_apply_flows(self):
|
||||
self.mox.StubOutWithMock(self.br, 'add_or_mod_flow_str')
|
||||
self.br.add_or_mod_flow_str(
|
||||
flow='added_flow_1').AndReturn('added_flow_1')
|
||||
self.br.add_or_mod_flow_str(
|
||||
flow='added_flow_2').AndReturn('added_flow_2')
|
||||
add_mod_flow = mock.patch.object(self.br,
|
||||
'add_or_mod_flow_str').start()
|
||||
add_mod_flow.side_effect = ['added_flow_1', 'added_flow_2']
|
||||
|
||||
self.mox.StubOutWithMock(self.br, '_build_flow_expr_arr')
|
||||
self.br._build_flow_expr_arr(delete=True,
|
||||
flow='deleted_flow_1'
|
||||
).AndReturn(['deleted_flow_1'])
|
||||
self.mox.StubOutWithMock(self.br, 'run_ofctl')
|
||||
self.br.run_ofctl('add-flows', ['-'], 'added_flow_1\nadded_flow_2\n')
|
||||
self.br.run_ofctl('del-flows', ['-'], 'deleted_flow_1\n')
|
||||
self.mox.ReplayAll()
|
||||
flow_expr = mock.patch.object(self.br, '_build_flow_expr_arr').start()
|
||||
flow_expr.return_value = ['deleted_flow_1']
|
||||
run_ofctl = mock.patch.object(self.br, 'run_ofctl').start()
|
||||
|
||||
self.br.defer_apply_on()
|
||||
self.br.add_flow(flow='added_flow_1')
|
||||
@ -295,7 +277,16 @@ class OVS_Lib_Test(base.BaseTestCase):
|
||||
self.br.add_flow(flow='added_flow_2')
|
||||
self.br.delete_flows(flow='deleted_flow_1')
|
||||
self.br.defer_apply_off()
|
||||
self.mox.VerifyAll()
|
||||
|
||||
add_mod_flow.assert_has_calls([
|
||||
mock.call(flow='added_flow_1'),
|
||||
mock.call(flow='added_flow_2')
|
||||
])
|
||||
flow_expr.assert_called_once_with(delete=True, flow='deleted_flow_1')
|
||||
run_ofctl.assert_has_calls([
|
||||
mock.call('add-flows', ['-'], 'added_flow_1\nadded_flow_2\n'),
|
||||
mock.call('del-flows', ['-'], 'deleted_flow_1\n')
|
||||
])
|
||||
|
||||
def test_add_tunnel_port(self):
|
||||
pname = "tap99"
|
||||
@ -303,51 +294,69 @@ class OVS_Lib_Test(base.BaseTestCase):
|
||||
remote_ip = "9.9.9.9"
|
||||
ofport = "6"
|
||||
|
||||
utils.execute(["ovs-vsctl", self.TO, '--', "--may-exist", "add-port",
|
||||
self.BR_NAME, pname], root_helper=self.root_helper)
|
||||
utils.execute(["ovs-vsctl", self.TO, "set", "Interface",
|
||||
pname, "type=gre"], root_helper=self.root_helper)
|
||||
utils.execute(["ovs-vsctl", self.TO, "set", "Interface",
|
||||
# Each element is a tuple of (expected mock call, return_value)
|
||||
expected_calls_and_values = [
|
||||
(mock.call(["ovs-vsctl", self.TO, '--', "--may-exist", "add-port",
|
||||
self.BR_NAME, pname], root_helper=self.root_helper),
|
||||
None),
|
||||
(mock.call(["ovs-vsctl", self.TO, "set", "Interface",
|
||||
pname, "type=gre"], root_helper=self.root_helper),
|
||||
None),
|
||||
(mock.call(["ovs-vsctl", self.TO, "set", "Interface",
|
||||
pname, "options:remote_ip=" + remote_ip],
|
||||
root_helper=self.root_helper)
|
||||
utils.execute(["ovs-vsctl", self.TO, "set", "Interface",
|
||||
root_helper=self.root_helper),
|
||||
None),
|
||||
(mock.call(["ovs-vsctl", self.TO, "set", "Interface",
|
||||
pname, "options:local_ip=" + local_ip],
|
||||
root_helper=self.root_helper)
|
||||
utils.execute(["ovs-vsctl", self.TO, "set", "Interface",
|
||||
root_helper=self.root_helper),
|
||||
None),
|
||||
(mock.call(["ovs-vsctl", self.TO, "set", "Interface",
|
||||
pname, "options:in_key=flow"],
|
||||
root_helper=self.root_helper)
|
||||
utils.execute(["ovs-vsctl", self.TO, "set", "Interface",
|
||||
root_helper=self.root_helper),
|
||||
None),
|
||||
(mock.call(["ovs-vsctl", self.TO, "set", "Interface",
|
||||
pname, "options:out_key=flow"],
|
||||
root_helper=self.root_helper)
|
||||
utils.execute(["ovs-vsctl", self.TO, "get",
|
||||
root_helper=self.root_helper),
|
||||
None),
|
||||
(mock.call(["ovs-vsctl", self.TO, "get",
|
||||
"Interface", pname, "ofport"],
|
||||
root_helper=self.root_helper).AndReturn(ofport)
|
||||
self.mox.ReplayAll()
|
||||
root_helper=self.root_helper),
|
||||
ofport),
|
||||
]
|
||||
tools.setup_mock_calls(self.execute, expected_calls_and_values)
|
||||
|
||||
self.assertEqual(
|
||||
self.br.add_tunnel_port(pname, remote_ip, local_ip),
|
||||
ofport)
|
||||
self.mox.VerifyAll()
|
||||
|
||||
tools.verify_mock_calls(self.execute, expected_calls_and_values)
|
||||
|
||||
def test_add_patch_port(self):
|
||||
pname = "tap99"
|
||||
peer = "bar10"
|
||||
ofport = "6"
|
||||
|
||||
utils.execute(["ovs-vsctl", self.TO, "add-port",
|
||||
self.BR_NAME, pname], root_helper=self.root_helper)
|
||||
utils.execute(["ovs-vsctl", self.TO, "set", "Interface",
|
||||
pname, "type=patch"], root_helper=self.root_helper)
|
||||
utils.execute(["ovs-vsctl", self.TO, "set",
|
||||
# Each element is a tuple of (expected mock call, return_value)
|
||||
expected_calls_and_values = [
|
||||
(mock.call(["ovs-vsctl", self.TO, "add-port",
|
||||
self.BR_NAME, pname], root_helper=self.root_helper),
|
||||
None),
|
||||
(mock.call(["ovs-vsctl", self.TO, "set", "Interface",
|
||||
pname, "type=patch"], root_helper=self.root_helper),
|
||||
None),
|
||||
(mock.call(["ovs-vsctl", self.TO, "set",
|
||||
"Interface", pname, "options:peer=" + peer],
|
||||
root_helper=self.root_helper)
|
||||
utils.execute(["ovs-vsctl", self.TO, "get",
|
||||
root_helper=self.root_helper),
|
||||
None),
|
||||
(mock.call(["ovs-vsctl", self.TO, "get",
|
||||
"Interface", pname, "ofport"],
|
||||
root_helper=self.root_helper).AndReturn(ofport)
|
||||
self.mox.ReplayAll()
|
||||
root_helper=self.root_helper),
|
||||
ofport)
|
||||
]
|
||||
tools.setup_mock_calls(self.execute, expected_calls_and_values)
|
||||
|
||||
self.assertEqual(self.br.add_patch_port(pname, peer), ofport)
|
||||
self.mox.VerifyAll()
|
||||
tools.verify_mock_calls(self.execute, expected_calls_and_values)
|
||||
|
||||
def _test_get_vif_ports(self, is_xen=False):
|
||||
pname = "tap99"
|
||||
@ -355,9 +364,6 @@ class OVS_Lib_Test(base.BaseTestCase):
|
||||
vif_id = uuidutils.generate_uuid()
|
||||
mac = "ca:fe:de:ad:be:ef"
|
||||
|
||||
utils.execute(["ovs-vsctl", self.TO, "list-ports", self.BR_NAME],
|
||||
root_helper=self.root_helper).AndReturn("%s\n" % pname)
|
||||
|
||||
if is_xen:
|
||||
external_ids = ('{xs-vif-uuid="%s", attached-mac="%s"}'
|
||||
% (vif_id, mac))
|
||||
@ -365,17 +371,28 @@ class OVS_Lib_Test(base.BaseTestCase):
|
||||
external_ids = ('{iface-id="%s", attached-mac="%s"}'
|
||||
% (vif_id, mac))
|
||||
|
||||
utils.execute(["ovs-vsctl", self.TO, "get",
|
||||
# Each element is a tuple of (expected mock call, return_value)
|
||||
expected_calls_and_values = [
|
||||
(mock.call(["ovs-vsctl", self.TO, "list-ports", self.BR_NAME],
|
||||
root_helper=self.root_helper),
|
||||
"%s\n" % pname),
|
||||
(mock.call(["ovs-vsctl", self.TO, "get",
|
||||
"Interface", pname, "external_ids"],
|
||||
root_helper=self.root_helper).AndReturn(external_ids)
|
||||
utils.execute(["ovs-vsctl", self.TO, "get",
|
||||
root_helper=self.root_helper),
|
||||
external_ids),
|
||||
(mock.call(["ovs-vsctl", self.TO, "get",
|
||||
"Interface", pname, "ofport"],
|
||||
root_helper=self.root_helper).AndReturn(ofport)
|
||||
root_helper=self.root_helper),
|
||||
ofport),
|
||||
]
|
||||
if is_xen:
|
||||
utils.execute(["xe", "vif-param-get", "param-name=other-config",
|
||||
expected_calls_and_values.append(
|
||||
(mock.call(["xe", "vif-param-get", "param-name=other-config",
|
||||
"param-key=nicira-iface-id", "uuid=" + vif_id],
|
||||
root_helper=self.root_helper).AndReturn(vif_id)
|
||||
self.mox.ReplayAll()
|
||||
root_helper=self.root_helper),
|
||||
vif_id)
|
||||
)
|
||||
tools.setup_mock_calls(self.execute, expected_calls_and_values)
|
||||
|
||||
ports = self.br.get_vif_ports()
|
||||
self.assertEqual(1, len(ports))
|
||||
@ -384,7 +401,7 @@ class OVS_Lib_Test(base.BaseTestCase):
|
||||
self.assertEqual(ports[0].vif_id, vif_id)
|
||||
self.assertEqual(ports[0].vif_mac, mac)
|
||||
self.assertEqual(ports[0].switch.br_name, self.BR_NAME)
|
||||
self.mox.VerifyAll()
|
||||
tools.verify_mock_calls(self.execute, expected_calls_and_values)
|
||||
|
||||
def _encode_ovs_json(self, headings, data):
|
||||
# See man ovs-vsctl(8) for the encoding details.
|
||||
@ -403,9 +420,6 @@ class OVS_Lib_Test(base.BaseTestCase):
|
||||
return jsonutils.dumps(r)
|
||||
|
||||
def _test_get_vif_port_set(self, is_xen):
|
||||
utils.execute(["ovs-vsctl", self.TO, "list-ports", self.BR_NAME],
|
||||
root_helper=self.root_helper).AndReturn('tap99\ntun22')
|
||||
|
||||
if is_xen:
|
||||
id_key = 'xs-vif-uuid'
|
||||
else:
|
||||
@ -421,21 +435,29 @@ class OVS_Lib_Test(base.BaseTestCase):
|
||||
['tun22', {}],
|
||||
]
|
||||
|
||||
utils.execute(["ovs-vsctl", self.TO, "--format=json",
|
||||
# Each element is a tuple of (expected mock call, return_value)
|
||||
expected_calls_and_values = [
|
||||
(mock.call(["ovs-vsctl", self.TO, "list-ports", self.BR_NAME],
|
||||
root_helper=self.root_helper),
|
||||
'tap99\ntun22'),
|
||||
(mock.call(["ovs-vsctl", self.TO, "--format=json",
|
||||
"--", "--columns=name,external_ids",
|
||||
"list", "Interface"],
|
||||
root_helper=self.root_helper).AndReturn(
|
||||
self._encode_ovs_json(headings, data))
|
||||
root_helper=self.root_helper),
|
||||
self._encode_ovs_json(headings, data)),
|
||||
]
|
||||
tools.setup_mock_calls(self.execute, expected_calls_and_values)
|
||||
|
||||
if is_xen:
|
||||
self.mox.StubOutWithMock(self.br, 'get_xapi_iface_id')
|
||||
self.br.get_xapi_iface_id('tap99id').AndReturn('tap99id')
|
||||
|
||||
self.mox.ReplayAll()
|
||||
get_xapi_iface_id = mock.patch.object(self.br,
|
||||
'get_xapi_iface_id').start()
|
||||
get_xapi_iface_id.return_value = 'tap99id'
|
||||
|
||||
port_set = self.br.get_vif_port_set()
|
||||
self.assertEqual(set(['tap99id']), port_set)
|
||||
self.mox.VerifyAll()
|
||||
tools.verify_mock_calls(self.execute, expected_calls_and_values)
|
||||
if is_xen:
|
||||
get_xapi_iface_id.assert_called_once_with('tap99id')
|
||||
|
||||
def test_get_vif_ports_nonxen(self):
|
||||
self._test_get_vif_ports(False)
|
||||
@ -450,35 +472,41 @@ class OVS_Lib_Test(base.BaseTestCase):
|
||||
self._test_get_vif_port_set(True)
|
||||
|
||||
def test_get_vif_port_set_list_ports_error(self):
|
||||
utils.execute(["ovs-vsctl", self.TO, "list-ports", self.BR_NAME],
|
||||
root_helper=self.root_helper).AndRaise(RuntimeError())
|
||||
utils.execute(["ovs-vsctl", self.TO, "--format=json",
|
||||
expected_calls_and_values = [
|
||||
(mock.call(["ovs-vsctl", self.TO, "list-ports", self.BR_NAME],
|
||||
root_helper=self.root_helper),
|
||||
RuntimeError()),
|
||||
(mock.call(["ovs-vsctl", self.TO, "--format=json",
|
||||
"--", "--columns=name,external_ids",
|
||||
"list", "Interface"],
|
||||
root_helper=self.root_helper).AndReturn(
|
||||
root_helper=self.root_helper),
|
||||
self._encode_ovs_json(['name', 'external_ids'], []))
|
||||
self.mox.ReplayAll()
|
||||
]
|
||||
tools.setup_mock_calls(self.execute, expected_calls_and_values)
|
||||
self.assertEqual(set(), self.br.get_vif_port_set())
|
||||
self.mox.VerifyAll()
|
||||
tools.verify_mock_calls(self.execute, expected_calls_and_values)
|
||||
|
||||
def test_get_vif_port_set_list_interface_error(self):
|
||||
utils.execute(["ovs-vsctl", self.TO, "list-ports", self.BR_NAME],
|
||||
root_helper=self.root_helper).AndRaise('tap99\n')
|
||||
utils.execute(["ovs-vsctl", self.TO, "--format=json",
|
||||
expected_calls_and_values = [
|
||||
(mock.call(["ovs-vsctl", self.TO, "list-ports", self.BR_NAME],
|
||||
root_helper=self.root_helper),
|
||||
'tap99\n'),
|
||||
(mock.call(["ovs-vsctl", self.TO, "--format=json",
|
||||
"--", "--columns=name,external_ids",
|
||||
"list", "Interface"],
|
||||
root_helper=self.root_helper).AndRaise(RuntimeError())
|
||||
self.mox.ReplayAll()
|
||||
root_helper=self.root_helper),
|
||||
RuntimeError()),
|
||||
]
|
||||
tools.setup_mock_calls(self.execute, expected_calls_and_values)
|
||||
self.assertEqual(set(), self.br.get_vif_port_set())
|
||||
self.mox.VerifyAll()
|
||||
tools.verify_mock_calls(self.execute, expected_calls_and_values)
|
||||
|
||||
def test_clear_db_attribute(self):
|
||||
pname = "tap77"
|
||||
utils.execute(["ovs-vsctl", self.TO, "clear", "Port",
|
||||
pname, "tag"], root_helper=self.root_helper)
|
||||
self.mox.ReplayAll()
|
||||
self.br.clear_db_attribute("Port", pname, "tag")
|
||||
self.mox.VerifyAll()
|
||||
self.execute.assert_called_once_with(
|
||||
["ovs-vsctl", self.TO, "clear", "Port", pname, "tag"],
|
||||
root_helper=self.root_helper)
|
||||
|
||||
def test_port_id_regex(self):
|
||||
result = ('external_ids : {attached-mac="fa:16:3e:23:5b:f2",'
|
||||
@ -499,55 +527,55 @@ class OVS_Lib_Test(base.BaseTestCase):
|
||||
iface = 'tap0'
|
||||
br = 'br-int'
|
||||
root_helper = 'sudo'
|
||||
utils.execute(["ovs-vsctl", self.TO, "iface-to-br", iface],
|
||||
root_helper=root_helper).AndReturn('br-int')
|
||||
self.execute.return_value = 'br-int'
|
||||
|
||||
self.mox.ReplayAll()
|
||||
self.assertEqual(ovs_lib.get_bridge_for_iface(root_helper, iface), br)
|
||||
self.mox.VerifyAll()
|
||||
self.execute.assert_called_once_with(
|
||||
["ovs-vsctl", self.TO, "iface-to-br", iface],
|
||||
root_helper=root_helper)
|
||||
|
||||
def test_iface_to_br_handles_ovs_vsctl_exception(self):
|
||||
iface = 'tap0'
|
||||
root_helper = 'sudo'
|
||||
utils.execute(["ovs-vsctl", self.TO, "iface-to-br", iface],
|
||||
root_helper=root_helper).AndRaise(Exception)
|
||||
self.execute.side_effect = Exception
|
||||
|
||||
self.mox.ReplayAll()
|
||||
self.assertIsNone(ovs_lib.get_bridge_for_iface(root_helper, iface))
|
||||
self.mox.VerifyAll()
|
||||
self.execute.assert_called_once_with(
|
||||
["ovs-vsctl", self.TO, "iface-to-br", iface],
|
||||
root_helper=root_helper)
|
||||
|
||||
def test_delete_all_ports(self):
|
||||
self.mox.StubOutWithMock(self.br, 'get_port_name_list')
|
||||
self.br.get_port_name_list().AndReturn(['port1'])
|
||||
self.mox.StubOutWithMock(self.br, 'delete_port')
|
||||
self.br.delete_port('port1')
|
||||
self.mox.ReplayAll()
|
||||
with mock.patch.object(self.br, 'get_port_name_list',
|
||||
return_value=['port1']) as get_port:
|
||||
with mock.patch.object(self.br, 'delete_port') as delete_port:
|
||||
self.br.delete_ports(all_ports=True)
|
||||
self.mox.VerifyAll()
|
||||
get_port.assert_called_once_with()
|
||||
delete_port.assert_called_once_with('port1')
|
||||
|
||||
def test_delete_neutron_ports(self):
|
||||
port1 = ovs_lib.VifPort('tap1234', 1, uuidutils.generate_uuid(),
|
||||
'ca:fe:de:ad:be:ef', 'br')
|
||||
port2 = ovs_lib.VifPort('tap5678', 2, uuidutils.generate_uuid(),
|
||||
'ca:ee:de:ad:be:ef', 'br')
|
||||
self.mox.StubOutWithMock(self.br, 'get_vif_ports')
|
||||
self.br.get_vif_ports().AndReturn([port1, port2])
|
||||
self.mox.StubOutWithMock(self.br, 'delete_port')
|
||||
self.br.delete_port('tap1234')
|
||||
self.br.delete_port('tap5678')
|
||||
self.mox.ReplayAll()
|
||||
with mock.patch.object(self.br, 'get_vif_ports',
|
||||
return_value=[port1, port2]) as get_ports:
|
||||
with mock.patch.object(self.br, 'delete_port') as delete_port:
|
||||
self.br.delete_ports(all_ports=False)
|
||||
self.mox.VerifyAll()
|
||||
get_ports.assert_called_once_with()
|
||||
delete_port.assert_has_calls([
|
||||
mock.call('tap1234'),
|
||||
mock.call('tap5678')
|
||||
])
|
||||
|
||||
def test_get_bridges(self):
|
||||
bridges = ['br-int', 'br-ex']
|
||||
root_helper = 'sudo'
|
||||
utils.execute(["ovs-vsctl", self.TO, "list-br"],
|
||||
root_helper=root_helper).AndReturn('br-int\nbr-ex\n')
|
||||
self.execute.return_value = 'br-int\nbr-ex\n'
|
||||
|
||||
self.mox.ReplayAll()
|
||||
self.assertEqual(ovs_lib.get_bridges(root_helper), bridges)
|
||||
self.mox.VerifyAll()
|
||||
self.execute.assert_called_once_with(
|
||||
["ovs-vsctl", self.TO, "list-br"],
|
||||
root_helper=root_helper)
|
||||
|
||||
def test_get_local_port_mac_succeeds(self):
|
||||
with mock.patch('neutron.agent.linux.ip_lib.IpLinkCommand',
|
||||
|
@ -16,7 +16,9 @@
|
||||
#
|
||||
# @author: Dave Lapsley, Nicira Networks, Inc.
|
||||
|
||||
import mox
|
||||
import contextlib
|
||||
|
||||
import mock
|
||||
from oslo.config import cfg
|
||||
|
||||
from neutron.agent.linux import ip_lib
|
||||
@ -68,8 +70,7 @@ class TunnelTest(base.BaseTestCase):
|
||||
cfg.CONF.set_override('rpc_backend',
|
||||
'neutron.openstack.common.rpc.impl_fake')
|
||||
cfg.CONF.set_override('report_interval', 0, 'AGENT')
|
||||
self.mox = mox.Mox()
|
||||
self.addCleanup(self.mox.UnsetStubs)
|
||||
self.addCleanup(mock.patch.stopall)
|
||||
|
||||
self.INT_BRIDGE = 'integration_bridge'
|
||||
self.TUN_BRIDGE = 'tunnel_bridge'
|
||||
@ -79,60 +80,89 @@ class TunnelTest(base.BaseTestCase):
|
||||
self.TUN_OFPORT = 22222
|
||||
self.MAP_TUN_OFPORT = 33333
|
||||
self.VETH_MTU = None
|
||||
self.inta = self.mox.CreateMock(ip_lib.IPDevice)
|
||||
self.intb = self.mox.CreateMock(ip_lib.IPDevice)
|
||||
self.inta.link = self.mox.CreateMock(ip_lib.IpLinkCommand)
|
||||
self.intb.link = self.mox.CreateMock(ip_lib.IpLinkCommand)
|
||||
self.inta = mock.Mock()
|
||||
self.intb = mock.Mock()
|
||||
|
||||
self.mox.StubOutClassWithMocks(ovs_lib, 'OVSBridge')
|
||||
self.mock_int_bridge = ovs_lib.OVSBridge(self.INT_BRIDGE, 'sudo')
|
||||
self.mock_int_bridge.get_local_port_mac().AndReturn('000000000001')
|
||||
self.mock_int_bridge.delete_port('patch-tun')
|
||||
self.mock_int_bridge.remove_all_flows()
|
||||
self.mock_int_bridge.add_flow(priority=1, actions='normal')
|
||||
self.ovs_bridges = {self.INT_BRIDGE: mock.Mock(),
|
||||
self.TUN_BRIDGE: mock.Mock(),
|
||||
self.MAP_TUN_BRIDGE: mock.Mock(),
|
||||
}
|
||||
|
||||
self.mock_map_tun_bridge = ovs_lib.OVSBridge(
|
||||
self.MAP_TUN_BRIDGE, 'sudo')
|
||||
self.mock_bridge = mock.patch.object(ovs_lib, 'OVSBridge').start()
|
||||
self.mock_bridge.side_effect = (lambda br_name, root_helper:
|
||||
self.ovs_bridges[br_name])
|
||||
self.mock_bridge_expected = [
|
||||
mock.call(self.INT_BRIDGE, 'sudo'),
|
||||
mock.call(self.MAP_TUN_BRIDGE, 'sudo'),
|
||||
mock.call(self.TUN_BRIDGE, 'sudo'),
|
||||
]
|
||||
|
||||
self.mock_int_bridge = self.ovs_bridges[self.INT_BRIDGE]
|
||||
self.mock_int_bridge.get_local_port_mac.return_value = '000000000001'
|
||||
self.mock_int_bridge_expected = [
|
||||
mock.call.get_local_port_mac(),
|
||||
mock.call.delete_port('patch-tun'),
|
||||
mock.call.remove_all_flows(),
|
||||
mock.call.add_flow(priority=1, actions='normal'),
|
||||
]
|
||||
|
||||
self.mock_map_tun_bridge = self.ovs_bridges[self.MAP_TUN_BRIDGE]
|
||||
self.mock_map_tun_bridge.br_name = self.MAP_TUN_BRIDGE
|
||||
self.mock_map_tun_bridge.remove_all_flows()
|
||||
self.mock_map_tun_bridge.add_flow(priority=1, actions='normal')
|
||||
self.mock_int_bridge.delete_port('int-tunnel_bridge_mapping')
|
||||
self.mock_map_tun_bridge.delete_port('phy-tunnel_bridge_mapping')
|
||||
self.mock_int_bridge.add_port(self.inta)
|
||||
self.mock_map_tun_bridge.add_port(self.intb)
|
||||
self.inta.link.set_up()
|
||||
self.intb.link.set_up()
|
||||
self.mock_map_tun_bridge.add_port.return_value = None
|
||||
self.mock_map_tun_bridge_expected = [
|
||||
mock.call.remove_all_flows(),
|
||||
mock.call.add_flow(priority=1, actions='normal'),
|
||||
mock.call.delete_port('phy-tunnel_bridge_mapping'),
|
||||
mock.call.add_port(self.intb),
|
||||
]
|
||||
self.mock_int_bridge.add_port.return_value = None
|
||||
self.mock_int_bridge_expected += [
|
||||
mock.call.delete_port('int-tunnel_bridge_mapping'),
|
||||
mock.call.add_port(self.inta)
|
||||
]
|
||||
self.inta_expected = [mock.call.link.set_up()]
|
||||
self.intb_expected = [mock.call.link.set_up()]
|
||||
|
||||
self.mock_int_bridge.add_flow(priority=2, in_port=None, actions='drop')
|
||||
self.mock_map_tun_bridge.add_flow(
|
||||
priority=2, in_port=None, actions='drop')
|
||||
self.mock_int_bridge_expected += [
|
||||
mock.call.add_flow(priority=2, in_port=None, actions='drop')
|
||||
]
|
||||
self.mock_map_tun_bridge_expected += [
|
||||
mock.call.add_flow(priority=2, in_port=None, actions='drop')
|
||||
]
|
||||
|
||||
self.mock_tun_bridge = ovs_lib.OVSBridge(self.TUN_BRIDGE, 'sudo')
|
||||
self.mock_tun_bridge.reset_bridge()
|
||||
self.mock_int_bridge.add_patch_port(
|
||||
'patch-tun', 'patch-int').AndReturn(self.TUN_OFPORT)
|
||||
self.mock_tun_bridge.add_patch_port(
|
||||
'patch-int', 'patch-tun').AndReturn(self.INT_OFPORT)
|
||||
self.mock_tun_bridge = self.ovs_bridges[self.TUN_BRIDGE]
|
||||
self.mock_tun_bridge_expected = [
|
||||
mock.call.reset_bridge(),
|
||||
mock.call.add_patch_port('patch-int', 'patch-tun'),
|
||||
]
|
||||
self.mock_int_bridge_expected += [
|
||||
mock.call.add_patch_port('patch-tun', 'patch-int')
|
||||
]
|
||||
self.mock_int_bridge.add_patch_port.return_value = self.TUN_OFPORT
|
||||
self.mock_tun_bridge.add_patch_port.return_value = self.INT_OFPORT
|
||||
|
||||
self.mock_tun_bridge.remove_all_flows()
|
||||
self.mock_tun_bridge.add_flow(priority=1,
|
||||
self.mock_tun_bridge_expected += [
|
||||
mock.call.remove_all_flows(),
|
||||
mock.call.add_flow(priority=1,
|
||||
in_port=self.INT_OFPORT,
|
||||
actions="resubmit(,%s)" %
|
||||
constants.PATCH_LV_TO_TUN)
|
||||
self.mock_tun_bridge.add_flow(priority=0, actions='drop')
|
||||
self.mock_tun_bridge.add_flow(table=constants.PATCH_LV_TO_TUN,
|
||||
constants.PATCH_LV_TO_TUN),
|
||||
mock.call.add_flow(priority=0, actions='drop'),
|
||||
mock.call.add_flow(table=constants.PATCH_LV_TO_TUN,
|
||||
dl_dst=UCAST_MAC,
|
||||
actions="resubmit(,%s)" %
|
||||
constants.UCAST_TO_TUN)
|
||||
self.mock_tun_bridge.add_flow(table=constants.PATCH_LV_TO_TUN,
|
||||
constants.UCAST_TO_TUN),
|
||||
mock.call.add_flow(table=constants.PATCH_LV_TO_TUN,
|
||||
dl_dst=BCAST_MAC,
|
||||
actions="resubmit(,%s)" %
|
||||
constants.FLOOD_TO_TUN)
|
||||
constants.FLOOD_TO_TUN),
|
||||
]
|
||||
for tunnel_type in constants.TUNNEL_NETWORK_TYPES:
|
||||
self.mock_tun_bridge.add_flow(
|
||||
self.mock_tun_bridge_expected.append(
|
||||
mock.call.add_flow(
|
||||
table=constants.TUN_TABLE[tunnel_type],
|
||||
priority=0,
|
||||
actions="drop")
|
||||
actions="drop"))
|
||||
learned_flow = ("table=%s,"
|
||||
"priority=1,"
|
||||
"hard_timeout=300,"
|
||||
@ -142,73 +172,105 @@ class TunnelTest(base.BaseTestCase):
|
||||
"load:NXM_NX_TUN_ID[]->NXM_NX_TUN_ID[],"
|
||||
"output:NXM_OF_IN_PORT[]" %
|
||||
constants.UCAST_TO_TUN)
|
||||
self.mock_tun_bridge.add_flow(table=constants.LEARN_FROM_TUN,
|
||||
self.mock_tun_bridge_expected += [
|
||||
mock.call.add_flow(table=constants.LEARN_FROM_TUN,
|
||||
priority=1,
|
||||
actions="learn(%s),output:%s" %
|
||||
(learned_flow, self.INT_OFPORT))
|
||||
self.mock_tun_bridge.add_flow(table=constants.UCAST_TO_TUN,
|
||||
(learned_flow, self.INT_OFPORT)),
|
||||
mock.call.add_flow(table=constants.UCAST_TO_TUN,
|
||||
priority=0,
|
||||
actions="resubmit(,%s)" %
|
||||
constants.FLOOD_TO_TUN)
|
||||
self.mock_tun_bridge.add_flow(table=constants.FLOOD_TO_TUN,
|
||||
constants.FLOOD_TO_TUN),
|
||||
mock.call.add_flow(table=constants.FLOOD_TO_TUN,
|
||||
priority=0,
|
||||
actions="drop")
|
||||
]
|
||||
|
||||
self.mox.StubOutWithMock(ip_lib, 'device_exists')
|
||||
ip_lib.device_exists('tunnel_bridge_mapping', 'sudo').AndReturn(True)
|
||||
ip_lib.device_exists(
|
||||
'int-tunnel_bridge_mapping', 'sudo').AndReturn(True)
|
||||
self.device_exists = mock.patch.object(ip_lib, 'device_exists').start()
|
||||
self.device_exists.return_value = True
|
||||
self.device_exists_expected = [
|
||||
mock.call('tunnel_bridge_mapping', 'sudo'),
|
||||
mock.call('int-tunnel_bridge_mapping', 'sudo'),
|
||||
]
|
||||
|
||||
self.mox.StubOutWithMock(ip_lib.IpLinkCommand, 'delete')
|
||||
ip_lib.IPDevice('int-tunnel_bridge_mapping').link.delete()
|
||||
self.ipdevice = mock.patch.object(ip_lib, 'IPDevice').start()
|
||||
self.ipdevice_expected = [
|
||||
mock.call('int-tunnel_bridge_mapping', 'sudo'),
|
||||
mock.call().link.delete()
|
||||
]
|
||||
self.ipwrapper = mock.patch.object(ip_lib, 'IPWrapper').start()
|
||||
add_veth = self.ipwrapper.return_value.add_veth
|
||||
add_veth.return_value = [self.inta, self.intb]
|
||||
self.ipwrapper_expected = [
|
||||
mock.call('sudo'),
|
||||
mock.call().add_veth('int-tunnel_bridge_mapping',
|
||||
'phy-tunnel_bridge_mapping')
|
||||
]
|
||||
|
||||
self.mox.StubOutClassWithMocks(ip_lib, 'IPWrapper')
|
||||
ip_lib.IPWrapper('sudo').add_veth(
|
||||
'int-tunnel_bridge_mapping',
|
||||
'phy-tunnel_bridge_mapping').AndReturn([self.inta, self.intb])
|
||||
|
||||
self.mox.StubOutWithMock(ovs_lib, 'get_bridges')
|
||||
ovs_lib.get_bridges('sudo').AndReturn([self.INT_BRIDGE,
|
||||
self.get_bridges = mock.patch.object(ovs_lib, 'get_bridges').start()
|
||||
self.get_bridges.return_value = [self.INT_BRIDGE,
|
||||
self.TUN_BRIDGE,
|
||||
self.MAP_TUN_BRIDGE])
|
||||
self.MAP_TUN_BRIDGE]
|
||||
self.get_bridges_expected = [
|
||||
mock.call('sudo')
|
||||
]
|
||||
|
||||
def _verify_mock_call(self, mock_obj, expected):
|
||||
mock_obj.assert_has_calls(expected)
|
||||
self.assertEqual(len(mock_obj.mock_calls), len(expected))
|
||||
|
||||
def _verify_mock_calls(self):
|
||||
self._verify_mock_call(self.mock_bridge, self.mock_bridge_expected)
|
||||
self._verify_mock_call(self.mock_int_bridge,
|
||||
self.mock_int_bridge_expected)
|
||||
self._verify_mock_call(self.mock_map_tun_bridge,
|
||||
self.mock_map_tun_bridge_expected)
|
||||
self._verify_mock_call(self.mock_tun_bridge,
|
||||
self.mock_tun_bridge_expected)
|
||||
self._verify_mock_call(self.device_exists, self.device_exists_expected)
|
||||
self._verify_mock_call(self.ipdevice, self.ipdevice_expected)
|
||||
self._verify_mock_call(self.ipwrapper, self.ipwrapper_expected)
|
||||
self._verify_mock_call(self.get_bridges, self.get_bridges_expected)
|
||||
self._verify_mock_call(self.inta, self.inta_expected)
|
||||
self._verify_mock_call(self.intb, self.intb_expected)
|
||||
|
||||
def test_construct(self):
|
||||
self.mox.ReplayAll()
|
||||
ovs_neutron_agent.OVSNeutronAgent(self.INT_BRIDGE,
|
||||
self.TUN_BRIDGE,
|
||||
'10.0.0.1', self.NET_MAPPING,
|
||||
'sudo', 2, ['gre'],
|
||||
self.VETH_MTU)
|
||||
self.mox.VerifyAll()
|
||||
self._verify_mock_calls()
|
||||
|
||||
def test_construct_vxlan(self):
|
||||
self.mox.StubOutWithMock(ovs_lib, 'get_installed_ovs_klm_version')
|
||||
ovs_lib.get_installed_ovs_klm_version().AndReturn("1.10")
|
||||
self.mox.StubOutWithMock(ovs_lib, 'get_installed_ovs_usr_version')
|
||||
ovs_lib.get_installed_ovs_usr_version('sudo').AndReturn("1.10")
|
||||
self.mox.ReplayAll()
|
||||
with mock.patch.object(ovs_lib, 'get_installed_ovs_klm_version',
|
||||
return_value="1.10") as klm_ver:
|
||||
with mock.patch.object(ovs_lib, 'get_installed_ovs_usr_version',
|
||||
return_value="1.10") as usr_ver:
|
||||
ovs_neutron_agent.OVSNeutronAgent(self.INT_BRIDGE,
|
||||
self.TUN_BRIDGE,
|
||||
'10.0.0.1', self.NET_MAPPING,
|
||||
'sudo', 2, ['vxlan'],
|
||||
self.VETH_MTU)
|
||||
self.mox.VerifyAll()
|
||||
klm_ver.assert_called_once_with()
|
||||
usr_ver.assert_called_once_with('sudo')
|
||||
self._verify_mock_calls()
|
||||
|
||||
def test_provision_local_vlan(self):
|
||||
ofports = ','.join(TUN_OFPORTS[constants.TYPE_GRE].values())
|
||||
self.mock_tun_bridge.mod_flow(table=constants.FLOOD_TO_TUN,
|
||||
self.mock_tun_bridge_expected += [
|
||||
mock.call.mod_flow(table=constants.FLOOD_TO_TUN,
|
||||
priority=1,
|
||||
dl_vlan=LV_ID,
|
||||
actions="strip_vlan,"
|
||||
"set_tunnel:%s,output:%s" %
|
||||
(LS_ID, ofports))
|
||||
|
||||
self.mock_tun_bridge.add_flow(table=constants.TUN_TABLE['gre'],
|
||||
(LS_ID, ofports)),
|
||||
mock.call.add_flow(table=constants.TUN_TABLE['gre'],
|
||||
priority=1,
|
||||
tun_id=LS_ID,
|
||||
actions="mod_vlan_vid:%s,resubmit(,%s)" %
|
||||
(LV_ID, constants.LEARN_FROM_TUN))
|
||||
self.mox.ReplayAll()
|
||||
(LV_ID, constants.LEARN_FROM_TUN)),
|
||||
]
|
||||
|
||||
a = ovs_neutron_agent.OVSNeutronAgent(self.INT_BRIDGE,
|
||||
self.TUN_BRIDGE,
|
||||
@ -218,19 +280,18 @@ class TunnelTest(base.BaseTestCase):
|
||||
a.available_local_vlans = set([LV_ID])
|
||||
a.tun_br_ofports = TUN_OFPORTS
|
||||
a.provision_local_vlan(NET_UUID, constants.TYPE_GRE, None, LS_ID)
|
||||
self.mox.VerifyAll()
|
||||
self._verify_mock_calls()
|
||||
|
||||
def test_provision_local_vlan_flat(self):
|
||||
action_string = 'strip_vlan,normal'
|
||||
self.mock_map_tun_bridge.add_flow(
|
||||
priority=4, in_port=self.MAP_TUN_OFPORT,
|
||||
dl_vlan=LV_ID, actions=action_string)
|
||||
self.mock_map_tun_bridge_expected.append(
|
||||
mock.call.add_flow(priority=4, in_port=self.MAP_TUN_OFPORT,
|
||||
dl_vlan=LV_ID, actions=action_string))
|
||||
|
||||
action_string = 'mod_vlan_vid:%s,normal' % LV_ID
|
||||
self.mock_int_bridge.add_flow(priority=3, in_port=self.INT_OFPORT,
|
||||
dl_vlan=65535, actions=action_string)
|
||||
|
||||
self.mox.ReplayAll()
|
||||
self.mock_int_bridge_expected.append(
|
||||
mock.call.add_flow(priority=3, in_port=self.INT_OFPORT,
|
||||
dl_vlan=65535, actions=action_string))
|
||||
|
||||
a = ovs_neutron_agent.OVSNeutronAgent(self.INT_BRIDGE,
|
||||
self.TUN_BRIDGE,
|
||||
@ -242,29 +303,28 @@ class TunnelTest(base.BaseTestCase):
|
||||
a.phys_ofports['net1'] = self.MAP_TUN_OFPORT
|
||||
a.int_ofports['net1'] = self.INT_OFPORT
|
||||
a.provision_local_vlan(NET_UUID, constants.TYPE_FLAT, 'net1', LS_ID)
|
||||
self.mox.VerifyAll()
|
||||
self._verify_mock_calls()
|
||||
|
||||
def test_provision_local_vlan_flat_fail(self):
|
||||
self.mox.ReplayAll()
|
||||
a = ovs_neutron_agent.OVSNeutronAgent(self.INT_BRIDGE,
|
||||
self.TUN_BRIDGE,
|
||||
'10.0.0.1', self.NET_MAPPING,
|
||||
'sudo', 2, ['gre'],
|
||||
self.VETH_MTU)
|
||||
a.provision_local_vlan(NET_UUID, constants.TYPE_FLAT, 'net2', LS_ID)
|
||||
self.mox.VerifyAll()
|
||||
self._verify_mock_calls()
|
||||
|
||||
def test_provision_local_vlan_vlan(self):
|
||||
action_string = 'mod_vlan_vid:%s,normal' % LS_ID
|
||||
self.mock_map_tun_bridge.add_flow(
|
||||
priority=4, in_port=self.MAP_TUN_OFPORT,
|
||||
dl_vlan=LV_ID, actions=action_string)
|
||||
self.mock_map_tun_bridge_expected.append(
|
||||
mock.call.add_flow(priority=4, in_port=self.MAP_TUN_OFPORT,
|
||||
dl_vlan=LV_ID, actions=action_string))
|
||||
|
||||
action_string = 'mod_vlan_vid:%s,normal' % LS_ID
|
||||
self.mock_int_bridge.add_flow(priority=3, in_port=self.INT_OFPORT,
|
||||
dl_vlan=LV_ID, actions=action_string)
|
||||
self.mock_int_bridge_expected.append(
|
||||
mock.call.add_flow(priority=3, in_port=self.INT_OFPORT,
|
||||
dl_vlan=LV_ID, actions=action_string))
|
||||
|
||||
self.mox.ReplayAll()
|
||||
a = ovs_neutron_agent.OVSNeutronAgent(self.INT_BRIDGE,
|
||||
self.TUN_BRIDGE,
|
||||
'10.0.0.1', self.NET_MAPPING,
|
||||
@ -275,24 +335,24 @@ class TunnelTest(base.BaseTestCase):
|
||||
a.phys_ofports['net1'] = self.MAP_TUN_OFPORT
|
||||
a.int_ofports['net1'] = self.INT_OFPORT
|
||||
a.provision_local_vlan(NET_UUID, constants.TYPE_VLAN, 'net1', LS_ID)
|
||||
self.mox.VerifyAll()
|
||||
self._verify_mock_calls()
|
||||
|
||||
def test_provision_local_vlan_vlan_fail(self):
|
||||
self.mox.ReplayAll()
|
||||
a = ovs_neutron_agent.OVSNeutronAgent(self.INT_BRIDGE,
|
||||
self.TUN_BRIDGE,
|
||||
'10.0.0.1', self.NET_MAPPING,
|
||||
'sudo', 2, ['gre'],
|
||||
self.VETH_MTU)
|
||||
a.provision_local_vlan(NET_UUID, constants.TYPE_VLAN, 'net2', LS_ID)
|
||||
self.mox.VerifyAll()
|
||||
self._verify_mock_calls()
|
||||
|
||||
def test_reclaim_local_vlan(self):
|
||||
self.mock_tun_bridge.delete_flows(
|
||||
table=constants.TUN_TABLE['gre'], tun_id=LS_ID)
|
||||
self.mock_tun_bridge.delete_flows(dl_vlan=LVM.vlan)
|
||||
self.mock_tun_bridge_expected += [
|
||||
mock.call.delete_flows(
|
||||
table=constants.TUN_TABLE['gre'], tun_id=LS_ID),
|
||||
mock.call.delete_flows(dl_vlan=LVM.vlan)
|
||||
]
|
||||
|
||||
self.mox.ReplayAll()
|
||||
a = ovs_neutron_agent.OVSNeutronAgent(self.INT_BRIDGE,
|
||||
self.TUN_BRIDGE,
|
||||
'10.0.0.1', self.NET_MAPPING,
|
||||
@ -302,16 +362,16 @@ class TunnelTest(base.BaseTestCase):
|
||||
a.local_vlan_map[NET_UUID] = LVM
|
||||
a.reclaim_local_vlan(NET_UUID)
|
||||
self.assertIn(LVM.vlan, a.available_local_vlans)
|
||||
self.mox.VerifyAll()
|
||||
self._verify_mock_calls()
|
||||
|
||||
def test_reclaim_local_vlan_flat(self):
|
||||
self.mock_map_tun_bridge.delete_flows(
|
||||
in_port=self.MAP_TUN_OFPORT, dl_vlan=LVM_FLAT.vlan)
|
||||
self.mock_map_tun_bridge_expected.append(
|
||||
mock.call.delete_flows(
|
||||
in_port=self.MAP_TUN_OFPORT, dl_vlan=LVM_FLAT.vlan))
|
||||
self.mock_int_bridge_expected.append(
|
||||
mock.call.delete_flows(
|
||||
dl_vlan=65535, in_port=self.INT_OFPORT))
|
||||
|
||||
self.mock_int_bridge.delete_flows(
|
||||
dl_vlan=65535, in_port=self.INT_OFPORT)
|
||||
|
||||
self.mox.ReplayAll()
|
||||
a = ovs_neutron_agent.OVSNeutronAgent(self.INT_BRIDGE,
|
||||
self.TUN_BRIDGE,
|
||||
'10.0.0.1', self.NET_MAPPING,
|
||||
@ -325,16 +385,16 @@ class TunnelTest(base.BaseTestCase):
|
||||
a.local_vlan_map[NET_UUID] = LVM_FLAT
|
||||
a.reclaim_local_vlan(NET_UUID)
|
||||
self.assertIn(LVM_FLAT.vlan, a.available_local_vlans)
|
||||
self.mox.VerifyAll()
|
||||
self._verify_mock_calls()
|
||||
|
||||
def test_reclaim_local_vlan_vlan(self):
|
||||
self.mock_map_tun_bridge.delete_flows(
|
||||
in_port=self.MAP_TUN_OFPORT, dl_vlan=LVM_VLAN.vlan)
|
||||
self.mock_map_tun_bridge_expected.append(
|
||||
mock.call.delete_flows(
|
||||
in_port=self.MAP_TUN_OFPORT, dl_vlan=LVM_VLAN.vlan))
|
||||
self.mock_int_bridge_expected.append(
|
||||
mock.call.delete_flows(
|
||||
dl_vlan=LV_ID, in_port=self.INT_OFPORT))
|
||||
|
||||
self.mock_int_bridge.delete_flows(
|
||||
dl_vlan=LV_ID, in_port=self.INT_OFPORT)
|
||||
|
||||
self.mox.ReplayAll()
|
||||
a = ovs_neutron_agent.OVSNeutronAgent(self.INT_BRIDGE,
|
||||
self.TUN_BRIDGE,
|
||||
'10.0.0.1', self.NET_MAPPING,
|
||||
@ -348,14 +408,15 @@ class TunnelTest(base.BaseTestCase):
|
||||
a.local_vlan_map[NET_UUID] = LVM_VLAN
|
||||
a.reclaim_local_vlan(NET_UUID)
|
||||
self.assertIn(LVM_VLAN.vlan, a.available_local_vlans)
|
||||
self.mox.VerifyAll()
|
||||
self._verify_mock_calls()
|
||||
|
||||
def test_port_bound(self):
|
||||
self.mock_int_bridge.set_db_attribute('Port', VIF_PORT.port_name,
|
||||
'tag', str(LVM.vlan))
|
||||
self.mock_int_bridge.delete_flows(in_port=VIF_PORT.ofport)
|
||||
self.mock_int_bridge_expected += [
|
||||
mock.call.set_db_attribute('Port', VIF_PORT.port_name,
|
||||
'tag', str(LVM.vlan)),
|
||||
mock.call.delete_flows(in_port=VIF_PORT.ofport)
|
||||
]
|
||||
|
||||
self.mox.ReplayAll()
|
||||
a = ovs_neutron_agent.OVSNeutronAgent(self.INT_BRIDGE,
|
||||
self.TUN_BRIDGE,
|
||||
'10.0.0.1', self.NET_MAPPING,
|
||||
@ -363,15 +424,11 @@ class TunnelTest(base.BaseTestCase):
|
||||
self.VETH_MTU)
|
||||
a.local_vlan_map[NET_UUID] = LVM
|
||||
a.port_bound(VIF_PORT, NET_UUID, 'gre', None, LS_ID)
|
||||
self.mox.VerifyAll()
|
||||
self._verify_mock_calls()
|
||||
|
||||
def test_port_unbound(self):
|
||||
self.mox.StubOutWithMock(
|
||||
ovs_neutron_agent.OVSNeutronAgent, 'reclaim_local_vlan')
|
||||
ovs_neutron_agent.OVSNeutronAgent.reclaim_local_vlan(NET_UUID)
|
||||
|
||||
self.mox.ReplayAll()
|
||||
|
||||
with mock.patch.object(ovs_neutron_agent.OVSNeutronAgent,
|
||||
'reclaim_local_vlan') as reclaim_local_vlan:
|
||||
a = ovs_neutron_agent.OVSNeutronAgent(self.INT_BRIDGE,
|
||||
self.TUN_BRIDGE,
|
||||
'10.0.0.1', self.NET_MAPPING,
|
||||
@ -379,16 +436,19 @@ class TunnelTest(base.BaseTestCase):
|
||||
self.VETH_MTU)
|
||||
a.local_vlan_map[NET_UUID] = LVM
|
||||
a.port_unbound(VIF_ID, NET_UUID)
|
||||
self.mox.VerifyAll()
|
||||
|
||||
reclaim_local_vlan.assert_called_once_with(NET_UUID)
|
||||
self._verify_mock_calls()
|
||||
|
||||
def test_port_dead(self):
|
||||
self.mock_int_bridge.set_db_attribute(
|
||||
'Port', VIF_PORT.port_name, 'tag', ovs_neutron_agent.DEAD_VLAN_TAG)
|
||||
|
||||
self.mock_int_bridge.add_flow(priority=2, in_port=VIF_PORT.ofport,
|
||||
self.mock_int_bridge_expected += [
|
||||
mock.call.set_db_attribute(
|
||||
'Port', VIF_PORT.port_name,
|
||||
'tag', ovs_neutron_agent.DEAD_VLAN_TAG),
|
||||
mock.call.add_flow(priority=2, in_port=VIF_PORT.ofport,
|
||||
actions='drop')
|
||||
]
|
||||
|
||||
self.mox.ReplayAll()
|
||||
a = ovs_neutron_agent.OVSNeutronAgent(self.INT_BRIDGE,
|
||||
self.TUN_BRIDGE,
|
||||
'10.0.0.1', self.NET_MAPPING,
|
||||
@ -397,34 +457,37 @@ class TunnelTest(base.BaseTestCase):
|
||||
a.available_local_vlans = set([LV_ID])
|
||||
a.local_vlan_map[NET_UUID] = LVM
|
||||
a.port_dead(VIF_PORT)
|
||||
self.mox.VerifyAll()
|
||||
self._verify_mock_calls()
|
||||
|
||||
def test_tunnel_update(self):
|
||||
self.mock_tun_bridge.add_tunnel_port('gre-1', '10.0.10.1', '10.0.0.1',
|
||||
'gre', 4789).AndReturn('9999')
|
||||
self.mock_tun_bridge.add_flow(actions='resubmit(,2)', in_port='9999',
|
||||
priority=1)
|
||||
self.mox.ReplayAll()
|
||||
tunnel_port = '9999'
|
||||
self.mock_tun_bridge.add_tunnel_port.return_value = tunnel_port
|
||||
self.mock_tun_bridge_expected += [
|
||||
mock.call.add_tunnel_port('gre-1', '10.0.10.1', '10.0.0.1',
|
||||
'gre', 4789),
|
||||
mock.call.add_flow(priority=1, in_port=tunnel_port,
|
||||
actions='resubmit(,2)')
|
||||
]
|
||||
|
||||
a = ovs_neutron_agent.OVSNeutronAgent(self.INT_BRIDGE,
|
||||
self.TUN_BRIDGE,
|
||||
'10.0.0.1', self.NET_MAPPING,
|
||||
'sudo', 2, ['gre'],
|
||||
self.VETH_MTU)
|
||||
a.tunnel_update(
|
||||
mox.MockAnything, tunnel_id='1', tunnel_ip='10.0.10.1',
|
||||
mock.sentinel.ctx, tunnel_id='1', tunnel_ip='10.0.10.1',
|
||||
tunnel_type=constants.TYPE_GRE)
|
||||
self.mox.VerifyAll()
|
||||
self._verify_mock_calls()
|
||||
|
||||
def test_tunnel_update_self(self):
|
||||
self.mox.ReplayAll()
|
||||
a = ovs_neutron_agent.OVSNeutronAgent(self.INT_BRIDGE,
|
||||
self.TUN_BRIDGE,
|
||||
'10.0.0.1', self.NET_MAPPING,
|
||||
'sudo', 2, ['gre'],
|
||||
self.VETH_MTU)
|
||||
a.tunnel_update(
|
||||
mox.MockAnything, tunnel_id='1', tunnel_ip='10.0.0.1')
|
||||
self.mox.VerifyAll()
|
||||
mock.sentinel.ctx, tunnel_id='1', tunnel_ip='10.0.0.1')
|
||||
self._verify_mock_calls()
|
||||
|
||||
def test_daemon_loop(self):
|
||||
reply2 = {'current': set(['tap0']),
|
||||
@ -435,28 +498,19 @@ class TunnelTest(base.BaseTestCase):
|
||||
'added': set([]),
|
||||
'removed': set([])}
|
||||
|
||||
self.mox.StubOutWithMock(log.ContextAdapter, 'exception')
|
||||
log.ContextAdapter.exception(
|
||||
_("Error in agent event loop")).AndRaise(
|
||||
Exception('Fake exception to get out of the loop'))
|
||||
with contextlib.nested(
|
||||
mock.patch.object(log.ContextAdapter, 'exception'),
|
||||
mock.patch.object(ovs_neutron_agent.OVSNeutronAgent,
|
||||
'update_ports'),
|
||||
mock.patch.object(ovs_neutron_agent.OVSNeutronAgent,
|
||||
'process_network_ports')
|
||||
) as (log_exception, update_ports, process_network_ports):
|
||||
log_exception.side_effect = Exception(
|
||||
'Fake exception to get out of the loop')
|
||||
update_ports.side_effect = [reply2, reply3]
|
||||
process_network_ports.side_effect = [
|
||||
False, Exception('Fake exception to get out of the loop')]
|
||||
|
||||
self.mox.StubOutWithMock(
|
||||
ovs_neutron_agent.OVSNeutronAgent, 'update_ports')
|
||||
ovs_neutron_agent.OVSNeutronAgent.update_ports(set()).AndReturn(reply2)
|
||||
ovs_neutron_agent.OVSNeutronAgent.update_ports(
|
||||
set(['tap0'])).AndReturn(reply3)
|
||||
self.mox.StubOutWithMock(
|
||||
ovs_neutron_agent.OVSNeutronAgent, 'process_network_ports')
|
||||
ovs_neutron_agent.OVSNeutronAgent.process_network_ports(
|
||||
{'current': set(['tap0']),
|
||||
'removed': set([]),
|
||||
'added': set([])}).AndReturn(False)
|
||||
ovs_neutron_agent.OVSNeutronAgent.process_network_ports(
|
||||
{'current': set(['tap0']),
|
||||
'removed': set([]),
|
||||
'added': set([])}).AndRaise(
|
||||
Exception('Fake exception to get out of the loop'))
|
||||
self.mox.ReplayAll()
|
||||
q_agent = ovs_neutron_agent.OVSNeutronAgent(self.INT_BRIDGE,
|
||||
self.TUN_BRIDGE,
|
||||
'10.0.0.1',
|
||||
@ -466,13 +520,26 @@ class TunnelTest(base.BaseTestCase):
|
||||
|
||||
# Hack to test loop
|
||||
# We start method and expect it will raise after 2nd loop
|
||||
# If something goes wrong, mox.VerifyAll() will catch it
|
||||
# If something goes wrong, assert_has_calls below will catch it
|
||||
try:
|
||||
q_agent.daemon_loop()
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
self.mox.VerifyAll()
|
||||
log_exception.assert_called_once_with("Error in agent event loop")
|
||||
update_ports.assert_has_calls([
|
||||
mock.call(set()),
|
||||
mock.call(set(['tap0']))
|
||||
])
|
||||
process_network_ports.assert_has_calls([
|
||||
mock.call({'current': set(['tap0']),
|
||||
'removed': set([]),
|
||||
'added': set([])}),
|
||||
mock.call({'current': set(['tap2']),
|
||||
'removed': set([]),
|
||||
'added': set([])})
|
||||
])
|
||||
self._verify_mock_calls()
|
||||
|
||||
|
||||
class TunnelTestWithMTU(TunnelTest):
|
||||
@ -480,5 +547,5 @@ class TunnelTestWithMTU(TunnelTest):
|
||||
def setUp(self):
|
||||
super(TunnelTestWithMTU, self).setUp()
|
||||
self.VETH_MTU = 1500
|
||||
self.inta.link.set_mtu(self.VETH_MTU)
|
||||
self.intb.link.set_mtu(self.VETH_MTU)
|
||||
self.inta_expected.append(mock.call.link.set_mtu(self.VETH_MTU))
|
||||
self.intb_expected.append(mock.call.link.set_mtu(self.VETH_MTU))
|
||||
|
@ -20,10 +20,11 @@
|
||||
import inspect
|
||||
import os
|
||||
|
||||
import mox
|
||||
import mock
|
||||
|
||||
from neutron.agent.linux import iptables_manager
|
||||
from neutron.tests import base
|
||||
from neutron.tests import tools
|
||||
|
||||
|
||||
IPTABLES_ARG = {'bn': iptables_manager.binary_name}
|
||||
@ -67,12 +68,11 @@ class IptablesManagerStateFulTestCase(base.BaseTestCase):
|
||||
|
||||
def setUp(self):
|
||||
super(IptablesManagerStateFulTestCase, self).setUp()
|
||||
self.mox = mox.Mox()
|
||||
self.root_helper = 'sudo'
|
||||
self.iptables = (iptables_manager.
|
||||
IptablesManager(root_helper=self.root_helper))
|
||||
self.mox.StubOutWithMock(self.iptables, "execute")
|
||||
self.addCleanup(self.mox.UnsetStubs)
|
||||
self.execute = mock.patch.object(self.iptables, "execute").start()
|
||||
self.addCleanup(mock.patch.stopall)
|
||||
|
||||
def test_binary_name(self):
|
||||
self.assertEqual(iptables_manager.binary_name,
|
||||
@ -94,10 +94,7 @@ class IptablesManagerStateFulTestCase(base.BaseTestCase):
|
||||
self.iptables = (iptables_manager.
|
||||
IptablesManager(root_helper=self.root_helper,
|
||||
binary_name=bn))
|
||||
self.mox.StubOutWithMock(self.iptables, "execute")
|
||||
|
||||
self.iptables.execute(['iptables-save', '-c'],
|
||||
root_helper=self.root_helper).AndReturn('')
|
||||
self.execute = mock.patch.object(self.iptables, "execute").start()
|
||||
|
||||
iptables_args = {'bn': bn[:16]}
|
||||
|
||||
@ -154,18 +151,23 @@ class IptablesManagerStateFulTestCase(base.BaseTestCase):
|
||||
'COMMIT\n'
|
||||
'# Completed by iptables_manager\n' % iptables_args)
|
||||
|
||||
self.iptables.execute(['iptables-restore', '-c'],
|
||||
expected_calls_and_values = [
|
||||
(mock.call(['iptables-save', '-c'],
|
||||
root_helper=self.root_helper),
|
||||
''),
|
||||
(mock.call(['iptables-restore', '-c'],
|
||||
process_input=nat_dump + filter_dump_mod,
|
||||
root_helper=self.root_helper).AndReturn(None)
|
||||
|
||||
self.iptables.execute(['iptables-save', '-c'],
|
||||
root_helper=self.root_helper).AndReturn('')
|
||||
|
||||
self.iptables.execute(['iptables-restore', '-c'],
|
||||
root_helper=self.root_helper),
|
||||
None),
|
||||
(mock.call(['iptables-save', '-c'],
|
||||
root_helper=self.root_helper),
|
||||
''),
|
||||
(mock.call(['iptables-restore', '-c'],
|
||||
process_input=nat_dump + filter_dump,
|
||||
root_helper=self.root_helper).AndReturn(None)
|
||||
|
||||
self.mox.ReplayAll()
|
||||
root_helper=self.root_helper),
|
||||
None),
|
||||
]
|
||||
tools.setup_mock_calls(self.execute, expected_calls_and_values)
|
||||
|
||||
self.iptables.ipv4['filter'].add_chain('filter')
|
||||
self.iptables.apply()
|
||||
@ -173,7 +175,7 @@ class IptablesManagerStateFulTestCase(base.BaseTestCase):
|
||||
self.iptables.ipv4['filter'].empty_chain('filter')
|
||||
self.iptables.apply()
|
||||
|
||||
self.mox.VerifyAll()
|
||||
tools.verify_mock_calls(self.execute, expected_calls_and_values)
|
||||
|
||||
def test_empty_chain_custom_binary_name(self):
|
||||
bn = ("abcdef" * 5)[:16]
|
||||
@ -181,10 +183,7 @@ class IptablesManagerStateFulTestCase(base.BaseTestCase):
|
||||
self.iptables = (iptables_manager.
|
||||
IptablesManager(root_helper=self.root_helper,
|
||||
binary_name=bn))
|
||||
self.mox.StubOutWithMock(self.iptables, "execute")
|
||||
|
||||
self.iptables.execute(['iptables-save', '-c'],
|
||||
root_helper=self.root_helper).AndReturn('')
|
||||
self.execute = mock.patch.object(self.iptables, "execute").start()
|
||||
|
||||
iptables_args = {'bn': bn}
|
||||
|
||||
@ -241,18 +240,23 @@ class IptablesManagerStateFulTestCase(base.BaseTestCase):
|
||||
'COMMIT\n'
|
||||
'# Completed by iptables_manager\n' % iptables_args)
|
||||
|
||||
self.iptables.execute(['iptables-restore', '-c'],
|
||||
expected_calls_and_values = [
|
||||
(mock.call(['iptables-save', '-c'],
|
||||
root_helper=self.root_helper),
|
||||
''),
|
||||
(mock.call(['iptables-restore', '-c'],
|
||||
process_input=nat_dump + filter_dump_mod,
|
||||
root_helper=self.root_helper).AndReturn(None)
|
||||
|
||||
self.iptables.execute(['iptables-save', '-c'],
|
||||
root_helper=self.root_helper).AndReturn('')
|
||||
|
||||
self.iptables.execute(['iptables-restore', '-c'],
|
||||
root_helper=self.root_helper),
|
||||
None),
|
||||
(mock.call(['iptables-save', '-c'],
|
||||
root_helper=self.root_helper),
|
||||
''),
|
||||
(mock.call(['iptables-restore', '-c'],
|
||||
process_input=nat_dump + filter_dump,
|
||||
root_helper=self.root_helper).AndReturn(None)
|
||||
|
||||
self.mox.ReplayAll()
|
||||
root_helper=self.root_helper),
|
||||
None),
|
||||
]
|
||||
tools.setup_mock_calls(self.execute, expected_calls_and_values)
|
||||
|
||||
self.iptables.ipv4['filter'].add_chain('filter')
|
||||
self.iptables.ipv4['filter'].add_rule('filter',
|
||||
@ -262,12 +266,9 @@ class IptablesManagerStateFulTestCase(base.BaseTestCase):
|
||||
self.iptables.ipv4['filter'].remove_chain('filter')
|
||||
self.iptables.apply()
|
||||
|
||||
self.mox.VerifyAll()
|
||||
tools.verify_mock_calls(self.execute, expected_calls_and_values)
|
||||
|
||||
def test_add_and_remove_chain(self):
|
||||
self.iptables.execute(['iptables-save', '-c'],
|
||||
root_helper=self.root_helper).AndReturn('')
|
||||
|
||||
filter_dump_mod = ('# Generated by iptables_manager\n'
|
||||
'*filter\n'
|
||||
':neutron-filter-top - [0:0]\n'
|
||||
@ -286,18 +287,23 @@ class IptablesManagerStateFulTestCase(base.BaseTestCase):
|
||||
'# Completed by iptables_manager\n'
|
||||
% IPTABLES_ARG)
|
||||
|
||||
self.iptables.execute(['iptables-restore', '-c'],
|
||||
expected_calls_and_values = [
|
||||
(mock.call(['iptables-save', '-c'],
|
||||
root_helper=self.root_helper),
|
||||
''),
|
||||
(mock.call(['iptables-restore', '-c'],
|
||||
process_input=NAT_DUMP + filter_dump_mod,
|
||||
root_helper=self.root_helper).AndReturn(None)
|
||||
|
||||
self.iptables.execute(['iptables-save', '-c'],
|
||||
root_helper=self.root_helper).AndReturn('')
|
||||
|
||||
self.iptables.execute(['iptables-restore', '-c'],
|
||||
root_helper=self.root_helper),
|
||||
None),
|
||||
(mock.call(['iptables-save', '-c'],
|
||||
root_helper=self.root_helper),
|
||||
''),
|
||||
(mock.call(['iptables-restore', '-c'],
|
||||
process_input=NAT_DUMP + FILTER_DUMP,
|
||||
root_helper=self.root_helper).AndReturn(None)
|
||||
|
||||
self.mox.ReplayAll()
|
||||
root_helper=self.root_helper),
|
||||
None),
|
||||
]
|
||||
tools.setup_mock_calls(self.execute, expected_calls_and_values)
|
||||
|
||||
self.iptables.ipv4['filter'].add_chain('filter')
|
||||
self.iptables.apply()
|
||||
@ -305,12 +311,9 @@ class IptablesManagerStateFulTestCase(base.BaseTestCase):
|
||||
self.iptables.ipv4['filter'].remove_chain('filter')
|
||||
self.iptables.apply()
|
||||
|
||||
self.mox.VerifyAll()
|
||||
tools.verify_mock_calls(self.execute, expected_calls_and_values)
|
||||
|
||||
def test_add_filter_rule(self):
|
||||
self.iptables.execute(['iptables-save', '-c'],
|
||||
root_helper=self.root_helper).AndReturn('')
|
||||
|
||||
filter_dump_mod = ('# Generated by iptables_manager\n'
|
||||
'*filter\n'
|
||||
':neutron-filter-top - [0:0]\n'
|
||||
@ -332,19 +335,24 @@ class IptablesManagerStateFulTestCase(base.BaseTestCase):
|
||||
'# Completed by iptables_manager\n'
|
||||
% IPTABLES_ARG)
|
||||
|
||||
self.iptables.execute(['iptables-restore', '-c'],
|
||||
expected_calls_and_values = [
|
||||
(mock.call(['iptables-save', '-c'],
|
||||
root_helper=self.root_helper),
|
||||
''),
|
||||
(mock.call(['iptables-restore', '-c'],
|
||||
process_input=NAT_DUMP + filter_dump_mod,
|
||||
root_helper=self.root_helper).AndReturn(None)
|
||||
|
||||
self.iptables.execute(['iptables-save', '-c'],
|
||||
root_helper=self.root_helper).AndReturn('')
|
||||
|
||||
self.iptables.execute(['iptables-restore', '-c'],
|
||||
root_helper=self.root_helper),
|
||||
None),
|
||||
(mock.call(['iptables-save', '-c'],
|
||||
root_helper=self.root_helper),
|
||||
''),
|
||||
(mock.call(['iptables-restore', '-c'],
|
||||
process_input=NAT_DUMP + FILTER_DUMP,
|
||||
root_helper=self.root_helper
|
||||
).AndReturn(None)
|
||||
|
||||
self.mox.ReplayAll()
|
||||
),
|
||||
None),
|
||||
]
|
||||
tools.setup_mock_calls(self.execute, expected_calls_and_values)
|
||||
|
||||
self.iptables.ipv4['filter'].add_chain('filter')
|
||||
self.iptables.ipv4['filter'].add_rule('filter', '-j DROP')
|
||||
@ -361,7 +369,8 @@ class IptablesManagerStateFulTestCase(base.BaseTestCase):
|
||||
self.iptables.ipv4['filter'].remove_chain('filter')
|
||||
|
||||
self.iptables.apply()
|
||||
self.mox.VerifyAll()
|
||||
|
||||
tools.verify_mock_calls(self.execute, expected_calls_and_values)
|
||||
|
||||
def test_add_nat_rule(self):
|
||||
nat_dump = ('# Generated by iptables_manager\n'
|
||||
@ -405,21 +414,24 @@ class IptablesManagerStateFulTestCase(base.BaseTestCase):
|
||||
'# Completed by iptables_manager\n'
|
||||
% IPTABLES_ARG)
|
||||
|
||||
self.iptables.execute(['iptables-save', '-c'],
|
||||
root_helper=self.root_helper).AndReturn('')
|
||||
|
||||
self.iptables.execute(['iptables-restore', '-c'],
|
||||
expected_calls_and_values = [
|
||||
(mock.call(['iptables-save', '-c'],
|
||||
root_helper=self.root_helper),
|
||||
''),
|
||||
(mock.call(['iptables-restore', '-c'],
|
||||
process_input=nat_dump_mod + FILTER_DUMP,
|
||||
root_helper=self.root_helper).AndReturn(None)
|
||||
|
||||
self.iptables.execute(['iptables-save', '-c'],
|
||||
root_helper=self.root_helper).AndReturn('')
|
||||
|
||||
self.iptables.execute(['iptables-restore', '-c'],
|
||||
root_helper=self.root_helper),
|
||||
None),
|
||||
(mock.call(['iptables-save', '-c'],
|
||||
root_helper=self.root_helper),
|
||||
''),
|
||||
(mock.call(['iptables-restore', '-c'],
|
||||
process_input=nat_dump + FILTER_DUMP,
|
||||
root_helper=self.root_helper).AndReturn(None)
|
||||
root_helper=self.root_helper),
|
||||
None),
|
||||
]
|
||||
tools.setup_mock_calls(self.execute, expected_calls_and_values)
|
||||
|
||||
self.mox.ReplayAll()
|
||||
self.iptables.ipv4['nat'].add_chain('nat')
|
||||
self.iptables.ipv4['nat'].add_rule('PREROUTING',
|
||||
'-d 192.168.0.3 -j '
|
||||
@ -439,57 +451,37 @@ class IptablesManagerStateFulTestCase(base.BaseTestCase):
|
||||
self.iptables.ipv4['nat'].remove_chain('nat')
|
||||
|
||||
self.iptables.apply()
|
||||
self.mox.VerifyAll()
|
||||
|
||||
tools.verify_mock_calls(self.execute, expected_calls_and_values)
|
||||
|
||||
def test_add_rule_to_a_nonexistent_chain(self):
|
||||
self.assertRaises(LookupError, self.iptables.ipv4['filter'].add_rule,
|
||||
'nonexistent', '-j DROP')
|
||||
|
||||
def test_remove_nonexistent_chain(self):
|
||||
self.mox.StubOutWithMock(iptables_manager, "LOG")
|
||||
iptables_manager.LOG.warn(('Attempted to remove chain %s which does '
|
||||
'not exist'), 'nonexistent')
|
||||
self.mox.ReplayAll()
|
||||
with mock.patch.object(iptables_manager, "LOG") as log:
|
||||
self.iptables.ipv4['filter'].remove_chain('nonexistent')
|
||||
self.mox.VerifyAll()
|
||||
log.warn.assert_called_once_with(
|
||||
'Attempted to remove chain %s which does not exist',
|
||||
'nonexistent')
|
||||
|
||||
def test_remove_nonexistent_rule(self):
|
||||
self.mox.StubOutWithMock(iptables_manager, "LOG")
|
||||
iptables_manager.LOG.warn('Tried to remove rule that was not there: '
|
||||
'%(chain)r %(rule)r %(wrap)r %(top)r',
|
||||
{'wrap': True, 'top': False,
|
||||
'rule': '-j DROP',
|
||||
'chain': 'nonexistent'})
|
||||
self.mox.ReplayAll()
|
||||
with mock.patch.object(iptables_manager, "LOG") as log:
|
||||
self.iptables.ipv4['filter'].remove_rule('nonexistent', '-j DROP')
|
||||
self.mox.VerifyAll()
|
||||
log.warn.assert_called_once_with(
|
||||
'Tried to remove rule that was not there: '
|
||||
'%(chain)r %(rule)r %(wrap)r %(top)r',
|
||||
{'wrap': True, 'top': False, 'rule': '-j DROP',
|
||||
'chain': 'nonexistent'})
|
||||
|
||||
def test_get_traffic_counters_chain_notexists(self):
|
||||
iptables_dump = (
|
||||
'Chain OUTPUT (policy ACCEPT 400 packets, 65901 bytes)\n'
|
||||
' pkts bytes target prot opt in out source'
|
||||
' destination \n'
|
||||
' 400 65901 chain1 all -- * * 0.0.0.0/0'
|
||||
' 0.0.0.0/0 \n'
|
||||
' 400 65901 chain2 all -- * * 0.0.0.0/0'
|
||||
' 0.0.0.0/0 \n')
|
||||
|
||||
self.iptables.execute(['iptables', '-t', 'filter', '-L', 'OUTPUT',
|
||||
'-n', '-v', '-x'],
|
||||
root_helper=self.root_helper
|
||||
).AndReturn(iptables_dump)
|
||||
self.iptables.execute(['iptables', '-t', 'nat', '-L', 'OUTPUT', '-n',
|
||||
'-v', '-x'],
|
||||
root_helper=self.root_helper
|
||||
).AndReturn('')
|
||||
self.iptables.execute(['ip6tables', '-t', 'filter', '-L', 'OUTPUT',
|
||||
'-n', '-v', '-x'],
|
||||
root_helper=self.root_helper
|
||||
).AndReturn(iptables_dump)
|
||||
|
||||
self.mox.ReplayAll()
|
||||
with mock.patch.object(iptables_manager, "LOG") as log:
|
||||
acc = self.iptables.get_traffic_counters('chain1')
|
||||
self.assertIsNone(acc)
|
||||
self.assertEqual(0, self.execute.call_count)
|
||||
log.warn.assert_called_once_with(
|
||||
'Attempted to get traffic counters of chain %s which '
|
||||
'does not exist', 'chain1')
|
||||
|
||||
def test_get_traffic_counters(self):
|
||||
iptables_dump = (
|
||||
@ -501,26 +493,27 @@ class IptablesManagerStateFulTestCase(base.BaseTestCase):
|
||||
' 400 65901 chain2 all -- * * 0.0.0.0/0'
|
||||
' 0.0.0.0/0 \n')
|
||||
|
||||
self.iptables.execute(['iptables', '-t', 'filter', '-L', 'OUTPUT',
|
||||
expected_calls_and_values = [
|
||||
(mock.call(['iptables', '-t', 'filter', '-L', 'OUTPUT',
|
||||
'-n', '-v', '-x'],
|
||||
root_helper=self.root_helper
|
||||
).AndReturn(iptables_dump)
|
||||
self.iptables.execute(['iptables', '-t', 'nat', '-L', 'OUTPUT', '-n',
|
||||
root_helper=self.root_helper),
|
||||
iptables_dump),
|
||||
(mock.call(['iptables', '-t', 'nat', '-L', 'OUTPUT', '-n',
|
||||
'-v', '-x'],
|
||||
root_helper=self.root_helper
|
||||
).AndReturn('')
|
||||
|
||||
self.iptables.execute(['ip6tables', '-t', 'filter', '-L', 'OUTPUT',
|
||||
root_helper=self.root_helper),
|
||||
''),
|
||||
(mock.call(['ip6tables', '-t', 'filter', '-L', 'OUTPUT',
|
||||
'-n', '-v', '-x'],
|
||||
root_helper=self.root_helper
|
||||
).AndReturn(iptables_dump)
|
||||
root_helper=self.root_helper),
|
||||
iptables_dump),
|
||||
]
|
||||
tools.setup_mock_calls(self.execute, expected_calls_and_values)
|
||||
|
||||
self.mox.ReplayAll()
|
||||
acc = self.iptables.get_traffic_counters('OUTPUT')
|
||||
self.assertEqual(acc['pkts'], 1600)
|
||||
self.assertEqual(acc['bytes'], 263604)
|
||||
|
||||
self.mox.VerifyAll()
|
||||
tools.verify_mock_calls(self.execute, expected_calls_and_values)
|
||||
|
||||
def test_get_traffic_counters_with_zero(self):
|
||||
iptables_dump = (
|
||||
@ -532,26 +525,27 @@ class IptablesManagerStateFulTestCase(base.BaseTestCase):
|
||||
' 400 65901 chain2 all -- * * 0.0.0.0/0'
|
||||
' 0.0.0.0/0 \n')
|
||||
|
||||
self.iptables.execute(['iptables', '-t', 'filter', '-L', 'OUTPUT',
|
||||
expected_calls_and_values = [
|
||||
(mock.call(['iptables', '-t', 'filter', '-L', 'OUTPUT',
|
||||
'-n', '-v', '-x', '-Z'],
|
||||
root_helper=self.root_helper
|
||||
).AndReturn(iptables_dump)
|
||||
self.iptables.execute(['iptables', '-t', 'nat', '-L', 'OUTPUT', '-n',
|
||||
root_helper=self.root_helper),
|
||||
iptables_dump),
|
||||
(mock.call(['iptables', '-t', 'nat', '-L', 'OUTPUT', '-n',
|
||||
'-v', '-x', '-Z'],
|
||||
root_helper=self.root_helper
|
||||
).AndReturn('')
|
||||
|
||||
self.iptables.execute(['ip6tables', '-t', 'filter', '-L', 'OUTPUT',
|
||||
root_helper=self.root_helper),
|
||||
''),
|
||||
(mock.call(['ip6tables', '-t', 'filter', '-L', 'OUTPUT',
|
||||
'-n', '-v', '-x', '-Z'],
|
||||
root_helper=self.root_helper
|
||||
).AndReturn(iptables_dump)
|
||||
root_helper=self.root_helper),
|
||||
iptables_dump),
|
||||
]
|
||||
tools.setup_mock_calls(self.execute, expected_calls_and_values)
|
||||
|
||||
self.mox.ReplayAll()
|
||||
acc = self.iptables.get_traffic_counters('OUTPUT', zero=True)
|
||||
self.assertEqual(acc['pkts'], 1600)
|
||||
self.assertEqual(acc['bytes'], 263604)
|
||||
|
||||
self.mox.VerifyAll()
|
||||
tools.verify_mock_calls(self.execute, expected_calls_and_values)
|
||||
|
||||
|
||||
class IptablesManagerStateLessTestCase(base.BaseTestCase):
|
||||
|
@ -19,8 +19,8 @@ from contextlib import nested
|
||||
|
||||
import mock
|
||||
from mock import call
|
||||
import mox
|
||||
from oslo.config import cfg
|
||||
from testtools import matchers
|
||||
import webob.exc
|
||||
|
||||
from neutron.agent import firewall as firewall_base
|
||||
@ -1212,13 +1212,11 @@ class TestSecurityGroupAgentWithIptables(base.BaseTestCase):
|
||||
|
||||
def setUp(self):
|
||||
super(TestSecurityGroupAgentWithIptables, self).setUp()
|
||||
self.mox = mox.Mox()
|
||||
cfg.CONF.set_override(
|
||||
'firewall_driver',
|
||||
self.FIREWALL_DRIVER,
|
||||
group='SECURITYGROUP')
|
||||
self.addCleanup(mock.patch.stopall)
|
||||
self.addCleanup(self.mox.UnsetStubs)
|
||||
|
||||
self.agent = sg_rpc.SecurityGroupAgentRpcMixin()
|
||||
self.agent.context = None
|
||||
@ -1228,7 +1226,13 @@ class TestSecurityGroupAgentWithIptables(base.BaseTestCase):
|
||||
self.agent.init_firewall()
|
||||
|
||||
self.iptables = self.agent.firewall.iptables
|
||||
self.mox.StubOutWithMock(self.iptables, "execute")
|
||||
self.iptables_execute = mock.patch.object(self.iptables,
|
||||
"execute").start()
|
||||
self.iptables_execute_return_values = []
|
||||
self.expected_call_count = 0
|
||||
self.expected_calls = []
|
||||
self.expected_process_inputs = []
|
||||
self.iptables_execute.side_effect = self.iptables_execute_return_values
|
||||
|
||||
self.rpc = mock.Mock()
|
||||
self.agent.plugin_rpc = self.rpc
|
||||
@ -1300,36 +1304,65 @@ class TestSecurityGroupAgentWithIptables(base.BaseTestCase):
|
||||
value = value.replace('[', '\[')
|
||||
value = value.replace(']', '\]')
|
||||
value = value.replace('*', '\*')
|
||||
return mox.Regex(value)
|
||||
return value
|
||||
|
||||
def _register_mock_call(self, *args, **kwargs):
|
||||
return_value = kwargs.pop('return_value', None)
|
||||
self.iptables_execute_return_values.append(return_value)
|
||||
|
||||
has_process_input = 'process_input' in kwargs
|
||||
process_input = kwargs.get('process_input')
|
||||
self.expected_process_inputs.append((has_process_input, process_input))
|
||||
|
||||
if has_process_input:
|
||||
kwargs['process_input'] = mock.ANY
|
||||
self.expected_calls.append(call(*args, **kwargs))
|
||||
self.expected_call_count += 1
|
||||
|
||||
def _verify_mock_calls(self):
|
||||
self.assertEqual(self.expected_call_count,
|
||||
self.iptables_execute.call_count)
|
||||
self.iptables_execute.assert_has_calls(self.expected_calls)
|
||||
|
||||
for i, expected in enumerate(self.expected_process_inputs):
|
||||
check, expected_regex = expected
|
||||
if not check:
|
||||
continue
|
||||
# The second or later arguments of self.iptables.execute
|
||||
# are keyword parameter, so keyword argument is extracted by [1]
|
||||
kwargs = self.iptables_execute.call_args_list[i][1]
|
||||
self.assertThat(kwargs['process_input'],
|
||||
matchers.MatchesRegex(expected_regex))
|
||||
|
||||
def _replay_iptables(self, v4_filter, v6_filter):
|
||||
self.iptables.execute(
|
||||
self._register_mock_call(
|
||||
['iptables-save', '-c'],
|
||||
root_helper=self.root_helper).AndReturn('')
|
||||
|
||||
self.iptables.execute(
|
||||
root_helper=self.root_helper,
|
||||
return_value='')
|
||||
self._register_mock_call(
|
||||
['iptables-restore', '-c'],
|
||||
process_input=(self._regex(IPTABLES_NAT + v4_filter)),
|
||||
root_helper=self.root_helper).AndReturn('')
|
||||
|
||||
self.iptables.execute(
|
||||
process_input=self._regex(IPTABLES_NAT + v4_filter),
|
||||
root_helper=self.root_helper,
|
||||
return_value='')
|
||||
self._register_mock_call(
|
||||
['ip6tables-save', '-c'],
|
||||
root_helper=self.root_helper).AndReturn('')
|
||||
|
||||
self.iptables.execute(
|
||||
root_helper=self.root_helper,
|
||||
return_value='')
|
||||
self._register_mock_call(
|
||||
['ip6tables-restore', '-c'],
|
||||
process_input=self._regex(v6_filter),
|
||||
root_helper=self.root_helper).AndReturn('')
|
||||
root_helper=self.root_helper,
|
||||
return_value='')
|
||||
|
||||
def test_prepare_remove_port(self):
|
||||
self.rpc.security_group_rules_for_devices.return_value = self.devices1
|
||||
self._replay_iptables(IPTABLES_FILTER_1, IPTABLES_FILTER_V6_1)
|
||||
self._replay_iptables(IPTABLES_FILTER_EMPTY, IPTABLES_FILTER_V6_EMPTY)
|
||||
self.mox.ReplayAll()
|
||||
|
||||
self.agent.prepare_devices_filter(['tap_port1'])
|
||||
self.agent.remove_devices_filter(['tap_port1'])
|
||||
self.mox.VerifyAll()
|
||||
|
||||
self._verify_mock_calls()
|
||||
|
||||
def test_security_group_member_updated(self):
|
||||
self.rpc.security_group_rules_for_devices.return_value = self.devices1
|
||||
@ -1339,7 +1372,6 @@ class TestSecurityGroupAgentWithIptables(base.BaseTestCase):
|
||||
self._replay_iptables(IPTABLES_FILTER_2_2, IPTABLES_FILTER_V6_2)
|
||||
self._replay_iptables(IPTABLES_FILTER_1, IPTABLES_FILTER_V6_1)
|
||||
self._replay_iptables(IPTABLES_FILTER_EMPTY, IPTABLES_FILTER_V6_EMPTY)
|
||||
self.mox.ReplayAll()
|
||||
|
||||
self.agent.prepare_devices_filter(['tap_port1'])
|
||||
self.rpc.security_group_rules_for_devices.return_value = self.devices2
|
||||
@ -1349,19 +1381,19 @@ class TestSecurityGroupAgentWithIptables(base.BaseTestCase):
|
||||
self.agent.security_groups_member_updated(['security_group1'])
|
||||
self.agent.remove_devices_filter(['tap_port2'])
|
||||
self.agent.remove_devices_filter(['tap_port1'])
|
||||
self.mox.VerifyAll()
|
||||
|
||||
self._verify_mock_calls()
|
||||
|
||||
def test_security_group_rule_updated(self):
|
||||
self.rpc.security_group_rules_for_devices.return_value = self.devices2
|
||||
self._replay_iptables(IPTABLES_FILTER_2, IPTABLES_FILTER_V6_2)
|
||||
self._replay_iptables(IPTABLES_FILTER_2_3, IPTABLES_FILTER_V6_2)
|
||||
self.mox.ReplayAll()
|
||||
|
||||
self.agent.prepare_devices_filter(['tap_port1', 'tap_port3'])
|
||||
self.rpc.security_group_rules_for_devices.return_value = self.devices3
|
||||
self.agent.security_groups_rule_updated(['security_group1'])
|
||||
|
||||
self.mox.VerifyAll()
|
||||
self._verify_mock_calls()
|
||||
|
||||
|
||||
class SGNotificationTestMixin():
|
||||
|
Loading…
Reference in New Issue
Block a user