1. Parent class should have instance variables used in its method(s).
2. remove some unused imports
Patch 2: fix according to comments
Patch 4: fix problem under pep8 1.1
Change-Id: Iafd89a6017b30484fb8da50219be6b4ae073f083
"""
Base class for Quantum API routes.
"""
+ _version = None
def __init__(self, options=None):
mapper = self._mapper()
class QuantumController(object):
""" Base controller class for Quantum API """
+ # _resource_name will be redefined in sub concrete controller
+ _resource_name = None
def __init__(self, plugin):
self._plugin = plugin
class Controller(common.QuantumController):
""" Port API controller for Quantum API """
-
+ _resource_name = 'attachment'
+ # version will be redefined by in child class
+ version = None
_attachment_ops_param_list = [
{
'param-name': 'id',
},
}
- def __init__(self, plugin):
- self._resource_name = 'attachment'
- super(Controller, self).__init__(plugin)
-
@common.APIFaultWrapper([exception.NetworkNotFound,
exception.PortNotFound])
def get_resource(self, request, tenant_id, network_id, id):
class ControllerV10(Controller):
"""Attachment resources controller for Quantum v1.0 API"""
-
- def __init__(self, plugin):
- self.version = "1.0"
- super(ControllerV10, self).__init__(plugin)
+ version = "1.0"
class ControllerV11(Controller):
"""Attachment resources controller for Quantum v1.1 API"""
-
- def __init__(self, plugin):
- self.version = "1.1"
- super(ControllerV11, self).__init__(plugin)
+ version = "1.1"
import logging
-from webob import exc
from quantum.api import api_common as common
-from quantum.api import faults
from quantum.api.views import filters
from quantum.api.views import networks as networks_view
from quantum.common import exceptions as exception
class Controller(common.QuantumController):
""" Network API controller for Quantum API """
-
+ _resource_name = 'network'
+ # version will be redefined in child class
+ version = None
_network_ops_param_list = [
{'param-name': 'name', 'required': True},
]
- def __init__(self, plugin):
- self._resource_name = 'network'
- super(Controller, self).__init__(plugin)
-
def _item(self, request, tenant_id, network_id,
net_details=True, port_details=False):
# We expect get_network_details to return information
},
}
- def __init__(self, plugin):
- self.version = "1.0"
- super(ControllerV10, self).__init__(plugin)
+ version = "1.0"
class ControllerV11(Controller):
},
}
- def __init__(self, plugin):
- self.version = "1.1"
- super(ControllerV11, self).__init__(plugin)
+ version = "1.1"
class Controller(common.QuantumController):
""" Port API controller for Quantum API """
-
+ _resource_name = 'port'
+ # version will be redefined in child class
+ version = None
_port_ops_param_list = [
{'param-name': 'state', 'default-value': 'DOWN', 'required': False},
]
- def __init__(self, plugin):
- self._resource_name = 'port'
- super(Controller, self).__init__(plugin)
-
def _items(self, request, tenant_id, network_id,
port_details=False):
""" Returns a list of ports.
},
}
- def __init__(self, plugin):
- self.version = "1.0"
- super(ControllerV10, self).__init__(plugin)
+ version = "1.0"
class ControllerV11(Controller):
},
}
- def __init__(self, plugin):
- self.version = "1.1"
- super(ControllerV11, self).__init__(plugin)
+ version = "1.1"
LOG = logging.getLogger(__name__)
-class Versions(wsgi.Application):
+class Versions(object):
+
+ @classmethod
+ def factory(cls, global_config, **local_config):
+ return cls()
@webob.dec.wsgify(RequestClass=wsgi.Request)
def __call__(self, req):
},
{
"id": "v1.1",
- "status": "PROPOSED",
+ "status": "CURRENT",
},
]
"""Utilities and helper functions."""
-import base64
import ConfigParser
import datetime
-import functools
import inspect
import json
import logging
import os
import random
-import re
-import socket
-import string
-import struct
import subprocess
import sys
-import time
-import types
from quantum.common import exceptions as exception
from quantum.common.exceptions import ProcessExecutionError
resources = []
resources.append(ResourceExtension('extensions',
ExtensionController(self)))
- for alias, ext in self.extensions.iteritems():
+ for ext in self.extensions.itervalues():
try:
resources.extend(ext.get_resources())
except AttributeError:
def get_actions(self):
"""Returns a list of ActionExtension objects."""
actions = []
- for alias, ext in self.extensions.iteritems():
+ for ext in self.extensions.itervalues():
try:
actions.extend(ext.get_actions())
except AttributeError:
def get_request_extensions(self):
"""Returns a list of RequestExtension objects."""
request_exts = []
- for alias, ext in self.extensions.iteritems():
+ for ext in self.extensions.itervalues():
try:
request_exts.extend(ext.get_request_extensions())
except AttributeError:
def network_id(net_name):
session = get_session()
try:
- return session.query(models.Network).\
+ return session.query(models.Network).\
options(joinedload(models.Network.ports)). \
filter_by(name=net_name).\
all()
def network_get(net_id):
session = get_session()
try:
- return session.query(models.Network).\
+ return session.query(models.Network).\
options(joinedload(models.Network.ports)). \
filter_by(uuid=net_id).\
one()
def validate_network_ownership(tenant_id, net_id):
session = get_session()
try:
- return session.query(models.Network).\
+ return session.query(models.Network).\
filter_by(uuid=net_id).\
filter_by(tenant_id=tenant_id).\
one()
network_get(net_id)
session = get_session()
try:
- return session.query(models.Port).\
+ return session.query(models.Port).\
filter_by(uuid=port_id).\
filter_by(network_id=net_id).\
one()
def port_get_by_id(port_id):
session = get_session()
try:
- return session.query(models.Port).\
+ return session.query(models.Port).\
filter_by(uuid=port_id).one()
except exc.NoResultFound:
raise q_exc.PortNotFound(port_id=port_id)
items = dict([(k, v) for k, v in conf.items()
if k not in ('__file__', 'here')])
for key, value in sorted(items.items()):
- LOG.debug("%(key)-30s %(value)s" % locals())
+ LOG.debug("%(key)-30s %(value)s" % {'key': key,
+ 'value': value,
+ })
LOG.debug("*" * 80)
service = cls(app_name, conf_file, conf)
return service
# License for the specific language governing permissions and limitations
# under the License.
-from abc import abstractmethod
+from abc import abstractmethod
import json
from quantum.extensions import extensions
return resources
def get_actions(self):
- return [extensions.ActionExtension('dummy_resources',
+ return [extensions.ActionExtension('dummy_resources',
'FOXNSOX:add_tweedle',
self._add_tweedle_handler),
- extensions.ActionExtension('dummy_resources',
+ extensions.ActionExtension('dummy_resources',
'FOXNSOX:delete_tweedle',
self._delete_tweedle_handler)]
# under the License.
import json
-import os
-import sys
import unittest
import routes
StubBaseAppController,
StubExtension,
StubPlugin,
- StubPluginInterface,
)
import quantum.tests.unit.extensions
from quantum import wsgi
self.body_serializers.update(body_serializers or {})
self.headers_serializer = (headers_serializer or
- ResponseHeadersSerializer())
+ ResponseHeaderSerializer())
def serialize(self, response_data, content_type, action='default'):
"""Serialize a dict into a string and wrap in a wsgi.Request object.
Code Review / openstack-build / neutron-build.git / commitdiff