openstack/neutron
Code Issues Proposed changes

use service type constants from neutron_lib plugins

Browse Source 
The well known service type constants are in
neutron_lib.plugins.constants, but for legacy reasons a few still exist
and are referenced from neutron_lib.constants that we'd like to remove.
This patch switches references over to neutron_lib's plugin constants.

Change-Id: I1861448cec303725b30cef8f42029f467f9e03a3
This commit is contained in:
Boden R 2017-06-23 11:04:34 -06:00
parent e3855c8b7c
commit 0a1405794f
49 changed files with 161 additions and 130 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
neutron/api/rpc/agentnotifiers/l3_rpc_agent_api.py
View File

@ -16,6 +16,7 @@
import random
from neutron_lib import constants
from neutron_lib.plugins import constants as plugin_constants
from neutron_lib.plugins import directory
from oslo_log import log as logging
import oslo_messaging
@ -56,7 +57,7 @@ class L3AgentNotifyAPI(object):
shuffle_agents):
"""Notify changed routers to hosting l3 agents."""
adminContext = context if context.is_admin else context.elevated()
plugin = directory.get_plugin(constants.L3)
plugin = directory.get_plugin(plugin_constants.L3)
for router_id in router_ids:
hosts = plugin.get_hosts_to_notify(adminContext, router_id)
if shuffle_agents:
@ -85,7 +86,7 @@ class L3AgentNotifyAPI(object):
def _notification(self, context, method, router_ids, operation,
shuffle_agents, schedule_routers=True):
"""Notify all the agents that are hosting the routers."""
plugin = directory.get_plugin(constants.L3)
plugin = directory.get_plugin(plugin_constants.L3)
if not plugin:
LOG.error(_LE('No plugin for L3 routing registered. Cannot notify '
'agents with the message %s'), method)

5
neutron/api/rpc/agentnotifiers/metering_rpc_agent_api.py
View File

@ -13,6 +13,7 @@
# under the License.
from neutron_lib import constants
from neutron_lib.plugins import constants as plugin_constants
from neutron_lib.plugins import directory
from oslo_log import log as logging
import oslo_messaging
@ -36,7 +37,7 @@ class MeteringAgentNotifyAPI(object):
def _agent_notification(self, context, method, routers):
"""Notify l3 metering agents hosted by l3 agent hosts."""
adminContext = context if context.is_admin else context.elevated()
plugin = directory.get_plugin(constants.L3)
plugin = directory.get_plugin(plugin_constants.L3)
l3_routers = {}
state = agentschedulers_db.get_admin_state_up_filter()
@ -71,7 +72,7 @@ class MeteringAgentNotifyAPI(object):
def _notification(self, context, method, routers):
"""Notify all the agents that are hosting the routers."""
plugin = directory.get_plugin(constants.L3)
plugin = directory.get_plugin(plugin_constants.L3)
if utils.is_extension_supported(
plugin, constants.L3_AGENT_SCHEDULER_EXT_ALIAS):
self._agent_notification(context, method, routers)

3
neutron/api/rpc/handlers/l3_rpc.py
View File

@ -17,6 +17,7 @@ from neutron_lib.api.definitions import portbindings
from neutron_lib import constants
from neutron_lib import context as neutron_context
from neutron_lib import exceptions
from neutron_lib.plugins import constants as plugin_constants
from neutron_lib.plugins import directory
from oslo_config import cfg
from oslo_log import log as logging
@ -56,7 +57,7 @@ class L3RpcCallback(object):
@property
def l3plugin(self):
if not hasattr(self, '_l3plugin'):
self._l3plugin = directory.get_plugin(constants.L3)
self._l3plugin = directory.get_plugin(plugin_constants.L3)
return self._l3plugin
def _update_ha_network_port_status(self, context, host_id):

2
neutron/api/v2/resource_helper.py
View File

@ -13,7 +13,7 @@
# License for the specific language governing permissions and limitations
# under the License.
from neutron_lib import constants
from neutron_lib.plugins import constants
from neutron_lib.plugins import directory
from oslo_log import log as logging

2
neutron/cmd/eventlet/usage_audit.py
View File

@ -19,8 +19,8 @@ subnets.
import sys
from neutron_lib import constants
from neutron_lib import context
from neutron_lib.plugins import constants
from neutron_lib.plugins import directory
from neutron.common import config

2
neutron/db/availability_zone/router.py
View File

@ -14,7 +14,7 @@
from neutron_lib.callbacks import events
from neutron_lib.callbacks import registry
from neutron_lib.callbacks import resources
from neutron_lib import constants
from neutron_lib.plugins import constants
from neutron_lib.plugins import directory
from neutron.common import utils

7
neutron/db/db_base_plugin_v2.py
View File

