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

neutron/tests/unit/services/loadbalancer/test_loadbalancer_plugin.py

@@ -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 test_api_v2_extension
-from neutron.tests.unit import testlib_api
 
 
 _uuid = uuidutils.generate_uuid
 _get_path = test_api_v2._get_path
 
 
-class LoadBalancerTestExtensionManager(object):
+class LoadBalancerExtensionTestCase(test_api_v2_extension.ExtensionTestCase):
-
-    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):
     fmt = 'json'
 
     def setUp(self):
         super(LoadBalancerExtensionTestCase, self).setUp()
-        plugin = 'neutron.extensions.loadbalancer.LoadBalancerPluginBase'
+        self._setUpExtension(
-
+            'neutron.extensions.loadbalancer.LoadBalancerPluginBase',
-        # Ensure existing ExtensionManager is not used
+            constants.LOADBALANCER, loadbalancer.RESOURCE_ATTRIBUTE_MAP,
-        extensions.PluginAwareExtensionManager._instance = None
+            loadbalancer.Loadbalancer, 'lb', use_quota=True)
-
-        # 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()
 
     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')
 

neutron/tests/unit/services/vpn/test_vpnaas_extension.py

@@ -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 test_api_v2_extension
-from neutron.tests.unit import testlib_api
 
 
 _uuid = uuidutils.generate_uuid
 _get_path = test_api_v2._get_path
 
 
-class VpnaasTestExtensionManager(object):
+class VpnaasExtensionTestCase(test_api_v2_extension.ExtensionTestCase):
-
-    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):
     fmt = 'json'
 
     def setUp(self):
         super(VpnaasExtensionTestCase, self).setUp()
-        plugin = 'neutron.extensions.vpnaas.VPNPluginBase'
+        plural_mappings = {'ipsecpolicy': 'ipsecpolicies',
-
+                           'ikepolicy': 'ikepolicies',
-        # Ensure existing ExtensionManager is not used
+                           'ipsec_site_connection': 'ipsec-site-connections'}
-        extensions.PluginAwareExtensionManager._instance = None
+        self._setUpExtension(
-
+            'neutron.extensions.vpnaas.VPNPluginBase', constants.VPN,
-        # Create the default configurations
+            vpnaas.RESOURCE_ATTRIBUTE_MAP, vpnaas.Vpnaas,
-        args = ['--config-file', test_api_v2.etcdir('neutron.conf.test')]
+            'vpn', plural_mappings=plural_mappings,
-        config.parse(args)
+            use_quota=True)
-
-        #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()
 
     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')

neutron/tests/unit/test_api_v2_extension.py

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

neutron/tests/unit/test_extension_firewall.py

@@ -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 test_api_v2_extension
-from neutron.tests.unit import testlib_api
 
 
 _uuid = uuidutils.generate_uuid
 _get_path = test_api_v2._get_path
 
 
-class FirewallTestExtensionManager(object):
+class FirewallExtensionTestCase(test_api_v2_extension.ExtensionTestCase):
-
-    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):
     fmt = 'json'
 
     def setUp(self):
         super(FirewallExtensionTestCase, self).setUp()
-        plugin = 'neutron.extensions.firewall.FirewallPluginBase'
+        plural_mappings = {'firewall_policy': 'firewall_policies'}
-
+        self._setUpExtension(
-        # Ensure existing ExtensionManager is not used
+            'neutron.extensions.firewall.FirewallPluginBase',
-        extensions.PluginAwareExtensionManager._instance = None
+            constants.FIREWALL, firewall.RESOURCE_ATTRIBUTE_MAP,
-
+            firewall.Firewall, 'fw', plural_mappings=plural_mappings)
-        # 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)
 
     def test_create_firewall(self):
         fw_id = _uuid()

neutron/tests/unit/test_l3_plugin.py

@@ -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()
-
+        self._setUpExtension(
-        plugin = 'neutron.extensions.l3.RouterPluginBase'
+            'neutron.extensions.l3.RouterPluginBase', None,
-        # Ensure 'stale' patched copies of the plugin are never returned
+            l3.RESOURCE_ATTRIBUTE_MAP, l3.L3, '',
-        NeutronManager._instance = None
+            allow_pagination=True, allow_sorting=True,
-
+            supported_extension_aliases=['router'],
-        # Ensure existing ExtensionManager is not used
+            use_quota=True)
-        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()
 
     def test_router_create(self):
         router_id = _uuid()