--- /dev/null
+# 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)
import os
import mock
-import mox
from oslo.config import cfg
-import stubout
import testtools
from neutron import context
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()
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()
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
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."""
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):
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()
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)
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()
'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):
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())
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):
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)
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()
'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)
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()
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)
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)
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()
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])),
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:
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'
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)
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'
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
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."""
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):
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):
'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)
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):
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):
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):
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):
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)
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'
# @author: Dan Wendlandt, Nicira, Inc.
import mock
-import mox
import testtools
from neutron.agent.linux import ovs_lib
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):
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
# 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",
- "del-port", self.BR_NAME, pname],
- root_helper=self.root_helper)
-
- self.mox.ReplayAll()
self.br.delete_port(pname)
- self.mox.VerifyAll()
+ self.execute.assert_called_once_with(
+ ["ovs-vsctl", self.TO, "--", "--if-exists",
+ "del-port", self.BR_NAME, pname],
+ root_helper=self.root_helper)
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,
+
+ self.br.add_flow(priority=2, dl_src="ca:fe:de:ad:be:ef",
+ actions="strip_vlan,output:0")
+ self.br.add_flow(priority=1, actions="normal")
+ self.br.add_flow(priority=2, actions="drop")
+ self.br.add_flow(priority=2, in_port=ofport, actions="drop")
+
+ self.br.add_flow(priority=4, in_port=ofport, dl_vlan=vid,
+ actions="strip_vlan,set_tunnel:%s,normal" %
+ (lsw_id))
+ self.br.add_flow(priority=3, tun_id=lsw_id,
+ actions="mod_vlan_vid:%s,output:%s" %
+ (vid, ofport))
+ self.br.add_flow(priority=4, proto='arp', nw_src=cidr, actions='drop')
+ 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"],
- root_helper=self.root_helper)
- utils.execute(["ovs-ofctl", "add-flow", self.BR_NAME,
+ 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"],
- root_helper=self.root_helper)
- utils.execute(["ovs-ofctl", "add-flow", self.BR_NAME,
+ 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"],
- root_helper=self.root_helper)
- utils.execute(["ovs-ofctl", "add-flow", self.BR_NAME,
+ 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],
- root_helper=self.root_helper)
- utils.execute(["ovs-ofctl", "add-flow", self.BR_NAME,
+ 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)],
- root_helper=self.root_helper)
- utils.execute(["ovs-ofctl", "add-flow", self.BR_NAME,
+ 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)], root_helper=self.root_helper)
- utils.execute(["ovs-ofctl", "add-flow", self.BR_NAME,
+ % (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],
- 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")
- self.br.add_flow(priority=1, actions="normal")
- self.br.add_flow(priority=2, actions="drop")
- self.br.add_flow(priority=2, in_port=ofport, actions="drop")
-
- self.br.add_flow(priority=4, in_port=ofport, dl_vlan=vid,
- actions="strip_vlan,set_tunnel:%s,normal" %
- (lsw_id))
- self.br.add_flow(priority=3, tun_id=lsw_id,
- 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()
+ 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')
-
- 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()
+ 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']
+
+ 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')
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"
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",
- pname, "options:remote_ip=" + remote_ip],
- root_helper=self.root_helper)
- utils.execute(["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",
- pname, "options:in_key=flow"],
- root_helper=self.root_helper)
- utils.execute(["ovs-vsctl", self.TO, "set", "Interface",
- pname, "options:out_key=flow"],
- root_helper=self.root_helper)
- utils.execute(["ovs-vsctl", self.TO, "get",
- "Interface", pname, "ofport"],
- root_helper=self.root_helper).AndReturn(ofport)
- self.mox.ReplayAll()
+ # 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),
+ None),
+ (mock.call(["ovs-vsctl", self.TO, "set", "Interface",
+ pname, "options:local_ip=" + local_ip],
+ root_helper=self.root_helper),
+ None),
+ (mock.call(["ovs-vsctl", self.TO, "set", "Interface",
+ pname, "options:in_key=flow"],
+ root_helper=self.root_helper),
+ None),
+ (mock.call(["ovs-vsctl", self.TO, "set", "Interface",
+ pname, "options:out_key=flow"],
+ root_helper=self.root_helper),
+ None),
+ (mock.call(["ovs-vsctl", self.TO, "get",
+ "Interface", pname, "ofport"],
+ 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",
- "Interface", pname, "options:peer=" + peer],
- root_helper=self.root_helper)
- utils.execute(["ovs-vsctl", self.TO, "get",
- "Interface", pname, "ofport"],
- root_helper=self.root_helper).AndReturn(ofport)
- self.mox.ReplayAll()
+ # 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),
+ None),
+ (mock.call(["ovs-vsctl", self.TO, "get",
+ "Interface", pname, "ofport"],
+ 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"
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))
external_ids = ('{iface-id="%s", attached-mac="%s"}'
% (vif_id, mac))
- utils.execute(["ovs-vsctl", self.TO, "get",
- "Interface", pname, "external_ids"],
- root_helper=self.root_helper).AndReturn(external_ids)
- utils.execute(["ovs-vsctl", self.TO, "get",
- "Interface", pname, "ofport"],
- root_helper=self.root_helper).AndReturn(ofport)
+ # 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),
+ external_ids),
+ (mock.call(["ovs-vsctl", self.TO, "get",
+ "Interface", pname, "ofport"],
+ root_helper=self.root_helper),
+ ofport),
+ ]
if is_xen:
- utils.execute(["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()
+ 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),
+ vif_id)
+ )
+ tools.setup_mock_calls(self.execute, expected_calls_and_values)
ports = self.br.get_vif_ports()
self.assertEqual(1, len(ports))
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.
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:
['tun22', {}],
]
- utils.execute(["ovs-vsctl", self.TO, "--format=json",
- "--", "--columns=name,external_ids",
- "list", "Interface"],
- root_helper=self.root_helper).AndReturn(
- self._encode_ovs_json(headings, data))
+ # 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),
+ 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)
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",
- "--", "--columns=name,external_ids",
- "list", "Interface"],
- root_helper=self.root_helper).AndReturn(
- self._encode_ovs_json(['name', 'external_ids'], []))
- self.mox.ReplayAll()
+ 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),
+ self._encode_ovs_json(['name', 'external_ids'], []))
+ ]
+ 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",
- "--", "--columns=name,external_ids",
- "list", "Interface"],
- root_helper=self.root_helper).AndRaise(RuntimeError())
- self.mox.ReplayAll()
+ 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),
+ 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",'
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()
- self.br.delete_ports(all_ports=True)
- self.mox.VerifyAll()
+ 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)
+ 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()
- self.br.delete_ports(all_ports=False)
- self.mox.VerifyAll()
+ 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)
+ 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',
#
# @author: Dave Lapsley, Nicira Networks, Inc.
-import mox
+import contextlib
+
+import mock
from oslo.config import cfg
from neutron.agent.linux import ip_lib
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'
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.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.mock_map_tun_bridge = ovs_lib.OVSBridge(
- self.MAP_TUN_BRIDGE, 'sudo')
+ self.inta = mock.Mock()
+ self.intb = mock.Mock()
+
+ self.ovs_bridges = {self.INT_BRIDGE: mock.Mock(),
+ self.TUN_BRIDGE: mock.Mock(),
+ self.MAP_TUN_BRIDGE: mock.Mock(),
+ }
+
+ 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_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_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.remove_all_flows()
- self.mock_tun_bridge.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,
- dl_dst=UCAST_MAC,
- actions="resubmit(,%s)" %
- constants.UCAST_TO_TUN)
- self.mock_tun_bridge.add_flow(table=constants.PATCH_LV_TO_TUN,
- dl_dst=BCAST_MAC,
- actions="resubmit(,%s)" %
- constants.FLOOD_TO_TUN)
+ 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_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 = 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_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),
+ 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),
+ mock.call.add_flow(table=constants.PATCH_LV_TO_TUN,
+ dl_dst=BCAST_MAC,
+ actions="resubmit(,%s)" %
+ constants.FLOOD_TO_TUN),
+ ]
for tunnel_type in constants.TUNNEL_NETWORK_TYPES:
- self.mock_tun_bridge.add_flow(
- table=constants.TUN_TABLE[tunnel_type],
- priority=0,
- actions="drop")
+ self.mock_tun_bridge_expected.append(
+ mock.call.add_flow(
+ table=constants.TUN_TABLE[tunnel_type],
+ priority=0,
+ actions="drop"))
learned_flow = ("table=%s,"
"priority=1,"
"hard_timeout=300,"
"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,
- priority=1,
- actions="learn(%s),output:%s" %
- (learned_flow, self.INT_OFPORT))
- self.mock_tun_bridge.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,
- 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.mox.StubOutWithMock(ip_lib.IpLinkCommand, 'delete')
- ip_lib.IPDevice('int-tunnel_bridge_mapping').link.delete()
-
- 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.TUN_BRIDGE,
- self.MAP_TUN_BRIDGE])
+ 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)),
+ mock.call.add_flow(table=constants.UCAST_TO_TUN,
+ priority=0,
+ actions="resubmit(,%s)" %
+ constants.FLOOD_TO_TUN),
+ mock.call.add_flow(table=constants.FLOOD_TO_TUN,
+ priority=0,
+ actions="drop")
+ ]
+
+ 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.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.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.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()
- 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()
+ 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)
+ 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,
- 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'],
- priority=1,
- tun_id=LS_ID,
- actions="mod_vlan_vid:%s,resubmit(,%s)" %
- (LV_ID, constants.LEARN_FROM_TUN))
- self.mox.ReplayAll()
+ 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)),
+ 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)),
+ ]
a = ovs_neutron_agent.OVSNeutronAgent(self.INT_BRIDGE,
self.TUN_BRIDGE,
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,
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,
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,
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_int_bridge.delete_flows(
- dl_vlan=65535, in_port=self.INT_OFPORT)
+ 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.mox.ReplayAll()
a = ovs_neutron_agent.OVSNeutronAgent(self.INT_BRIDGE,
self.TUN_BRIDGE,
'10.0.0.1', self.NET_MAPPING,
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,
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,
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()
-
- 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.local_vlan_map[NET_UUID] = LVM
- a.port_unbound(VIF_ID, NET_UUID)
- self.mox.VerifyAll()
+ 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,
+ 'sudo', 2, ['gre'],
+ self.VETH_MTU)
+ a.local_vlan_map[NET_UUID] = LVM
+ a.port_unbound(VIF_ID, NET_UUID)
+
+ 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_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.mock_int_bridge.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,
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']),
'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'))
-
- 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',
- self.NET_MAPPING,
- 'sudo', 2, ['gre'],
- self.VETH_MTU)
-
- # Hack to test loop
- # We start method and expect it will raise after 2nd loop
- # If something goes wrong, mox.VerifyAll() will catch it
- try:
- q_agent.daemon_loop()
- except Exception:
- pass
-
- self.mox.VerifyAll()
+ 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')]
+
+ q_agent = ovs_neutron_agent.OVSNeutronAgent(self.INT_BRIDGE,
+ self.TUN_BRIDGE,
+ '10.0.0.1',
+ self.NET_MAPPING,
+ 'sudo', 2, ['gre'],
+ self.VETH_MTU)
+
+ # Hack to test loop
+ # We start method and expect it will raise after 2nd loop
+ # If something goes wrong, assert_has_calls below will catch it
+ try:
+ q_agent.daemon_loop()
+ except Exception:
+ pass
+
+ 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):
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))
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}
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,
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]}
'COMMIT\n'
'# Completed by iptables_manager\n' % iptables_args)
- self.iptables.execute(['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'],
- process_input=nat_dump + filter_dump,
- root_helper=self.root_helper).AndReturn(None)
-
- self.mox.ReplayAll()
+ 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),
+ 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),
+ None),
+ ]
+ tools.setup_mock_calls(self.execute, expected_calls_and_values)
self.iptables.ipv4['filter'].add_chain('filter')
self.iptables.apply()
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]
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}
'COMMIT\n'
'# Completed by iptables_manager\n' % iptables_args)
- self.iptables.execute(['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'],
- process_input=nat_dump + filter_dump,
- root_helper=self.root_helper).AndReturn(None)
-
- self.mox.ReplayAll()
+ 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),
+ 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),
+ 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',
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'
'# Completed by iptables_manager\n'
% IPTABLES_ARG)
- self.iptables.execute(['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'],
- process_input=NAT_DUMP + FILTER_DUMP,
- root_helper=self.root_helper).AndReturn(None)
-
- self.mox.ReplayAll()
+ 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),
+ 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),
+ None),
+ ]
+ tools.setup_mock_calls(self.execute, expected_calls_and_values)
self.iptables.ipv4['filter'].add_chain('filter')
self.iptables.apply()
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'
'# Completed by iptables_manager\n'
% IPTABLES_ARG)
- self.iptables.execute(['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'],
- process_input=NAT_DUMP + FILTER_DUMP,
- root_helper=self.root_helper
- ).AndReturn(None)
-
- self.mox.ReplayAll()
+ 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),
+ 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
+ ),
+ 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')
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'
'# 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'],
- 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'],
- process_input=nat_dump + FILTER_DUMP,
- root_helper=self.root_helper).AndReturn(None)
+ 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),
+ 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),
+ 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 '
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()
- self.iptables.ipv4['filter'].remove_chain('nonexistent')
- self.mox.VerifyAll()
+ with mock.patch.object(iptables_manager, "LOG") as log:
+ self.iptables.ipv4['filter'].remove_chain('nonexistent')
+ 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()
- self.iptables.ipv4['filter'].remove_rule('nonexistent', '-j DROP')
- self.mox.VerifyAll()
+ with mock.patch.object(iptables_manager, "LOG") as log:
+ self.iptables.ipv4['filter'].remove_rule('nonexistent', '-j DROP')
+ 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()
- acc = self.iptables.get_traffic_counters('chain1')
- self.assertIsNone(acc)
+ 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 = (
' 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()
+ expected_calls_and_values = [
+ (mock.call(['iptables', '-t', 'filter', '-L', 'OUTPUT',
+ '-n', '-v', '-x'],
+ root_helper=self.root_helper),
+ iptables_dump),
+ (mock.call(['iptables', '-t', 'nat', '-L', 'OUTPUT', '-n',
+ '-v', '-x'],
+ root_helper=self.root_helper),
+ ''),
+ (mock.call(['ip6tables', '-t', 'filter', '-L', 'OUTPUT',
+ '-n', '-v', '-x'],
+ root_helper=self.root_helper),
+ iptables_dump),
+ ]
+ tools.setup_mock_calls(self.execute, expected_calls_and_values)
+
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 = (
' 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', '-Z'],
- root_helper=self.root_helper
- ).AndReturn(iptables_dump)
- self.iptables.execute(['iptables', '-t', 'nat', '-L', 'OUTPUT', '-n',
- '-v', '-x', '-Z'],
- root_helper=self.root_helper
- ).AndReturn('')
-
- self.iptables.execute(['ip6tables', '-t', 'filter', '-L', 'OUTPUT',
- '-n', '-v', '-x', '-Z'],
- root_helper=self.root_helper
- ).AndReturn(iptables_dump)
-
- self.mox.ReplayAll()
+ expected_calls_and_values = [
+ (mock.call(['iptables', '-t', 'filter', '-L', 'OUTPUT',
+ '-n', '-v', '-x', '-Z'],
+ root_helper=self.root_helper),
+ iptables_dump),
+ (mock.call(['iptables', '-t', 'nat', '-L', 'OUTPUT', '-n',
+ '-v', '-x', '-Z'],
+ root_helper=self.root_helper),
+ ''),
+ (mock.call(['ip6tables', '-t', 'filter', '-L', 'OUTPUT',
+ '-n', '-v', '-x', '-Z'],
+ root_helper=self.root_helper),
+ iptables_dump),
+ ]
+ tools.setup_mock_calls(self.execute, expected_calls_and_values)
+
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):
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
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
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
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
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
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():
Code Review / openstack-build / neutron-build.git / commitdiff