x/vmware-nsx
Code Issues Proposed changes

Merging from lp:quantum.

Browse Source
This commit is contained in:
Sumit Naiksatam 2011-08-15 20:27:49 -07:00
commit 4315a65f74
24 changed files with 1361 additions and 108 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

31
README
View File

@ -105,4 +105,35 @@ There are a few requirements to writing your own plugin:
4) Launch the Quantum Service, and your plug-in is configured and ready to
manage a Cloud Networking Fabric.
# -- Extensions
1) Creating Extensions:
a) Extension files should be placed under ./extensions folder.
b) The extension file should have a class with the same name as the filename.
This class should implement the contract required by the extension framework.
See ExtensionDescriptor class in ./quantum/common/extensions.py for details
c) To stop a file in ./extensions folder from being loaded as an extension,
the filename should start with an "_"
For an example of an extension file look at Foxinsocks class in
./tests/unit/extensions/foxinsocks.py
The unit tests in ./tests/unit/test_extensions.py document all the ways in
which you can use extensions
2) Associating plugins with extensions:
a) A Plugin can advertize all the extensions it supports through the
'supported_extension_aliases' attribute. Eg:
class SomePlugin:
...
supported_extension_aliases = ['extension1_alias',
'extension2_alias',
'extension3_alias']
Any extension not in this list will not be loaded for the plugin
b) Extension Interfaces for plugins (optional)
The extension can mandate an interface that plugins have to support with the
'get_plugin_interface' method in the extension.
For an example see the FoxInSocksPluginInterface in foxinsocks.py.
The QuantumEchoPlugin lists foxinsox in its supported_extension_aliases
and implements the method from FoxInSocksPluginInterface.

11
etc/quantum.conf
View File

@ -11,13 +11,22 @@ bind_host = 0.0.0.0
# Port the bind the API server to
bind_port = 9696
# Path to the extensions
api_extensions_path = extensions
[composite:quantum]
use = egg:Paste#urlmap
/: quantumversions
/v0.1: quantumapi
[pipeline:quantumapi]
pipeline = extensions quantumapiapp
[filter:extensions]
paste.filter_factory = quantum.common.extensions:plugin_aware_extension_middleware_factory
[app:quantumversions]
paste.app_factory = quantum.api.versions:Versions.factory
[app:quantumapi]
[app:quantumapiapp]
paste.app_factory = quantum.api:APIRouterV01.factory

25
etc/quantum.conf.sample
View File

@ -5,11 +5,28 @@ verbose = True
# Show debugging output in logs (sets DEBUG log level output)
debug = False
[app:quantum]
paste.app_factory = quantum.service:app_factory
# Address to bind the API server
bind_host = 0.0.0.0
# Port the bind the API server to
bind_port = 9696
bind_port = 9696
# Path to the extensions
api_extensions_path = extensions
[composite:quantum]
use = egg:Paste#urlmap
/: quantumversions
/v0.1: quantumapi
[pipeline:quantumapi]
pipeline = extensions quantumapiapp
[filter:extensions]
paste.filter_factory = quantum.common.extensions:plugin_aware_extension_middleware_factory
[app:quantumversions]
paste.app_factory = quantum.api.versions:Versions.factory
[app:quantumapiapp]
paste.app_factory = quantum.api:APIRouterV01.factory

15
etc/quantum.conf.test
View File

@ -5,11 +5,20 @@ verbose = True
# Show debugging output in logs (sets DEBUG log level output)
debug = False
[app:quantum]
paste.app_factory = quantum.l2Network.service:app_factory
# Address to bind the API server
bind_host = 0.0.0.0
# Port the bind the API server to
bind_port = 9696
# 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:plugin_aware_extension_middleware_factory
[app:extensions_test_app]
paste.app_factory = tests.unit.test_extensions:app_factory

15
extensions/__init__.py Normal file
View File

@ -0,0 +1,15 @@
# vim: tabstop=4 shiftwidth=4 softtabstop=4
# Copyright 2011 OpenStack LLC
#
# 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.

3
quantum/api/__init__.py
View File

