tests/service: consolidate setUp/tearDown logic

Consolidate the duplicate code for extension api tests.
This will be also used by servicevm extensions later.

At the same time, a bug in test_extension_firewall.py is also fixed.
double call of super().setUp().

Closes-bug: #1275654
Change-Id: I72f185ef6e5af55729ced51fc0b72fae799d46ac
This commit is contained in:
Isaku Yamahata 2014-02-03 15:44:02 +09:00
parent b1243f6d6e
commit fdbaba877f
5 changed files with 159 additions and 266 deletions

View File

@ -18,83 +18,29 @@
import copy
import mock
from oslo.config import cfg
from webob import exc
import webtest
from neutron.api import extensions
from neutron.api.v2 import attributes as attr
from neutron.common import config
from neutron.extensions import loadbalancer
from neutron.openstack.common import uuidutils
from neutron.plugins.common import constants
from neutron import quota
from neutron.tests.unit import test_api_v2
from neutron.tests.unit import test_extensions
from neutron.tests.unit import testlib_api
from neutron.tests.unit import test_api_v2_extension
_uuid = uuidutils.generate_uuid
_get_path = test_api_v2._get_path
class LoadBalancerTestExtensionManager(object):
def get_resources(self):
# Add the resources to the global attribute map
# This is done here as the setup process won't
# initialize the main API router which extends
# the global attribute map
attr.RESOURCE_ATTRIBUTE_MAP.update(
loadbalancer.RESOURCE_ATTRIBUTE_MAP)
return loadbalancer.Loadbalancer.get_resources()
def get_actions(self):
return []
def get_request_extensions(self):
return []
class LoadBalancerExtensionTestCase(testlib_api.WebTestCase):
class LoadBalancerExtensionTestCase(test_api_v2_extension.ExtensionTestCase):
fmt = 'json'
def setUp(self):
super(LoadBalancerExtensionTestCase, self).setUp()
plugin = 'neutron.extensions.loadbalancer.LoadBalancerPluginBase'
# Ensure existing ExtensionManager is not used
extensions.PluginAwareExtensionManager._instance = None
# Create the default configurations
args = ['--config-file', test_api_v2.etcdir('neutron.conf.test')]
config.parse(args)
#just stubbing core plugin with LoadBalancer plugin
self.setup_coreplugin(plugin)
cfg.CONF.set_override('service_plugins', [plugin])
self._plugin_patcher = mock.patch(plugin, autospec=True)
self.plugin = self._plugin_patcher.start()
instance = self.plugin.return_value
instance.get_plugin_type.return_value = constants.LOADBALANCER
ext_mgr = LoadBalancerTestExtensionManager()
self.ext_mdw = test_extensions.setup_extensions_middleware(ext_mgr)
self.api = webtest.TestApp(self.ext_mdw)
quota.QUOTAS._driver = None
cfg.CONF.set_override('quota_driver', quota.QUOTA_CONF_DRIVER,
group='QUOTAS')
super(LoadBalancerExtensionTestCase, self).setUp()
def tearDown(self):
self._plugin_patcher.stop()
self.api = None
self.plugin = None
cfg.CONF.reset()
super(LoadBalancerExtensionTestCase, self).tearDown()
self._setUpExtension(
'neutron.extensions.loadbalancer.LoadBalancerPluginBase',
constants.LOADBALANCER, loadbalancer.RESOURCE_ATTRIBUTE_MAP,
loadbalancer.Loadbalancer, 'lb', use_quota=True)
def test_vip_create(self):
vip_id = _uuid()
@ -182,15 +128,6 @@ class LoadBalancerExtensionTestCase(testlib_api.WebTestCase):
self.assertIn('vip', res)
self.assertEqual(res['vip'], return_value)
def _test_entity_delete(self, entity):
"""Does the entity deletion based on naming convention."""
entity_id = _uuid()
res = self.api.delete(_get_path('lb/' + entity + 's', id=entity_id,
fmt=self.fmt))
delete_entity = getattr(self.plugin.return_value, "delete_" + entity)
delete_entity.assert_called_with(mock.ANY, entity_id)
self.assertEqual(res.status_int, exc.HTTPNoContent.code)
def test_vip_delete(self):
self._test_entity_delete('vip')

View File