@ -24,6 +24,7 @@ from neutron_lib.callbacks import resources
from neutron_lib import constants
from neutron_lib import context as ctx
from neutron_lib import exceptions as exc
from neutron_lib.plugins import constants as plugin_constants
from neutron_lib.plugins import directory
from oslo_config import cfg
from oslo_db import exception as os_db_exc
@ -621,7 +622,7 @@ class NeutronDbPluginV2(db_base_plugin_common.DbBasePluginCommon,
raise exc.BadRequest(resource='subnets', msg=reason)
def _update_router_gw_ports(self, context, network, subnet):
l3plugin = directory.get_plugin(constants.L3)
l3plugin = directory.get_plugin(plugin_constants.L3)
if l3plugin:
gw_ports = self._get_router_gw_ports_by_network(context,
network['id'])
@ -635,7 +636,7 @@ class NeutronDbPluginV2(db_base_plugin_common.DbBasePluginCommon,
{'id': id, 's': subnet})
def _update_router_gw_port(self, context, router_id, network, subnet):
l3plugin = directory.get_plugin(constants.L3)
l3plugin = directory.get_plugin(plugin_constants.L3)
ctx_admin = context.elevated()
ext_subnets_dict = {s['id']: s for s in network['subnets']}
router = l3plugin.get_router(ctx_admin, router_id)
@ -1406,7 +1407,7 @@ class NeutronDbPluginV2(db_base_plugin_common.DbBasePluginCommon,
except l3.RouterNotFound:
return
else:
l3plugin = directory.get_plugin(constants.L3)
l3plugin = directory.get_plugin(plugin_constants.L3)
if l3plugin:
try:
ctx_admin = context.elevated()

3
neutron/db/external_net_db.py
View File

@ -19,6 +19,7 @@ from neutron_lib.callbacks import registry
from neutron_lib.callbacks import resources
from neutron_lib import constants
from neutron_lib import exceptions as n_exc
from neutron_lib.plugins import constants as plugin_constants
from neutron_lib.plugins import directory
from sqlalchemy.sql import expression as expr
@ -139,7 +140,7 @@ class External_net_db_mixin(object):
net_data[external_net.EXTERNAL] = False
def _process_l3_delete(self, context, network_id):
l3plugin = directory.get_plugin(constants.L3)
l3plugin = directory.get_plugin(plugin_constants.L3)
if l3plugin:
l3plugin.delete_disassociated_floatingips(context, network_id)

5
neutron/db/l3_agentschedulers_db.py
View File

@ -14,6 +14,7 @@
# under the License.
from neutron_lib import constants
from neutron_lib.plugins import constants as plugin_constants
from neutron_lib.plugins import directory
from oslo_config import cfg
from oslo_db import exception as db_exc
@ -152,7 +153,7 @@ class L3AgentSchedulerDbMixin(l3agentscheduler.L3AgentSchedulerPluginBase,
router_id = router['id']
agent_id = agent['id']
if self.router_scheduler:
plugin = directory.get_plugin(constants.L3)
plugin = directory.get_plugin(plugin_constants.L3)
try:
if router.get('ha'):
self.router_scheduler.create_ha_port_and_bind(
@ -199,7 +200,7 @@ class L3AgentSchedulerDbMixin(l3agentscheduler.L3AgentSchedulerPluginBase,
self._unbind_router(context, router_id, agent_id)
router = self.get_router(context, router_id)
plugin = directory.get_plugin(constants.L3)
plugin = directory.get_plugin(plugin_constants.L3)
if router.get('ha'):
plugin.delete_ha_interfaces_on_host(context, router_id, agent.host)
# NOTE(Swami): Need to verify if there are DVR serviceable

17
neutron/db/l3_db.py
View File

@ -26,6 +26,7 @@ from neutron_lib.callbacks import resources
from neutron_lib import constants
from neutron_lib import context as n_ctx
from neutron_lib import exceptions as n_exc
from neutron_lib.plugins import constants as plugin_constants
from neutron_lib.plugins import directory
from neutron_lib.services import base as base_services
from oslo_log import log as logging
@ -95,7 +96,7 @@ class L3_NAT_dbonly_mixin(l3.RouterPluginBase,
context = kwargs['context']
port_id = kwargs['port_id']
port_check = kwargs['port_check']
l3plugin = directory.get_plugin(constants.L3)
l3plugin = directory.get_plugin(plugin_constants.L3)
if l3plugin and port_check:
l3plugin.prevent_l3_port_deletion(context, port_id)
@ -271,7 +272,7 @@ class L3_NAT_dbonly_mixin(l3.RouterPluginBase,
candidates = None
router_db = self._update_router_db(context, id, r)
if candidates:
l3_plugin = directory.get_plugin(constants.L3)
l3_plugin = directory.get_plugin(plugin_constants.L3)
l3_plugin.reschedule_router(context, id, candidates)
updated = self._make_router_dict(router_db)
registry.notify(resources.ROUTER, events.AFTER_UPDATE, self,
@ -307,7 +308,7 @@ class L3_NAT_dbonly_mixin(l3.RouterPluginBase,
# first get plugin supporting l3 agent scheduling
# (either l3 service plugin or core_plugin)
l3_plugin = directory.get_plugin(constants.L3)
l3_plugin = directory.get_plugin(plugin_constants.L3)
if (not utils.is_extension_supported(
l3_plugin,
constants.L3_AGENT_SCHEDULER_EXT_ALIAS) or
@ -1753,16 +1754,16 @@ class L3RpcNotifierMixin(object):
def _notify_routers_callback(resource, event, trigger, **kwargs):
context = kwargs['context']
router_ids = kwargs['router_ids']
l3plugin = directory.get_plugin(constants.L3)
l3plugin = directory.get_plugin(plugin_constants.L3)
if l3plugin:
l3plugin.notify_routers_updated(context, router_ids)
else:
LOG.debug('%s not configured', constants.L3)
LOG.debug('%s not configured', plugin_constants.L3)
@staticmethod
@registry.receives(resources.SUBNET, [events.AFTER_UPDATE])
def _notify_subnet_gateway_ip_update(resource, event, trigger, **kwargs):
l3plugin = directory.get_plugin(constants.L3)
l3plugin = directory.get_plugin(plugin_constants.L3)
if not l3plugin:
return
context = kwargs['context']
@ -1800,11 +1801,11 @@ class L3RpcNotifierMixin(object):
query = query.distinct()
router_ids = [r[0] for r in query]
l3plugin = directory.get_plugin(constants.L3)
l3plugin = directory.get_plugin(plugin_constants.L3)
if l3plugin:
l3plugin.notify_routers_updated(context, router_ids)
else:
LOG.debug('%s not configured', constants.L3)
LOG.debug('%s not configured', plugin_constants.L3)
@property
def l3_rpc_notifier(self):

5
neutron/db/l3_dvr_db.py
View File

@ -21,6 +21,7 @@ from neutron_lib.callbacks import registry
from neutron_lib.callbacks import resources
from neutron_lib import constants as const
from neutron_lib import exceptions as n_exc
from neutron_lib.plugins import constants as plugin_constants
from neutron_lib.plugins import directory
from oslo_config import cfg
from oslo_log import helpers as log_helper
@ -70,7 +71,7 @@ class DVRResourceOperationHandler(object):
@property
def l3plugin(self):
return directory.get_plugin(const.L3)
return directory.get_plugin(plugin_constants.L3)
@registry.receives(resources.ROUTER, [events.PRECOMMIT_CREATE])
def _set_distributed_flag(self, resource, event, trigger, context,
@ -500,7 +501,7 @@ class DVRResourceOperationHandler(object):
if not router.extra_attributes.distributed:
return
plugin = directory.get_plugin(const.L3)
plugin = directory.get_plugin(plugin_constants.L3)
# we calculate which hosts to notify by checking the hosts for
# the removed port's subnets and then subtract out any hosts still

9
neutron/db/l3_dvrscheduler_db.py
View File

@ -18,6 +18,7 @@ from neutron_lib.callbacks import events
from neutron_lib.callbacks import registry
from neutron_lib.callbacks import resources
from neutron_lib import constants as n_const
from neutron_lib.plugins import constants as plugin_constants
from neutron_lib.plugins import directory
from oslo_log import log as logging
from sqlalchemy import or_
@ -94,7 +95,7 @@ class L3_DVRsch_db_mixin(l3agent_sch_db.L3AgentSchedulerDbMixin):
# Make sure we create the floatingip agent gateway port
# for the destination node if fip is associated with this
# fixed port
l3plugin = directory.get_plugin(n_const.L3)
l3plugin = directory.get_plugin(plugin_constants.L3)
(
l3plugin.
check_for_fip_and_create_agent_gw_port_on_host_if_not_exists(
@ -373,7 +374,7 @@ def _notify_l3_agent_new_port(resource, event, trigger, **kwargs):
return
if n_utils.is_dvr_serviced(port['device_owner']):
l3plugin = directory.get_plugin(n_const.L3)
l3plugin = directory.get_plugin(plugin_constants.L3)
context = kwargs['context']
l3plugin.dvr_handle_new_service_port(context, port)
l3plugin.update_arp_entry_for_dvr_service_port(context, port)
@ -382,7 +383,7 @@ def _notify_l3_agent_new_port(resource, event, trigger, **kwargs):
def _notify_port_delete(event, resource, trigger, **kwargs):
context = kwargs['context']
port = kwargs['port']
l3plugin = directory.get_plugin(n_const.L3)
l3plugin = directory.get_plugin(plugin_constants.L3)
if port:
port_host = port.get(portbindings.HOST_ID)
allowed_address_pairs_list = port.get('allowed_address_pairs')
@ -405,7 +406,7 @@ def _notify_l3_agent_port_update(resource, event, trigger, **kwargs):
original_device_owner = original_port.get('device_owner', '')
new_device_owner = new_port.get('device_owner', '')
is_new_device_dvr_serviced = n_utils.is_dvr_serviced(new_device_owner)
l3plugin = directory.get_plugin(n_const.L3)
l3plugin = directory.get_plugin(plugin_constants.L3)
context = kwargs['context']
is_port_no_longer_serviced = (
n_utils.is_dvr_serviced(original_device_owner) and

3
neutron/db/l3_hascheduler_db.py
View File

@ -17,6 +17,7 @@ from neutron_lib.callbacks import events
from neutron_lib.callbacks import registry
from neutron_lib.callbacks import resources
from neutron_lib import constants
from neutron_lib.plugins import constants as plugin_constants
from neutron_lib.plugins import directory
from neutron.db import l3_agentschedulers_db as l3_sch_db
@ -64,7 +65,7 @@ def _notify_l3_agent_ha_port_update(resource, event, trigger, **kwargs):
if (new_device_owner == constants.DEVICE_OWNER_ROUTER_HA_INTF and
new_port['status'] == constants.PORT_STATUS_ACTIVE and
original_port['status'] != new_port['status']):
l3plugin = directory.get_plugin(constants.L3)
l3plugin = directory.get_plugin(plugin_constants.L3)
l3plugin.l3_rpc_notifier.routers_updated_on_host(
context, [new_port['device_id']], host)

3
neutron/db/metering/metering_rpc.py
View File

@ -13,6 +13,7 @@
# under the License.
from neutron_lib import constants as consts
from neutron_lib.plugins import constants as plugin_constants
from neutron_lib.plugins import directory
from oslo_log import log as logging
import oslo_messaging
@ -31,7 +32,7 @@ class MeteringRpcCallbacks(object):
self.meter_plugin = meter_plugin
def get_sync_data_metering(self, context, **kwargs):
l3_plugin = directory.get_plugin(consts.L3)
l3_plugin = directory.get_plugin(plugin_constants.L3)
if not l3_plugin:
return

2
neutron/extensions/l3.py
View File

@ -17,9 +17,9 @@ import abc
from neutron_lib.api import converters
from neutron_lib.api import extensions
from neutron_lib import constants
from neutron_lib.db import constants as db_const
from neutron_lib import exceptions as nexception
from neutron_lib.plugins import constants
import six
from neutron._i18n import _

5
neutron/extensions/l3agentscheduler.py
View File

@ -18,6 +18,7 @@ import abc
from neutron_lib.api import extensions as api_extensions
from neutron_lib import constants
from neutron_lib import exceptions
from neutron_lib.plugins import constants as plugin_constants
from neutron_lib.plugins import directory
from oslo_log import log as logging
import six
@ -44,7 +45,7 @@ L3_AGENTS = L3_AGENT + 's'
class RouterSchedulerController(wsgi.Controller):
def get_plugin(self):
plugin = directory.get_plugin(constants.L3)
plugin = directory.get_plugin(plugin_constants.L3)
if not plugin:
LOG.error(_LE('No plugin for L3 routing registered to handle '
'router scheduling'))
@ -86,7 +87,7 @@ class RouterSchedulerController(wsgi.Controller):
class L3AgentsHostingRouterController(wsgi.Controller):
def get_plugin(self):
plugin = directory.get_plugin(constants.L3)
plugin = directory.get_plugin(plugin_constants.L3)
if not plugin:
LOG.error(_LE('No plugin for L3 routing registered to handle '
'router scheduling'))

2
neutron/manager.py
View File

@ -14,7 +14,7 @@
# under the License.
from collections import defaultdict
from neutron_lib import constants as lib_const
from neutron_lib.plugins import constants as lib_const
from neutron_lib.plugins import directory
from oslo_config import cfg
from oslo_log import log as logging

2
neutron/plugins/common/constants.py
View File

@ -13,7 +13,7 @@
# License for the specific language governing permissions and limitations
# under the License.
from neutron_lib import constants
from neutron_lib.plugins import constants
# Neutron well-known service type constants:

3
neutron/plugins/ml2/drivers/l2pop/mech_driver.py
View File

@ -16,6 +16,7 @@
from neutron_lib import constants as const
from neutron_lib import context as n_context
from neutron_lib import exceptions
from neutron_lib.plugins import constants as plugin_constants
from neutron_lib.plugins import directory
from neutron_lib.plugins.ml2 import api
from oslo_config import cfg
@ -242,7 +243,7 @@ class L2populationMechanismDriver(api.MechanismDriver):
def update_port_down(self, context):
port = context.current
agent_host = context.host
l3plugin = directory.get_plugin(const.L3)
l3plugin = directory.get_plugin(plugin_constants.L3)
# when agent transitions to backup, don't remove flood flows
if agent_host and l3plugin and getattr(
l3plugin, "list_router_ids_on_host", None):

3
neutron/plugins/ml2/plugin.py
View File

@ -26,6 +26,7 @@ from neutron_lib.callbacks import resources
from neutron_lib import constants as const
from neutron_lib import exceptions as exc
from neutron_lib.exceptions import port_security as psec_exc
from neutron_lib.plugins import constants as plugin_constants
from neutron_lib.plugins import directory
from neutron_lib.plugins.ml2 import api
from oslo_config import cfg
@ -1431,7 +1432,7 @@ class Ml2Plugin(db_base_plugin_v2.NeutronDbPluginV2,
self._pre_delete_port(context, id, l3_port_check)
# TODO(armax): get rid of the l3 dependency in the with block
router_ids = []
l3plugin = directory.get_plugin(const.L3)
l3plugin = directory.get_plugin(plugin_constants.L3)
with db_api.context_manager.writer.using(context):
try:

2
neutron/services/auto_allocate/db.py
View File

@ -17,8 +17,8 @@
from neutron_lib.callbacks import events
from neutron_lib.callbacks import registry
from neutron_lib.callbacks import resources
from neutron_lib import constants
from neutron_lib import exceptions as n_exc
from neutron_lib.plugins import constants
from neutron_lib.plugins import directory
from oslo_log import log as logging

3
neutron/services/l3_router/l3_router_plugin.py
View File

@ -14,6 +14,7 @@
# under the License.
from neutron_lib import constants as n_const
from neutron_lib.plugins import constants as plugin_constants
from neutron_lib.services import base as service_base
from oslo_config import cfg
from oslo_log import helpers as log_helpers
@ -116,7 +117,7 @@ class L3RouterPlugin(service_base.ServicePluginBase,
@classmethod
def get_plugin_type(cls):
return n_const.L3
return plugin_constants.L3
def get_plugin_description(self):
"""returns string description of the plugin."""

13
neutron/services/l3_router/service_providers/driver_controller.py
View File

@ -17,6 +17,7 @@ from neutron_lib.callbacks import registry
from neutron_lib.callbacks import resources
from neutron_lib import constants as lib_const
from neutron_lib import exceptions as lib_exc
from neutron_lib.plugins import constants as plugin_constants
from neutron_lib.plugins import directory
from oslo_config import cfg
from oslo_log import log as logging
@ -46,12 +47,12 @@ class DriverController(object):
self.l3_plugin = l3_plugin
self._stm = st_db.ServiceTypeManager.get_instance()
self._stm.add_provider_configuration(
lib_const.L3, _LegacyPlusProviderConfiguration())
plugin_constants.L3, _LegacyPlusProviderConfiguration())
self._load_drivers()
def _load_drivers(self):
self.drivers, self.default_provider = (
service_base.load_drivers(lib_const.L3, self.l3_plugin))
service_base.load_drivers(plugin_constants.L3, self.l3_plugin))
# store the provider name on each driver to make finding inverse easy
for provider_name, driver in self.drivers.items():
setattr(driver, 'name', provider_name)
@ -81,7 +82,7 @@ class DriverController(object):
if _flavor_specified(router):
router_db.flavor_id = router['flavor_id']
drv = self._get_provider_for_create(context, router)
self._stm.add_resource_association(context, lib_const.L3,
self._stm.add_resource_association(context, plugin_constants.L3,
drv.name, router['id'])
@registry.receives(resources.ROUTER, [events.PRECOMMIT_DELETE])
@ -140,7 +141,7 @@ class DriverController(object):
with context.session.begin(subtransactions=True):
self._stm.del_resource_associations(context, [router_id])
self._stm.add_resource_association(
context, lib_const.L3, new_drv.name, router_id)
context, plugin_constants.L3, new_drv.name, router_id)
def get_provider_for_router(self, context, router_id):
"""Return the provider driver handle for a router id."""
@ -152,7 +153,7 @@ class DriverController(object):
router = self.l3_plugin.get_router(context, router_id)
driver = self._attrs_to_driver(router)
driver_name = driver.name
self._stm.add_resource_association(context, lib_const.L3,
self._stm.add_resource_association(context, plugin_constants.L3,
driver_name, router_id)
return self.drivers[driver_name]
@ -208,7 +209,7 @@ class _LegacyPlusProviderConfiguration(
('single_node', 'single_node.SingleNodeDriver')):
path = 'neutron.services.l3_router.service_providers.%s' % driver
try:
self.add_provider({'service_type': lib_const.L3,
self.add_provider({'service_type': plugin_constants.L3,
'name': name, 'driver': path,
'default': False})
except lib_exc.Invalid:

7
neutron/tests/functional/pecan_wsgi/test_controllers.py
View File

@ -13,6 +13,7 @@
import mock
from neutron_lib import constants as n_const
from neutron_lib import context
from neutron_lib.plugins import constants as plugin_constants
from neutron_lib.plugins import directory
from oslo_config import cfg
from oslo_db import exception as db_exc
@ -689,7 +690,7 @@ class TestRequestProcessing(TestRootController):
self.assertEqual('router', self.req_stash['resource_type'])
# make sure the core plugin was identified as the handler for ports
self.assertEqual(
directory.get_plugin(n_const.L3),
directory.get_plugin(plugin_constants.L3),
self.req_stash['plugin'])
def test_service_plugin_uri(self):
@ -719,7 +720,7 @@ class TestRouterController(TestResourceController):
self.addCleanup(policy.reset)
plugin = directory.get_plugin()
ctx = context.get_admin_context()
l3_plugin = directory.get_plugin(n_const.L3)
l3_plugin = directory.get_plugin(plugin_constants.L3)
network_id = pecan_utils.create_network(ctx, plugin)['id']
self.subnet = pecan_utils.create_subnet(ctx, plugin, network_id)
self.router = pecan_utils.create_router(ctx, l3_plugin)
@ -820,7 +821,7 @@ class TestL3AgentShimControllers(test_functional.PecanFunctionalTest):
'get_l3-routers': 'role:admin'}),
overwrite=False)
ctx = context.get_admin_context()
l3_plugin = directory.get_plugin(n_const.L3)
l3_plugin = directory.get_plugin(plugin_constants.L3)
self.router = pecan_utils.create_router(ctx, l3_plugin)
self.agent = helpers.register_l3_agent()
# NOTE(blogan): Not sending notifications because this test is for

2
neutron/tests/functional/pecan_wsgi/test_functional.py
View File

@ -16,9 +16,9 @@
import os
import mock
from neutron_lib import constants
from neutron_lib import context
from neutron_lib import exceptions as n_exc
from neutron_lib.plugins import constants
from oslo_config import cfg
from oslo_middleware import base
from oslo_service import wsgi

7
neutron/tests/functional/scheduler/test_l3_agent_scheduler.py
View File

@ -18,6 +18,7 @@ import random
from neutron_lib import constants
from neutron_lib import context
from neutron_lib.plugins import constants as plugin_constants
from neutron_lib.plugins import directory
from oslo_utils import uuidutils
import testscenarios
@ -46,7 +47,7 @@ class L3SchedulerBaseTest(test_db_base_plugin_v2.NeutronDbPluginV2TestCase):
super(L3SchedulerBaseTest, self).setUp(PLUGIN_NAME)
self.l3_plugin = l3_router_plugin.L3RouterPlugin()
directory.add_plugin(constants.L3, self.l3_plugin)
directory.add_plugin(plugin_constants.L3, self.l3_plugin)
self.adminContext = context.get_admin_context()
self.adminContext.tenant_id = _uuid()
@ -298,9 +299,9 @@ class L3AZSchedulerBaseTest(test_db_base_plugin_v2.NeutronDbPluginV2TestCase):
super(L3AZSchedulerBaseTest, self).setUp(plugin='ml2')
self.l3_plugin = l3_router_plugin.L3RouterPlugin()
directory.add_plugin(constants.L3, self.l3_plugin)
directory.add_plugin(plugin_constants.L3, self.l3_plugin)
self.l3_plugin.router_scheduler = None
directory.add_plugin(constants.L3, self.l3_plugin)
directory.add_plugin(plugin_constants.L3, self.l3_plugin)
self.adminContext = context.get_admin_context()
self.adminContext.tenant_id = '_func_test_tenant_'

2
neutron/tests/tempest/api/admin/test_routers_flavors.py
View File

@ -11,7 +11,7 @@
# License for the specific language governing permissions and limitations
# under the License.
from neutron_lib import constants
from neutron_lib.plugins import constants
from tempest.lib import decorators
from tempest.lib import exceptions as lib_exc
import testtools

3
neutron/tests/unit/agent/l3/test_agent.py
View File

@ -23,6 +23,7 @@ import netaddr
from neutron_lib.api.definitions import portbindings
from neutron_lib import constants as lib_constants
from neutron_lib import exceptions as exc
from neutron_lib.plugins import constants as plugin_constants
from oslo_config import cfg
from oslo_log import log
import oslo_messaging
@ -2534,7 +2535,7 @@ class TestBasicRouterOperations(BasicRouterOperationsFramework):
self.assertEqual(2, mocked_func.call_count)
def test_get_service_plugin_list(self):
service_plugins = [lib_constants.L3]
service_plugins = [plugin_constants.L3]
self.plugin_api.get_service_plugin_list.return_value = service_plugins
agent = l3_agent.L3NATAgent(HOSTNAME, self.conf)
self.assertEqual(service_plugins, agent.neutron_service_plugins)

3
neutron/tests/unit/api/rpc/handlers/test_dhcp_rpc.py
View File

@ -18,6 +18,7 @@ from neutron_lib.api.definitions import portbindings
from neutron_lib.callbacks import resources
from neutron_lib import constants
from neutron_lib import exceptions as n_exc
from neutron_lib.plugins import constants as plugin_constants
from neutron_lib.plugins import directory
from oslo_db import exception as db_exc
@ -34,7 +35,7 @@ class TestDhcpRpcCallback(base.BaseTestCase):
def setUp(self):
super(TestDhcpRpcCallback, self).setUp()
self.plugin = mock.MagicMock()
directory.add_plugin(constants.CORE, self.plugin)
directory.add_plugin(plugin_constants.CORE, self.plugin)
self.callbacks = dhcp_rpc.DhcpRpcCallback()
self.log_p = mock.patch('neutron.api.rpc.handlers.dhcp_rpc.LOG')
self.log = self.log_p.start()

2
neutron/tests/unit/api/test_extensions.py
View File

@ -17,7 +17,7 @@ import copy
import fixtures
import mock
from neutron_lib import constants as lib_const
from neutron_lib.plugins import constants as lib_const
from neutron_lib.plugins import directory
from neutron_lib.services import base as service_base
from oslo_config import cfg

2
neutron/tests/unit/db/quota/test_api.py
View File

@ -15,8 +15,8 @@
import datetime
import mock
from neutron_lib import constants as const
from neutron_lib import context
from neutron_lib.plugins import constants as const
from neutron_lib.plugins import directory
from oslo_config import cfg

19
neutron/tests/unit/db/test_agentschedulers_db.py
View File

@ -18,6 +18,7 @@ import datetime
import mock
from neutron_lib import constants
from neutron_lib import context
from neutron_lib.plugins import constants as plugin_constants
from neutron_lib.plugins import directory
from oslo_config import cfg
from oslo_db import exception as db_exc
@ -259,7 +260,7 @@ class OvsAgentSchedulerTestCaseBase(test_l3.L3NatTestCaseMixin,
# the global attribute map
attributes.RESOURCE_ATTRIBUTE_MAP.update(
agent.RESOURCE_ATTRIBUTE_MAP)
self.l3plugin = directory.get_plugin(constants.L3)
self.l3plugin = directory.get_plugin(plugin_constants.L3)
self.l3_notify_p = mock.patch(
'neutron.extensions.l3agentscheduler.notify')
self.patched_l3_notify = self.l3_notify_p.start()
@ -675,7 +676,7 @@ class OvsAgentSchedulerTestCase(OvsAgentSchedulerTestCaseBase):
agt.heartbeat_timestamp - datetime.timedelta(hours=1))
self.adminContext.session.commit()
plugin = directory.get_plugin(constants.L3)
plugin = directory.get_plugin(plugin_constants.L3)
plugin.reschedule_routers_from_down_agents()
def _set_agent_admin_state_up(self, host, state):
@ -692,7 +693,7 @@ class OvsAgentSchedulerTestCase(OvsAgentSchedulerTestCaseBase):
# schedule the router to host A
l3_rpc_cb.get_router_ids(self.adminContext, host=L3_HOSTA)
plugin = directory.get_plugin(constants.L3)
plugin = directory.get_plugin(plugin_constants.L3)
mock.patch.object(
plugin, 'reschedule_router',
side_effect=[
@ -713,12 +714,12 @@ class OvsAgentSchedulerTestCase(OvsAgentSchedulerTestCaseBase):
mock_ctx = mock.Mock()
get_ctx.return_value = mock_ctx
mock_ctx.session.query.side_effect = db_exc.DBError()
plugin = directory.get_plugin(constants.L3)
plugin = directory.get_plugin(plugin_constants.L3)
# check that no exception is raised
plugin.reschedule_routers_from_down_agents()
def test_router_rescheduler_iterates_after_reschedule_failure(self):
plugin = directory.get_plugin(constants.L3)
plugin = directory.get_plugin(plugin_constants.L3)
l3_rpc_cb = l3_rpc.L3RpcCallback()
self._register_agent_states()
with self.router() as r1, self.router() as r2:
@ -750,7 +751,7 @@ class OvsAgentSchedulerTestCase(OvsAgentSchedulerTestCaseBase):
self.assertFalse(rr.called)
def test_router_is_not_rescheduled_if_agent_is_back_online(self):
plugin = directory.get_plugin(constants.L3)
plugin = directory.get_plugin(plugin_constants.L3)
l3_rpc_cb = l3_rpc.L3RpcCallback()
agent = helpers.register_l3_agent(host=L3_HOSTA)
with self.router(),\
@ -1487,7 +1488,7 @@ class OvsL3AgentNotifierTestCase(test_l3.L3NatTestCaseMixin,
fake_notifier.reset()
def test_router_add_to_l3_agent_notification(self):
l3_plugin = directory.get_plugin(constants.L3)
l3_plugin = directory.get_plugin(plugin_constants.L3)
l3_notifier = l3_plugin.agent_notifiers[constants.AGENT_TYPE_L3]
with mock.patch.object(
l3_notifier.client,
@ -1509,7 +1510,7 @@ class OvsL3AgentNotifierTestCase(test_l3.L3NatTestCaseMixin,
self._assert_notify(notifications, expected_event_type)
def test_router_remove_from_l3_agent_notification(self):
l3_plugin = directory.get_plugin(constants.L3)
l3_plugin = directory.get_plugin(plugin_constants.L3)
l3_notifier = l3_plugin.agent_notifiers[constants.AGENT_TYPE_L3]
with mock.patch.object(
l3_notifier.client,
@ -1534,7 +1535,7 @@ class OvsL3AgentNotifierTestCase(test_l3.L3NatTestCaseMixin,
self._assert_notify(notifications, expected_event_type)
def test_agent_updated_l3_agent_notification(self):
l3_plugin = directory.get_plugin(constants.L3)
l3_plugin = directory.get_plugin(plugin_constants.L3)
l3_notifier = l3_plugin.agent_notifiers[constants.AGENT_TYPE_L3]
with mock.patch.object(
l3_notifier.client,

2
neutron/tests/unit/db/test_extraroute_db.py
View File

@ -14,8 +14,8 @@
# under the License.
import mock
from neutron_lib import constants
from neutron_lib import context
from neutron_lib.plugins import constants
from neutron_lib.plugins import directory
from neutron.db import common_db_mixin

13
neutron/tests/unit/db/test_l3_dvr_db.py
View File

@ -21,6 +21,7 @@ from neutron_lib.callbacks import resources
from neutron_lib import constants as const
from neutron_lib import context
from neutron_lib import exceptions
from neutron_lib.plugins import constants as plugin_constants
from neutron_lib.plugins import directory
from oslo_utils import uuidutils
@ -48,7 +49,7 @@ class L3DvrTestCase(test_db_base_plugin_v2.NeutronDbPluginV2TestCase):
self.core_plugin = directory.get_plugin()
self.ctx = context.get_admin_context()
self.mixin = FakeL3Plugin()
directory.add_plugin(const.L3, self.mixin)
directory.add_plugin(plugin_constants.L3, self.mixin)
def _create_router(self, router):
with self.ctx.session.begin(subtransactions=True):
@ -173,7 +174,7 @@ class L3DvrTestCase(test_db_base_plugin_v2.NeutronDbPluginV2TestCase):
def test__get_agent_gw_ports_exist_for_network(self):
plugin = mock.Mock()
directory.add_plugin(const.CORE, plugin)
directory.add_plugin(plugin_constants.CORE, plugin)
plugin.get_ports.return_value = []
self.mixin._get_agent_gw_ports_exist_for_network(
self.ctx, 'network_id', 'host', 'agent_id')
@ -184,7 +185,7 @@ class L3DvrTestCase(test_db_base_plugin_v2.NeutronDbPluginV2TestCase):
def _test_prepare_direct_delete_dvr_internal_ports(self, port):
plugin = mock.Mock()
directory.add_plugin(const.CORE, plugin)
directory.add_plugin(plugin_constants.CORE, plugin)
plugin.get_port.return_value = port
self.mixin._router_exists = mock.Mock(return_value=True)
self.assertRaises(exceptions.ServicePortInUse,
@ -249,7 +250,7 @@ class L3DvrTestCase(test_db_base_plugin_v2.NeutronDbPluginV2TestCase):
'device_owner': const.DEVICE_OWNER_ROUTER_GW
}]
plugin = mock.Mock()
directory.add_plugin(const.CORE, plugin)
directory.add_plugin(plugin_constants.CORE, plugin)
plugin.get_ports.return_value = ports
self.mixin.delete_floatingip_agent_gateway_port(
self.ctx, port_host, 'ext_network_id')
@ -286,7 +287,7 @@ class L3DvrTestCase(test_db_base_plugin_v2.NeutronDbPluginV2TestCase):
router.gw_port = None
plugin = mock.Mock()
directory.add_plugin(const.CORE, plugin)
directory.add_plugin(plugin_constants.CORE, plugin)
with mock.patch.object(l3_dvr_db.l3_db.L3_NAT_db_mixin,
'router_gw_port_has_floating_ips',
return_value=False),\
@ -690,7 +691,7 @@ class L3DvrTestCase(test_db_base_plugin_v2.NeutronDbPluginV2TestCase):
'distributed': True}
router = self._create_router(router_dict)
plugin = mock.Mock()
directory.add_plugin(const.CORE, plugin)
directory.add_plugin(plugin_constants.CORE, plugin)
l3_notify = self.mixin.l3_rpc_notifier = mock.Mock()
port = {
'id': 'my_port_id',

3
neutron/tests/unit/db/test_l3_hamode_db.py
View File

@ -23,6 +23,7 @@ from neutron_lib.callbacks import resources
from neutron_lib import constants
from neutron_lib import context
from neutron_lib import exceptions as n_exc
from neutron_lib.plugins import constants as plugin_constants
from neutron_lib.plugins import directory
from oslo_config import cfg
from oslo_db import exception as db_exc
@ -69,7 +70,7 @@ class L3HATestFramework(testlib_api.SqlTestCase):
cfg.CONF.set_override('allow_overlapping_ips', True)
self.plugin = FakeL3PluginWithAgents()
directory.add_plugin(constants.L3, self.plugin)
directory.add_plugin(plugin_constants.L3, self.plugin)
self.plugin.router_scheduler = l3_agent_scheduler.ChanceScheduler()
self.agent1 = helpers.register_l3_agent()
self.agent2 = helpers.register_l3_agent(

2
neutron/tests/unit/db/test_portsecurity_db.py
View File

@ -11,7 +11,7 @@
# under the License.
import mock
from neutron_lib import constants
from neutron_lib.plugins import constants
from neutron_lib.plugins import directory
from neutron.db import portsecurity_db as pd

3
neutron/tests/unit/extensions/test_external_net.py
View File

@ -16,6 +16,7 @@
import mock
from neutron_lib import constants
from neutron_lib import context
from neutron_lib.plugins import constants as plugin_constants
from neutron_lib.plugins import directory
from oslo_utils import uuidutils
import testtools
@ -185,7 +186,7 @@ class ExtNetDBTestCase(test_db_base_plugin_v2.NeutronDbPluginV2TestCase):
def test_delete_network_check_disassociated_floatingips(self):
l3_mock = mock.Mock()
directory.add_plugin(constants.L3, l3_mock)
directory.add_plugin(plugin_constants.L3, l3_mock)
with self.network() as net:
req = self.new_delete_request('networks', net['network']['id'])
res = req.get_response(self.api)

45
neutron/tests/unit/extensions/test_l3.py
View File

@ -27,6 +27,7 @@ from neutron_lib.callbacks import resources
from neutron_lib import constants as lib_constants
from neutron_lib import context
from neutron_lib import exceptions as n_exc
from neutron_lib.plugins import constants as plugin_constants
from neutron_lib.plugins import directory
from oslo_config import cfg
from oslo_utils import importutils
@ -100,7 +101,7 @@ class L3NatExtensionTestCase(test_extensions_base.ExtensionTestCase):
super(L3NatExtensionTestCase, self).setUp()
self._setUpExtension(
'neutron.services.l3_router.l3_router_plugin.L3RouterPlugin',
lib_constants.L3, l3.RESOURCE_ATTRIBUTE_MAP,
plugin_constants.L3, l3.RESOURCE_ATTRIBUTE_MAP,
l3.L3, '', allow_pagination=True, allow_sorting=True,
supported_extension_aliases=['router'],
use_quota=True)
@ -269,7 +270,7 @@ class TestL3NatBasePlugin(db_base_plugin_v2.NeutronDbPluginV2,
return net
def delete_port(self, context, id, l3_port_check=True):
plugin = directory.get_plugin(lib_constants.L3)
plugin = directory.get_plugin(plugin_constants.L3)
if plugin:
if l3_port_check:
plugin.prevent_l3_port_deletion(context, id)
@ -322,7 +323,7 @@ class TestL3NatServicePlugin(common_db_mixin.CommonDbMixin,
@classmethod
def get_plugin_type(cls):
return lib_constants.L3
return plugin_constants.L3
def get_plugin_description(self):
return "L3 Routing Service Plugin for testing"
@ -571,7 +572,7 @@ class ExtraAttributesMixinTestCase(testlib_api.SqlTestCase):
def setUp(self):
super(ExtraAttributesMixinTestCase, self).setUp()
self.mixin = l3_attrs_db.ExtraAttributesMixin()
directory.add_plugin(lib_constants.L3, self.mixin)
directory.add_plugin(plugin_constants.L3, self.mixin)
self.ctx = context.get_admin_context()
self.router = l3_models.Router()
with self.ctx.session.begin():
@ -901,7 +902,7 @@ class L3NatTestCaseBase(L3NatTestCaseMixin):
r['router']['id'],
n['network']['id'],
ext_ips=[{'subnet_id': s1['subnet']['id']}])
plugin = directory.get_plugin(lib_constants.L3)
plugin = directory.get_plugin(plugin_constants.L3)
mock.patch.object(
plugin, 'update_router',
side_effect=l3.RouterNotFound(router_id='1')).start()
@ -2131,7 +2132,7 @@ class L3NatTestCaseBase(L3NatTestCaseMixin):
self.assertIsNotNone(body['floatingip']['router_id'])
def test_create_floatingip_non_admin_context_agent_notification(self):
plugin = directory.get_plugin(lib_constants.L3)
plugin = directory.get_plugin(plugin_constants.L3)
if not hasattr(plugin, 'l3_rpc_notifier'):
self.skipTest("Plugin does not support l3_rpc_notifier")
@ -2972,7 +2973,7 @@ class L3NatTestCaseBase(L3NatTestCaseMixin):
def test_floatingip_via_router_interface_returns_201(self):
# Override get_router_for_floatingip, as
# networking-midonet's L3 service plugin would do.
plugin = directory.get_plugin(lib_constants.L3)
plugin = directory.get_plugin(plugin_constants.L3)
with mock.patch.object(plugin, "get_router_for_floatingip",
self._get_router_for_floatingip_without_device_owner_check):
self._test_floatingip_via_router_interface(exc.HTTPCreated.code)
@ -3112,7 +3113,7 @@ class L3NatTestCaseBase(L3NatTestCaseMixin):
self.assertEqual(exc.HTTPConflict.code, res.status_int)
def test_router_specify_id_backend(self):
plugin = directory.get_plugin(lib_constants.L3)
plugin = directory.get_plugin(plugin_constants.L3)
router_req = {'router': {'id': _uuid(), 'name': 'router',
'tenant_id': 'foo',
'admin_state_up': True}}
@ -3167,7 +3168,7 @@ class L3NatTestCaseBase(L3NatTestCaseMixin):
def test_create_router_gateway_fails_nested(self):
# Force _update_router_gw_info failure
plugin = directory.get_plugin(lib_constants.L3)
plugin = directory.get_plugin(plugin_constants.L3)
if not isinstance(plugin, l3_db.L3_NAT_dbonly_mixin):
self.skipTest("Plugin is not L3_NAT_dbonly_mixin")
ctx = context.Context('', 'foo')
@ -3198,7 +3199,7 @@ class L3NatTestCaseBase(L3NatTestCaseMixin):
def test_create_router_gateway_fails_nested_delete_router_failed(self):
# Force _update_router_gw_info failure
plugin = directory.get_plugin(lib_constants.L3)
plugin = directory.get_plugin(plugin_constants.L3)
if not isinstance(plugin, l3_db.L3_NAT_dbonly_mixin):
self.skipTest("Plugin is not L3_NAT_dbonly_mixin")
ctx = context.Context('', 'foo')
@ -3235,7 +3236,7 @@ class L3NatTestCaseBase(L3NatTestCaseMixin):
def test_router_add_interface_by_port_fails_nested(self):
# Force _validate_router_port_info failure
plugin = directory.get_plugin(lib_constants.L3)
plugin = directory.get_plugin(plugin_constants.L3)
if not isinstance(plugin, l3_db.L3_NAT_dbonly_mixin):
self.skipTest("Plugin is not L3_NAT_dbonly_mixin")
orig_update_port = self.plugin.update_port
@ -3288,7 +3289,7 @@ class L3NatTestCaseBase(L3NatTestCaseMixin):
"""Test to make sure notification to routers occurs when the gateway
ip address of a subnet of the external network is changed.
"""
plugin = directory.get_plugin(lib_constants.L3)
plugin = directory.get_plugin(plugin_constants.L3)
if not hasattr(plugin, 'l3_rpc_notifier'):
self.skipTest("Plugin does not support l3_rpc_notifier")
# make sure the callback is registered.
@ -3320,7 +3321,7 @@ class L3NatTestCaseBase(L3NatTestCaseMixin):
['fake_device'], None)
def test__notify_subnetpool_address_scope_update(self):
plugin = directory.get_plugin(lib_constants.L3)
plugin = directory.get_plugin(plugin_constants.L3)
tenant_id = _uuid()
with mock.patch.object(
@ -3350,7 +3351,7 @@ class L3NatTestCaseBase(L3NatTestCaseMixin):
chk_method.assert_called_with(admin_ctx, [router['router']['id']])
def test_janitor_clears_orphaned_floatingip_port(self):
plugin = directory.get_plugin(lib_constants.L3)
plugin = directory.get_plugin(plugin_constants.L3)
with self.network() as n:
# floating IP ports are initially created with a device ID of
# PENDING and are updated after the floating IP is actually
@ -3375,7 +3376,7 @@ class L3NatTestCaseBase(L3NatTestCaseMixin):
# if a server dies after the floating IP is created but before it
# updates the floating IP port device ID, the janitor will be
# responsible for updating the device ID to the correct value.
plugin = directory.get_plugin(lib_constants.L3)
plugin = directory.get_plugin(plugin_constants.L3)
with self.floatingip_with_assoc() as fip:
fip_port = self._list('ports',
query_params='device_owner=network:floatingip')['ports'][0]
@ -3397,7 +3398,7 @@ class L3NatTestCaseBase(L3NatTestCaseMixin):
def test_janitor_doesnt_delete_if_fixed_in_interim(self):
# here we ensure that the janitor doesn't delete the port on the second
# call if the conditions have been fixed
plugin = directory.get_plugin(lib_constants.L3)
plugin = directory.get_plugin(plugin_constants.L3)
with self.network() as n:
port_res = self._create_port(
self.fmt, n['network']['id'],
@ -3533,7 +3534,7 @@ class L3AgentDbTestCaseBase(L3NatTestCaseMixin):
l3_rpc_agent_api_str = (
'neutron.api.rpc.agentnotifiers.l3_rpc_agent_api.L3AgentNotifyAPI')
with mock.patch(l3_rpc_agent_api_str):
plugin = directory.get_plugin(lib_constants.L3)
plugin = directory.get_plugin(plugin_constants.L3)
notifyApi = plugin.l3_rpc_notifier
kargs = [item for item in args]
kargs.append(notifyApi)
@ -3701,7 +3702,7 @@ class L3NatDBIntAgentSchedulingTestCase(L3BaseForIntTests,
self.adminContext = context.get_admin_context()
def _assert_router_on_agent(self, router_id, agent_host):
plugin = directory.get_plugin(lib_constants.L3)
plugin = directory.get_plugin(plugin_constants.L3)
agents = plugin.list_l3_agents_hosting_router(
self.adminContext, router_id)['agents']
self.assertEqual(1, len(agents))
@ -3758,7 +3759,7 @@ class L3NatDBIntAgentSchedulingTestCase(L3BaseForIntTests,
self._assert_router_on_agent(r['router']['id'], 'host2')
def test_router_update_gateway_scheduling_not_supported(self):
plugin = directory.get_plugin(lib_constants.L3)
plugin = directory.get_plugin(plugin_constants.L3)
mock.patch.object(plugin, 'router_supports_scheduling',
return_value=False).start()
with self.router() as r:
@ -3907,7 +3908,7 @@ class L3NatDBTestCaseMixin(object):
def setUp(self):
super(L3NatDBTestCaseMixin, self).setUp()
plugin = directory.get_plugin(lib_constants.L3)
plugin = directory.get_plugin(plugin_constants.L3)
if not isinstance(plugin, l3_db.L3_NAT_dbonly_mixin):
self.skipTest("Plugin is not L3_NAT_dbonly_mixin")
@ -3916,7 +3917,7 @@ class L3NatDBTestCaseMixin(object):
the exception is propagated.
"""
plugin = directory.get_plugin(lib_constants.L3)
plugin = directory.get_plugin(plugin_constants.L3)
ctx = context.Context('', 'foo')
class MyException(Exception):
@ -3949,7 +3950,7 @@ class L3NatDBSepTestCase(L3BaseForSepTests, L3NatTestCaseBase,
"""Unit tests for a separate L3 routing service plugin."""
def test_port_deletion_prevention_handles_missing_port(self):
pl = directory.get_plugin(lib_constants.L3)
pl = directory.get_plugin(plugin_constants.L3)
self.assertIsNone(
pl.prevent_l3_port_deletion(context.get_admin_context(), 'fakeid')
)

2
neutron/tests/unit/extensions/test_providernet.py
View File

@ -15,8 +15,8 @@
import mock
from neutron_lib.api.definitions import provider_net
from neutron_lib import constants
from neutron_lib import context
from neutron_lib.plugins import constants
from neutron_lib.plugins import directory
from oslo_config import cfg
from oslo_utils import uuidutils

2
neutron/tests/unit/extensions/test_router_availability_zone.py
View File

@ -11,7 +11,7 @@
# License for the specific language governing permissions and limitations
# under the License.
from neutron_lib import constants
from neutron_lib.plugins import constants
from neutron.db.availability_zone import router as router_az_db
from neutron.db import common_db_mixin

3
neutron/tests/unit/notifiers/test_nova.py
View File

@ -17,6 +17,7 @@
import mock
from neutron_lib import constants as n_const
from neutron_lib import exceptions as n_exc
from neutron_lib.plugins import constants as plugin_constants
from neutron_lib.plugins import directory
from novaclient import api_versions
from novaclient import exceptions as nova_exceptions
@ -44,7 +45,7 @@ class TestNovaNotify(base.BaseTestCase):
'id': port_id}
self.nova_notifier = nova.Notifier()
directory.add_plugin(n_const.CORE, FakePlugin())
directory.add_plugin(plugin_constants.CORE, FakePlugin())
def test_notify_port_status_all_values(self):
states = [n_const.PORT_STATUS_ACTIVE, n_const.PORT_STATUS_DOWN,

2
neutron/tests/unit/plugins/ml2/base.py
View File

@ -12,7 +12,7 @@
# License for the specific language governing permissions and limitations
# under the License.
from neutron_lib import constants
from neutron_lib.plugins import constants
from neutron_lib.plugins import directory
from neutron.tests.unit.plugins.ml2 import test_plugin

7
neutron/tests/unit/plugins/ml2/drivers/l2pop/test_mech_driver.py
View File

@ -20,6 +20,7 @@ from neutron_lib.api.definitions import provider_net as pnet
from neutron_lib import constants
from neutron_lib import context
from neutron_lib import exceptions
from neutron_lib.plugins import constants as plugin_constants
from neutron_lib.plugins import directory
from oslo_serialization import jsonutils
import testtools
@ -279,7 +280,7 @@ class TestL2PopulationRpcTestCase(test_plugin.Ml2PluginV2TestCase):
# is added on HOST4.
# HOST4 should get flood entries for HOST1 and HOST2
router = self._create_ha_router()
directory.add_plugin(constants.L3, self.plugin)
directory.add_plugin(plugin_constants.L3, self.plugin)
with self.subnet(network=self._network, enable_dhcp=False) as snet:
subnet = snet['subnet']
port = self._add_router_interface(subnet, router, HOST)
@ -312,7 +313,7 @@ class TestL2PopulationRpcTestCase(test_plugin.Ml2PluginV2TestCase):
# Remove_fdb should carry flood entry of only HOST2 and not HOST
router = self._create_ha_router()
directory.add_plugin(constants.L3, self.plugin)
directory.add_plugin(plugin_constants.L3, self.plugin)
with self.subnet(network=self._network, enable_dhcp=False) as snet:
host_arg = {portbindings.HOST_ID: HOST, 'admin_state_up': True}
with self.port(subnet=snet,
@ -345,7 +346,7 @@ class TestL2PopulationRpcTestCase(test_plugin.Ml2PluginV2TestCase):
# Both HA agents should be notified to other agents.
router = self._create_ha_router()
directory.add_plugin(constants.L3, self.plugin)
directory.add_plugin(plugin_constants.L3, self.plugin)
with self.subnet(network=self._network, enable_dhcp=False) as snet:
host_arg = {portbindings.HOST_ID: HOST_4, 'admin_state_up': True}
with self.port(subnet=snet,

17
neutron/tests/unit/plugins/ml2/test_plugin.py
View File

@ -29,6 +29,7 @@ from neutron_lib.callbacks import resources
from neutron_lib import constants
from neutron_lib import context
from neutron_lib import exceptions as exc
from neutron_lib.plugins import constants as plugin_constants
from neutron_lib.plugins import directory
from neutron_lib.plugins.ml2 import api as driver_api
from oslo_db import exception as db_exc
@ -772,7 +773,7 @@ class TestMl2PortsV2(test_plugin.TestPortsV2, Ml2PluginV2TestCase):
'create_port_postcommit',
side_effect=ml2_exc.MechanismDriverError(
method='create_port_postcommit')):
l3_plugin = directory.get_plugin(constants.L3)
l3_plugin = directory.get_plugin(plugin_constants.L3)
data = {'router': {'name': 'router', 'admin_state_up': True,
'tenant_id': 'fake_tenant'}}
r = l3_plugin.create_router(self.context, data)
@ -789,7 +790,7 @@ class TestMl2PortsV2(test_plugin.TestPortsV2, Ml2PluginV2TestCase):
with mock.patch.object(ml2_plugin.Ml2Plugin, '_bind_port_if_needed',
side_effect=ml2_exc.MechanismDriverError(
method='_bind_port_if_needed')):
l3_plugin = directory.get_plugin(constants.L3)
l3_plugin = directory.get_plugin(plugin_constants.L3)
data = {'router': {'name': 'router', 'admin_state_up': True,
'tenant_id': 'fake_tenant'}}
r = l3_plugin.create_router(self.context, data)
@ -918,7 +919,7 @@ class TestMl2PortsV2(test_plugin.TestPortsV2, Ml2PluginV2TestCase):
])
def test_l3_cleanup_on_net_delete(self):
l3plugin = directory.get_plugin(constants.L3)
l3plugin = directory.get_plugin(plugin_constants.L3)
kwargs = {'arg_list': (external_net.EXTERNAL,),
external_net.EXTERNAL: True}
with self.network(**kwargs) as n:
@ -1045,7 +1046,7 @@ class TestMl2PortsV2(test_plugin.TestPortsV2, Ml2PluginV2TestCase):
def test_delete_port_no_notify_in_disassociate_floatingips(self):
ctx = context.get_admin_context()
plugin = directory.get_plugin()
l3plugin = directory.get_plugin(constants.L3)
l3plugin = directory.get_plugin(plugin_constants.L3)
with self.port() as port,\
mock.patch.object(
l3plugin,
@ -1101,7 +1102,7 @@ class TestMl2PortsV2(test_plugin.TestPortsV2, Ml2PluginV2TestCase):
def test_disassociate_floatingips_do_notify_returns_nothing(self):
ctx = context.get_admin_context()
l3plugin = directory.get_plugin(constants.L3)
l3plugin = directory.get_plugin(plugin_constants.L3)
with self.port() as port:
port_id = port['port']['id']
@ -1248,7 +1249,7 @@ class TestMl2PortsV2WithL3(test_plugin.TestPortsV2, Ml2PluginV2TestCase):
plugin = directory.get_plugin()
# enable subscription for events
l3_router_plugin.L3RouterPlugin()
l3plugin = directory.get_plugin(constants.L3)
l3plugin = directory.get_plugin(plugin_constants.L3)
host_arg = {portbindings.HOST_ID: HOST}
with mock.patch.object(l3plugin.l3_rpc_notifier,
'routers_updated_on_host') as mock_updated:
@ -1454,7 +1455,7 @@ class TestMl2DvrPortsV2(TestMl2PortsV2):
mock.PropertyMock(return_value=extensions))
def test_delete_port_notifies_l3_plugin(self, floating_ip=False):
directory.add_plugin(constants.L3, self.l3plugin)
directory.add_plugin(plugin_constants.L3, self.l3plugin)
ns_to_delete = {'host': 'myhost', 'agent_id': 'vm_l3_agent',
'router_id': 'my_router'}
router_ids = set()
@ -1484,7 +1485,7 @@ class TestMl2DvrPortsV2(TestMl2PortsV2):
self.test_delete_port_notifies_l3_plugin(floating_ip=True)
def test_concurrent_csnat_port_delete(self):
plugin = directory.get_plugin(constants.L3)
plugin = directory.get_plugin(plugin_constants.L3)
r = plugin.create_router(
self.context,
{'router': {'name': 'router', 'admin_state_up': True,

3
neutron/tests/unit/plugins/ml2/test_rpc.py
View File

@ -22,6 +22,7 @@ import collections
import mock
from neutron_lib.callbacks import resources
from neutron_lib import constants
from neutron_lib.plugins import constants as plugin_constants
from neutron_lib.plugins import directory
from oslo_config import cfg
from oslo_context import context as oslo_context
@ -49,7 +50,7 @@ class RpcCallbacksTestCase(base.BaseTestCase):
self.callbacks = plugin_rpc.RpcCallbacks(self.notifier,
self.type_manager)
self.plugin = mock.MagicMock()
directory.add_plugin(constants.CORE, self.plugin)
directory.add_plugin(plugin_constants.CORE, self.plugin)
def _test_update_device_up(self, host=None):
kwargs = {

27
neutron/tests/unit/scheduler/test_l3_agent_scheduler.py
View File

@ -21,6 +21,7 @@ import mock
from neutron_lib.api.definitions import portbindings
from neutron_lib import constants
from neutron_lib import context as n_context
from neutron_lib.plugins import constants as plugin_constants
from neutron_lib.plugins import directory
from oslo_config import cfg
from oslo_utils import importutils
@ -828,7 +829,7 @@ class L3DvrSchedulerTestCase(testlib_api.SqlTestCase):
port = kwargs.get('original_port')
port_addr_pairs = port['allowed_address_pairs']
l3plugin = mock.Mock()
directory.add_plugin(constants.L3, l3plugin)
directory.add_plugin(plugin_constants.L3, l3plugin)
l3_dvrscheduler_db._notify_l3_agent_port_update(
'port', 'after_update', mock.ANY, **kwargs)
l3plugin._get_allowed_address_pair_fixed_ips.return_value = (
@ -875,7 +876,7 @@ class L3DvrSchedulerTestCase(testlib_api.SqlTestCase):
port = kwargs.get('port')
port_addr_pairs = port['allowed_address_pairs']
l3plugin = mock.Mock()
directory.add_plugin(constants.L3, l3plugin)
directory.add_plugin(plugin_constants.L3, l3plugin)
l3_dvrscheduler_db._notify_l3_agent_port_update(
'port', 'after_update', mock.ANY, **kwargs)
self.assertTrue(
@ -911,7 +912,7 @@ class L3DvrSchedulerTestCase(testlib_api.SqlTestCase):
'router', constants.L3_AGENT_SCHEDULER_EXT_ALIAS,
constants.L3_DISTRIBUTED_EXT_ALIAS
]
directory.add_plugin(constants.L3, l3plugin)
directory.add_plugin(plugin_constants.L3, l3plugin)
l3_dvrscheduler_db._notify_l3_agent_port_update(
'port', 'after_update', plugin, **kwargs)
self.assertFalse(
@ -930,7 +931,7 @@ class L3DvrSchedulerTestCase(testlib_api.SqlTestCase):
},
}
l3plugin = mock.Mock()
directory.add_plugin(constants.L3, l3plugin)
directory.add_plugin(plugin_constants.L3, l3plugin)
l3_dvrscheduler_db._notify_l3_agent_new_port(
'port', 'after_create', mock.ANY, **kwargs)
l3plugin.update_arp_entry_for_dvr_service_port.\
@ -948,7 +949,7 @@ class L3DvrSchedulerTestCase(testlib_api.SqlTestCase):
}
}
l3plugin = mock.Mock()
directory.add_plugin(constants.L3, l3plugin)
directory.add_plugin(plugin_constants.L3, l3plugin)
l3_dvrscheduler_db._notify_l3_agent_new_port(
'port', 'after_create', mock.ANY, **kwargs)
self.assertFalse(
@ -970,7 +971,7 @@ class L3DvrSchedulerTestCase(testlib_api.SqlTestCase):
},
}
l3plugin = mock.Mock()
directory.add_plugin(constants.L3, l3plugin)
directory.add_plugin(plugin_constants.L3, l3plugin)
l3_dvrscheduler_db._notify_l3_agent_port_update(
'port', 'after_update', mock.ANY, **kwargs)
l3plugin.dvr_handle_new_service_port.assert_called_once_with(
@ -992,7 +993,7 @@ class L3DvrSchedulerTestCase(testlib_api.SqlTestCase):
},
}
l3plugin = mock.Mock()
directory.add_plugin(constants.L3, l3plugin)
directory.add_plugin(plugin_constants.L3, l3plugin)
l3_dvrscheduler_db._notify_l3_agent_port_update(
'port', 'after_update', mock.ANY, **kwargs)
@ -1017,7 +1018,7 @@ class L3DvrSchedulerTestCase(testlib_api.SqlTestCase):
'mac_address_updated': True
}
l3plugin = mock.Mock()
directory.add_plugin(constants.L3, l3plugin)
directory.add_plugin(plugin_constants.L3, l3plugin)
l3_dvrscheduler_db._notify_l3_agent_port_update(
'port', 'after_update', mock.ANY, **kwargs)
@ -1063,7 +1064,7 @@ class L3DvrSchedulerTestCase(testlib_api.SqlTestCase):
},
}
l3plugin = mock.Mock()
directory.add_plugin(constants.L3, l3plugin)
directory.add_plugin(plugin_constants.L3, l3plugin)
with mock.patch.object(l3plugin, 'get_dvr_routers_to_remove',
return_value=routers_to_remove),\
mock.patch.object(l3plugin, '_get_floatingips_by_port_id',
@ -1110,7 +1111,7 @@ class L3DvrSchedulerTestCase(testlib_api.SqlTestCase):
'router', constants.L3_AGENT_SCHEDULER_EXT_ALIAS,
constants.L3_DISTRIBUTED_EXT_ALIAS
]
directory.add_plugin(constants.L3, l3plugin)
directory.add_plugin(plugin_constants.L3, l3plugin)
with mock.patch.object(l3plugin, 'get_dvr_routers_to_remove',
return_value=[{'agent_id': 'foo_agent',
'router_id': 'foo_id',
@ -1138,7 +1139,7 @@ class L3DvrSchedulerTestCase(testlib_api.SqlTestCase):
'router', constants.L3_AGENT_SCHEDULER_EXT_ALIAS,
constants.L3_DISTRIBUTED_EXT_ALIAS
]
directory.add_plugin(constants.L3, l3plugin)
directory.add_plugin(plugin_constants.L3, l3plugin)
port = {
'id': uuidutils.generate_uuid(),
'device_id': 'abcd',
@ -1314,7 +1315,7 @@ class L3HAPlugin(db_v2.NeutronDbPluginV2,
@classmethod
def get_plugin_type(cls):
return constants.L3
return plugin_constants.L3
def get_plugin_description(self):
return "L3 Routing Service Plugin for testing"
@ -1337,7 +1338,7 @@ class L3HATestCaseMixin(testlib_api.SqlTestCase,
cfg.CONF.set_override('max_l3_agents_per_router', 0)
manager.init()
self.plugin = directory.get_plugin(constants.L3)
self.plugin = directory.get_plugin(plugin_constants.L3)
self.plugin.router_scheduler = importutils.import_object(
'neutron.scheduler.l3_agent_scheduler.ChanceScheduler'
)

2
neutron/tests/unit/services/revisions/test_revision_plugin.py
View File

@ -14,8 +14,8 @@
#
import netaddr
from neutron_lib import constants
from neutron_lib import context as nctx
from neutron_lib.plugins import constants
from neutron_lib.plugins import directory
from oslo_utils import uuidutils

2
neutron/tests/unit/test_manager.py
View File

@ -16,7 +16,7 @@
import weakref
import fixtures
from neutron_lib import constants as lib_const
from neutron_lib.plugins import constants as lib_const
from neutron_lib.plugins import directory
from oslo_config import cfg

3
neutron/tests/unit/test_policy.py
View File

@ -19,6 +19,7 @@ import mock
from neutron_lib import constants
from neutron_lib import context
from neutron_lib import exceptions
from neutron_lib.plugins import constants as plugin_constants
from neutron_lib.plugins import directory
from oslo_config import cfg
from oslo_db import exception as db_exc
@ -222,7 +223,7 @@ class NeutronPolicyTestCase(base.BaseTestCase):
self.context = context.Context('fake', 'fake', roles=['user'])
plugin_klass = importutils.import_class(
"neutron.db.db_base_plugin_v2.NeutronDbPluginV2")
directory.add_plugin(constants.CORE, plugin_klass())
directory.add_plugin(plugin_constants.CORE, plugin_klass())
def _set_rules(self, **kwargs):
rules_dict = {