from neutron.agent.linux import ovs_lib
from neutron.agent.linux import utils
from neutron.common import exceptions
-from neutron.extensions.flavor import (FLAVOR_NETWORK)
+from neutron.extensions import flavor
from neutron.openstack.common import importutils
from neutron.openstack.common import log as logging
region_name=self.conf.auth_region
)
self.flavor_driver_map = {}
- for flavor, driver_name in [
+ for net_flavor, driver_name in [
driver_set.split(':')
for driver_set in
self.conf.meta_flavor_driver_mappings.split(',')]:
- self.flavor_driver_map[flavor] = self._load_driver(driver_name)
+ self.flavor_driver_map[net_flavor] = self._load_driver(driver_name)
def _get_flavor_by_network_id(self, network_id):
network = self.neutron.show_network(network_id)
- return network['network'][FLAVOR_NETWORK]
+ return network['network'][flavor.FLAVOR_NETWORK]
def _get_driver_by_network_id(self, network_id):
- flavor = self._get_flavor_by_network_id(network_id)
- return self.flavor_driver_map[flavor]
+ net_flavor = self._get_flavor_by_network_id(network_id)
+ return self.flavor_driver_map[net_flavor]
def _set_device_plugin_tag(self, network_id, device_name, namespace=None):
plugin_tag = self._get_flavor_by_network_id(network_id)
# License for the specific language governing permissions and limitations
# under the License.
-from abc import ABCMeta
+import abc
import imp
import itertools
import os
from neutron.api.v2 import attributes
from neutron.common import exceptions
import neutron.extensions
-from neutron.manager import NeutronManager
+from neutron import manager
from neutron.openstack.common import log as logging
from neutron import policy
from neutron import wsgi
LOG = logging.getLogger(__name__)
-@six.add_metaclass(ABCMeta)
+@six.add_metaclass(abc.ABCMeta)
class PluginInterface(object):
@classmethod
def get_instance(cls):
if cls._instance is None:
cls._instance = cls(get_extensions_path(),
- NeutronManager.get_service_plugins())
+ manager.NeutronManager.get_service_plugins())
return cls._instance
def check_if_plugin_extensions_loaded(self):
from neutron.openstack.common.db import options as db_options
from neutron.openstack.common import log as logging
from neutron.openstack.common import rpc
-from neutron.version import version_info as neutron_version
+from neutron import version
LOG = logging.getLogger(__name__)
def parse(args):
cfg.CONF(args=args, project='neutron',
- version='%%prog %s' % neutron_version.release_string())
+ version='%%prog %s' % version.version_info.release_string())
# Validate that the base_mac is of the correct format
msg = attributes._validate_regex(cfg.CONF.base_mac,
import copy
-from datetime import datetime
+import datetime
from neutron.db import api as db_api
from neutron.openstack.common import context as common_context
self.read_deleted = read_deleted
if not timestamp:
- timestamp = datetime.utcnow()
+ timestamp = datetime.datetime.utcnow()
self.timestamp = timestamp
self._session = None
self.roles = roles or []
from cliff import lister
from neutronclient.common import utils
-from neutronclient.neutron.v2_0 import NeutronCommand
-from neutronclient.neutron.v2_0.port import _format_fixed_ips
+from neutronclient.neutron import v2_0 as client
+from neutronclient.neutron.v2_0 import port
from neutron.openstack.common import log as logging
-class ProbeCommand(NeutronCommand):
+class ProbeCommand(client.NeutronCommand):
log = logging.getLogger(__name__ + '.ProbeCommand')
def get_debug_agent(self):
def run(self, parsed_args):
self.log.debug('run(%s)' % parsed_args)
debug_agent = self.get_debug_agent()
- port = debug_agent.create_probe(parsed_args.id,
- parsed_args.device_owner)
- self.log.info(_('Probe created : %s '), port.id)
+ probe_port = debug_agent.create_probe(parsed_args.id,
+ parsed_args.device_owner)
+ self.log.info(_('Probe created : %s '), probe_port.id)
class DeleteProbe(ProbeCommand):
self.log.info(_('Probe %s deleted'), parsed_args.id)
-class ListProbe(NeutronCommand, lister.Lister):
+class ListProbe(client.NeutronCommand, lister.Lister):
"""List probes."""
log = logging.getLogger(__name__ + '.ListProbe')
- _formatters = {'fixed_ips': _format_fixed_ips, }
+ _formatters = {'fixed_ips': port._format_fixed_ips, }
def get_debug_agent(self):
return self.app.debug_agent
from oslo.config import cfg
from neutron.agent.common import config
-from neutron.agent.linux.dhcp import DictModel
+from neutron.agent.linux import dhcp
from neutron.agent.linux import ip_lib
from neutron.agent.linux import utils
from neutron.openstack.common import log as logging
def _get_subnet(self, subnet_id):
subnet_dict = self.client.show_subnet(subnet_id)['subnet']
- return DictModel(subnet_dict)
+ return dhcp.DictModel(subnet_dict)
def _get_network(self, network_id):
network_dict = self.client.show_network(network_id)['network']
- network = DictModel(network_dict)
+ network = dhcp.DictModel(network_dict)
network.external = network_dict.get('router:external')
obj_subnet = [self._get_subnet(s_id) for s_id in network.subnets]
network.subnets = obj_subnet
self.delete_probe(port['id'])
def delete_probe(self, port_id):
- port = DictModel(self.client.show_port(port_id)['port'])
+ port = dhcp.DictModel(self.client.show_port(port_id)['port'])
network = self._get_network(port.network_id)
bridge = None
if network.external:
DEVICE_OWNER_COMPUTE_PROBE])
info = ports['ports']
for port in info:
- port['device_name'] = self.driver.get_device_name(DictModel(port))
+ port['device_name'] = self.driver.get_device_name(
+ dhcp.DictModel(port))
return info
def exec_command(self, port_id, command=None):
- port = DictModel(self.client.show_port(port_id)['port'])
+ port = dhcp.DictModel(self.client.show_port(port_id)['port'])
ip = ip_lib.IPWrapper(self.root_helper)
namespace = self._get_namespace(port)
if self.conf.use_namespaces:
device_owner=DEVICE_OWNER_NETWORK_PROBE)
info = ports.get('ports', [])
if info:
- return DictModel(info[0])
+ return dhcp.DictModel(info[0])
else:
return self.create_probe(network_id)
'fixed_ips': [dict(subnet_id=s.id)
for s in network.subnets]}}
port_dict = self.client.create_port(body)['port']
- port = DictModel(port_dict)
+ port = dhcp.DictModel(port_dict)
port.network = network
for fixed_ip in port.fixed_ips:
fixed_ip.subnet = self._get_subnet(fixed_ip.subnet_id)
from neutron.agent.common import config
from neutron.agent.linux import interface
-from neutron.debug.debug_agent import NeutronDebugAgent
+from neutron.debug import debug_agent
from neutron.openstack.common import importutils
from neutronclient.common import exceptions as exc
from neutronclient.common import utils
" either --config-file or env[NEUTRON_TEST_CONFIG_FILE]"))
client = self.client_manager.neutron
cfg.CONF.register_opts(interface.OPTS)
- cfg.CONF.register_opts(NeutronDebugAgent.OPTS)
+ cfg.CONF.register_opts(debug_agent.NeutronDebugAgent.OPTS)
config.register_interface_driver_opts_helper(cfg.CONF)
config.register_use_namespaces_opts_helper(cfg.CONF)
config.register_root_helper(cfg.CONF)
cfg.CONF(['--config-file', self.options.config_file])
config.setup_logging(cfg.CONF)
driver = importutils.import_object(cfg.CONF.interface_driver, cfg.CONF)
- self.debug_agent = NeutronDebugAgent(cfg.CONF, client, driver)
+ self.debug_agent = debug_agent.NeutronDebugAgent(cfg.CONF,
+ client,
+ driver)
def main(argv=None):
[flake8]
# E125 continuation line does not distinguish itself from next logical line
-# H302 import only modules
# TODO(marun) H404 multi line docstring should start with a summary
-ignore = E125,H302,H404
+ignore = E125,H404
show-source = true
builtins = _
exclude = .venv,.git,.tox,dist,doc,*openstack/common*,*lib/python*,*egg,build,tools,.ropeproject