@ -20,82 +20,32 @@
import copy
import mock
from oslo.config import cfg
from webob import exc
import webtest
from neutron.api import extensions
from neutron.api.v2 import attributes
from neutron.common import config
from neutron.extensions import vpnaas
from neutron.openstack.common import uuidutils
from neutron.plugins.common import constants
from neutron import quota
from neutron.tests.unit import test_api_v2
from neutron.tests.unit import test_extensions
from neutron.tests.unit import testlib_api
from neutron.tests.unit import test_api_v2_extension
_uuid = uuidutils.generate_uuid
_get_path = test_api_v2._get_path
class VpnaasTestExtensionManager(object):
def get_resources(self):
# Add the resources to the global attribute map
# This is done here as the setup process won't
# initialize the main API router which extends
# the global attribute map
attributes.RESOURCE_ATTRIBUTE_MAP.update(
vpnaas.RESOURCE_ATTRIBUTE_MAP)
return vpnaas.Vpnaas.get_resources()
def get_actions(self):
return []
def get_request_extensions(self):
return []
class VpnaasExtensionTestCase(testlib_api.WebTestCase):
class VpnaasExtensionTestCase(test_api_v2_extension.ExtensionTestCase):
fmt = 'json'
def setUp(self):
super(VpnaasExtensionTestCase, self).setUp()
plugin = 'neutron.extensions.vpnaas.VPNPluginBase'
# Ensure existing ExtensionManager is not used
extensions.PluginAwareExtensionManager._instance = None
# Create the default configurations
args = ['--config-file', test_api_v2.etcdir('neutron.conf.test')]
config.parse(args)
#just stubbing core plugin with LoadBalancer plugin
self.setup_coreplugin(plugin)
cfg.CONF.set_override('service_plugins', [plugin])
self._plugin_patcher = mock.patch(plugin, autospec=True)
self.plugin = self._plugin_patcher.start()
instance = self.plugin.return_value
instance.get_plugin_type.return_value = constants.VPN
ext_mgr = VpnaasTestExtensionManager()
self.ext_mdw = test_extensions.setup_extensions_middleware(ext_mgr)
self.api = webtest.TestApp(self.ext_mdw)
super(VpnaasExtensionTestCase, self).setUp()
quota.QUOTAS._driver = None
cfg.CONF.set_override('quota_driver', 'neutron.quota.ConfDriver',
group='QUOTAS')
def tearDown(self):
self._plugin_patcher.stop()
self.api = None
self.plugin = None
cfg.CONF.reset()
super(VpnaasExtensionTestCase, self).tearDown()
plural_mappings = {'ipsecpolicy': 'ipsecpolicies',
'ikepolicy': 'ikepolicies',
'ipsec_site_connection': 'ipsec-site-connections'}
self._setUpExtension(
'neutron.extensions.vpnaas.VPNPluginBase', constants.VPN,
vpnaas.RESOURCE_ATTRIBUTE_MAP, vpnaas.Vpnaas,
'vpn', plural_mappings=plural_mappings,
use_quota=True)
def test_ikepolicy_create(self):
"""Test case to create an ikepolicy."""
@ -422,20 +372,6 @@ class VpnaasExtensionTestCase(testlib_api.WebTestCase):
self.assertIn('vpnservice', res)
self.assertEqual(res['vpnservice'], return_value)
def _test_entity_delete(self, entity):
"""does the entity deletion based on naming convention."""
entity_id = _uuid()
path_map = {'ipsecpolicy': 'vpn/ipsecpolicies',
'ikepolicy': 'vpn/ikepolicies',
'ipsec_site_connection': 'vpn/ipsec-site-connections'}
path = path_map.get(entity, 'vpn/' + entity + 's')
res = self.api.delete(_get_path(path,
id=entity_id,
fmt=self.fmt))
delete_entity = getattr(self.plugin.return_value, "delete_" + entity)
delete_entity.assert_called_with(mock.ANY, entity_id)
self.assertEqual(res.status_int, exc.HTTPNoContent.code)
def test_vpnservice_delete(self):
"""Test case to delete a vpnservice."""
self._test_entity_delete('vpnservice')

View File

