class ExtensionMiddleware(wsgi.Middleware):
"""Extensions middleware for WSGI."""
+ def __init__(self, application, config_params,
+ ext_mgr=None):
+
+ self.ext_mgr = (ext_mgr
+ or PluginAwareExtensionManager(
+ config_params.get('api_extensions_path', '')))
+
+ 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."""
return request_ext_controllers
- def __init__(self, application, config_params,
- ext_mgr=None):
-
- self.ext_mgr = (ext_mgr
- or PluginAwareExtensionManager(
- config_params.get('api_extensions_path', '')))
-
- 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)
-
@webob.dec.wsgify(RequestClass=wsgi.Request)
def __call__(self, req):
"""Route the incoming request with router."""
self._plugin_implements_interface(extension))
def _plugin_supports(self, extension):
- return (hasattr(self.plugin, "supports_extension") and
- self.plugin.supports_extension(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
from quantum.common.extensions import (ExtensionManager,
PluginAwareExtensionManager)
-extension_index_response = "Try to say this Mr. Knox, sir..."
test_conf_file = os.path.join(os.path.dirname(__file__), os.pardir,
os.pardir, 'etc', 'quantum.conf.test')
-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 ResourceExtensionTest(unittest.TestCase):
-
- def test_resource_extension(self):
- res_ext = extensions.ResourceExtension('tweedles', StubController(
- extension_index_response))
- test_app = setup_extensions_test_app(StubExtensionManager(res_ext))
-
- response = test_app.get("/tweedles")
- self.assertEqual(200, response.status_int)
- self.assertEqual(extension_index_response, response.body)
-
- def test_resource_extension_with_custom_member_action(self):
- controller = StubController(extension_index_response)
- member = {'custom_member_action': "GET"}
- res_ext = extensions.ResourceExtension('tweedles', controller,
- member_actions=member)
- test_app = setup_extensions_test_app(StubExtensionManager(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 = StubController(extension_index_response)
- collections = {'custom_collection_action': "GET"}
- res_ext = extensions.ResourceExtension('tweedles', controller,
- collection_actions=collections)
- test_app = setup_extensions_test_app(StubExtensionManager(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(StubExtensionManager(None))
-
- response = test_app.get("/non_extistant_extension", status='*')
-
- self.assertEqual(404, response.status_int)
-
-
class StubExtension(object):
def __init__(self, alias="stub_extension"):
class StubPlugin(object):
def __init__(self, supported_extensions=[]):
- self.supported_extensions = supported_extensions
-
- def supports_extension(self, extension):
- return extension.get_alias() in self.supported_extensions
+ self.supported_extension_aliases = supported_extensions
class ExtensionExpectingPluginInterface(StubExtension):
pass
-class ExtensionManagerTest(unittest.TestCase):
-
- def test_invalid_extensions_are_not_registered(self):
+class StubBaseAppController(wsgi.Controller):
- 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 index(self, request):
+ return "base app index"
- def setUp(self):
- self.ext_mgr = PluginAwareExtensionManager('')
+ def show(self, request, id):
+ return {'fort': 'knox'}
- def test_unsupported_extensions_are_not_loaded(self):
- self.ext_mgr.plugin = StubPlugin(supported_extensions=["e1", "e3"])
+ def update(self, request, id):
+ return {'uneditable': 'original_value'}
- 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)
+class ExtensionsTestApp(wsgi.Router):
- 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
+ def __init__(self, options={}):
+ mapper = routes.Mapper()
+ controller = StubBaseAppController()
+ mapper.resource("dummy_resource", "/dummy_resources",
+ controller=controller)
+ super(ExtensionsTestApp, self).__init__(mapper)
- self.ext_mgr.plugin = ExtensionUnawarePlugin()
- self.ext_mgr.add_extension(StubExtension("e1"))
- self.assertFalse("e1" in self.ext_mgr.extensions)
+class ResourceExtensionTest(unittest.TestCase):
- def test_extensions_not_loaded_for_plugin_without_expected_interface(self):
+ class ResourceExtensionController(wsgi.Controller):
- class PluginWithoutExpectedInterface(object):
- """
- Plugin does not implement get_foo method as expected by extension
- """
- def supports_extension(self, true):
- return true
+ def index(self, request):
+ return "resource index"
- self.ext_mgr.plugin = PluginWithoutExpectedInterface()
- self.ext_mgr.add_extension(ExtensionExpectingPluginInterface("e1"))
+ def show(self, request, id):
+ return {'data': {'id': id}}
- self.assertFalse("e1" in self.ext_mgr.extensions)
+ def custom_member_action(self, request, id):
+ return {'member_action': 'value'}
- def test_extensions_are_loaded_for_plugin_with_expected_interface(self):
+ def custom_collection_action(self, request):
+ return {'collection': 'value'}
- class PluginWithExpectedInterface(object):
- """
- This Plugin implements get_foo method as expected by extension
- """
- def supports_extension(self, true):
- return true
+ 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))
- def get_foo(self, bar=None):
- pass
+ index_response = test_app.get("/tweedles")
+ self.assertEqual(200, index_response.status_int)
+ self.assertEqual("resource index", index_response.body)
- self.ext_mgr.plugin = PluginWithExpectedInterface()
- self.ext_mgr.add_extension(ExtensionExpectingPluginInterface("e1"))
+ show_response = test_app.get("/tweedles/25266")
+ self.assertEqual({'data': {'id': "25266"}}, show_response.json)
- self.assertTrue("e1" in self.ext_mgr.extensions)
+ 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))
- 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
+ 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")
- self.ext_mgr.plugin = StubPlugin(supported_extensions=["e1"])
- self.ext_mgr.add_extension(ExtensionForQuamtumPluginInterface("e1"))
+ 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))
- self.assertTrue("e1" in self.ext_mgr.extensions)
+ response = test_app.get("/tweedles/custom_collection_action")
+ self.assertEqual(200, response.status_int)
+ self.assertEqual(json.loads(response.body)['collection'], "value")
- 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
+ def test_returns_404_for_non_existant_extension(self):
+ test_app = setup_extensions_test_app(SimpleExtensionManager(None))
- self.ext_mgr.plugin = StubPlugin(supported_extensions=["e1"])
- self.ext_mgr.add_extension(ExtensionWithNoNeedForPluginInterface("e1"))
+ response = test_app.get("/non_extistant_extension", status='*')
- self.assertTrue("e1" in self.ext_mgr.extensions)
+ self.assertEqual(404, response.status_int)
class ActionExtensionTest(unittest.TestCase):
def _setup_app_with_request_handler(self, handler, verb):
req_ext = extensions.RequestExtension(verb,
'/dummy_resources/:(id)', handler)
- manager = StubExtensionManager(None, None, req_ext)
+ manager = SimpleExtensionManager(None, None, req_ext)
return setup_extensions_test_app(manager)
-class TestExtensionMiddlewareFactory(unittest.TestCase):
+class ExtensionManagerTest(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)
+ def test_invalid_extensions_are_not_registered(self):
- response = TestApp(quantum_app).get("/extensions")
- self.assertEqual(response.status_int, 200)
+ 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"))
-class ExtensionsTestApp(wsgi.Router):
+ self.assertTrue('valid_extension' in ext_mgr.extensions)
+ self.assertFalse('invalid_extension' in ext_mgr.extensions)
- def __init__(self, options={}):
- mapper = routes.Mapper()
- controller = StubController(extension_index_response)
- mapper.resource("dummy_resource", "/dummy_resources",
- controller=controller)
- super(ExtensionsTestApp, self).__init__(mapper)
+class PluginAwareExtensionManagerTest(unittest.TestCase):
-class StubController(wsgi.Controller):
+ def setUp(self):
+ self.ext_mgr = PluginAwareExtensionManager('')
- def __init__(self, body):
- self.body = body
+ def test_unsupported_extensions_are_not_loaded(self):
+ self.ext_mgr.plugin = StubPlugin(supported_extensions=["e1", "e3"])
- def index(self, request):
- return self.body
+ self.ext_mgr.add_extension(StubExtension("e1"))
+ self.ext_mgr.add_extension(StubExtension("e2"))
+ self.ext_mgr.add_extension(StubExtension("e3"))
- def show(self, request, id):
- return {'fort': 'knox'}
+ 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 update(self, request, id):
- return {'uneditable': 'original_value'}
+ 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
- def custom_member_action(self, request, id):
- return {'member_action': 'value'}
+ self.ext_mgr.plugin = StubPlugin(supported_extensions=["e1"])
+ self.ext_mgr.add_extension(ExtensionForQuamtumPluginInterface("e1"))
- def custom_collection_action(self, request):
- return {'collection': 'value'}
+ 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):
return TestApp(setup_extensions_middleware(extension_manager))
-class StubExtensionManager(object):
+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_name(self):
- return "Tweedle Beetle Extension"
-
- def get_alias(self):
- return "TWDLBETL"
-
- def get_description(self):
- return "Provides access to Tweedle Beetles"
-
def get_resources(self):
resource_exts = []
if self.resource_ext:
Code Review / openstack-build / neutron-build.git / commitdiff