From 11bd2dcdb92327af3de2a61cc9af9ffe69f9b6a3 Mon Sep 17 00:00:00 2001 From: Akihiro MOTOKI Date: Sun, 27 Oct 2013 22:42:12 +0900 Subject: [PATCH] Replace mox in unit tests with mock blueprint remove-mox Removing mox from test-requirements.txt will be done in another patch because stubout module is a part of mox and it is used in several modules. Change-Id: I9370e8da815d2234a032a8875da73a86d4126d46 --- neutron/tests/tools.py | 47 ++ .../tests/unit/metaplugin/test_metaplugin.py | 13 +- neutron/tests/unit/nec/test_pfc_driver.py | 158 +++--- neutron/tests/unit/nec/test_trema_driver.py | 117 ++--- .../tests/unit/openvswitch/test_ovs_lib.py | 432 ++++++++------- .../tests/unit/openvswitch/test_ovs_tunnel.py | 491 ++++++++++-------- neutron/tests/unit/test_iptables_manager.py | 306 ++++++----- .../tests/unit/test_security_groups_rpc.py | 78 ++- 8 files changed, 857 insertions(+), 785 deletions(-) create mode 100644 neutron/tests/tools.py diff --git a/neutron/tests/tools.py b/neutron/tests/tools.py new file mode 100644 index 00000000000..f80b3050892 --- /dev/null +++ b/neutron/tests/tools.py @@ -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) diff --git a/neutron/tests/unit/metaplugin/test_metaplugin.py b/neutron/tests/unit/metaplugin/test_metaplugin.py index 9edff614202..14f94881979 100644 --- a/neutron/tests/unit/metaplugin/test_metaplugin.py +++ b/neutron/tests/unit/metaplugin/test_metaplugin.py @@ -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() diff --git a/neutron/tests/unit/nec/test_pfc_driver.py b/neutron/tests/unit/nec/test_pfc_driver.py index f45e0bf5c7f..ba517a05f81 100644 --- a/neutron/tests/unit/nec/test_pfc_driver.py +++ b/neutron/tests/unit/nec/test_pfc_driver.py @@ -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' diff --git a/neutron/tests/unit/nec/test_trema_driver.py b/neutron/tests/unit/nec/test_trema_driver.py index c6710ef1b41..f08beca17e6 100644 --- a/neutron/tests/unit/nec/test_trema_driver.py +++ b/neutron/tests/unit/nec/test_trema_driver.py @@ -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' diff --git a/neutron/tests/unit/openvswitch/test_ovs_lib.py b/neutron/tests/unit/openvswitch/test_ovs_lib.py index 415cd14a961..f313a12db33 100644 --- a/neutron/tests/unit/openvswitch/test_ovs_lib.py +++ b/neutron/tests/unit/openvswitch/test_ovs_lib.py @@ -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", - "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, - "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", - 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" @@ -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", - "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)) @@ -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", - "--", "--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) @@ -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", - "--", "--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",' @@ -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() - 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', diff --git a/neutron/tests/unit/openvswitch/test_ovs_tunnel.py b/neutron/tests/unit/openvswitch/test_ovs_tunnel.py index 3715fb07bdc..5c571322ffd 100644 --- a/neutron/tests/unit/openvswitch/test_ovs_tunnel.py +++ b/neutron/tests/unit/openvswitch/test_ovs_tunnel.py @@ -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, - 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_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," @@ -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, - 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.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.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.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') + ] - self.mox.StubOutWithMock(ovs_lib, 'get_bridges') - ovs_lib.get_bridges('sudo').AndReturn([self.INT_BRIDGE, - self.TUN_BRIDGE, - self.MAP_TUN_BRIDGE]) + 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, @@ -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,32 +424,31 @@ 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) + 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) - 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() + 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, @@ -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,44 +498,48 @@ 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', - self.NET_MAPPING, - 'sudo', 2, ['gre'], - self.VETH_MTU) + 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 + # 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 - 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)) diff --git a/neutron/tests/unit/test_iptables_manager.py b/neutron/tests/unit/test_iptables_manager.py index fda9a3e4710..be05ee82966 100644 --- a/neutron/tests/unit/test_iptables_manager.py +++ b/neutron/tests/unit/test_iptables_manager.py @@ -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'], - 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() @@ -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'], - 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', @@ -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'], - 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() @@ -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'], - 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') @@ -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('') + 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.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) - - 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() - 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 = ( @@ -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', - '-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('') + 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) - 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('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', - '-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('') + 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) - self.iptables.execute(['ip6tables', '-t', 'filter', '-L', 'OUTPUT', - '-n', '-v', '-x', '-Z'], - root_helper=self.root_helper - ).AndReturn(iptables_dump) - - 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): diff --git a/neutron/tests/unit/test_security_groups_rpc.py b/neutron/tests/unit/test_security_groups_rpc.py index 9952fb06816..d0256ebf689 100644 --- a/neutron/tests/unit/test_security_groups_rpc.py +++ b/neutron/tests/unit/test_security_groups_rpc.py @@ -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():