@ -0,0 +1,128 @@
# Copyright 2014 Intel Corporation.
# Copyright 2014 Isaku Yamahata <isaku.yamahata at intel com>
# <isaku.yamahata at gmail com>
# 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: Isaku Yamahata, Intel Corporation.
import uuid
import mock
from oslo.config import cfg
from webob import exc
import webtest
from neutron.api import extensions
from neutron.api.v2 import attributes
from neutron.common import config
from neutron import quota
from neutron.tests.unit import test_api_v2
from neutron.tests.unit import test_extensions
from neutron.tests.unit import testlib_api
class ExtensionTestCase(testlib_api.WebTestCase):
def _resotre_attr_map(self):
attributes.RESOURCE_ATTRIBUTE_MAP = self._saved_attr_map
def _setUpExtension(self, plugin, service_type,
resource_attribute_map, extension_class,
resource_prefix, plural_mappings=None,
translate_resource_name=False,
allow_pagination=False, allow_sorting=False,
supported_extension_aliases=None,
use_quota=False,
):
self.addCleanup(mock.patch.stopall)
self._resource_prefix = resource_prefix
self._plural_mappings = plural_mappings or {}
self._translate_resource_name = translate_resource_name
# Ensure existing ExtensionManager is not used
extensions.PluginAwareExtensionManager._instance = None
# Save the global RESOURCE_ATTRIBUTE_MAP
self._saved_attr_map = attributes.RESOURCE_ATTRIBUTE_MAP.copy()
# Restore the global RESOURCE_ATTRIBUTE_MAP
self.addCleanup(self._resotre_attr_map)
# Create the default configurations
args = ['--config-file', test_api_v2.etcdir('neutron.conf.test')]
config.parse(args)
#just stubbing core plugin with plugin
self.setup_coreplugin(plugin)
cfg.CONF.set_override('core_plugin', plugin)
if service_type:
cfg.CONF.set_override('service_plugins', [plugin])
self._plugin_patcher = mock.patch(plugin, autospec=True)
self.plugin = self._plugin_patcher.start()
instance = self.plugin.return_value
if service_type:
instance.get_plugin_type.return_value = service_type
if supported_extension_aliases is not None:
instance.supported_extension_aliases = supported_extension_aliases
if allow_pagination:
cfg.CONF.set_override('allow_pagination', True)
# instance.__native_pagination_support = True
native_pagination_attr_name = ("_%s__native_pagination_support"
% instance.__class__.__name__)
setattr(instance, native_pagination_attr_name, True)
if allow_sorting:
cfg.CONF.set_override('allow_sorting', True)
# instance.__native_sorting_support = True
native_sorting_attr_name = ("_%s__native_sorting_support"
% instance.__class__.__name__)
setattr(instance, native_sorting_attr_name, True)
if use_quota:
quota.QUOTAS._driver = None
cfg.CONF.set_override('quota_driver', 'neutron.quota.ConfDriver',
group='QUOTAS')
class ExtensionTestExtensionManager(object):
def get_resources(self):
# Add the resources to the global attribute map
# This is done here as the setup process won't
# initialize the main API router which extends
# the global attribute map
attributes.RESOURCE_ATTRIBUTE_MAP.update(
resource_attribute_map)
return extension_class.get_resources()
def get_actions(self):
return []
def get_request_extensions(self):
return []
ext_mgr = ExtensionTestExtensionManager()
self.ext_mdw = test_extensions.setup_extensions_middleware(ext_mgr)
self.api = webtest.TestApp(self.ext_mdw)
def _test_entity_delete(self, entity):
"""Does the entity deletion based on naming convention."""
entity_id = str(uuid.uuid4())
path = self._resource_prefix + '/' if self._resource_prefix else ''
path += self._plural_mappings.get(entity, entity + 's')
if self._translate_resource_name:
path = path.replace('_', '-')
res = self.api.delete(
test_api_v2._get_path(path, id=entity_id, fmt=self.fmt))
delete_entity = getattr(self.plugin.return_value, "delete_" + entity)
delete_entity.assert_called_with(mock.ANY, entity_id)
self.assertEqual(res.status_int, exc.HTTPNoContent.code)

View File

@ -20,94 +20,31 @@
import copy
import mock
from oslo.config import cfg
from webob import exc
import webtest
from neutron.api import extensions
from neutron.api.v2 import attributes
from neutron.common import config
from neutron.extensions import firewall
from neutron.openstack.common import uuidutils
from neutron.plugins.common import constants
from neutron.tests import base
from neutron.tests.unit import test_api_v2
from neutron.tests.unit import test_extensions
from neutron.tests.unit import testlib_api
from neutron.tests.unit import test_api_v2_extension
_uuid = uuidutils.generate_uuid
_get_path = test_api_v2._get_path
class FirewallTestExtensionManager(object):
def get_resources(self):
# Add the resources to the global attribute map
# This is done here as the setup process won't
# initialize the main API router which extends
# the global attribute map
attributes.RESOURCE_ATTRIBUTE_MAP.update(
firewall.RESOURCE_ATTRIBUTE_MAP)
return firewall.Firewall.get_resources()
def get_actions(self):
return []
def get_request_extensions(self):
return []
class FirewallExtensionTestCase(testlib_api.WebTestCase):
class FirewallExtensionTestCase(test_api_v2_extension.ExtensionTestCase):
fmt = 'json'
def setUp(self):
super(FirewallExtensionTestCase, self).setUp()
plugin = 'neutron.extensions.firewall.FirewallPluginBase'
# Ensure existing ExtensionManager is not used
extensions.PluginAwareExtensionManager._instance = None
# Create the default configurations
args = ['--config-file', test_api_v2.etcdir('neutron.conf.test')]
config.parse(args)
# Stubbing core plugin with Firewall plugin
self.setup_coreplugin(plugin)
cfg.CONF.set_override('service_plugins', [plugin])
self._plugin_patcher = mock.patch(plugin, autospec=True)
self.plugin = self._plugin_patcher.start()
instance = self.plugin.return_value
instance.get_plugin_type.return_value = constants.FIREWALL
ext_mgr = FirewallTestExtensionManager()
self.ext_mdw = test_extensions.setup_extensions_middleware(ext_mgr)
self.api = webtest.TestApp(self.ext_mdw)
super(FirewallExtensionTestCase, self).setUp()
def tearDown(self):
self._plugin_patcher.stop()
self.api = None
self.plugin = None
cfg.CONF.reset()
super(FirewallExtensionTestCase, self).tearDown()
def _test_entity_delete(self, entity):
"""Does the entity deletion based on naming convention."""
entity_id = _uuid()
path_prefix = 'fw/'
if entity == 'firewall_policy':
entity_plural = 'firewall_policies'
else:
entity_plural = entity + 's'
res = self.api.delete(_get_path(path_prefix + entity_plural,
id=entity_id, fmt=self.fmt))
delete_entity = getattr(self.plugin.return_value, "delete_" + entity)
delete_entity.assert_called_with(mock.ANY, entity_id)
self.assertEqual(res.status_int, exc.HTTPNoContent.code)
plural_mappings = {'firewall_policy': 'firewall_policies'}
self._setUpExtension(
'neutron.extensions.firewall.FirewallPluginBase',
constants.FIREWALL, firewall.RESOURCE_ATTRIBUTE_MAP,
firewall.Firewall, 'fw', plural_mappings=plural_mappings)
def test_create_firewall(self):
fw_id = _uuid()

