from neutron.openstack.common import importutils
from neutronclient.common import exceptions as exc
from neutronclient.common import utils
-from neutronclient.shell import env, NeutronShell, NEUTRON_API_VERSION
+from neutronclient import shell
COMMAND_V2 = {
'probe-create': utils.import_class(
COMMANDS = {'2.0': COMMAND_V2}
-class NeutronDebugShell(NeutronShell):
+class NeutronDebugShell(shell.NeutronShell):
def __init__(self, api_version):
super(NeutronDebugShell, self).__init__(api_version)
for k, v in COMMANDS[api_version].items():
parser = super(NeutronDebugShell, self).build_option_parser(
description, version)
default = (
- env('NEUTRON_TEST_CONFIG_FILE') or env('QUANTUM_TEST_CONFIG_FILE')
+ shell.env('NEUTRON_TEST_CONFIG_FILE') or
+ shell.env('QUANTUM_TEST_CONFIG_FILE')
)
parser.add_argument(
'--config-file',
def main(argv=None):
- return NeutronDebugShell(NEUTRON_API_VERSION).run(argv or sys.argv[1:])
+ return NeutronDebugShell(shell.NEUTRON_API_VERSION).run(
+ argv or sys.argv[1:])
methods that needs to be implemented by a v2 Neutron Plug-in.
"""
-from abc import ABCMeta, abstractmethod
-
+import abc
import six
-@six.add_metaclass(ABCMeta)
+@six.add_metaclass(abc.ABCMeta)
class NeutronPluginBaseV2(object):
- @abstractmethod
+ @abc.abstractmethod
def create_subnet(self, context, subnet):
"""Create a subnet.
"""
pass
- @abstractmethod
+ @abc.abstractmethod
def update_subnet(self, context, id, subnet):
"""Update values of a subnet.
"""
pass
- @abstractmethod
+ @abc.abstractmethod
def get_subnet(self, context, id, fields=None):
"""Retrieve a subnet.
"""
pass
- @abstractmethod
+ @abc.abstractmethod
def get_subnets(self, context, filters=None, fields=None,
sorts=None, limit=None, marker=None, page_reverse=False):
"""Retrieve a list of subnets.
"""
raise NotImplementedError
- @abstractmethod
+ @abc.abstractmethod
def delete_subnet(self, context, id):
"""Delete a subnet.
"""
pass
- @abstractmethod
+ @abc.abstractmethod
def create_network(self, context, network):
"""Create a network.
"""
pass
- @abstractmethod
+ @abc.abstractmethod
def update_network(self, context, id, network):
"""Update values of a network.
"""
pass
- @abstractmethod
+ @abc.abstractmethod
def get_network(self, context, id, fields=None):
"""Retrieve a network.
"""
pass
- @abstractmethod
+ @abc.abstractmethod
def get_networks(self, context, filters=None, fields=None,
sorts=None, limit=None, marker=None, page_reverse=False):
"""Retrieve a list of networks.
"""
raise NotImplementedError
- @abstractmethod
+ @abc.abstractmethod
def delete_network(self, context, id):
"""Delete a network.
"""
pass
- @abstractmethod
+ @abc.abstractmethod
def create_port(self, context, port):
"""Create a port.
"""
pass
- @abstractmethod
+ @abc.abstractmethod
def update_port(self, context, id, port):
"""Update values of a port.
"""
pass
- @abstractmethod
+ @abc.abstractmethod
def get_port(self, context, id, fields=None):
"""Retrieve a port.
"""
pass
- @abstractmethod
+ @abc.abstractmethod
def get_ports(self, context, filters=None, fields=None,
sorts=None, limit=None, marker=None, page_reverse=False):
"""Retrieve a list of ports.
"""
raise NotImplementedError
- @abstractmethod
+ @abc.abstractmethod
def delete_port(self, context, id):
"""Delete a port.
#
# @author: Sumit Naiksatam, Cisco Systems, Inc.
-from abc import ABCMeta, abstractmethod
+import abc
import inspect
-
import six
-@six.add_metaclass(ABCMeta)
+@six.add_metaclass(abc.ABCMeta)
class L2DevicePluginBase(object):
"""Base class for a device-specific plugin.
the configuration on each device.
"""
- @abstractmethod
+ @abc.abstractmethod
def create_network(self, tenant_id, net_name, net_id, vlan_name, vlan_id,
**kwargs):
"""Create network.
"""
pass
- @abstractmethod
+ @abc.abstractmethod
def delete_network(self, tenant_id, net_id, **kwargs):
"""Delete network.
"""
pass
- @abstractmethod
+ @abc.abstractmethod
def update_network(self, tenant_id, net_id, name, **kwargs):
"""Update network.
"""
pass
- @abstractmethod
+ @abc.abstractmethod
def create_port(self, tenant_id, net_id, port_state, port_id, **kwargs):
"""Create port.
"""
pass
- @abstractmethod
+ @abc.abstractmethod
def delete_port(self, tenant_id, net_id, port_id, **kwargs):
"""Delete port.
"""
pass
- @abstractmethod
+ @abc.abstractmethod
def update_port(self, tenant_id, net_id, port_id, **kwargs):
"""Update port.
"""
pass
- @abstractmethod
+ @abc.abstractmethod
def plug_interface(self, tenant_id, net_id, port_id, remote_interface_id,
**kwargs):
"""Plug interface.
"""
pass
- @abstractmethod
+ @abc.abstractmethod
def unplug_interface(self, tenant_id, net_id, port_id, **kwargs):
"""Unplug interface.
from neutron.db import extraroute_db
from neutron.db import l3_db
from neutron.db import models_v2
-from neutron.extensions.flavor import (FLAVOR_NETWORK, FLAVOR_ROUTER)
+from neutron.extensions import flavor as ext_flavor
from neutron.openstack.common import importutils
from neutron.openstack.common import log as logging
from neutron.plugins.metaplugin.common import config # noqa
from neutron.plugins.metaplugin import meta_db_v2
-from neutron.plugins.metaplugin.meta_models_v2 import (NetworkFlavor,
- RouterFlavor)
+from neutron.plugins.metaplugin import meta_models_v2
LOG = logging.getLogger(__name__)
# Hooks used to select records which belong a target plugin.
def _meta_network_model_hook(context, original_model, query):
- return query.outerjoin(NetworkFlavor,
- NetworkFlavor.network_id == models_v2.Network.id)
+ return query.outerjoin(meta_models_v2.NetworkFlavor,
+ meta_models_v2.NetworkFlavor.network_id ==
+ models_v2.Network.id)
def _meta_port_model_hook(context, original_model, query):
- return query.join(NetworkFlavor,
- NetworkFlavor.network_id == models_v2.Port.network_id)
+ return query.join(meta_models_v2.NetworkFlavor,
+ meta_models_v2.NetworkFlavor.network_id ==
+ models_v2.Port.network_id)
def _meta_flavor_filter_hook(query, filters):
- if FLAVOR_NETWORK in filters:
- return query.filter(NetworkFlavor.flavor ==
- filters[FLAVOR_NETWORK][0])
+ if ext_flavor.FLAVOR_NETWORK in filters:
+ return query.filter(meta_models_v2.NetworkFlavor.flavor ==
+ filters[ext_flavor.FLAVOR_NETWORK][0])
return query
def _extend_network_dict(self, context, network):
flavor = self._get_flavor_by_network_id(context, network['id'])
- network[FLAVOR_NETWORK] = flavor
+ network[ext_flavor.FLAVOR_NETWORK] = flavor
def create_network(self, context, network):
n = network['network']
- flavor = n.get(FLAVOR_NETWORK)
+ flavor = n.get(ext_flavor.FLAVOR_NETWORK)
if str(flavor) not in self.plugins:
flavor = self.default_flavor
plugin = self._get_plugin(flavor)
plugin = self._get_plugin(flavor)
net = plugin.get_network(context, id, fields)
net['id'] = id
- if not fields or FLAVOR_NETWORK in fields:
+ if not fields or ext_flavor.FLAVOR_NETWORK in fields:
self._extend_network_dict(context, net)
if fields and 'id' not in fields:
del net['id']
def get_networks(self, context, filters=None, fields=None):
nets = []
for flavor, plugin in self.plugins.items():
- if (filters and FLAVOR_NETWORK in filters and
- not flavor in filters[FLAVOR_NETWORK]):
+ if (filters and ext_flavor.FLAVOR_NETWORK in filters and
+ not flavor in filters[ext_flavor.FLAVOR_NETWORK]):
continue
if filters:
#NOTE: copy each time since a target plugin may modify
plugin_filters = filters.copy()
else:
plugin_filters = {}
- plugin_filters[FLAVOR_NETWORK] = [flavor]
+ plugin_filters[ext_flavor.FLAVOR_NETWORK] = [flavor]
plugin_nets = plugin.get_networks(context, plugin_filters, fields)
for net in plugin_nets:
- if not fields or FLAVOR_NETWORK in fields:
- net[FLAVOR_NETWORK] = flavor
+ if not fields or ext_flavor.FLAVOR_NETWORK in fields:
+ net[ext_flavor.FLAVOR_NETWORK] = flavor
nets.append(net)
return nets
plugin_filters = filters.copy()
else:
plugin_filters = {}
- plugin_filters[FLAVOR_NETWORK] = [flavor]
+ plugin_filters[ext_flavor.FLAVOR_NETWORK] = [flavor]
ports = plugin.get_ports(context, plugin_filters, fields)
all_ports += ports
return all_ports
def _extend_router_dict(self, context, router):
flavor = self._get_flavor_by_router_id(context, router['id'])
- router[FLAVOR_ROUTER] = flavor
+ router[ext_flavor.FLAVOR_ROUTER] = flavor
def create_router(self, context, router):
r = router['router']
- flavor = r.get(FLAVOR_ROUTER)
+ flavor = r.get(ext_flavor.FLAVOR_ROUTER)
if str(flavor) not in self.l3_plugins:
flavor = self.default_l3_flavor
plugin = self._get_l3_plugin(flavor)
flavor = meta_db_v2.get_flavor_by_router(context.session, id)
plugin = self._get_l3_plugin(flavor)
router = plugin.get_router(context, id, fields)
- if not fields or FLAVOR_ROUTER in fields:
+ if not fields or ext_flavor.FLAVOR_ROUTER in fields:
self._extend_router_dict(context, router)
return router
def get_routers_with_flavor(self, context, filters=None,
fields=None):
collection = self._model_query(context, l3_db.Router)
- r_model = RouterFlavor
+ r_model = meta_models_v2.RouterFlavor
collection = collection.join(r_model,
l3_db.Router.id == r_model.router_id)
if filters:
for key, value in filters.iteritems():
- if key == FLAVOR_ROUTER:
- column = RouterFlavor.flavor
+ if key == ext_flavor.FLAVOR_ROUTER:
+ column = meta_models_v2.RouterFlavor.flavor
else:
column = getattr(l3_db.Router, key, None)
if column:
# License for the specific language governing permissions and limitations
# under the License.
-from abc import ABCMeta, abstractmethod, abstractproperty
-
+import abc
import six
# The following keys are used in the segment dictionaries passed via
SEGMENTATION_ID = 'segmentation_id'
-@six.add_metaclass(ABCMeta)
+@six.add_metaclass(abc.ABCMeta)
class TypeDriver(object):
"""Define stable abstract interface for ML2 type drivers.
either be excluded or stored as None.
"""
- @abstractmethod
+ @abc.abstractmethod
def get_type(self):
"""Get driver's network type.
"""
pass
- @abstractmethod
+ @abc.abstractmethod
def initialize(self):
"""Perform driver initialization.
"""
pass
- @abstractmethod
+ @abc.abstractmethod
def validate_provider_segment(self, segment):
"""Validate attributes of a provider network segment.
"""
pass
- @abstractmethod
+ @abc.abstractmethod
def reserve_provider_segment(self, session, segment):
"""Reserve resource associated with a provider network segment.
"""
pass
- @abstractmethod
+ @abc.abstractmethod
def allocate_tenant_segment(self, session):
"""Allocate resource for a new tenant network segment.
"""
pass
- @abstractmethod
+ @abc.abstractmethod
def release_segment(self, session, segment):
"""Release network segment.
pass
-@six.add_metaclass(ABCMeta)
+@six.add_metaclass(abc.ABCMeta)
class NetworkContext(object):
"""Context passed to MechanismDrivers for changes to network resources.
MechanismDrivers can freely access the same information.
"""
- @abstractproperty
+ @abc.abstractproperty
def current(self):
"""Return the current state of the network.
"""
pass
- @abstractproperty
+ @abc.abstractproperty
def original(self):
"""Return the original state of the network.
"""
pass
- @abstractproperty
+ @abc.abstractproperty
def network_segments(self):
"""Return the segments associated with this network resource."""
pass
-@six.add_metaclass(ABCMeta)
+@six.add_metaclass(abc.ABCMeta)
class SubnetContext(object):
"""Context passed to MechanismDrivers for changes to subnet resources.
MechanismDrivers can freely access the same information.
"""
- @abstractproperty
+ @abc.abstractproperty
def current(self):
"""Return the current state of the subnet.
"""
pass
- @abstractproperty
+ @abc.abstractproperty
def original(self):
"""Return the original state of the subnet.
pass
-@six.add_metaclass(ABCMeta)
+@six.add_metaclass(abc.ABCMeta)
class PortContext(object):
"""Context passed to MechanismDrivers for changes to port resources.
freely access the same information.
"""
- @abstractproperty
+ @abc.abstractproperty
def current(self):
"""Return the current state of the port.
"""
pass
- @abstractproperty
+ @abc.abstractproperty
def original(self):
"""Return the original state of the port.
"""
pass
- @abstractproperty
+ @abc.abstractproperty
def network(self):
"""Return the NetworkContext associated with this port."""
pass
- @abstractproperty
+ @abc.abstractproperty
def bound_segment(self):
"""Return the currently bound segment dictionary."""
pass
- @abstractproperty
+ @abc.abstractproperty
def original_bound_segment(self):
"""Return the original bound segment dictionary.
"""
pass
- @abstractproperty
+ @abc.abstractproperty
def bound_driver(self):
"""Return the currently bound mechanism driver name."""
pass
- @abstractproperty
+ @abc.abstractproperty
def original_bound_driver(self):
"""Return the original bound mechanism driver name.
"""
pass
- @abstractmethod
+ @abc.abstractmethod
def host_agents(self, agent_type):
"""Get agents of the specified type on port's host.
"""
pass
- @abstractmethod
+ @abc.abstractmethod
def set_binding(self, segment_id, vif_type, vif_details):
"""Set the binding for the port.
pass
-@six.add_metaclass(ABCMeta)
+@six.add_metaclass(abc.ABCMeta)
class MechanismDriver(object):
"""Define stable abstract interface for ML2 mechanism drivers.
methods that are part of the database transaction.
"""
- @abstractmethod
+ @abc.abstractmethod
def initialize(self):
"""Perform driver initialization.
# License for the specific language governing permissions and limitations
# under the License.
-from abc import ABCMeta, abstractmethod
-
+import abc
import six
from neutron.extensions import portbindings
LOG = log.getLogger(__name__)
-@six.add_metaclass(ABCMeta)
+@six.add_metaclass(abc.ABCMeta)
class AgentMechanismDriverBase(api.MechanismDriver):
"""Base class for drivers that attach to networks using an L2 agent.
LOG.warning(_("Attempting to bind with dead agent: %s"),
agent)
- @abstractmethod
+ @abc.abstractmethod
def try_to_bind_segment_for_agent(self, context, segment, agent):
"""Try to bind with segment for agent.
"""
-@six.add_metaclass(ABCMeta)
+@six.add_metaclass(abc.ABCMeta)
class SimpleAgentMechanismDriverBase(AgentMechanismDriverBase):
"""Base class for simple drivers using an L2 agent.
else:
return False
- @abstractmethod
+ @abc.abstractmethod
def check_segment_for_agent(self, segment, agent):
"""Check if segment can be bound for agent.
# 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 ABCMeta, abstractmethod
-
+import abc
import six
from neutron.common import exceptions as exc
TUNNEL = 'tunnel'
-@six.add_metaclass(ABCMeta)
+@six.add_metaclass(abc.ABCMeta)
class TunnelTypeDriver(api.TypeDriver):
"""Define stable abstract interface for ML2 type drivers.
methods to manage these endpoints.
"""
- @abstractmethod
+ @abc.abstractmethod
def add_endpoint(self, ip):
"""Register the endpoint in the type_driver database.
"""
pass
- @abstractmethod
+ @abc.abstractmethod
def get_endpoints(self):
"""Get every endpoint managed by the type_driver
# @author: Ryota MIBU
# @author: Akihiro MOTOKI
-from abc import ABCMeta, abstractmethod
-
+import abc
import six
-@six.add_metaclass(ABCMeta)
+@six.add_metaclass(abc.ABCMeta)
class OFCDriverBase(object):
"""OpenFlow Controller (OFC) Driver Specification.
It would be better that other methods like update_* are implemented.
"""
- @abstractmethod
+ @abc.abstractmethod
def create_tenant(self, description, tenant_id=None):
"""Create a new tenant at OpenFlow Controller.
"""
pass
- @abstractmethod
+ @abc.abstractmethod
def delete_tenant(self, ofc_tenant_id):
"""Delete a tenant at OpenFlow Controller.
"""
pass
- @abstractmethod
+ @abc.abstractmethod
def create_network(self, ofc_tenant_id, description, network_id=None):
"""Create a new network on specified OFC tenant at OpenFlow Controller.
"""
pass
- @abstractmethod
+ @abc.abstractmethod
def delete_network(self, ofc_network_id):
"""Delete a netwrok at OpenFlow Controller.
"""
pass
- @abstractmethod
+ @abc.abstractmethod
def create_port(self, ofc_network_id, portinfo,
port_id=None, filters=None):
"""Create a new port on specified network at OFC.
"""
pass
- @abstractmethod
+ @abc.abstractmethod
def delete_port(self, ofc_port_id):
"""Delete a port at OpenFlow Controller.
from neutron.db import api as db
from neutron.db import db_base_plugin_v2
from neutron.db import models_v2
-from neutron.extensions.flavor import (FLAVOR_NETWORK, FLAVOR_ROUTER)
+from neutron.extensions import flavor as ext_flavor
from neutron.openstack.common import uuidutils
from neutron.plugins.metaplugin.meta_neutron_plugin import (
FaildToAddFlavorBinding)
'shared': False,
'router:external': [],
'tenant_id': self.fake_tenant_id,
- FLAVOR_NETWORK: flavor}}
+ ext_flavor.FLAVOR_NETWORK: flavor}}
return data
def _fake_port(self, net_id):
def _fake_router(self, flavor):
data = {'router': {'name': flavor, 'admin_state_up': True,
'tenant_id': self.fake_tenant_id,
- FLAVOR_ROUTER: flavor,
+ ext_flavor.FLAVOR_ROUTER: flavor,
'external_gateway_info': None}}
return data
def test_create_delete_network(self):
network1 = self._fake_network('fake1')
ret1 = self.plugin.create_network(self.context, network1)
- self.assertEqual('fake1', ret1[FLAVOR_NETWORK])
+ self.assertEqual('fake1', ret1[ext_flavor.FLAVOR_NETWORK])
network2 = self._fake_network('fake2')
ret2 = self.plugin.create_network(self.context, network2)
- self.assertEqual('fake2', ret2[FLAVOR_NETWORK])
+ self.assertEqual('fake2', ret2[ext_flavor.FLAVOR_NETWORK])
network3 = self._fake_network('proxy')
ret3 = self.plugin.create_network(self.context, network3)
- self.assertEqual('proxy', ret3[FLAVOR_NETWORK])
+ self.assertEqual('proxy', ret3[ext_flavor.FLAVOR_NETWORK])
db_ret1 = self.plugin.get_network(self.context, ret1['id'])
self.assertEqual('fake1', db_ret1['name'])
db_ret4 = self.plugin.get_networks(self.context)
self.assertEqual(3, len(db_ret4))
- db_ret5 = self.plugin.get_networks(self.context,
- {FLAVOR_NETWORK: ['fake1']})
+ db_ret5 = self.plugin.get_networks(
+ self.context,
+ {ext_flavor.FLAVOR_NETWORK: ['fake1']})
self.assertEqual(1, len(db_ret5))
self.assertEqual('fake1', db_ret5[0]['name'])
self.plugin.delete_network(self.context, ret1['id'])
router2 = self._fake_router('fake2')
router_ret2 = self.plugin.create_router(self.context, router2)
- self.assertEqual('fake1', router_ret1[FLAVOR_ROUTER])
- self.assertEqual('fake2', router_ret2[FLAVOR_ROUTER])
+ self.assertEqual('fake1', router_ret1[ext_flavor.FLAVOR_ROUTER])
+ self.assertEqual('fake2', router_ret2[ext_flavor.FLAVOR_ROUTER])
router_in_db1 = self.plugin.get_router(self.context, router_ret1['id'])
router_in_db2 = self.plugin.get_router(self.context, router_ret2['id'])
- self.assertEqual('fake1', router_in_db1[FLAVOR_ROUTER])
- self.assertEqual('fake2', router_in_db2[FLAVOR_ROUTER])
+ self.assertEqual('fake1', router_in_db1[ext_flavor.FLAVOR_ROUTER])
+ self.assertEqual('fake2', router_in_db2[ext_flavor.FLAVOR_ROUTER])
self.plugin.delete_router(self.context, router_ret1['id'])
self.plugin.delete_router(self.context, router_ret2['id'])
import mock
import requests
-from neutron.services.loadbalancer.drivers.netscaler import (
- ncc_client, netscaler_driver
-)
+from neutron.services.loadbalancer.drivers.netscaler import ncc_client
+from neutron.services.loadbalancer.drivers.netscaler import netscaler_driver
from neutron.tests.unit import testlib_api
NCC_CLIENT_CLASS = ('neutron.services.loadbalancer.drivers'
# E711/E712 comparison to False should be 'if cond is False:' or 'if not cond:'
# query = query.filter(Component.disabled == False)
# E125 continuation line does not distinguish itself from next logical line
-# H301 one import per line
# H302 import only modules
# TODO(marun) H404 multi line docstring should start with a summary
-ignore = E711,E712,E125,H301,H302,H404
+ignore = E711,E712,E125,H302,H404
show-source = true
builtins = _
exclude=.venv,.git,.tox,dist,doc,*openstack/common*,*lib/python*,*egg,tools
Code Review / openstack-build / neutron-build.git / commitdiff