# Port the bind the API server to
bind_port = 9696
- [app:quantum]
- paste.app_factory = quantum.l2Network.service:app_factory
+
+# Path to the extensions
+api_extensions_path = unit/extensions
+
+[pipeline:extensions_app_with_filter]
+pipeline = extensions extensions_test_app
+
+[filter:extensions]
+paste.filter_factory = quantum.common.extensions:ExtensionMiddleware.factory
+
+[app:extensions_test_app]
+paste.app_factory = tests.unit.test_extensions:app_factory
+
++[composite:quantum]
++use = egg:Paste#urlmap
++/: quantumversions
++/v0.1: quantumapi
+
++[pipeline:quantumapi]
++pipeline = extensions quantumapiapp
++
++[filter:extensions]
++paste.filter_factory = quantum.common.extensions:ExtensionMiddleware.factory
++
++[app:quantumversions]
++paste.app_factory = quantum.api.versions:Versions.factory
++
++[app:quantumapiapp]
++paste.app_factory = quantum.api:APIRouterV01.factory
Routes requests on the Quantum API to the appropriate controller
"""
- def __init__(self, ext_mgr=None):
+ def __init__(self, options=None):
mapper = routes.Mapper()
- self._setup_routes(mapper)
+ self._setup_routes(mapper, options)
super(APIRouterV01, self).__init__(mapper)
- def _setup_routes(self, mapper):
+ def _setup_routes(self, mapper, options):
# Loads the quantum plugin
- plugin = manager.QuantumManager.get_plugin()
+ plugin = manager.QuantumManager(options).get_plugin()
++
uri_prefix = '/tenants/{tenant_id}/'
mapper.resource('network', 'networks',
controller=networks.Controller(plugin),
--- /dev/null
- or PluginAwareExtensionManager(
+
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+
+# Copyright 2011 OpenStack LLC.
+# Copyright 2011 Justin Santa Barbara
+# 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.
+
+import imp
+import os
+import routes
+import logging
+import webob.dec
+import webob.exc
+from gettext import gettext as _
+from abc import ABCMeta, abstractmethod
+
+from quantum.manager import QuantumManager
+from quantum.common import exceptions
+from quantum.common import wsgi
+
+LOG = logging.getLogger('quantum.common.extensions')
+
+
+class PluginInterface(object):
+ __metaclass__ = ABCMeta
+
+ @classmethod
+ def __subclasshook__(cls, klass):
+ """
+ The __subclasshook__ method is a class method
+ that will be called everytime a class is tested
+ using issubclass(klass, PluginInterface).
+ In that case, it will check that every method
+ marked with the abstractmethod decorator is
+ provided by the plugin class.
+ """
+ for method in cls.__abstractmethods__:
+ if any(method in base.__dict__ for base in klass.__mro__):
+ continue
+ return NotImplemented
+ return True
+
+
+class ExtensionDescriptor(object):
+ """Base class that defines the contract for extensions.
+
+ Note that you don't have to derive from this class to have a valid
+ extension; it is purely a convenience.
+
+ """
+
+ def get_name(self):
+ """The name of the extension.
+
+ e.g. 'Fox In Socks'
+
+ """
+ raise NotImplementedError()
+
+ def get_alias(self):
+ """The alias for the extension.
+
+ e.g. 'FOXNSOX'
+
+ """
+ raise NotImplementedError()
+
+ def get_description(self):
+ """Friendly description for the extension.
+
+ e.g. 'The Fox In Socks Extension'
+
+ """
+ raise NotImplementedError()
+
+ def get_namespace(self):
+ """The XML namespace for the extension.
+
+ e.g. 'http://www.fox.in.socks/api/ext/pie/v1.0'
+
+ """
+ raise NotImplementedError()
+
+ def get_updated(self):
+ """The timestamp when the extension was last updated.
+
+ e.g. '2011-01-22T13:25:27-06:00'
+
+ """
+ # NOTE(justinsb): Not sure of the purpose of this is, vs the XML NS
+ raise NotImplementedError()
+
+ def get_resources(self):
+ """List of extensions.ResourceExtension extension objects.
+
+ Resources define new nouns, and are accessible through URLs.
+
+ """
+ resources = []
+ return resources
+
+ def get_actions(self):
+ """List of extensions.ActionExtension extension objects.
+
+ Actions are verbs callable from the API.
+
+ """
+ actions = []
+ return actions
+
+ def get_request_extensions(self):
+ """List of extensions.RequestException extension objects.
+
+ Request extensions are used to handle custom request data.
+
+ """
+ request_exts = []
+ return request_exts
+
+ def get_plugin_interface(self):
+ """
+ Returns an abstract class which defines contract for the plugin.
+ The abstract class should inherit from extesnions.PluginInterface,
+ Methods in this abstract class should be decorated as abstractmethod
+ """
+ return None
+
+
+class ActionExtensionController(wsgi.Controller):
+
+ def __init__(self, application):
+
+ self.application = application
+ self.action_handlers = {}
+
+ def add_action(self, action_name, handler):
+ self.action_handlers[action_name] = handler
+
+ def action(self, request, id):
+
+ input_dict = self._deserialize(request.body,
+ request.get_content_type())
+ for action_name, handler in self.action_handlers.iteritems():
+ if action_name in input_dict:
+ return handler(input_dict, request, id)
+ # no action handler found (bump to downstream application)
+ response = self.application
+ return response
+
+
+class RequestExtensionController(wsgi.Controller):
+
+ def __init__(self, application):
+ self.application = application
+ self.handlers = []
+
+ def add_handler(self, handler):
+ self.handlers.append(handler)
+
+ def process(self, request, *args, **kwargs):
+ res = request.get_response(self.application)
+ # currently request handlers are un-ordered
+ for handler in self.handlers:
+ response = handler(request, res)
+ return response
+
+
+class ExtensionController(wsgi.Controller):
+
+ def __init__(self, extension_manager):
+ self.extension_manager = extension_manager
+
+ def _translate(self, ext):
+ ext_data = {}
+ ext_data['name'] = ext.get_name()
+ ext_data['alias'] = ext.get_alias()
+ ext_data['description'] = ext.get_description()
+ ext_data['namespace'] = ext.get_namespace()
+ ext_data['updated'] = ext.get_updated()
+ ext_data['links'] = [] # TODO(dprince): implement extension links
+ return ext_data
+
+ def index(self, request):
+ extensions = []
+ for _alias, ext in self.extension_manager.extensions.iteritems():
+ extensions.append(self._translate(ext))
+ return dict(extensions=extensions)
+
+ def show(self, request, id):
+ # NOTE(dprince): the extensions alias is used as the 'id' for show
+ ext = self.extension_manager.extensions[id]
+ return self._translate(ext)
+
+ def delete(self, request, id):
+ raise webob.exc.HTTPNotFound()
+
+ def create(self, request):
+ raise webob.exc.HTTPNotFound()
+
+
+class ExtensionMiddleware(wsgi.Middleware):
+ """Extensions middleware for WSGI."""
+ def __init__(self, application, config_params,
+ ext_mgr=None):
+
+ self.ext_mgr = (ext_mgr
-
++ or ExtensionManager(
+ config_params.get('api_extensions_path', '')))
- def __init__(self, path):
- self.plugin = QuantumManager.get_plugin()
+ mapper = routes.Mapper()
+
+ # extended resources
+ for resource in self.ext_mgr.get_resources():
+ LOG.debug(_('Extended resource: %s'),
+ resource.collection)
+ mapper.resource(resource.collection, resource.collection,
+ controller=resource.controller,
+ collection=resource.collection_actions,
+ member=resource.member_actions,
+ parent_resource=resource.parent)
+
+ # extended actions
+ action_controllers = self._action_ext_controllers(application,
+ self.ext_mgr, mapper)
+ for action in self.ext_mgr.get_actions():
+ LOG.debug(_('Extended action: %s'), action.action_name)
+ controller = action_controllers[action.collection]
+ controller.add_action(action.action_name, action.handler)
+
+ # extended requests
+ req_controllers = self._request_ext_controllers(application,
+ self.ext_mgr, mapper)
+ for request_ext in self.ext_mgr.get_request_extensions():
+ LOG.debug(_('Extended request: %s'), request_ext.key)
+ controller = req_controllers[request_ext.key]
+ controller.add_handler(request_ext.handler)
+
+ self._router = routes.middleware.RoutesMiddleware(self._dispatch,
+ mapper)
+
+ super(ExtensionMiddleware, self).__init__(application)
+
+ @classmethod
+ def factory(cls, global_config, **local_config):
+ """Paste factory."""
+ def _factory(app):
+ return cls(app, global_config, **local_config)
+ return _factory
+
+ def _action_ext_controllers(self, application, ext_mgr, mapper):
+ """Return a dict of ActionExtensionController-s by collection."""
+ action_controllers = {}
+ for action in ext_mgr.get_actions():
+ if not action.collection in action_controllers.keys():
+ controller = ActionExtensionController(application)
+ mapper.connect("/%s/:(id)/action.:(format)" %
+ action.collection,
+ action='action',
+ controller=controller,
+ conditions=dict(method=['POST']))
+ mapper.connect("/%s/:(id)/action" % action.collection,
+ action='action',
+ controller=controller,
+ conditions=dict(method=['POST']))
+ action_controllers[action.collection] = controller
+
+ return action_controllers
+
+ def _request_ext_controllers(self, application, ext_mgr, mapper):
+ """Returns a dict of RequestExtensionController-s by collection."""
+ request_ext_controllers = {}
+ for req_ext in ext_mgr.get_request_extensions():
+ if not req_ext.key in request_ext_controllers.keys():
+ controller = RequestExtensionController(application)
+ mapper.connect(req_ext.url_route + '.:(format)',
+ action='process',
+ controller=controller,
+ conditions=req_ext.conditions)
+
+ mapper.connect(req_ext.url_route,
+ action='process',
+ controller=controller,
+ conditions=req_ext.conditions)
+ request_ext_controllers[req_ext.key] = controller
+
+ return request_ext_controllers
+
+ @webob.dec.wsgify(RequestClass=wsgi.Request)
+ def __call__(self, req):
+ """Route the incoming request with router."""
+ req.environ['extended.app'] = self.application
+ return self._router
+
+ @staticmethod
+ @webob.dec.wsgify(RequestClass=wsgi.Request)
+ def _dispatch(req):
+ """Dispatch the request.
+
+ Returns the routed WSGI app's response or defers to the extended
+ application.
+
+ """
+ match = req.environ['wsgiorg.routing_args'][1]
+ if not match:
+ return req.environ['extended.app']
+ app = match['controller']
+ return app
+
+
++class PluginAwareExtensionMiddleware(ExtensionMiddleware):
++
++ def __init__(self, application, config_params, ext_mgr=None,
++ plugin_options=None):
++ plugin_aware_extension_mgr = PluginAwareExtensionManager(
++ config_params.get('api_extensions_path', ''),
++ plugin_options)
++ ext_mgr = (ext_mgr or plugin_aware_extension_mgr)
++ super(PluginAwareExtensionMiddleware, self).__init__(
++ application, config_params, ext_mgr)
++
++
+class ExtensionManager(object):
+ """Load extensions from the configured extension path.
+
+ See tests/unit/extensions/foxinsocks.py for an
+ example extension implementation.
+
+ """
+ def __init__(self, path):
+ LOG.info(_('Initializing extension manager.'))
+ self.path = path
+ self.extensions = {}
+ self._load_all_extensions()
+
+ def get_resources(self):
+ """Returns a list of ResourceExtension objects."""
+ resources = []
+ resources.append(ResourceExtension('extensions',
+ ExtensionController(self)))
+ for alias, ext in self.extensions.iteritems():
+ try:
+ resources.extend(ext.get_resources())
+ except AttributeError:
+ # NOTE(dprince): Extension aren't required to have resource
+ # extensions
+ pass
+ return resources
+
+ def get_actions(self):
+ """Returns a list of ActionExtension objects."""
+ actions = []
+ for alias, ext in self.extensions.iteritems():
+ try:
+ actions.extend(ext.get_actions())
+ except AttributeError:
+ # NOTE(dprince): Extension aren't required to have action
+ # extensions
+ pass
+ return actions
+
+ def get_request_extensions(self):
+ """Returns a list of RequestExtension objects."""
+ request_exts = []
+ for alias, ext in self.extensions.iteritems():
+ try:
+ request_exts.extend(ext.get_request_extensions())
+ except AttributeError:
+ # NOTE(dprince): Extension aren't required to have request
+ # extensions
+ pass
+ return request_exts
+
+ def _check_extension(self, extension):
+ """Checks for required methods in extension objects."""
+ try:
+ LOG.debug(_('Ext name: %s'), extension.get_name())
+ LOG.debug(_('Ext alias: %s'), extension.get_alias())
+ LOG.debug(_('Ext description: %s'), extension.get_description())
+ LOG.debug(_('Ext namespace: %s'), extension.get_namespace())
+ LOG.debug(_('Ext updated: %s'), extension.get_updated())
+ except AttributeError as ex:
+ LOG.exception(_("Exception loading extension: %s"), unicode(ex))
+ return False
+ return True
+
+ def _load_all_extensions(self):
+ """Load extensions from the configured path.
+
+ Load extensions from the configured path. The extension name is
+ constructed from the module_name. If your extension module was named
+ widgets.py the extension class within that module should be
+ 'Widgets'.
+
+ In addition, extensions are loaded from the 'contrib' directory.
+
+ See tests/unit/extensions/foxinsocks.py for an example
+ extension implementation.
+
+ """
+ if os.path.exists(self.path):
+ self._load_all_extensions_from_path(self.path)
+
+ contrib_path = os.path.join(os.path.dirname(__file__), "contrib")
+ if os.path.exists(contrib_path):
+ self._load_all_extensions_from_path(contrib_path)
+
+ def _load_all_extensions_from_path(self, path):
+ for f in os.listdir(path):
+ LOG.info(_('Loading extension file: %s'), f)
+ mod_name, file_ext = os.path.splitext(os.path.split(f)[-1])
+ ext_path = os.path.join(path, f)
+ if file_ext.lower() == '.py' and not mod_name.startswith('_'):
+ mod = imp.load_source(mod_name, ext_path)
+ ext_name = mod_name[0].upper() + mod_name[1:]
+ new_ext_class = getattr(mod, ext_name, None)
+ if not new_ext_class:
+ LOG.warn(_('Did not find expected name '
+ '"%(ext_name)s" in %(file)s'),
+ {'ext_name': ext_name,
+ 'file': ext_path})
+ continue
+ new_ext = new_ext_class()
+ self.add_extension(new_ext)
+
+ def add_extension(self, ext):
+ # Do nothing if the extension doesn't check out
+ if not self._check_extension(ext):
+ return
+
+ alias = ext.get_alias()
+ LOG.warn(_('Loaded extension: %s'), alias)
+
+ if alias in self.extensions:
+ raise exceptions.Error("Found duplicate extension: %s"
+ % alias)
+ self.extensions[alias] = ext
+
+
+class PluginAwareExtensionManager(ExtensionManager):
+
++ def __init__(self, path, plugin_options=None):
++ self.plugin = QuantumManager(plugin_options).get_plugin()
+ super(PluginAwareExtensionManager, self).__init__(path)
+
+ def _check_extension(self, extension):
+ """Checks if plugin supports extension and implements the contract."""
+ extension_is_valid = super(PluginAwareExtensionManager,
+ self)._check_extension(extension)
+ return (extension_is_valid and
+ self._plugin_supports(extension) and
+ self._plugin_implements_interface(extension))
+
+ def _plugin_supports(self, extension):
+ alias = extension.get_alias()
+ return (hasattr(self.plugin, "supported_extension_aliases") and
+ alias in self.plugin.supported_extension_aliases)
+
+ def _plugin_implements_interface(self, extension):
+ if(not hasattr(extension, "get_plugin_interface") or
+ extension.get_plugin_interface() is None):
+ return True
+ return isinstance(self.plugin, extension.get_plugin_interface())
+
+
+class RequestExtension(object):
+ """Extend requests and responses of core Quantum OpenStack API controllers.
+
+ Provide a way to add data to responses and handle custom request data
+ that is sent to core Quantum OpenStack API controllers.
+
+ """
+ def __init__(self, method, url_route, handler):
+ self.url_route = url_route
+ self.handler = handler
+ self.conditions = dict(method=[method])
+ self.key = "%s-%s" % (method, url_route)
+
+
+class ActionExtension(object):
+ """Add custom actions to core Quantum OpenStack API controllers."""
+
+ def __init__(self, collection, action_name, handler):
+ self.collection = collection
+ self.action_name = action_name
+ self.handler = handler
+
+
+class ResourceExtension(object):
+ """Add top level resources to the OpenStack API in Quantum."""
+
+ def __init__(self, collection, controller, parent=None,
+ collection_actions={}, member_actions={}):
+ self.collection = collection
+ self.controller = controller
+ self.parent = parent
+ self.collection_actions = collection_actions
+ self.member_actions = member_actions
The caller should make sure that QuantumManager is a singleton.
"""
import gettext
+ import logging
import os
-
+import logging
gettext.install('quantum', unicode=1)
from common import utils
from quantum_plugin_base import QuantumPluginBase
+LOG = logging.getLogger('quantum.manager')
CONFIG_FILE = "plugins.ini"
+ LOG = logging.getLogger('quantum.manager')
def find_config(basepath):
class QuantumManager(object):
- _instance = None
-
- def __init__(self, config=None):
- if config == None:
+
+ def __init__(self, options=None, config_file=None):
+ if config_file == None:
self.configuration_file = find_config(
os.path.abspath(os.path.dirname(__file__)))
else:
client/cli/api development
"""
- #static data for networks and ports
- _port_dict_1 = {
- 1: {'port-id': 1,
- 'port-state': 'DOWN',
- 'attachment': None},
- 2: {'port-id': 2,
- 'port-state': 'UP',
- 'attachment': None}}
- _port_dict_2 = {
- 1: {'port-id': 1,
- 'port-state': 'UP',
- 'attachment': 'SomeFormOfVIFID'},
- 2: {'port-id': 2,
- 'port-state': 'DOWN',
- 'attachment': None}}
- _networks = {'001':
- {
- 'net-id': '001',
- 'net-name': 'pippotest',
- 'net-ports': _port_dict_1},
- '002':
- {
- 'net-id': '002',
- 'net-name': 'cicciotest',
- 'net-ports': _port_dict_2}}
+ def __init__(self):
+ db.configure_db({'sql_connection': 'sqlite:///:memory:'})
+ FakePlugin._net_counter = 0
- def __init__(self):
- FakePlugin._net_counter = len(FakePlugin._networks)
-
+ supported_extension_aliases = ["FOXNSOX"]
+
def _get_network(self, tenant_id, network_id):
- network = FakePlugin._networks.get(network_id)
- if not network:
+ try:
+ network = db.network_get(network_id)
+ except:
raise exc.NetworkNotFound(net_id=network_id)
return network
--- /dev/null
- PluginAwareExtensionManager)
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+# Copyright 2011 OpenStack LLC.
+# 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.
+import json
+import unittest
+import routes
+import os.path
+from tests.unit import BaseTest
+from abc import abstractmethod
+
+from webtest import TestApp
+from quantum.common import extensions
+from quantum.common import wsgi
+from quantum.common import config
+from quantum.common.extensions import (ExtensionManager,
- self.ext_mgr = PluginAwareExtensionManager('')
++ PluginAwareExtensionManager,
++ PluginAwareExtensionMiddleware)
+
+test_conf_file = os.path.join(os.path.dirname(__file__), os.pardir,
+ os.pardir, 'etc', 'quantum.conf.test')
+
++plugin_options = {'plugin_provider': "quantum.plugins.SamplePlugin.FakePlugin"}
++
+
+class StubExtension(object):
+
+ def __init__(self, alias="stub_extension"):
+ self.alias = alias
+
+ def get_name(self):
+ return "Stub Extension"
+
+ def get_alias(self):
+ return self.alias
+
+ def get_description(self):
+ return ""
+
+ def get_namespace(self):
+ return ""
+
+ def get_updated(self):
+ return ""
+
+
+class StubPlugin(object):
+
+ def __init__(self, supported_extensions=[]):
+ self.supported_extension_aliases = supported_extensions
+
+
+class ExtensionExpectingPluginInterface(StubExtension):
+ """
+ This extension expects plugin to implement all the methods defined
+ in StubPluginInterface
+ """
+
+ def get_plugin_interface(self):
+ return StubPluginInterface
+
+
+class StubPluginInterface(extensions.PluginInterface):
+
+ @abstractmethod
+ def get_foo(self, bar=None):
+ pass
+
+
+class StubBaseAppController(wsgi.Controller):
+
+ def index(self, request):
+ return "base app index"
+
+ def show(self, request, id):
+ return {'fort': 'knox'}
+
+ def update(self, request, id):
+ return {'uneditable': 'original_value'}
+
+
+class ExtensionsTestApp(wsgi.Router):
+
+ def __init__(self, options={}):
+ mapper = routes.Mapper()
+ controller = StubBaseAppController()
+ mapper.resource("dummy_resource", "/dummy_resources",
+ controller=controller)
+ super(ExtensionsTestApp, self).__init__(mapper)
+
+
+class ResourceExtensionTest(unittest.TestCase):
+
+ class ResourceExtensionController(wsgi.Controller):
+
+ def index(self, request):
+ return "resource index"
+
+ def show(self, request, id):
+ return {'data': {'id': id}}
+
+ def custom_member_action(self, request, id):
+ return {'member_action': 'value'}
+
+ def custom_collection_action(self, request):
+ return {'collection': 'value'}
+
+ def test_resource_can_be_added_as_extension(self):
+ res_ext = extensions.ResourceExtension('tweedles',
+ self.ResourceExtensionController())
+ test_app = setup_extensions_test_app(SimpleExtensionManager(res_ext))
+
+ index_response = test_app.get("/tweedles")
+ self.assertEqual(200, index_response.status_int)
+ self.assertEqual("resource index", index_response.body)
+
+ show_response = test_app.get("/tweedles/25266")
+ self.assertEqual({'data': {'id': "25266"}}, show_response.json)
+
+ def test_resource_extension_with_custom_member_action(self):
+ controller = self.ResourceExtensionController()
+ member = {'custom_member_action': "GET"}
+ res_ext = extensions.ResourceExtension('tweedles', controller,
+ member_actions=member)
+ test_app = setup_extensions_test_app(SimpleExtensionManager(res_ext))
+
+ response = test_app.get("/tweedles/some_id/custom_member_action")
+ self.assertEqual(200, response.status_int)
+ self.assertEqual(json.loads(response.body)['member_action'], "value")
+
+ def test_resource_extension_with_custom_collection_action(self):
+ controller = self.ResourceExtensionController()
+ collections = {'custom_collection_action': "GET"}
+ res_ext = extensions.ResourceExtension('tweedles', controller,
+ collection_actions=collections)
+ test_app = setup_extensions_test_app(SimpleExtensionManager(res_ext))
+
+ response = test_app.get("/tweedles/custom_collection_action")
+ self.assertEqual(200, response.status_int)
+ self.assertEqual(json.loads(response.body)['collection'], "value")
+
+ def test_returns_404_for_non_existant_extension(self):
+ test_app = setup_extensions_test_app(SimpleExtensionManager(None))
+
+ response = test_app.get("/non_extistant_extension", status='*')
+
+ self.assertEqual(404, response.status_int)
+
+
+class ActionExtensionTest(unittest.TestCase):
+
+ def setUp(self):
+ super(ActionExtensionTest, self).setUp()
+ self.extension_app = setup_extensions_test_app()
+
+ def test_extended_action_for_adding_extra_data(self):
+ action_name = 'add_tweedle'
+ action_params = dict(name='Beetle')
+ req_body = json.dumps({action_name: action_params})
+ response = self.extension_app.post('/dummy_resources/1/action',
+ req_body, content_type='application/json')
+ self.assertEqual("Tweedle Beetle Added.", response.body)
+
+ def test_extended_action_for_deleting_extra_data(self):
+ action_name = 'delete_tweedle'
+ action_params = dict(name='Bailey')
+ req_body = json.dumps({action_name: action_params})
+ response = self.extension_app.post("/dummy_resources/1/action",
+ req_body, content_type='application/json')
+ self.assertEqual("Tweedle Bailey Deleted.", response.body)
+
+ def test_returns_404_for_non_existant_action(self):
+ non_existant_action = 'blah_action'
+ action_params = dict(name="test")
+ req_body = json.dumps({non_existant_action: action_params})
+
+ response = self.extension_app.post("/dummy_resources/1/action",
+ req_body, content_type='application/json',
+ status='*')
+
+ self.assertEqual(404, response.status_int)
+
+ def test_returns_404_for_non_existant_resource(self):
+ action_name = 'add_tweedle'
+ action_params = dict(name='Beetle')
+ req_body = json.dumps({action_name: action_params})
+
+ response = self.extension_app.post("/asdf/1/action", req_body,
+ content_type='application/json', status='*')
+ self.assertEqual(404, response.status_int)
+
+
+class RequestExtensionTest(BaseTest):
+
+ def test_headers_can_be_extended(self):
+ def extend_headers(req, res):
+ assert req.headers['X-NEW-REQUEST-HEADER'] == "sox"
+ res.headers['X-NEW-RESPONSE-HEADER'] = "response_header_data"
+ return res
+
+ app = self._setup_app_with_request_handler(extend_headers, 'GET')
+ response = app.get("/dummy_resources/1",
+ headers={'X-NEW-REQUEST-HEADER': "sox"})
+
+ self.assertEqual(response.headers['X-NEW-RESPONSE-HEADER'],
+ "response_header_data")
+
+ def test_extend_get_resource_response(self):
+ def extend_response_data(req, res):
+ data = json.loads(res.body)
+ data['extended_key'] = req.GET.get('extended_key')
+ res.body = json.dumps(data)
+ return res
+
+ app = self._setup_app_with_request_handler(extend_response_data, 'GET')
+ response = app.get("/dummy_resources/1?extended_key=extended_data")
+
+ self.assertEqual(200, response.status_int)
+ response_data = json.loads(response.body)
+ self.assertEqual('extended_data', response_data['extended_key'])
+ self.assertEqual('knox', response_data['fort'])
+
+ def test_get_resources(self):
+ app = setup_extensions_test_app()
+
+ response = app.get("/dummy_resources/1?chewing=newblue")
+
+ response_data = json.loads(response.body)
+ self.assertEqual('newblue', response_data['googoose'])
+ self.assertEqual("Pig Bands!", response_data['big_bands'])
+
+ def test_edit_previously_uneditable_field(self):
+
+ def _update_handler(req, res):
+ data = json.loads(res.body)
+ data['uneditable'] = req.params['uneditable']
+ res.body = json.dumps(data)
+ return res
+
+ base_app = TestApp(setup_base_app())
+ response = base_app.put("/dummy_resources/1",
+ {'uneditable': "new_value"})
+ self.assertEqual(response.json['uneditable'], "original_value")
+
+ ext_app = self._setup_app_with_request_handler(_update_handler,
+ 'PUT')
+ ext_response = ext_app.put("/dummy_resources/1",
+ {'uneditable': "new_value"})
+ self.assertEqual(ext_response.json['uneditable'], "new_value")
+
+ def _setup_app_with_request_handler(self, handler, verb):
+ req_ext = extensions.RequestExtension(verb,
+ '/dummy_resources/:(id)', handler)
+ manager = SimpleExtensionManager(None, None, req_ext)
+ return setup_extensions_test_app(manager)
+
+
+class ExtensionManagerTest(unittest.TestCase):
+
+ def test_invalid_extensions_are_not_registered(self):
+
+ class InvalidExtension(object):
+ """
+ This Extension doesn't implement extension methods :
+ get_name, get_description, get_namespace and get_updated
+ """
+ def get_alias(self):
+ return "invalid_extension"
+
+ ext_mgr = ExtensionManager('')
+ ext_mgr.add_extension(InvalidExtension())
+ ext_mgr.add_extension(StubExtension("valid_extension"))
+
+ self.assertTrue('valid_extension' in ext_mgr.extensions)
+ self.assertFalse('invalid_extension' in ext_mgr.extensions)
+
+
+class PluginAwareExtensionManagerTest(unittest.TestCase):
+
+ def setUp(self):
- return extensions.ExtensionMiddleware(app, conf, extension_manager)
++ self.ext_mgr = PluginAwareExtensionManager('', plugin_options)
+
+ def test_unsupported_extensions_are_not_loaded(self):
+ self.ext_mgr.plugin = StubPlugin(supported_extensions=["e1", "e3"])
+
+ self.ext_mgr.add_extension(StubExtension("e1"))
+ self.ext_mgr.add_extension(StubExtension("e2"))
+ self.ext_mgr.add_extension(StubExtension("e3"))
+
+ self.assertTrue("e1" in self.ext_mgr.extensions)
+ self.assertFalse("e2" in self.ext_mgr.extensions)
+ self.assertTrue("e3" in self.ext_mgr.extensions)
+
+ def test_extensions_are_not_loaded_for_plugins_unaware_of_extensions(self):
+ class ExtensionUnawarePlugin(object):
+ """
+ This plugin does not implement supports_extension method.
+ Extensions will not be loaded when this plugin is used.
+ """
+ pass
+
+ self.ext_mgr.plugin = ExtensionUnawarePlugin()
+ self.ext_mgr.add_extension(StubExtension("e1"))
+
+ self.assertFalse("e1" in self.ext_mgr.extensions)
+
+ def test_extensions_not_loaded_for_plugin_without_expected_interface(self):
+
+ class PluginWithoutExpectedInterface(object):
+ """
+ Plugin does not implement get_foo method as expected by extension
+ """
+ supported_extension_aliases = ["supported_extension"]
+
+ self.ext_mgr.plugin = PluginWithoutExpectedInterface()
+ self.ext_mgr.add_extension(
+ ExtensionExpectingPluginInterface("supported_extension"))
+
+ self.assertFalse("e1" in self.ext_mgr.extensions)
+
+ def test_extensions_are_loaded_for_plugin_with_expected_interface(self):
+
+ class PluginWithExpectedInterface(object):
+ """
+ This Plugin implements get_foo method as expected by extension
+ """
+ supported_extension_aliases = ["supported_extension"]
+
+ def get_foo(self, bar=None):
+ pass
+
+ self.ext_mgr.plugin = PluginWithExpectedInterface()
+ self.ext_mgr.add_extension(
+ ExtensionExpectingPluginInterface("supported_extension"))
+
+ self.assertTrue("supported_extension" in self.ext_mgr.extensions)
+
+ def test_extensions_expecting_quantum_plugin_interface_are_loaded(self):
+ class ExtensionForQuamtumPluginInterface(StubExtension):
+ """
+ This Extension does not implement get_plugin_interface method.
+ This will work with any plugin implementing QuantumPluginBase
+ """
+ pass
+
+ self.ext_mgr.plugin = StubPlugin(supported_extensions=["e1"])
+ self.ext_mgr.add_extension(ExtensionForQuamtumPluginInterface("e1"))
+
+ self.assertTrue("e1" in self.ext_mgr.extensions)
+
+ def test_extensions_without_need_for__plugin_interface_are_loaded(self):
+ class ExtensionWithNoNeedForPluginInterface(StubExtension):
+ """
+ This Extension does not need any plugin interface.
+ This will work with any plugin implementing QuantumPluginBase
+ """
+ def get_plugin_interface(self):
+ return None
+
+ self.ext_mgr.plugin = StubPlugin(supported_extensions=["e1"])
+ self.ext_mgr.add_extension(ExtensionWithNoNeedForPluginInterface("e1"))
+
+ self.assertTrue("e1" in self.ext_mgr.extensions)
+
+
+class ExtensionControllerTest(unittest.TestCase):
+
+ def setUp(self):
+ super(ExtensionControllerTest, self).setUp()
+ self.test_app = setup_extensions_test_app()
+
+ def test_index_gets_all_registerd_extensions(self):
+ response = self.test_app.get("/extensions")
+ foxnsox = response.json["extensions"][0]
+
+ self.assertEqual(foxnsox["alias"], "FOXNSOX")
+ self.assertEqual(foxnsox["namespace"],
+ "http://www.fox.in.socks/api/ext/pie/v1.0")
+
+ def test_extension_can_be_accessed_by_alias(self):
+ foxnsox_extension = self.test_app.get("/extensions/FOXNSOX").json
+
+ self.assertEqual(foxnsox_extension["alias"], "FOXNSOX")
+ self.assertEqual(foxnsox_extension["namespace"],
+ "http://www.fox.in.socks/api/ext/pie/v1.0")
+
+
+class TestExtensionMiddlewareFactory(unittest.TestCase):
+
+ def test_app_configured_with_extensions_as_filter(self):
+ conf, quantum_app = config.load_paste_app('extensions_app_with_filter',
+ {"config_file": test_conf_file}, None)
+
+ response = TestApp(quantum_app).get("/extensions")
+ self.assertEqual(response.status_int, 200)
+
+
+def app_factory(global_conf, **local_conf):
+ conf = global_conf.copy()
+ conf.update(local_conf)
+ return ExtensionsTestApp(conf)
+
+
+def setup_base_app():
+ options = {'config_file': test_conf_file}
+ conf, app = config.load_paste_app('extensions_test_app', options, None)
+ return app
+
+
+def setup_extensions_middleware(extension_manager=None):
+ options = {'config_file': test_conf_file}
+ conf, app = config.load_paste_app('extensions_test_app', options, None)
++ return PluginAwareExtensionMiddleware(app, conf, ext_mgr=extension_manager,
++ plugin_options=plugin_options)
+
+
+def setup_extensions_test_app(extension_manager=None):
+ return TestApp(setup_extensions_middleware(extension_manager))
+
+
+class SimpleExtensionManager(object):
+
+ def __init__(self, resource_ext=None, action_ext=None, request_ext=None):
+ self.resource_ext = resource_ext
+ self.action_ext = action_ext
+ self.request_ext = request_ext
+
+ def get_resources(self):
+ resource_exts = []
+ if self.resource_ext:
+ resource_exts.append(self.resource_ext)
+ return resource_exts
+
+ def get_actions(self):
+ action_exts = []
+ if self.action_ext:
+ action_exts.append(self.action_ext)
+ return action_exts
+
+ def get_request_extensions(self):
+ request_extensions = []
+ if self.request_ext:
+ request_extensions.append(self.request_ext)
+ return request_extensions
PasteDeploy
pep8==0.5.0
python-gflags
--routes
simplejson
++sqlalchemy
webob
webtest
Code Review / openstack-build / neutron-build.git / commitdiff