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()
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')
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."""
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')
--- /dev/null
+# 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)
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()
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
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__)
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()
Code Review / openstack-build / neutron-build.git / commitdiff