@ -48,7 +48,8 @@ class APIRouterV01(wsgi.Router):
def _setup_routes(self, mapper, options):
# Loads the quantum plugin
plugin = manager.QuantumManager(options).get_plugin()
plugin = manager.QuantumManager.get_plugin(options)
uri_prefix = '/tenants/{tenant_id}/'
mapper.resource('network', 'networks',
controller=networks.Controller(plugin),

6
quantum/cli.py
View File

@ -104,7 +104,7 @@ def detail_net(manager, *args):
def api_detail_net(client, *args):
tid, nid = args
try:
res = client.list_network_details(nid)["networks"]["network"]
res = client.show_network_details(nid)["networks"]["network"]
except Exception, e:
LOG.error("Failed to get network details: %s" % e)
return
@ -119,7 +119,7 @@ def api_detail_net(client, *args):
print "Remote Interfaces on Virtual Network:%s\n" % nid
for port in ports["ports"]:
pid = port["id"]
res = client.list_port_attachments(nid, pid)
res = client.show_port_attachment(nid, pid)
LOG.debug(res)
remote_iface = res["attachment"]
print "\tRemote interface:%s" % remote_iface
@ -214,7 +214,7 @@ def detail_port(manager, *args):
def api_detail_port(client, *args):
tid, nid, pid = args
try:
port = client.list_port_details(nid, pid)["ports"]["port"]
port = client.show_port_details(nid, pid)["ports"]["port"]
except Exception, e:
LOG.error("Failed to get port details: %s" % e)
return

77
quantum/client.py
View File

@ -57,7 +57,8 @@ class Client(object):
attachment_path = "/networks/%s/ports/%s/attachment"
def __init__(self, host="127.0.0.1", port=9696, use_ssl=False, tenant=None,
format="xml", testingStub=None, key_file=None, cert_file=None):
format="xml", testingStub=None, key_file=None, cert_file=None,
logger=None):
"""
Creates a new client to some service.
@ -79,6 +80,7 @@ class Client(object):
self.testingStub = testingStub
self.key_file = key_file
self.cert_file = cert_file
self.logger = logger
def get_connection_type(self):
"""
@ -118,6 +120,9 @@ class Client(object):
if type(params) is dict:
action += '?' + urllib.urlencode(params)
if body:
body = self.serialize(body)
try:
connection_type = self.get_connection_type()
headers = headers or {"Content-Type":
@ -132,14 +137,26 @@ class Client(object):
else:
c = connection_type(self.host, self.port)
if self.logger:
self.logger.debug("Quantum Client Request:\n" \
+ method + " " + action + "\n")
if body:
self.logger.debug(body)
c.request(method, action, body, headers)
res = c.getresponse()
status_code = self.get_status_code(res)
data = res.read()
if self.logger:
self.logger.debug("Quantum Client Reply (code = %s) :\n %s" \
% (str(status_code), data))
if status_code in (httplib.OK,
httplib.CREATED,
httplib.ACCEPTED,
httplib.NO_CONTENT):
return self.deserialize(res)
return self.deserialize(data, status_code)
else:
raise Exception("Server returned error: %s" % res.read())
@ -158,13 +175,18 @@ class Client(object):
return response.status
def serialize(self, data):
if type(data) is dict:
if data is None:
return None
elif type(data) is dict:
return Serializer().serialize(data, self.content_type())
else:
raise Exception("unable to deserialize object of type = '%s'" \
% type(data))
def deserialize(self, data):
if self.get_status_code(data) == 202:
return data.read()
return Serializer().deserialize(data.read(), self.content_type())
def deserialize(self, data, status_code):
if status_code == 202:
return data
return Serializer().deserialize(data, self.content_type())
def content_type(self, format=None):
if not format:
@ -174,97 +196,94 @@ class Client(object):
@api_call
def list_networks(self):
"""
Queries the server for a list of networks
Fetches a list of all networks for a tenant
"""
return self.do_request("GET", self.networks_path)
@api_call
def list_network_details(self, network):
def show_network_details(self, network):
"""
Queries the server for the details of a certain network
Fetches the details of a certain network
"""
return self.do_request("GET", self.network_path % (network))
@api_call
def create_network(self, body=None):
"""
Creates a new network on the server
Creates a new network
"""
body = self.serialize(body)
return self.do_request("POST", self.networks_path, body=body)
@api_call
def update_network(self, network, body=None):
"""
Updates a network on the server
Updates a network
"""
body = self.serialize(body)
return self.do_request("PUT", self.network_path % (network), body=body)
@api_call
def delete_network(self, network):
"""
Deletes a network on the server
Deletes the specified network
"""
return self.do_request("DELETE", self.network_path % (network))
@api_call
def list_ports(self, network):
"""
Queries the server for a list of ports on a given network
Fetches a list of ports on a given network
"""
return self.do_request("GET", self.ports_path % (network))
@api_call
def list_port_details(self, network, port):
def show_port_details(self, network, port):
"""
Queries the server for a list of ports on a given network
Fetches the details of a certain port
"""
return self.do_request("GET", self.port_path % (network, port))
@api_call
def create_port(self, network):
def create_port(self, network, body=None):
"""
Creates a new port on a network on the server
Creates a new port on a given network
"""
return self.do_request("POST", self.ports_path % (network))
body = self.serialize(body)
return self.do_request("POST", self.ports_path % (network), body=body)
@api_call
def delete_port(self, network, port):
"""
Deletes a port from a network on the server
Deletes the specified port from a network
"""
return self.do_request("DELETE", self.port_path % (network, port))
@api_call
def set_port_state(self, network, port, body=None):
"""
Sets the state of a port on the server
Sets the state of the specified port
"""
body = self.serialize(body)
return self.do_request("PUT",
self.port_path % (network, port), body=body)
@api_call
def list_port_attachments(self, network, port):
def show_port_attachment(self, network, port):
"""
Deletes a port from a network on the server
Fetches the attachment-id associated with the specified port
"""
return self.do_request("GET", self.attachment_path % (network, port))
@api_call
def attach_resource(self, network, port, body=None):
"""
Deletes a port from a network on the server
Sets the attachment-id of the specified port
"""
body = self.serialize(body)
return self.do_request("PUT",
self.attachment_path % (network, port), body=body)
@api_call
def detach_resource(self, network, port):
"""
Deletes a port from a network on the server
Removes the attachment-id of the specified port
"""
return self.do_request("DELETE",
self.attachment_path % (network, port))

518
quantum/common/extensions.py Normal file
View File

@ -0,0 +1,518 @@
# 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 logging
import os
import routes
import webob.dec
import webob.exc
from gettext import gettext as _
from abc import ABCMeta
from quantum.common import exceptions
from quantum.manager import QuantumManager
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.get(id, None)
if not ext:
raise webob.exc.HTTPNotFound(
_("Extension with alias %s does not exist") % 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', '')))
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
def plugin_aware_extension_middleware_factory(global_config, **local_config):
"""Paste factory."""
def _factory(app):
extensions_path = global_config.get('api_extensions_path', '')
ext_mgr = PluginAwareExtensionManager(extensions_path,
QuantumManager().get_plugin())
return ExtensionMiddleware(app, global_config, ext_mgr=ext_mgr)
return _factory
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'.
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)
def _load_all_extensions_from_path(self, path):
for f in os.listdir(path):
try:
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)
except Exception as exception:
LOG.warn("extension file %s wasnt loaded due to %s",
f, exception)
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):
self.plugin = plugin
super(PluginAwareExtensionManager, self).__init__(path)
def _check_extension(self, extension):
"""Checks if plugin supports extension and implements the
extension 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()
supports_extension = (hasattr(self.plugin,
"supported_extension_aliases") and
alias in self.plugin.supported_extension_aliases)
if not supports_extension:
LOG.warn("extension %s not supported by plugin %s",
alias, self.plugin)
return supports_extension
def _plugin_implements_interface(self, extension):
if(not hasattr(extension, "get_plugin_interface") or
extension.get_plugin_interface() is None):
return True
plugin_has_interface = isinstance(self.plugin,
extension.get_plugin_interface())
if not plugin_has_interface:
LOG.warn("plugin %s does not implement extension's"
"plugin interface %s" % (self.plugin,
extension.get_alias()))
return plugin_has_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

4
quantum/common/flags.py
View File

@ -21,14 +21,12 @@ Wraps gflags.
Global flags should be defined here, the rest are defined where they're used.
"""
import getopt
import gflags
import os
import string
import sys
import gflags
class FlagValues(gflags.FlagValues):
"""Extension of gflags.FlagValues that allows undefined and runtime flags.

8
quantum/common/utils.py
View File

@ -18,8 +18,10 @@
"""
System-level utilities and helper functions.
"""
import ConfigParser
import datetime
import exceptions as exception
import flags
import inspect
import logging
import os
@ -27,10 +29,8 @@ import random
import subprocess
import socket
import sys
import ConfigParser
import exceptions as exception
import flags
from exceptions import ProcessExecutionError

8
quantum/common/wsgi.py
View File

@ -22,17 +22,17 @@ Utility methods for working with WSGI servers
import logging
import sys
from xml.dom import minidom
import eventlet.wsgi
eventlet.patcher.monkey_patch(all=False, socket=True)
import routes.middleware
import webob.dec
import webob.exc
from xml.dom import minidom
from quantum import utils
from quantum.common import exceptions as exception
from quantum import utils
LOG = logging.getLogger('quantum.common.wsgi')

1
quantum/db/api.py
View File

@ -23,6 +23,7 @@ from sqlalchemy.orm import sessionmaker, exc
from quantum.common import exceptions as q_exc
from quantum.db import models
_ENGINE = None
_MAKER = None
BASE = models.BASE

2
quantum/db/models.py
View File

@ -19,10 +19,12 @@
import uuid
from sqlalchemy import Column, String, ForeignKey
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relation, object_mapper
BASE = declarative_base()

13
quantum/manager.py
View File

@ -26,12 +26,13 @@ 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')
@ -44,6 +45,9 @@ def find_config(basepath):
class QuantumManager(object):
_instance = None
def __init__(self, options=None, config_file=None):
if config_file == None:
self.configuration_file = find_config(
@ -66,5 +70,8 @@ class QuantumManager(object):
"All compatibility tests passed")
self.plugin = plugin_klass()
def get_plugin(self):
return self.plugin
@classmethod
def get_plugin(cls, options=None, config_file=None):
if cls._instance is None:
cls._instance = cls(options, config_file)
return cls._instance.plugin

5
quantum/plugins/SamplePlugin.py
View File

@ -119,6 +119,11 @@ class QuantumEchoPlugin(object):
"""
print("unplug_interface() called\n")
supported_extension_aliases = ["FOXNSOX"]
def method_to_support_foxnsox_extension(self):
print("method_to_support_foxnsox_extension() called\n")
class DummyDataPlugin(object):