View File

@ -25,11 +25,8 @@ import mock
import netaddr
from oslo.config import cfg
from webob import exc
import webtest
from neutron.api import extensions
from neutron.api.v2 import attributes
from neutron.common import config
from neutron.common import constants as l3_constants
from neutron.common import exceptions as q_exc
from neutron import context
@ -46,11 +43,9 @@ from neutron.openstack.common.notifier import api as notifier_api
from neutron.openstack.common.notifier import test_notifier
from neutron.openstack.common import uuidutils
from neutron.plugins.common import constants as service_constants
from neutron import quota
from neutron.tests.unit import test_api_v2
from neutron.tests.unit import test_api_v2_extension
from neutron.tests.unit import test_db_plugin
from neutron.tests.unit import test_extensions
from neutron.tests.unit import testlib_api
LOG = logging.getLogger(__name__)
@ -77,57 +72,17 @@ class L3TestExtensionManager(object):
return []
class L3NatExtensionTestCase(testlib_api.WebTestCase):
class L3NatExtensionTestCase(test_api_v2_extension.ExtensionTestCase):
fmt = 'json'
def setUp(self):
super(L3NatExtensionTestCase, self).setUp()
plugin = 'neutron.extensions.l3.RouterPluginBase'
# Ensure 'stale' patched copies of the plugin are never returned
NeutronManager._instance = None
# Ensure existing ExtensionManager is not used
extensions.PluginAwareExtensionManager._instance = None
# Save the global RESOURCE_ATTRIBUTE_MAP
self.saved_attr_map = {}
for resource, attrs in attributes.RESOURCE_ATTRIBUTE_MAP.iteritems():
self.saved_attr_map[resource] = attrs.copy()
# Create the default configurations
args = ['--config-file', test_api_v2.etcdir('neutron.conf.test')]
config.parse(args=args)
# Update the plugin and extensions path
cfg.CONF.set_override('core_plugin', plugin)
cfg.CONF.set_override('allow_pagination', True)
cfg.CONF.set_override('allow_sorting', True)
self._plugin_patcher = mock.patch(plugin, autospec=True)
self.plugin = self._plugin_patcher.start()
instances = self.plugin.return_value
instances._RouterPluginBase__native_pagination_support = True
instances._RouterPluginBase__native_sorting_support = True
# Enable the 'router' extension
instances.supported_extension_aliases = ["router"]
ext_mgr = L3TestExtensionManager()
self.ext_mdw = test_extensions.setup_extensions_middleware(ext_mgr)
self.api = webtest.TestApp(self.ext_mdw)
quota.QUOTAS._driver = None
cfg.CONF.set_override('quota_driver', 'neutron.quota.ConfDriver',
group='QUOTAS')
def tearDown(self):
self._plugin_patcher.stop()
self.api = None
self.plugin = None
cfg.CONF.reset()
# Restore the global RESOURCE_ATTRIBUTE_MAP
attributes.RESOURCE_ATTRIBUTE_MAP = self.saved_attr_map
super(L3NatExtensionTestCase, self).tearDown()
self._setUpExtension(
'neutron.extensions.l3.RouterPluginBase', None,
l3.RESOURCE_ATTRIBUTE_MAP, l3.L3, '',
allow_pagination=True, allow_sorting=True,
supported_extension_aliases=['router'],
use_quota=True)
def test_router_create(self):
router_id = _uuid()