Browse Source

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
changes/44/54044/9
Akihiro MOTOKI 9 years ago
parent
commit
11bd2dcdb9
  1. 47
      neutron/tests/tools.py
  2. 13
      neutron/tests/unit/metaplugin/test_metaplugin.py
  3. 158
      neutron/tests/unit/nec/test_pfc_driver.py
  4. 117
      neutron/tests/unit/nec/test_trema_driver.py
  5. 426
      neutron/tests/unit/openvswitch/test_ovs_lib.py
  6. 517
      neutron/tests/unit/openvswitch/test_ovs_tunnel.py
  7. 310
      neutron/tests/unit/test_iptables_manager.py
  8. 78
      neutron/tests/unit/test_security_groups_rpc.py

47
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)

13
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()

158
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'

117
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'

426
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,157 +135,141 @@ 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,
self.br.add_flow(priority=2, dl_src="ca:fe:de:ad:be:ef",
actions="strip_vlan,output:0")
self.br.add_flow(priority=1, actions="normal")
self.br.add_flow(priority=2, actions="drop")
self.br.add_flow(priority=2, in_port=ofport, actions="drop")
self.br.add_flow(priority=4, in_port=ofport, dl_vlan=vid,
actions="strip_vlan,set_tunnel:%s,normal" %
(lsw_id))
self.br.add_flow(priority=3, tun_id=lsw_id,
actions="mod_vlan_vid:%s,output:%s" %
(vid, ofport))
self.br.add_flow(priority=4, proto='arp', nw_src=cidr, actions='drop')
expected_calls = [
mock.call(["ovs-ofctl", "add-flow", self.BR_NAME,
"hard_timeout=0,idle_timeout=0,"
"priority=2,dl_src=ca:fe:de:ad:be:ef"
",actions=strip_vlan,output:0"],
root_helper=self.root_helper)
utils.execute(["ovs-ofctl", "add-flow", self.BR_NAME,
process_input=None, root_helper=self.root_helper),
mock.call(["ovs-ofctl", "add-flow", self.BR_NAME,
"hard_timeout=0,idle_timeout=0,"
"priority=1,actions=normal"],
root_helper=self.root_helper)
utils.execute(["ovs-ofctl", "add-flow", self.BR_NAME,
process_input=None, root_helper=self.root_helper),
mock.call(["ovs-ofctl", "add-flow", self.BR_NAME,
"hard_timeout=0,idle_timeout=0,"
"priority=2,actions=drop"],
root_helper=self.root_helper)
utils.execute(["ovs-ofctl", "add-flow", self.BR_NAME,
process_input=None, root_helper=self.root_helper),
mock.call(["ovs-ofctl", "add-flow", self.BR_NAME,
"hard_timeout=0,idle_timeout=0,"
"priority=2,in_port=%s,actions=drop" % ofport],
root_helper=self.root_helper)
utils.execute(["ovs-ofctl", "add-flow", self.BR_NAME,
process_input=None, root_helper=self.root_helper),
mock.call(["ovs-ofctl", "add-flow", self.BR_NAME,
"hard_timeout=0,idle_timeout=0,"
"priority=4,in_port=%s,dl_vlan=%s,"
"actions=strip_vlan,set_tunnel:%s,normal"
% (ofport, vid, lsw_id)],
root_helper=self.root_helper)
utils.execute(["ovs-ofctl", "add-flow", self.BR_NAME,
process_input=None, root_helper=self.root_helper),
mock.call(["ovs-ofctl", "add-flow", self.BR_NAME,
"hard_timeout=0,idle_timeout=0,"
"priority=3,tun_id=%s,actions="
"mod_vlan_vid:%s,output:%s"
% (lsw_id, vid, ofport)], root_helper=self.root_helper)
utils.execute(["ovs-ofctl", "add-flow", self.BR_NAME,
% (lsw_id, vid, ofport)],
process_input=None, root_helper=self.root_helper),
mock.call(["ovs-ofctl", "add-flow", self.BR_NAME,
"hard_timeout=0,idle_timeout=0,"
"priority=4,arp,nw_src=%s,actions=drop" % cidr],
root_helper=self.root_helper)
self.mox.ReplayAll()
self.br.add_flow(priority=2, dl_src="ca:fe:de:ad:be:ef",
actions="strip_vlan,output:0")
self.br.add_flow(priority=1, actions="normal")
self.br.add_flow(priority=2, actions="drop")
self.br.add_flow(priority=2, in_port=ofport, actions="drop")
self.br.add_flow(priority=4, in_port=ofport, dl_vlan=vid,
actions="strip_vlan,set_tunnel:%s,normal" %
(lsw_id))
self.br.add_flow(priority=3, tun_id=lsw_id,
actions="mod_vlan_vid:%s,output:%s" %
(vid, ofport))
self.br.add_flow(priority=4, proto='arp', nw_src=cidr, actions='drop')
self.mox.VerifyAll()
process_input=None, root_helper=self.root_helper),
]
self.execute.assert_has_calls(expected_calls)
def test_get_port_ofport(self):
pname = "tap99"
ofport = "6"
utils.execute(["ovs-vsctl", self.TO, "get",
"Interface", pname, "ofport"],
root_helper=self.root_helper).AndReturn(ofport)
self.mox.ReplayAll()
self.execute.return_value = ofport
self.assertEqual(self.br.get_port_ofport(pname), ofport)
self.mox.VerifyAll()
self.execute.assert_called_once_with(
["ovs-vsctl", self.TO, "get", "Interface", pname, "ofport"],
root_helper=self.root_helper)
def test_get_datapath_id(self):
datapath_id = '"0000b67f4fbcc149"'
utils.execute(["ovs-vsctl", self.TO, "get",
"Bridge", self.BR_NAME, "datapath_id"],
root_helper=self.root_helper).AndReturn(datapath_id)
self.mox.ReplayAll()
self.execute.return_value = datapath_id
self.assertEqual(self.br.get_datapath_id(), datapath_id.strip('"'))
self.mox.VerifyAll()
self.execute.assert_called_once_with(
["ovs-vsctl", self.TO, "get",
"Bridge", self.BR_NAME, "datapath_id"],
root_helper=self.root_helper)
def test_count_flows(self):
utils.execute(["ovs-ofctl", "dump-flows", self.BR_NAME],
root_helper=self.root_helper,
process_input=None).AndReturn('ignore\nflow-1\n')
self.mox.ReplayAll()
self.execute.return_value = 'ignore\nflow-1\n'
# counts the number of flows as total lines of output - 2
self.assertEqual(self.br.count_flows(), 1)
self.mox.VerifyAll()
self.execute.assert_called_once_with(
["ovs-ofctl", "dump-flows", self.BR_NAME],
root_helper=self.root_helper,
process_input=None)
def test_delete_flow(self):
ofport = "5"
lsw_id = 40
vid = 39
utils.execute(["ovs-ofctl", "del-flows", self.BR_NAME,
"in_port=" + ofport], root_helper=self.root_helper)
utils.execute(["ovs-ofctl", "del-flows", self.BR_NAME,
"tun_id=%s" % lsw_id], root_helper=self.root_helper)
utils.execute(["ovs-ofctl", "del-flows", self.BR_NAME,
"dl_vlan=%s" % vid], root_helper=self.root_helper)
self.mox.ReplayAll()
self.br.delete_flows(in_port=ofport)
self.br.delete_flows(tun_id=lsw_id)
self.br.delete_flows(dl_vlan=vid)
self.mox.VerifyAll()
expected_calls = [
mock.call(["ovs-ofctl", "del-flows", self.BR_NAME,
"in_port=" + ofport],
process_input=None, root_helper=self.root_helper),
mock.call(["ovs-ofctl", "del-flows", self.BR_NAME,
"tun_id=%s" % lsw_id],
process_input=None, root_helper=self.root_helper),
mock.call(["ovs-ofctl", "del-flows", self.BR_NAME,
"dl_vlan=%s" % vid],
process_input=None, root_helper=self.root_helper),
]
self.execute.assert_has_calls(expected_calls)
def test_defer_apply_flows(self):
self.mox.StubOutWithMock(self.br, 'add_or_mod_flow_str')
self.br.add_or_mod_flow_str(
flow='added_flow_1').AndReturn('added_flow_1')
self.br.add_or_mod_flow_str(
flow='added_flow_2').AndReturn('added_flow_2')
self.mox.StubOutWithMock(self.br, '_build_flow_expr_arr')
self.br._build_flow_expr_arr(delete=True,
flow='deleted_flow_1'
).AndReturn(['deleted_flow_1'])
self.mox.StubOutWithMock(self.br, 'run_ofctl')
self.br.run_ofctl('add-flows', ['-'], 'added_flow_1\nadded_flow_2\n')
self.br.run_ofctl('del-flows', ['-'], 'deleted_flow_1\n')
self.mox.ReplayAll()
add_mod_flow = mock.patch.object(self.br,
'add_or_mod_flow_str').start()
add_mod_flow.side_effect = ['added_flow_1', 'added_flow_2']
flow_expr = mock.patch.object(self.br, '_build_flow_expr_arr').start()
flow_expr.return_value = ['deleted_flow_1']
run_ofctl = mock.patch.object(self.br, 'run_ofctl').start()
self.br.defer_apply_on()
self.br.add_flow(flow='added_flow_1')
@ -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)