19
tests/__init__.py
View File

@ -0,0 +1,19 @@
# 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 logging
logging.basicConfig()

75
tests/unit/extension_stubs.py Normal file
View File

@ -0,0 +1,75 @@
# 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.
from abc import abstractmethod
from quantum.common import extensions
from quantum.common import wsgi
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'}

15
tests/unit/extensions/__init__.py Normal file
View File

@ -0,0 +1,15 @@
# vim: tabstop=4 shiftwidth=4 softtabstop=4
# Copyright 2011 OpenStack LLC
#
# 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.

110
tests/unit/extensions/foxinsocks.py Normal file
View File

@ -0,0 +1,110 @@
# 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
from quantum.common import wsgi
from quantum.common import extensions
from abc import abstractmethod
class FoxInSocksController(wsgi.Controller):
def index(self, request):
return "Try to say this Mr. Knox, sir..."
class FoxInSocksPluginInterface(extensions.PluginInterface):
@abstractmethod
def method_to_support_foxnsox_extension(self):
pass
class Foxinsocks(object):
def __init__(self):
pass
def get_plugin_interface(self):
return FoxInSocksPluginInterface
def get_name(self):
return "Fox In Socks"
def get_alias(self):
return "FOXNSOX"
def get_description(self):
return "The Fox In Socks Extension"
def get_namespace(self):
return "http://www.fox.in.socks/api/ext/pie/v1.0"
def get_updated(self):
return "2011-01-22T13:25:27-06:00"
def get_resources(self):
resources = []
resource = extensions.ResourceExtension('foxnsocks',
FoxInSocksController())
resources.append(resource)
return resources
def get_actions(self):
return [extensions.ActionExtension('dummy_resources',
'FOXNSOX:add_tweedle',
self._add_tweedle_handler),
extensions.ActionExtension('dummy_resources',
'FOXNSOX:delete_tweedle',
self._delete_tweedle_handler)]
def get_request_extensions(self):
request_exts = []
def _goose_handler(req, res):
#NOTE: This only handles JSON responses.
# You can use content type header to test for XML.
data = json.loads(res.body)
data['FOXNSOX:googoose'] = req.GET.get('chewing')
res.body = json.dumps(data)
return res
req_ext1 = extensions.RequestExtension('GET', '/dummy_resources/:(id)',
_goose_handler)
request_exts.append(req_ext1)
def _bands_handler(req, res):
#NOTE: This only handles JSON responses.
# You can use content type header to test for XML.
data = json.loads(res.body)
data['FOXNSOX:big_bands'] = 'Pig Bands!'
res.body = json.dumps(data)
return res
req_ext2 = extensions.RequestExtension('GET', '/dummy_resources/:(id)',
_bands_handler)
request_exts.append(req_ext2)
return request_exts
def _add_tweedle_handler(self, input_dict, req, id):
return "Tweedle {0} Added.".format(
input_dict['FOXNSOX:add_tweedle']['name'])
def _delete_tweedle_handler(self, input_dict, req, id):
return "Tweedle {0} Deleted.".format(
input_dict['FOXNSOX:delete_tweedle']['name'])

2
tests/unit/test_api.py
View File

@ -20,8 +20,8 @@
import logging
import unittest
import tests.unit.testlib_api as testlib
import tests.unit.testlib_api as testlib
from quantum import api as server
from quantum.db import api as db
from quantum.common.wsgi import Serializer

102
tests/unit/api.py → tests/unit/test_clientlib.py
View File

@ -131,19 +131,19 @@ class APITest(unittest.TestCase):
LOG.debug("_test_list_networks - tenant:%s "\
"- format:%s - END", format, tenant)
def _test_list_network_details(self,
def _test_show_network_details(self,
tenant=TENANT_1, format='json', status=200):
LOG.debug("_test_list_network_details - tenant:%s "\
LOG.debug("_test_show_network_details - tenant:%s "\
"- format:%s - START", format, tenant)
self._assert_sanity(self.client.list_network_details,
self._assert_sanity(self.client.show_network_details,
status,
"GET",
"networks/001",
data=["001"],
params={'tenant': tenant, 'format': format})
LOG.debug("_test_list_network_details - tenant:%s "\
LOG.debug("_test_show_network_details - tenant:%s "\
"- format:%s - END", format, tenant)
def _test_create_network(self, tenant=TENANT_1, format='json', status=200):
@ -203,19 +203,19 @@ class APITest(unittest.TestCase):
LOG.debug("_test_list_ports - tenant:%s "\
"- format:%s - END", format, tenant)
def _test_list_port_details(self,
def _test_show_port_details(self,
tenant=TENANT_1, format='json', status=200):
LOG.debug("_test_list_port_details - tenant:%s "\
LOG.debug("_test_show_port_details - tenant:%s "\
"- format:%s - START", format, tenant)
self._assert_sanity(self.client.list_port_details,
self._assert_sanity(self.client.show_port_details,
status,
"GET",
"networks/001/ports/001",
data=["001", "001"],
params={'tenant': tenant, 'format': format})
LOG.debug("_test_list_port_details - tenant:%s "\
LOG.debug("_test_show_port_details - tenant:%s "\
"- format:%s - END", format, tenant)
def _test_create_port(self, tenant=TENANT_1, format='json', status=200):
@ -261,19 +261,19 @@ class APITest(unittest.TestCase):
LOG.debug("_test_set_port_state - tenant:%s "\
"- format:%s - END", format, tenant)
def _test_list_port_attachments(self,
def _test_show_port_attachment(self,
tenant=TENANT_1, format='json', status=200):
LOG.debug("_test_list_port_attachments - tenant:%s "\
LOG.debug("_test_show_port_attachment - tenant:%s "\
"- format:%s - START", format, tenant)
self._assert_sanity(self.client.list_port_attachments,
self._assert_sanity(self.client.show_port_attachment,
status,
"GET",
"networks/001/ports/001/attachment",
data=["001", "001"],
params={'tenant': tenant, 'format': format})
LOG.debug("_test_list_port_attachments - tenant:%s "\
LOG.debug("_test_show_port_attachment - tenant:%s "\
"- format:%s - END", format, tenant)
def _test_attach_resource(self, tenant=TENANT_1,
@ -345,23 +345,23 @@ class APITest(unittest.TestCase):
def test_list_networks_error_401(self):
self._test_list_networks(status=401)
def test_list_network_details_json(self):
self._test_list_network_details(format='json')
def test_show_network_details_json(self):
self._test_show_network_details(format='json')
def test_list_network_details_xml(self):
self._test_list_network_details(format='xml')
def test_show_network_details_xml(self):
self._test_show_network_details(format='xml')
def test_list_network_details_alt_tenant(self):
self._test_list_network_details(tenant=TENANT_2)
def test_show_network_details_alt_tenant(self):
self._test_show_network_details(tenant=TENANT_2)
def test_list_network_details_error_470(self):
self._test_list_network_details(status=470)
def test_show_network_details_error_470(self):
self._test_show_network_details(status=470)
def test_list_network_details_error_401(self):
self._test_list_network_details(status=401)
def test_show_network_details_error_401(self):
self._test_show_network_details(status=401)
def test_list_network_details_error_420(self):
self._test_list_network_details(status=420)
def test_show_network_details_error_420(self):
self._test_show_network_details(status=420)
def test_create_network_json(self):
self._test_create_network(format='json')
@ -447,26 +447,26 @@ class APITest(unittest.TestCase):
def test_list_ports_error_420(self):
self._test_list_ports(status=420)
def test_list_port_details_json(self):
def test_show_port_details_json(self):
self._test_list_ports(format='json')
def test_list_port_details_xml(self):
def test_show_port_details_xml(self):
self._test_list_ports(format='xml')
def test_list_port_details_alt_tenant(self):
def test_show_port_details_alt_tenant(self):
self._test_list_ports(tenant=TENANT_2)
def test_list_port_details_error_470(self):
self._test_list_port_details(status=470)
def test_show_port_details_error_470(self):
self._test_show_port_details(status=470)
def test_list_port_details_error_401(self):
self._test_list_ports(status=401)
def test_show_port_details_error_401(self):
self._test_show_port_details(status=401)
def test_list_port_details_error_420(self):
self._test_list_ports(status=420)
def test_show_port_details_error_420(self):
self._test_show_port_details(status=420)
def test_list_port_details_error_430(self):
self._test_list_ports(status=430)
def test_show_port_details_error_430(self):
self._test_show_port_details(status=430)
def test_create_port_json(self):
self._test_create_port(format='json')
@ -546,29 +546,29 @@ class APITest(unittest.TestCase):
def test_set_port_state_error_431(self):
self._test_set_port_state(status=431)
def test_list_port_attachments_json(self):
self._test_list_port_attachments(format='json')
def test_show_port_attachment_json(self):
self._test_show_port_attachment(format='json')
def test_list_port_attachments_xml(self):
self._test_list_port_attachments(format='xml')
def test_show_port_attachment_xml(self):
self._test_show_port_attachment(format='xml')
def test_list_port_attachments_alt_tenant(self):
self._test_list_port_attachments(tenant=TENANT_2)
def test_show_port_attachment_alt_tenant(self):
self._test_show_port_attachment(tenant=TENANT_2)
def test_list_port_attachments_error_470(self):
self._test_list_port_attachments(status=470)
def test_show_port_attachment_error_470(self):
self._test_show_port_attachment(status=470)
def test_list_port_attachments_error_401(self):
self._test_list_port_attachments(status=401)
def test_show_port_attachment_error_401(self):
self._test_show_port_attachment(status=401)
def test_list_port_attachments_error_400(self):
self._test_list_port_attachments(status=400)
def test_show_port_attachment_error_400(self):
self._test_show_port_attachment(status=400)
def test_list_port_attachments_error_420(self):
self._test_list_port_attachments(status=420)
def test_show_port_attachment_error_420(self):
self._test_show_port_attachment(status=420)
def test_list_port_attachments_error_430(self):
self._test_list_port_attachments(status=430)
def test_show_port_attachment_error_430(self):
self._test_show_port_attachment(status=430)
def test_attach_resource_json(self):
self._test_attach_resource(format='json')

402
tests/unit/test_extensions.py Normal file
View File

@ -0,0 +1,402 @@
# 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 os.path
import routes
import unittest
from tests.unit import BaseTest
from webtest import TestApp
from quantum.common import wsgi
from quantum.common import config
from quantum.common import extensions
from quantum.plugins.SamplePlugin import QuantumEchoPlugin
from tests.unit.extension_stubs import (StubExtension, StubPlugin,
StubPluginInterface,
StubBaseAppController,
ExtensionExpectingPluginInterface)
from quantum.common.extensions import (ExtensionManager,
PluginAwareExtensionManager,
ExtensionMiddleware)
test_conf_file = os.path.join(os.path.dirname(__file__), os.pardir,
os.pardir, 'etc', 'quantum.conf.test')
extensions_path = os.path.join(os.path.dirname(__file__), "extensions")
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 = 'FOXNSOX: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 = 'FOXNSOX: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['FOXNSOX: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['FOXNSOX: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['FOXNSOX:googoose'])
self.assertEqual("Pig Bands!", response_data['FOXNSOX: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 test_unsupported_extensions_are_not_loaded(self):
stub_plugin = StubPlugin(supported_extensions=["e1", "e3"])
ext_mgr = PluginAwareExtensionManager('', stub_plugin)
ext_mgr.add_extension(StubExtension("e1"))
ext_mgr.add_extension(StubExtension("e2"))
ext_mgr.add_extension(StubExtension("e3"))
self.assertTrue("e1" in ext_mgr.extensions)
self.assertFalse("e2" in ext_mgr.extensions)
self.assertTrue("e3" in 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
ext_mgr = PluginAwareExtensionManager('', ExtensionUnawarePlugin())
ext_mgr.add_extension(StubExtension("e1"))
self.assertFalse("e1" in 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"]
ext_mgr = PluginAwareExtensionManager('',
PluginWithoutExpectedInterface())
ext_mgr.add_extension(
ExtensionExpectingPluginInterface("supported_extension"))
self.assertFalse("e1" in 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
ext_mgr = PluginAwareExtensionManager('',
PluginWithExpectedInterface())
ext_mgr.add_extension(
ExtensionExpectingPluginInterface("supported_extension"))
self.assertTrue("supported_extension" in 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
stub_plugin = StubPlugin(supported_extensions=["e1"])
ext_mgr = PluginAwareExtensionManager('', stub_plugin)
ext_mgr.add_extension(ExtensionForQuamtumPluginInterface("e1"))
self.assertTrue("e1" in 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
stub_plugin = StubPlugin(supported_extensions=["e1"])
ext_mgr = PluginAwareExtensionManager('', stub_plugin)
ext_mgr.add_extension(ExtensionWithNoNeedForPluginInterface("e1"))
self.assertTrue("e1" in 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")
def test_show_returns_not_found_for_non_existant_extension(self):
response = self.test_app.get("/extensions/non_existant", status="*")
self.assertEqual(response.status_int, 404)
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):
extension_manager = (extension_manager or
PluginAwareExtensionManager(extensions_path,
QuantumEchoPlugin()))
options = {'config_file': test_conf_file}
conf, app = config.load_paste_app('extensions_test_app', options, None)
return ExtensionMiddleware(app, conf, ext_mgr=extension_manager)
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

2
tools/pip-requires
View File

@ -5,7 +5,7 @@ Paste
PasteDeploy
pep8>=0.5.0
python-gflags
sqlalchemy
simplejson
sqlalchemy
webob
webtest