openstack/neutron
Code Issues Proposed changes

Merge "remove neutron.common.exceptions"

Browse Source
This commit is contained in:
Zuul 2019-02-03 17:07:02 +00:00 committed by Gerrit Code Review
commit b86fa161ed
74 changed files with 286 additions and 402 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
neutron/agent/l3/agent.py
View File

@ -22,6 +22,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 context as n_context
from neutron_lib.exceptions import l3 as l3_exc
from oslo_config import cfg
from oslo_context import context as common_context
from oslo_log import log as logging
@ -52,7 +53,6 @@ from neutron.agent.linux import pd
from neutron.agent.metadata import driver as metadata_driver
from neutron.agent import rpc as agent_rpc
from neutron.common import constants as l3_constants
from neutron.common import exceptions as n_exc
from neutron.common import ipv6_utils
from neutron.common import rpc as n_rpc
from neutron.common import utils
@ -485,7 +485,7 @@ class L3NATAgent(ha.AgentMixin,
# Either ex_net_id or handle_internal_only_routers must be set
ex_net_id = (router['external_gateway_info'] or {}).get('network_id')
if not ex_net_id and not self.conf.handle_internal_only_routers:
raise n_exc.RouterNotCompatibleWithAgent(router_id=router['id'])
raise l3_exc.RouterNotCompatibleWithAgent(router_id=router['id'])
# If target_ex_net_id and ex_net_id are set they must be equal
target_ex_net_id = self._fetch_external_net_id()
@ -493,7 +493,7 @@ class L3NATAgent(ha.AgentMixin,
# Double check that our single external_net_id has not changed
# by forcing a check by RPC.
if ex_net_id != self._fetch_external_net_id(force=True):
raise n_exc.RouterNotCompatibleWithAgent(
raise l3_exc.RouterNotCompatibleWithAgent(
router_id=router['id'])
if router['id'] not in self.router_info:
@ -623,7 +623,7 @@ class L3NATAgent(ha.AgentMixin,
try:
self._process_router_if_compatible(router)
except n_exc.RouterNotCompatibleWithAgent as e:
except l3_exc.RouterNotCompatibleWithAgent as e:
log_verbose_exc(e.msg, router)
# Was the router previously handled by this agent?
if router['id'] in self.router_info:
@ -664,7 +664,7 @@ class L3NATAgent(ha.AgentMixin,
try:
with self.namespaces_manager as ns_manager:
self.fetch_and_sync_all_routers(context, ns_manager)
except n_exc.AbortSyncRouters:
except l3_exc.AbortSyncRouters:
self.fullsync = True
def fetch_and_sync_all_routers(self, context, ns_manager):
@ -720,7 +720,7 @@ class L3NATAgent(ha.AgentMixin,
failed_routers = chunk or router_ids
LOG.exception("Failed synchronizing routers '%s' "
"due to RPC error", failed_routers)
raise n_exc.AbortSyncRouters()
raise l3_exc.AbortSyncRouters()
self.fullsync = False
LOG.debug("periodic_sync_routers_task successfully completed")

4
neutron/agent/l3/dvr_fip_ns.py
View File

@ -16,6 +16,7 @@ import contextlib
import os
from neutron_lib import constants as lib_constants
from neutron_lib.exceptions import l3 as l3_exc
from neutron_lib.utils import runtime
from oslo_concurrency import lockutils
from oslo_log import log as logging
@ -29,7 +30,6 @@ from neutron.agent.l3 import router_info
from neutron.agent.linux import ip_lib
from neutron.agent.linux import iptables_manager
from neutron.common import constants
from neutron.common import exceptions as n_exc
from neutron.common import utils as common_utils
from neutron.ipam import utils as ipam_utils
@ -352,7 +352,7 @@ class FipNamespace(namespaces.Namespace):
'ns': ns_name})
msg = _('DVR: Gateway update route in FIP namespace failed, retry '
'should be attempted on next call')
raise n_exc.FloatingIpSetupException(msg)
raise l3_exc.FloatingIpSetupException(msg)
for fixed_ip in agent_gateway_port['fixed_ips']:
ip_lib.send_ip_addr_adv_notif(ns_name,

12
neutron/agent/l3/router_info.py
View File

@ -16,6 +16,7 @@ import collections
import netaddr
from neutron_lib import constants as lib_constants
from neutron_lib.exceptions import l3 as l3_exc
from neutron_lib.utils import helpers
from oslo_log import log as logging
@ -25,7 +26,6 @@ from neutron.agent.linux import ip_lib
from neutron.agent.linux import iptables_manager
from neutron.agent.linux import ra
from neutron.common import constants as n_const
from neutron.common import exceptions as n_exc
from neutron.common import ipv6_utils
from neutron.common import utils as common_utils
from neutron.ipam import utils as ipam_utils
@ -295,7 +295,7 @@ class RouterInfo(object):
# TODO(salv-orlando): Less broad catching
msg = _('L3 agent failure to setup NAT for floating IPs')
LOG.exception(msg)
raise n_exc.FloatingIpSetupException(msg)
raise l3_exc.FloatingIpSetupException(msg)
def _add_fip_addr_to_device(self, fip, device):
"""Configures the floating ip address on the device.
@ -413,7 +413,7 @@ class RouterInfo(object):
# TODO(salv-orlando): Less broad catching
msg = _('L3 agent failure to setup floating IPs')
LOG.exception(msg)
raise n_exc.FloatingIpSetupException(msg)
raise l3_exc.FloatingIpSetupException(msg)
def put_fips_in_error_state(self):
fip_statuses = {}
@ -924,7 +924,7 @@ class RouterInfo(object):
ex_gw_port)
fip_statuses = self.configure_fip_addresses(interface_name)
except n_exc.FloatingIpSetupException:
except l3_exc.FloatingIpSetupException:
# All floating IPs must be put in error state
LOG.exception("Failed to process floating IPs.")
fip_statuses = self.put_fips_in_error_state()
@ -949,8 +949,8 @@ class RouterInfo(object):
ex_gw_port)
fip_statuses = self.configure_fip_addresses(interface_name)
except (n_exc.FloatingIpSetupException,
n_exc.IpTablesApplyException):
except (l3_exc.FloatingIpSetupException,
l3_exc.IpTablesApplyException):
# All floating IPs must be put in error state
LOG.exception("Failed to process floating IPs.")
fip_statuses = self.put_fips_in_error_state()

2
neutron/agent/linux/daemon.py
View File

@ -22,10 +22,10 @@ import pwd
import signal
import sys
from neutron_lib import exceptions
from oslo_log import log as logging
from neutron._i18n import _
from neutron.common import exceptions
LOG = logging.getLogger(__name__)

2
neutron/agent/linux/interface.py
View File

@ -18,6 +18,7 @@ import time
import netaddr
from neutron_lib import constants
from neutron_lib import exceptions
from oslo_log import log as logging
from oslo_utils import excutils
import six
@ -25,7 +26,6 @@ import six
from neutron.agent.common import ovs_lib
from neutron.agent.linux import ip_lib
from neutron.common import constants as n_const
from neutron.common import exceptions
LOG = logging.getLogger(__name__)

6
neutron/agent/linux/ip_conntrack.py
View File

@ -15,12 +15,12 @@ import re
import eventlet
import netaddr
from neutron_lib import exceptions
from oslo_concurrency import lockutils
from oslo_log import log as logging
from neutron.agent.linux import utils as linux_utils
from neutron.common import constants as n_const
from neutron.common import exceptions as n_exc
LOG = logging.getLogger(__name__)
CONTRACK_MGRS = {}
@ -225,7 +225,7 @@ class IpConntrackManager(object):
"""Generates a unique conntrack zone for the passed in ID."""
try:
zone = self._find_open_zone()
except n_exc.CTZoneExhaustedError:
except exceptions.CTZoneExhaustedError:
# Free some zones and try again, repeat failure will not be caught
self._free_zones_from_removed_ports()
zone = self._find_open_zone()
@ -250,4 +250,4 @@ class IpConntrackManager(object):
# gap found, let's use it!
return index + ZONE_START
# conntrack zones exhausted :( :(
raise n_exc.CTZoneExhaustedError()
raise exceptions.CTZoneExhaustedError()

4
neutron/agent/linux/ip_lib.py
View File

@ -32,7 +32,6 @@ from pyroute2 import netns
from neutron._i18n import _
from neutron.agent.common import utils
from neutron.common import exceptions as n_exc
from neutron.common import ipv6_utils
from neutron.common import utils as common_utils
from neutron.privileged.agent.linux import ip_lib as privileged
@ -285,7 +284,8 @@ class IPWrapper(SubProcessBase):
if len(srcport) == 2 and srcport[0] <= srcport[1]:
kwargs['vxlan_port_range'] = (str(srcport[0]), str(srcport[1]))
else:
raise n_exc.NetworkVxlanPortRangeError(vxlan_range=srcport)
raise exceptions.NetworkVxlanPortRangeError(
vxlan_range=srcport)
if dstport:
kwargs['vxlan_port'] = dstport
privileged.create_interface(name, self.namespace, "vxlan", **kwargs)

11
neutron/agent/linux/iptables_manager.py
View File

@ -25,6 +25,8 @@ import os
import re
import sys
from neutron_lib import exceptions
from neutron_lib.exceptions import l3 as l3_exc
from neutron_lib.utils import runtime
from oslo_concurrency import lockutils
from oslo_config import cfg
@ -36,7 +38,6 @@ from neutron.agent.linux import ip_lib
from neutron.agent.linux import iptables_comments as ic
from neutron.agent.linux import utils as linux_utils
from neutron.common import constants
from neutron.common import exceptions as n_exc
from neutron.conf.agent import common as config
LOG = logging.getLogger(__name__)
@ -424,13 +425,13 @@ class IptablesManager(object):
finally:
try:
self.defer_apply_off()
except n_exc.IpTablesApplyException:
except l3_exc.IpTablesApplyException:
# already in the format we want, just reraise
raise
except Exception:
msg = _('Failure applying iptables rules')
LOG.exception(msg)
raise n_exc.IpTablesApplyException(msg)
raise l3_exc.IpTablesApplyException(msg)
def defer_apply_on(self):
self.iptables_apply_deferred = True
@ -462,7 +463,7 @@ class IptablesManager(object):
msg = (_("IPTables Rules did not converge. Diff: %s") %
'\n'.join(second))
LOG.error(msg)
raise n_exc.IpTablesApplyException(msg)
raise l3_exc.IpTablesApplyException(msg)
return first
def get_rules_for_table(self, table):
@ -495,7 +496,7 @@ class IptablesManager(object):
return self._do_run_restore(args, commands, lock=True)
err = self._do_run_restore(args, commands)
if (isinstance(err, n_exc.ProcessExecutionError) and
if (isinstance(err, exceptions.ProcessExecutionError) and
err.returncode == XTABLES_RESOURCE_PROBLEM_CODE):
# maybe we run on a platform that includes iptables commit
# 999eaa241212d3952ddff39a99d0d55a74e3639e (for example, latest

2
neutron/agent/linux/l3_tc_lib.py
View File

@ -13,11 +13,11 @@
import re
from neutron_lib import constants
from neutron_lib import exceptions
from oslo_log import log as logging
from neutron.agent.linux import ip_lib
from neutron.agent.linux import tc_lib
from neutron.common import exceptions
LOG = logging.getLogger(__name__)

2
neutron/agent/linux/utils.py
View File

@ -24,6 +24,7 @@ import time
import eventlet
from eventlet.green import subprocess
from neutron_lib import exceptions
from neutron_lib.utils import helpers
from oslo_config import cfg
from oslo_log import log as logging
@ -35,7 +36,6 @@ from six.moves import http_client as httplib
from neutron._i18n import _
from neutron.agent.linux import xenapi_root_helper
from neutron.common import exceptions
from neutron.common import utils
from neutron.conf.agent import common as config
from neutron import wsgi

3
neutron/agent/metadata/driver.py
View File

@ -21,6 +21,7 @@ import pwd
from neutron_lib.callbacks import events
from neutron_lib.callbacks import registry
from neutron_lib.callbacks import resources
from neutron_lib import exceptions
from oslo_config import cfg
from oslo_log import log as logging
@ -29,7 +30,7 @@ from neutron.agent.l3 import ha_router
from neutron.agent.l3 import namespaces
from neutron.agent.linux import external_process
from neutron.common import constants
from neutron.common import exceptions
LOG = logging.getLogger(__name__)

3
neutron/agent/windows/utils.py
View File

@ -18,13 +18,14 @@ import os
import eventlet
from eventlet import tpool
from neutron_lib import exceptions
from neutron_lib.utils import helpers
from oslo_log import log as logging
from oslo_utils import encodeutils
import six
from neutron._i18n import _
from neutron.common import exceptions
if os.name == 'nt':
import wmi

2
neutron/api/extensions.py
View File

@ -18,6 +18,7 @@ import imp
import os
from neutron_lib.api import extensions as api_extensions
from neutron_lib import exceptions
from neutron_lib.plugins import directory
from oslo_config import cfg
from oslo_log import log as logging
@ -27,7 +28,6 @@ import webob.dec
import webob.exc
from neutron._i18n import _
from neutron.common import exceptions
from neutron import extensions as core_extensions
from neutron.plugins.common import constants as const
from neutron.services import provider_configuration

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

@ -33,7 +33,6 @@ from oslo_utils import excutils
from neutron._i18n import _
from neutron.common import constants as n_const
from neutron.common import exceptions as n_exc
from neutron.common import utils
from neutron.db import provisioning_blocks
from neutron.extensions import segment as segment_ext
@ -298,7 +297,7 @@ class DhcpRpcCallback(object):
utils.get_dhcp_agent_device_id(network_id, host) or
not self._is_dhcp_agent_hosting_network(plugin, context, host,
network_id)):
raise n_exc.DhcpPortInUse(port_id=port['id'])
raise exceptions.DhcpPortInUse(port_id=port['id'])
LOG.debug('Update dhcp port %(port)s '
'from %(host)s.',
{'port': port,

3
neutron/api/v2/base.py
View File

@ -31,7 +31,6 @@ from neutron._i18n import _
from neutron.api import api_common
from neutron.api.v2 import resource as wsgi_resource
from neutron.common import constants as n_const
from neutron.common import exceptions as n_exc
from neutron.common import rpc as n_rpc
from neutron import policy
from neutron import quota
@ -485,7 +484,7 @@ class Controller(object):
{self._resource: delta},
self._plugin)
reservations.append(reservation)
except n_exc.QuotaResourceUnknown as e:
except exceptions.QuotaResourceUnknown as e:
# We don't want to quota this resource
LOG.debug(e)

102
neutron/common/exceptions.py
View File

@ -1,102 +0,0 @@
# Copyright 2011 VMware, Inc
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
from neutron_lib import exceptions
from neutron_lib.exceptions import l3
from neutron_lib.exceptions import qos
# TODO(boden): remove lib shims
SubnetPoolNotFound = exceptions.SubnetPoolNotFound
StateInvalid = exceptions.StateInvalid
DhcpPortInUse = exceptions.DhcpPortInUse
HostRoutesExhausted = exceptions.HostRoutesExhausted
DNSNameServersExhausted = exceptions.DNSNameServersExhausted
FlatNetworkInUse = exceptions.FlatNetworkInUse
NoNetworkFoundInMaximumAllowedAttempts = \
exceptions.NoNetworkFoundInMaximumAllowedAttempts
MalformedRequestBody = exceptions.MalformedRequestBody
InvalidAllocationPool = exceptions.InvalidAllocationPool
UnsupportedPortDeviceOwner = \
exceptions.UnsupportedPortDeviceOwner
OverlappingAllocationPools = exceptions.OverlappingAllocationPools
OutOfBoundsAllocationPool = exceptions.OutOfBoundsAllocationPool
BridgeDoesNotExist = exceptions.BridgeDoesNotExist
QuotaResourceUnknown = exceptions.QuotaResourceUnknown
QuotaMissingTenant = exceptions.QuotaMissingTenant
InvalidQuotaValue = exceptions.InvalidQuotaValue
InvalidSharedSetting = exceptions.InvalidSharedSetting
ExtensionsNotFound = exceptions.ExtensionsNotFound
GatewayConflictWithAllocationPools = \
exceptions.GatewayConflictWithAllocationPools
GatewayIpInUse = exceptions.GatewayIpInUse
NetworkVxlanPortRangeError = exceptions.NetworkVxlanPortRangeError
VxlanNetworkUnsupported = exceptions.VxlanNetworkUnsupported
DuplicatedExtension = exceptions.DuplicatedExtension
DriverCallError = exceptions.DriverCallError
DeviceIDNotOwnedByTenant = exceptions.DeviceIDNotOwnedByTenant
InvalidCIDR = exceptions.InvalidCIDR
FailToDropPrivilegesExit = exceptions.FailToDropPrivilegesExit
NetworkIdOrRouterIdRequiredError = exceptions.NetworkIdOrRouterIdRequiredError
EmptySubnetPoolPrefixList = exceptions.EmptySubnetPoolPrefixList
PrefixVersionMismatch = exceptions.PrefixVersionMismatch
UnsupportedMinSubnetPoolPrefix = exceptions.UnsupportedMinSubnetPoolPrefix
IllegalSubnetPoolPrefixBounds = exceptions.IllegalSubnetPoolPrefixBounds
IllegalSubnetPoolPrefixUpdate = exceptions.IllegalSubnetPoolPrefixUpdate
SubnetAllocationError = exceptions.SubnetAllocationError
AddressScopePrefixConflict = exceptions.AddressScopePrefixConflict
IllegalSubnetPoolAssociationToAddressScope = \
exceptions.IllegalSubnetPoolAssociationToAddressScope
IllegalSubnetPoolIpVersionAssociationToAddressScope = \
exceptions.IllegalSubnetPoolIpVersionAssociationToAddressScope
IllegalSubnetPoolUpdate = exceptions.IllegalSubnetPoolUpdate
MinPrefixSubnetAllocationError = exceptions.MinPrefixSubnetAllocationError
MaxPrefixSubnetAllocationError = exceptions.MaxPrefixSubnetAllocationError
SubnetPoolDeleteError = exceptions.SubnetPoolDeleteError
SubnetPoolQuotaExceeded = exceptions.SubnetPoolQuotaExceeded
NetworkSubnetPoolAffinityError = exceptions.NetworkSubnetPoolAffinityError
ObjectActionError = exceptions.ObjectActionError
CTZoneExhaustedError = exceptions.CTZoneExhaustedError
TenantQuotaNotFound = exceptions.TenantQuotaNotFound
MultipleFilterIDForIPFound = exceptions.MultipleFilterIDForIPFound
FilterIDForIPNotFound = exceptions.FilterIDForIPNotFound
FailedToAddQdiscToDevice = exceptions.FailedToAddQdiscToDevice
PortBindingNotFound = exceptions.PortBindingNotFound
PortBindingAlreadyActive = exceptions.PortBindingAlreadyActive
PortBindingAlreadyExists = exceptions.PortBindingAlreadyExists
PortBindingError = exceptions.PortBindingError
ProcessExecutionError = exceptions.ProcessExecutionError
QosPolicyNotFound = qos.QosPolicyNotFound
QosRuleNotFound = qos.QosRuleNotFound
QoSPolicyDefaultAlreadyExists = qos.QoSPolicyDefaultAlreadyExists
PortQosBindingNotFound = qos.PortQosBindingNotFound
PortQosBindingError = qos.PortQosBindingError
NetworkQosBindingNotFound = qos.NetworkQosBindingNotFound
FloatingIPQosBindingNotFound = qos.FloatingIPQosBindingNotFound
FloatingIPQosBindingError = qos.FloatingIPQosBindingError
NetworkQosBindingError = qos.NetworkQosBindingError
PolicyRemoveAuthorizationError = qos.PolicyRemoveAuthorizationError
QosPolicyInUse = qos.QosPolicyInUse
QosRuleNotSupported = qos.QosRuleNotSupported
QoSRuleParameterConflict = qos.QoSRuleParameterConflict
QoSRulesConflict = qos.QoSRulesConflict
RouterQosBindingNotFound = qos.RouterQosBindingNotFound
RouterQosBindingError = qos.RouterQosBindingError
RouterNotCompatibleWithAgent = l3.RouterNotCompatibleWithAgent
FloatingIpSetupException = l3.FloatingIpSetupException
IpTablesApplyException = l3.IpTablesApplyException
AbortSyncRouters = l3.AbortSyncRouters

5
neutron/core_extensions/qos.py
View File

@ -14,11 +14,11 @@
# under the License.
from neutron_lib.db import api as db_api
from neutron_lib.exceptions import qos as qos_exc
from neutron_lib.plugins import constants as plugin_constants
from neutron_lib.plugins import directory
from neutron_lib.services.qos import constants as qos_consts
from neutron.common import exceptions as n_exc
from neutron.core_extensions import base
from neutron.objects.qos import policy as policy_object
@ -42,7 +42,8 @@ class QosCoreResourceExtension(base.CoreResourceExtension):
Using is_accessible expresses these conditions.
"""
if not (policy_object.QosPolicy.is_accessible(context, old_policy)):
raise n_exc.PolicyRemoveAuthorizationError(policy_id=old_policy.id)
raise qos_exc.PolicyRemoveAuthorizationError(
policy_id=old_policy.id)
def _update_port_policy(self, context, port, port_changes):
old_policy = policy_object.QosPolicy.get_port_policy(

11
neutron/db/db_base_plugin_common.py
View File

@ -27,14 +27,13 @@ from neutron_lib.db import api as db_api
from neutron_lib.db import model_query
from neutron_lib.db import resource_extend
from neutron_lib.db import utils as db_utils
from neutron_lib import exceptions as n_exc
from neutron_lib import exceptions
from neutron_lib.utils import net
from oslo_config import cfg
from oslo_log import log as logging
from sqlalchemy.orm import exc
from neutron.common import constants as n_const
from neutron.common import exceptions
from neutron.db import common_db_mixin
from neutron.db import models_v2
from neutron.objects import base as base_obj
@ -225,7 +224,7 @@ class DbBasePluginCommon(common_db_mixin.CommonDbMixin):
try:
network = model_query.get_by_id(context, models_v2.Network, id)
except exc.NoResultFound:
raise n_exc.NetworkNotFound(net_id=id)
raise exceptions.NetworkNotFound(net_id=id)
return network
def _get_subnet(self, context, id):
@ -234,13 +233,13 @@ class DbBasePluginCommon(common_db_mixin.CommonDbMixin):
try:
subnet = model_query.get_by_id(context, models_v2.Subnet, id)
except exc.NoResultFound:
raise n_exc.SubnetNotFound(subnet_id=id)
raise exceptions.SubnetNotFound(subnet_id=id)
return subnet
def _get_subnet_object(self, context, id):
subnet = subnet_obj.Subnet.get_object(context, id=id)
if not subnet:
raise n_exc.SubnetNotFound(subnet_id=id)
raise exceptions.SubnetNotFound(subnet_id=id)
return subnet
def _get_subnetpool(self, context, id):
@ -254,7 +253,7 @@ class DbBasePluginCommon(common_db_mixin.CommonDbMixin):
try:
port = model_query.get_by_id(context, models_v2.Port, id)
except exc.NoResultFound:
raise n_exc.PortNotFound(port_id=id)
raise exceptions.PortNotFound(port_id=id)
return port
def _get_route_by_subnet(self, context, subnet_id):

29
neutron/db/db_base_plugin_v2.py
View File

@ -44,7 +44,6 @@ from sqlalchemy import not_
from neutron._i18n import _
from neutron.api.rpc.agentnotifiers import l3_rpc_agent_api
from neutron.common import exceptions as n_exc
from neutron.common import ipv6_utils
from neutron.common import utils
from neutron.db import db_base_plugin_common
@ -100,7 +99,7 @@ def _update_subnetpool_dict(orig_pool, new_pool):
if not orig_ip_set.issubset(new_ip_set):
msg = _("Existing prefixes must be "
"a subset of the new prefixes")
raise n_exc.IllegalSubnetPoolPrefixUpdate(msg=msg)
raise exc.IllegalSubnetPoolPrefixUpdate(msg=msg)
new_ip_set.compact()
updated['prefixes'] = [str(prefix.cidr)
for prefix in new_ip_set.iter_cidrs()]
@ -245,7 +244,7 @@ class NeutronDbPluginV2(db_base_plugin_common.DbBasePluginCommon,
return
ports = ports.filter(models_v2.Port.tenant_id == tenant_id)
if ports.count():
raise n_exc.InvalidSharedSetting(network=network_id)
raise exc.InvalidSharedSetting(network=network_id)
def set_ipam_backend(self):
self.ipam = ipam_pluggable_backend.IpamPluggableBackend()
@ -292,7 +291,7 @@ class NeutronDbPluginV2(db_base_plugin_common.DbBasePluginCommon,
if rbac.target_tenant != '*'}
allowed_projects.add(network.project_id)
if project_ids - allowed_projects:
raise n_exc.InvalidSharedSetting(network=network.name)
raise exc.InvalidSharedSetting(network=network.name)
def _validate_ipv6_attributes(self, subnet, cur_subnet):
if cur_subnet:
@ -617,13 +616,13 @@ class NeutronDbPluginV2(db_base_plugin_common.DbBasePluginCommon,
ipal.ip_address == gateway_ip,
ipal.subnet_id == cur_subnet['id']).first()
if allocated and allocated.port_id:
raise n_exc.GatewayIpInUse(
raise exc.GatewayIpInUse(
ip_address=gateway_ip,
port_id=allocated.port_id)
if validators.is_attr_set(s.get('dns_nameservers')):
if len(s['dns_nameservers']) > cfg.CONF.max_dns_nameservers:
raise n_exc.DNSNameServersExhausted(
raise exc.DNSNameServersExhausted(
subnet_id=s.get('id', _('new subnet')),
quota=cfg.CONF.max_dns_nameservers)
for dns in s['dns_nameservers']:
@ -637,7 +636,7 @@ class NeutronDbPluginV2(db_base_plugin_common.DbBasePluginCommon,
if validators.is_attr_set(s.get('host_routes')):
if len(s['host_routes']) > cfg.CONF.max_subnet_host_routes:
raise n_exc.HostRoutesExhausted(
raise exc.HostRoutesExhausted(
subnet_id=s.get('id', _('new subnet')),
quota=cfg.CONF.max_subnet_host_routes)
# check if the routes are all valid
@ -1102,14 +1101,14 @@ class NeutronDbPluginV2(db_base_plugin_common.DbBasePluginCommon,
if not self.is_address_scope_owned_by_tenant(context,
address_scope_id):
raise n_exc.IllegalSubnetPoolAssociationToAddressScope(
raise exc.IllegalSubnetPoolAssociationToAddressScope(
subnetpool_id=subnetpool_id, address_scope_id=address_scope_id)
as_ip_version = self.get_ip_version_for_address_scope(context,
address_scope_id)
if ip_version != as_ip_version:
raise n_exc.IllegalSubnetPoolIpVersionAssociationToAddressScope(
raise exc.IllegalSubnetPoolIpVersionAssociationToAddressScope(
subnetpool_id=subnetpool_id, address_scope_id=address_scope_id,
ip_version=as_ip_version)
@ -1122,7 +1121,7 @@ class NeutronDbPluginV2(db_base_plugin_common.DbBasePluginCommon,
continue
sp_set = netaddr.IPSet(sp.prefixes)
if sp_set.intersection(new_set):
raise n_exc.AddressScopePrefixConflict()
raise exc.AddressScopePrefixConflict()
def _check_subnetpool_update_allowed(self, context, subnetpool_id,
address_scope_id):
@ -1139,7 +1138,7 @@ class NeutronDbPluginV2(db_base_plugin_common.DbBasePluginCommon,
"%(address_scope_id)s") % {
'subnetpool_id': subnetpool_id,
'address_scope_id': address_scope_id}
raise n_exc.IllegalSubnetPoolUpdate(reason=msg)
raise exc.IllegalSubnetPoolUpdate(reason=msg)
def _check_default_subnetpool_exists(self, context, ip_version):
"""Check if a default already exists for the given IP version.
@ -1252,14 +1251,14 @@ class NeutronDbPluginV2(db_base_plugin_common.DbBasePluginCommon,
subnetpool = self._get_subnetpool(context, id=id)
if subnet_obj.Subnet.objects_exist(context, subnetpool_id=id):
reason = _("Subnet pool has existing allocations")
raise n_exc.SubnetPoolDeleteError(reason=reason)
raise exc.SubnetPoolDeleteError(reason=reason)
subnetpool.delete()
def _check_mac_addr_update(self, context, port, new_mac, device_owner):
if (device_owner and
device_owner.startswith(
constants.DEVICE_OWNER_NETWORK_PREFIX)):
raise n_exc.UnsupportedPortDeviceOwner(
raise exc.UnsupportedPortDeviceOwner(
op=_("mac address update"), port_id=id,
device_owner=device_owner)
@ -1539,7 +1538,7 @@ class NeutronDbPluginV2(db_base_plugin_common.DbBasePluginCommon,
return
else:
# raise as extension doesn't support L3 anyways.
raise n_exc.DeviceIDNotOwnedByTenant(
raise exc.DeviceIDNotOwnedByTenant(
device_id=device_id)
if tenant_id != router['tenant_id']:
raise n_exc.DeviceIDNotOwnedByTenant(device_id=device_id)
raise exc.DeviceIDNotOwnedByTenant(device_id=device_id)

13
neutron/db/ipam_backend_mixin.py
View File

@ -30,7 +30,6 @@ from oslo_log import log as logging
from sqlalchemy.orm import exc as orm_exc
from neutron._i18n import _
from neutron.common import exceptions as n_exc
from neutron.common import ipv6_utils
from neutron.common import utils as common_utils
from neutron.db import db_base_plugin_common
@ -70,7 +69,7 @@ class IpamBackendMixin(db_base_plugin_common.DbBasePluginCommon):
"%(start)s - %(end)s:",
{'start': ip_pool['start'],
'end': ip_pool['end']})
raise n_exc.InvalidAllocationPool(pool=ip_pool)
raise exc.InvalidAllocationPool(pool=ip_pool)
return ip_range_pools
def delete_subnet(self, context, subnet_id):
@ -257,7 +256,7 @@ class IpamBackendMixin(db_base_plugin_common.DbBasePluginCommon):
for subnet in network.subnets:
if (subnet.ip_version == ip_version and
new_subnetpool_id != subnet.subnetpool_id):
raise n_exc.NetworkSubnetPoolAffinityError()
raise exc.NetworkSubnetPoolAffinityError()
def validate_allocation_pools(self, ip_pools, subnet_cidr):
"""Validate IP allocation pools.
@ -281,12 +280,12 @@ class IpamBackendMixin(db_base_plugin_common.DbBasePluginCommon):
end_ip.version != subnet.version):
LOG.info("Specified IP addresses do not match "
"the subnet IP version")
raise n_exc.InvalidAllocationPool(pool=ip_pool)
raise exc.InvalidAllocationPool(pool=ip_pool)
if start_ip < subnet_first_ip or end_ip > subnet_last_ip:
LOG.info("Found pool larger than subnet "
"CIDR:%(start)s - %(end)s",
{'start': start_ip, 'end': end_ip})
raise n_exc.OutOfBoundsAllocationPool(
raise exc.OutOfBoundsAllocationPool(
pool=ip_pool,
subnet_cidr=subnet_cidr)
# Valid allocation pool
@ -307,7 +306,7 @@ class IpamBackendMixin(db_base_plugin_common.DbBasePluginCommon):
LOG.info("Found overlapping ranges: %(l_range)s and "
"%(r_range)s",
{'l_range': l_range, 'r_range': r_range})
raise n_exc.OverlappingAllocationPools(
raise exc.OverlappingAllocationPools(
pool_1=l_range,
pool_2=r_range,
subnet_cidr=subnet_cidr)
@ -401,7 +400,7 @@ class IpamBackendMixin(db_base_plugin_common.DbBasePluginCommon):
def validate_gw_out_of_pools(self, gateway_ip, pools):
for pool_range in pools:
if netaddr.IPAddress(gateway_ip) in pool_range:
raise n_exc.GatewayConflictWithAllocationPools(
raise exc.GatewayConflictWithAllocationPools(
pool=pool_range,
ip_address=gateway_ip)

5
neutron/db/quota/driver.py
View File

@ -20,7 +20,6 @@ from neutron_lib.plugins import constants
from neutron_lib.plugins import directory
from oslo_log import log
from neutron.common import exceptions as n_exc
from neutron.db.quota import api as quota_api
from neutron.objects import quota as quota_obj
from neutron.quota import resource as res
@ -122,7 +121,7 @@ class DbQuotaDriver(object):
if quota_obj.Quota.delete_objects(
context, project_id=tenant_id) < 1:
# No record deleted means the quota was not found
raise n_exc.TenantQuotaNotFound(tenant_id=tenant_id)
raise exceptions.TenantQuotaNotFound(tenant_id=tenant_id)
@staticmethod
@db_api.retry_if_session_inactive()
@ -300,7 +299,7 @@ class DbQuotaDriver(object):
# Ensure no value is less than zero
unders = [key for key, val in values.items() if val < 0]
if unders:
raise n_exc.InvalidQuotaValue(unders=sorted(unders))
raise exceptions.InvalidQuotaValue(unders=sorted(unders))
# Get the applicable quotas
quotas = self._get_quotas(context, tenant_id, resources)

5
neutron/extensions/quotasv2.py
View File

@ -17,7 +17,7 @@ from neutron_lib.api import converters
from neutron_lib.api import extensions as api_extensions
from neutron_lib.api import faults
from neutron_lib.db import constants as const
from neutron_lib import exceptions as n_exc
from neutron_lib import exceptions
from neutron_lib.plugins import directory
from oslo_config import cfg
from oslo_utils import importutils
@ -27,7 +27,6 @@ from neutron._i18n import _
from neutron.api import extensions
from neutron.api.v2 import base
from neutron.api.v2 import resource
from neutron.common import exceptions
from neutron import quota
from neutron.quota import resource_registry
from neutron import wsgi
@ -108,7 +107,7 @@ class QuotaSetsController(wsgi.Controller):
def _check_admin(self, context,
reason=_("Only admin can view or configure quota")):
if not context.is_admin:
raise n_exc.AdminRequired(reason=reason)
raise exceptions.AdminRequired(reason=reason)
def delete(self, request, id):
self._check_admin(request.context)

49
neutron/extensions/securitygroup.py
View File

@ -21,7 +21,7 @@ from neutron_lib.api import extensions as api_extensions
from neutron_lib.api import validators
from neutron_lib import constants as const
from neutron_lib.db import constants as db_const
from neutron_lib import exceptions as nexception
from neutron_lib import exceptions
from neutron_lib.plugins import directory
from oslo_utils import netutils
import six
@ -29,43 +29,42 @@ import six
from neutron._i18n import _
from neutron.api import extensions
from neutron.api.v2 import base
from neutron.common import exceptions
from neutron.conf import quota
from neutron.extensions import standardattrdescription as stdattr_ext
from neutron.quota import resource_registry
# Security group Exceptions
class SecurityGroupInvalidPortRange(nexception.InvalidInput):
class SecurityGroupInvalidPortRange(exceptions.InvalidInput):
message = _("For TCP/UDP protocols, port_range_min must be "
"<= port_range_max")
class SecurityGroupInvalidProtocolForPortRange(nexception.InvalidInput):
class SecurityGroupInvalidProtocolForPortRange(exceptions.InvalidInput):
message = _("Invalid protocol %(protocol)s for port range, only "
"supported for TCP, UDP, UDPLITE, SCTP and DCCP.")
class SecurityGroupInvalidPortValue(nexception.InvalidInput):
class SecurityGroupInvalidPortValue(exceptions.InvalidInput):
message = _("Invalid value for port %(port)s")
class SecurityGroupInvalidIcmpValue(nexception.InvalidInput):
class SecurityGroupInvalidIcmpValue(exceptions.InvalidInput):
message = _("Invalid value for ICMP %(field)s (%(attr)s) "
"%(value)s. It must be 0 to 255.")
class SecurityGroupEthertypeConflictWithProtocol(nexception.InvalidInput):
class SecurityGroupEthertypeConflictWithProtocol(exceptions.InvalidInput):
message = _("Invalid ethertype %(ethertype)s for protocol "
"%(protocol)s.")
class SecurityGroupMissingIcmpType(nexception.InvalidInput):
class SecurityGroupMissingIcmpType(exceptions.InvalidInput):
message = _("ICMP code (port-range-max) %(value)s is provided"
" but ICMP type (port-range-min) is missing.")
class SecurityGroupInUse(nexception.InUse):
class SecurityGroupInUse(exceptions.InUse):
message = _("Security Group %(id)s %(reason)s.")
def __init__(self, **kwargs):
@ -74,60 +73,60 @@ class SecurityGroupInUse(nexception.InUse):
super(SecurityGroupInUse, self).__init__(**kwargs)
class SecurityGroupCannotRemoveDefault(nexception.InUse):
class SecurityGroupCannotRemoveDefault(exceptions.InUse):
message = _("Insufficient rights for removing default security group.")
class SecurityGroupCannotUpdateDefault(nexception.InUse):
class SecurityGroupCannotUpdateDefault(exceptions.InUse):
message = _("Updating default security group not allowed.")
class SecurityGroupDefaultAlreadyExists(nexception.InUse):
class SecurityGroupDefaultAlreadyExists(exceptions.InUse):
message = _("Default security group already exists.")
class SecurityGroupRuleInvalidProtocol(nexception.InvalidInput):
class SecurityGroupRuleInvalidProtocol(exceptions.InvalidInput):
message = _("Security group rule protocol %(protocol)s not supported. "
"Only protocol values %(values)s and integer representations "
"[0 to 255] are supported.")
class SecurityGroupRulesNotSingleTenant(nexception.InvalidInput):
class SecurityGroupRulesNotSingleTenant(exceptions.InvalidInput):
message = _("Multiple tenant_ids in bulk security group rule create"
" not allowed")
class SecurityGroupRemoteGroupAndRemoteIpPrefix(nexception.InvalidInput):
class SecurityGroupRemoteGroupAndRemoteIpPrefix(exceptions.InvalidInput):
message = _("Only remote_ip_prefix or remote_group_id may "
"be provided.")
class SecurityGroupProtocolRequiredWithPorts(nexception.InvalidInput):
class SecurityGroupProtocolRequiredWithPorts(exceptions.InvalidInput):
message = _("Must also specify protocol if port range is given.")
class SecurityGroupNotSingleGroupRules(nexception.InvalidInput):
class SecurityGroupNotSingleGroupRules(exceptions.InvalidInput):
message = _("Only allowed to update rules for "
"one security profile at a time")
class SecurityGroupNotFound(nexception.NotFound):
class SecurityGroupNotFound(exceptions.NotFound):
message = _("Security group %(id)s does not exist")
class SecurityGroupRuleNotFound(nexception.NotFound):
class SecurityGroupRuleNotFound(exceptions.NotFound):
message = _("Security group rule %(id)s does not exist")
class DuplicateSecurityGroupRuleInPost(nexception.InUse):
class DuplicateSecurityGroupRuleInPost(exceptions.InUse):
message = _("Duplicate Security Group Rule in POST.")
class SecurityGroupRuleExists(nexception.InUse):
class SecurityGroupRuleExists(exceptions.InUse):
message = _("Security group rule already exists. Rule id is %(rule_id)s.")
class SecurityGroupRuleInUse(nexception.InUse):
class SecurityGroupRuleInUse(exceptions.InUse):
message = _("Security Group Rule %(id)s %(reason)s.")
def __init__(self, **kwargs):
@ -136,15 +135,15 @@ class SecurityGroupRuleInUse(nexception.InUse):
super(SecurityGroupRuleInUse, self).__init__(**kwargs)
class SecurityGroupRuleParameterConflict(nexception.InvalidInput):
class SecurityGroupRuleParameterConflict(exceptions.InvalidInput):
message = _("Conflicting value ethertype %(ethertype)s for CIDR %(cidr)s")
class SecurityGroupConflict(nexception.Conflict):
class SecurityGroupConflict(exceptions.Conflict):
message = _("Error %(reason)s while attempting the operation.")
class SecurityGroupRuleInvalidEtherType(nexception.InvalidInput):
class SecurityGroupRuleInvalidEtherType(exceptions.InvalidInput):
message = _("Security group rule for ethertype '%(ethertype)s' not "
"supported. Allowed values are %(values)s.")

37
neutron/ipam/subnet_alloc.py
View File

@ -19,12 +19,11 @@ import operator
import netaddr
from neutron_lib import constants
from neutron_lib.db import api as db_api
from neutron_lib import exceptions as lib_exc
from neutron_lib import exceptions
from oslo_db import exception as db_exc
from oslo_utils import uuidutils
from neutron._i18n import _
from neutron.common import exceptions as n_exc
from neutron.db import models_v2
from neutron.ipam import driver
from neutron.ipam import exceptions as ipam_exc
@ -56,7 +55,7 @@ class SubnetAllocator(driver.Pool):
.filter_by(id=self._subnetpool['id']).scalar())
if current_hash is None:
# NOTE(cbrandily): subnetpool has been deleted
raise n_exc.SubnetPoolNotFound(
raise exceptions.SubnetPoolNotFound(
subnetpool_id=self._subnetpool['id'])
new_hash = uuidutils.generate_uuid()
@ -70,7 +69,7 @@ class SubnetAllocator(driver.Pool):
count = query.update({'hash': new_hash})
if not count:
raise db_exc.RetryRequest(lib_exc.SubnetPoolInUse(
raise db_exc.RetryRequest(exceptions.SubnetPoolInUse(
subnet_pool_id=self._subnetpool['id']))
def _get_allocated_cidrs(self):
@ -118,7 +117,7 @@ class SubnetAllocator(driver.Pool):
quota_unit)
if used + requested_units > quota:
raise n_exc.SubnetPoolQuotaExceeded()
raise exceptions.SubnetPoolQuotaExceeded()
def _allocate_any_subnet(self, request):
with db_api.CONTEXT_WRITER.using(self._context):
@ -141,8 +140,8 @@ class SubnetAllocator(driver.Pool):
gateway_ip=gateway_ip,
allocation_pools=pools)
msg = _("Insufficient prefix space to allocate subnet size /%s")
raise n_exc.SubnetAllocationError(reason=msg %
str(request.prefixlen))
raise exceptions.SubnetAllocationError(
reason=msg % str(request.prefixlen))
def _allocate_specific_subnet(self, request):
with db_api.CONTEXT_WRITER.using(self._context):
@ -160,17 +159,17 @@ class SubnetAllocator(driver.Pool):
allocation_pools=request.allocation_pools)
msg = _("Cannot allocate requested subnet from the available "
"set of prefixes")
raise n_exc.SubnetAllocationError(reason=msg)
raise exceptions.SubnetAllocationError(reason=msg)
def allocate_subnet(self, request):
max_prefixlen = int(self._subnetpool['max_prefixlen'])
min_prefixlen = int(self._subnetpool['min_prefixlen'])
if request.prefixlen > max_prefixlen:
raise n_exc.MaxPrefixSubnetAllocationError(
raise exceptions.MaxPrefixSubnetAllocationError(
prefixlen=request.prefixlen,
max_prefixlen=max_prefixlen)
if request.prefixlen < min_prefixlen:
raise n_exc.MinPrefixSubnetAllocationError(
raise exceptions.MinPrefixSubnetAllocationError(
prefixlen=request.prefixlen,
min_prefixlen=min_prefixlen)
@ -180,7 +179,7 @@ class SubnetAllocator(driver.Pool):
return self._allocate_specific_subnet(request)
else:
msg = _("Unsupported request type")
raise n_exc.SubnetAllocationError(reason=msg)
raise exceptions.SubnetAllocationError(reason=msg)
def get_subnet(self, subnet_id):
raise NotImplementedError()
@ -336,14 +335,14 @@ class SubnetPoolReader(object):
def _read_prefix_info(self, subnetpool):
prefix_list = subnetpool['prefixes']
if not prefix_list:
raise n_exc.EmptySubnetPoolPrefixList()
raise exceptions.EmptySubnetPoolPrefixList()
ip_version = None
for prefix in prefix_list:
if not ip_version:
ip_version = netaddr.IPNetwork(prefix).version
elif netaddr.IPNetwork(prefix).version != ip_version:
raise n_exc.PrefixVersionMismatch()
raise exceptions.PrefixVersionMismatch()
self.default_quota = subnetpool.get('default_quota')
if self.default_quota is constants.ATTR_NOT_SPECIFIED:
@ -385,10 +384,10 @@ class SubnetPoolHelper(object):
def validate_min_prefixlen(self, min_prefixlen, max_prefixlen):
if min_prefixlen < 0:
raise n_exc.UnsupportedMinSubnetPoolPrefix(prefix=min_prefixlen,
version=4)
raise exceptions.UnsupportedMinSubnetPoolPrefix(
prefix=min_prefixlen, version=4)
if min_prefixlen > max_prefixlen:
raise n_exc.IllegalSubnetPoolPrefixBounds(
raise exceptions.IllegalSubnetPoolPrefixBounds(
prefix_type='min_prefixlen',
prefixlen=min_prefixlen,
base_prefix_type='max_prefixlen',
@ -397,7 +396,7 @@ class SubnetPoolHelper(object):
def validate_max_prefixlen(self, prefixlen, ip_version):
max = self._PREFIX_VERSION_INFO[ip_version]['max_prefixlen']
if prefixlen > max:
raise n_exc.IllegalSubnetPoolPrefixBounds(
raise exceptions.IllegalSubnetPoolPrefixBounds(
prefix_type='max_prefixlen',
prefixlen=prefixlen,
base_prefix_type='ip_version_max',
@ -408,13 +407,13 @@ class SubnetPoolHelper(object):
max_prefixlen,
default_prefixlen):
if default_prefixlen < min_prefixlen:
raise n_exc.IllegalSubnetPoolPrefixBounds(
raise exceptions.IllegalSubnetPoolPrefixBounds(
prefix_type='default_prefixlen',
prefixlen=default_prefixlen,
base_prefix_type='min_prefixlen',
base_prefixlen=min_prefixlen)
if default_prefixlen > max_prefixlen:
raise n_exc.IllegalSubnetPoolPrefixBounds(
raise exceptions.IllegalSubnetPoolPrefixBounds(
prefix_type='default_prefixlen',
prefixlen=default_prefixlen,
base_prefix_type='max_prefixlen',

52
neutron/objects/qos/policy.py
View File

@ -16,12 +16,12 @@
import itertools
from neutron_lib import constants as n_const
from neutron_lib.exceptions import qos as qos_exc
from oslo_db import exception as db_exc
from oslo_utils import versionutils
from oslo_versionedobjects import exception
from oslo_versionedobjects import fields as obj_fields
from neutron.common import exceptions
from neutron.db.models import l3
from neutron.db import models_v2
from neutron.db.qos import models as qos_db_model
@ -119,8 +119,8 @@ class QosPolicy(rbac_db.NeutronRbacObject):
for rule in self.rules:
if rule_id == rule.id:
return rule
raise exceptions.QosRuleNotFound(policy_id=self.id,
rule_id=rule_id)
raise qos_exc.QosRuleNotFound(policy_id=self.id,
rule_id=rule_id)
# TODO(hichihara): For tag mechanism. This will be removed in bug/1704137
def to_dict(self):
@ -148,7 +148,7 @@ class QosPolicy(rbac_db.NeutronRbacObject):
obj = cls.get_object(context, id=policy_id)
if obj is None:
raise exceptions.QosPolicyNotFound(policy_id=policy_id)
raise qos_exc.QosPolicyNotFound(policy_id=policy_id)
return obj
@classmethod
@ -240,7 +240,7 @@ class QosPolicy(rbac_db.NeutronRbacObject):
policy_id=self.id,
_pager=pager)
if binding_obj:
raise exceptions.QosPolicyInUse(
raise qos_exc.QosPolicyInUse(
policy_id=self.id,
object_type=object_type,
object_id=binding_obj[0]['%s_id' % object_type])
@ -255,9 +255,9 @@ class QosPolicy(rbac_db.NeutronRbacObject):
try:
network_binding_obj.create()
except db_exc.DBReferenceError as e:
raise exceptions.NetworkQosBindingError(policy_id=self.id,
net_id=network_id,
db_error=e)
raise qos_exc.NetworkQosBindingError(policy_id=self.id,
net_id=network_id,
db_error=e)
def attach_port(self, port_id):
port_binding_obj = binding.QosPolicyPortBinding(
@ -265,9 +265,9 @@ class QosPolicy(rbac_db.NeutronRbacObject):
try:
port_binding_obj.create()
except db_exc.DBReferenceError as e:
raise exceptions.PortQosBindingError(policy_id=self.id,
port_id=port_id,
db_error=e)
raise qos_exc.PortQosBindingError(policy_id=self.id,
port_id=port_id,
db_error=e)
def attach_floatingip(self, fip_id):
fip_binding_obj = binding.QosPolicyFloatingIPBinding(
@ -275,9 +275,9 @@ class QosPolicy(rbac_db.NeutronRbacObject):
try:
fip_binding_obj.create()
except db_exc.DBReferenceError as e:
raise exceptions.FloatingIPQosBindingError(policy_id=self.id,
fip_id=fip_id,
db_error=e)
raise qos_exc.FloatingIPQosBindingError(policy_id=self.id,
fip_id=fip_id,
db_error=e)
def attach_router(self, router_id):
router_binding_obj = binding.QosPolicyRouterGatewayIPBinding(
@ -285,37 +285,37 @@ class QosPolicy(rbac_db.NeutronRbacObject):
try:
router_binding_obj.create()
except db_exc.DBReferenceError as e:
raise exceptions.RouterQosBindingError(policy_id=self.id,
router_id=router_id,
db_error=e)
raise qos_exc.RouterQosBindingError(policy_id=self.id,
router_id=router_id,
db_error=e)
def detach_network(self, network_id):
deleted = binding.QosPolicyNetworkBinding.delete_objects(
self.obj_context, network_id=network_id)
if not deleted:
raise exceptions.NetworkQosBindingNotFound(net_id=network_id,
policy_id=self.id)
raise qos_exc.NetworkQosBindingNotFound(net_id=network_id,
policy_id=self.id)
def detach_port(self, port_id):
deleted = binding.QosPolicyPortBinding.delete_objects(self.obj_context,
port_id=port_id)
if not deleted:
raise exceptions.PortQosBindingNotFound(port_id=port_id,
policy_id=self.id)
raise qos_exc.PortQosBindingNotFound(port_id=port_id,
policy_id=self.id)
def detach_floatingip(self, fip_id):
deleted = binding.QosPolicyFloatingIPBinding.delete_objects(
self.obj_context, fip_id=fip_id)
if not deleted:
raise exceptions.FloatingIPQosBindingNotFound(fip_id=fip_id,
policy_id=self.id)
raise qos_exc.FloatingIPQosBindingNotFound(fip_id=fip_id,
policy_id=self.id)
def detach_router(self, router_id):
deleted = binding.QosPolicyRouterGatewayIPBinding.delete_objects(
self.obj_context, router_id=router_id)
if not deleted:
raise exceptions.RouterQosBindingNotFound(router_id=router_id,
policy_id=self.id)
raise qos_exc.RouterQosBindingNotFound(router_id=router_id,
policy_id=self.id)
def set_default(self):
if not self.get_default():
@ -324,7 +324,7 @@ class QosPolicy(rbac_db.NeutronRbacObject):
project_id=self.project_id)
qos_default_policy.create()
elif self.get_default() != self.id:
raise exceptions.QoSPolicyDefaultAlreadyExists(
raise qos_exc.QoSPolicyDefaultAlreadyExists(
project_id=self.project_id)
def unset_default(self):

9
neutron/objects/qos/qos_policy_validator.py
View File

@ -13,10 +13,9 @@
# License for the specific language governing permissions and limitations
# under the License.
from neutron_lib.exceptions import qos as qos_exc
from neutron_lib.services.qos import constants as qos_consts
from neutron.common import exceptions as n_exc
def check_bandwidth_rule_conflict(policy, rule_data):
"""Implementation of the QoS Rule checker.
@ -32,7 +31,7 @@ def check_bandwidth_rule_conflict(policy, rule_data):
elif rule.rule_type == qos_consts.RULE_TYPE_MINIMUM_BANDWIDTH:
if "max_kbps" in rule_data and (
int(rule.min_kbps) > int(rule_data["max_kbps"])):
raise n_exc.QoSRuleParameterConflict(
raise qos_exc.QoSRuleParameterConflict(
rule_value=rule_data["max_kbps"],
policy_id=policy["id"],
existing_rule=rule.rule_type,
@ -40,7 +39,7 @@ def check_bandwidth_rule_conflict(policy, rule_data):
elif rule.rule_type == qos_consts.RULE_TYPE_BANDWIDTH_LIMIT:
if "min_kbps" in rule_data and (
int(rule.max_kbps) < int(rule_data["min_kbps"])):
raise n_exc.QoSRuleParameterConflict(
raise qos_exc.QoSRuleParameterConflict(
rule_value=rule_data["min_kbps"],
policy_id=policy["id"],
existing_rule=rule.rule_type,
@ -62,7 +61,7 @@ def check_rules_conflict(policy, rule_obj):
if rule.id == getattr(rule_obj, "id", None):
continue
if rule.duplicates(rule_obj):
raise n_exc.QoSRulesConflict(
raise qos_exc.QoSRulesConflict(
new_rule_type=rule_obj.rule_type,
rule_id=rule.id,
policy_id=policy.id)

7
neutron/objects/rbac_db.py
View File

@ -18,13 +18,12 @@ import itertools
from neutron_lib.callbacks import events
from neutron_lib.callbacks import registry
from neutron_lib.callbacks import resources
from neutron_lib import exceptions as lib_exc
from neutron_lib import exceptions
from six import add_metaclass
from six import with_metaclass
from sqlalchemy import and_
from neutron._i18n import _
from neutron.common import exceptions as n_exc
from neutron.db import _utils as db_utils
from neutron.db import rbac_db_mixin
from neutron.db import rbac_db_models as models
@ -191,7 +190,7 @@ class RbacNeutronDbObjectMixin(rbac_db_mixin.RbacPluginMixin,
db_obj['tenant_id'] != context.tenant_id):
msg = _("Only admins can manipulate policies on objects "
"they do not own")
raise lib_exc.InvalidInput(error_message=msg)
raise exceptions.InvalidInput(error_message=msg)
callback_map = {events.BEFORE_UPDATE: cls.validate_rbac_policy_update,
events.BEFORE_DELETE: cls.validate_rbac_policy_delete}
if event in callback_map:
@ -290,7 +289,7 @@ class RbacNeutronMetaclass(type):
synthetic_attr = mcs.get_attribute('synthetic_fields', bases, dct)
dct['synthetic_fields'] = synthetic_attr or []
if 'shared' in dct['synthetic_fields']:
raise n_exc.ObjectActionError(
raise exceptions.ObjectActionError(
action=_('shared attribute switching to synthetic'),
reason=_('already a synthetic attribute'))
dct['synthetic_fields'].append('shared')

2
neutron/pecan_wsgi/hooks/quota_enforcement.py
View File

@ -16,10 +16,10 @@
import collections
from neutron_lib.db import api as db_api
from neutron_lib import exceptions
from oslo_log import log as logging
from pecan import hooks
from neutron.common import exceptions
from neutron import manager
from neutron import quota
from neutron.quota import resource_registry

4
neutron/plugins/ml2/drivers/helpers.py
View File

@ -17,6 +17,7 @@ import random
from neutron_lib import context as neutron_ctx
from neutron_lib.db import api as db_api
from neutron_lib import exceptions
from neutron_lib.plugins.ml2 import api
from neutron_lib.plugins import utils as p_utils
from neutron_lib.utils import helpers
@ -24,7 +25,6 @@ from oslo_config import cfg
from oslo_db import exception as db_exc
from oslo_log import log
from neutron.common import exceptions as exc
from neutron.objects import base as base_obj
@ -172,4 +172,4 @@ class SegmentTypeDriver(BaseTypeDriver):
"segment": raw_segment})
# saving real exception in case we exceeded amount of attempts
raise db_exc.RetryRequest(
exc.NoNetworkFoundInMaximumAllowedAttempts())
exceptions.NoNetworkFoundInMaximumAllowedAttempts())

2
neutron/plugins/ml2/drivers/linuxbridge/agent/linuxbridge_neutron_agent.py
View File

@ -24,6 +24,7 @@ import sys
import netaddr
from neutron_lib.agent import topics
from neutron_lib import constants
from neutron_lib import exceptions
from neutron_lib.plugins import utils as plugin_utils
from neutron_lib.utils import helpers
from oslo_config import cfg
@ -38,7 +39,6 @@ from neutron.agent.linux import bridge_lib
from neutron.agent.linux import ip_lib
from neutron.api.rpc.handlers import securitygroups_rpc as sg_rpc
from neutron.common import config as common_config
from neutron.common import exceptions
from neutron.common import profiler as setup_profiler
from neutron.common import utils
from neutron.conf.agent import common as agent_config

3
neutron/plugins/ml2/drivers/type_flat.py
View File

@ -22,7 +22,6 @@ from oslo_config import cfg
from oslo_log import log
from neutron._i18n import _
from neutron.common import exceptions as n_exc
from neutron.conf.plugins.ml2.drivers import driver_type
from neutron.objects.plugins.ml2 import flatallocation as flat_obj
from neutron.plugins.ml2.drivers import helpers
@ -96,7 +95,7 @@ class FlatTypeDriver(helpers.BaseTypeDriver):
physical_network=physical_network)
alloc.create()
except obj_base.NeutronDbObjectDuplicateEntry:
raise n_exc.FlatNetworkInUse(
raise exc.FlatNetworkInUse(
physical_network=physical_network)
segment[api.MTU] = self.get_mtu(alloc.physical_network)
return segment

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

@ -68,7 +68,6 @@ from neutron.api.rpc.handlers import metadata_rpc
from neutron.api.rpc.handlers import resources_rpc
from neutron.api.rpc.handlers import securitygroups_rpc
from neutron.common import constants as n_const
from neutron.common import exceptions as n_exc
from neutron.common import rpc as n_rpc
from neutron.common import utils
from neutron.db import address_scope_db
@ -2111,7 +2110,7 @@ class Ml2Plugin(db_base_plugin_v2.NeutronDbPluginV2,
self._validate_compute_port(port_db)
if self._get_binding_for_host(port_db.port_bindings,
attrs[pbe_ext.HOST]):
raise n_exc.PortBindingAlreadyExists(
raise exc.PortBindingAlreadyExists(
port_id=port_id, host=attrs[pbe_ext.HOST])
status = const.ACTIVE
is_active_binding = True
@ -2138,8 +2137,8 @@ class Ml2Plugin(db_base_plugin_v2.NeutronDbPluginV2,
mech_context, allow_commit=is_active_binding)
if (bind_context._binding.vif_type ==
portbindings.VIF_TYPE_BINDING_FAILED):
raise n_exc.PortBindingError(port_id=port_id,
host=attrs[pbe_ext.HOST])
raise exc.PortBindingError(port_id=port_id,
host=attrs[pbe_ext.HOST])
bind_context._binding.port_id = port_id
bind_context._binding.status = status
if not is_active_binding:
@ -2175,7 +2174,7 @@ class Ml2Plugin(db_base_plugin_v2.NeutronDbPluginV2,
binding = ports_obj.PortBinding.get_object(context, host=host,
port_id=port_id)
if not binding:
raise n_exc.PortBindingNotFound(port_id=port_id, host=host)
raise exc.PortBindingNotFound(port_id=port_id, host=host)
return self._make_port_binding_dict(binding, fields)
def _get_binding_for_host(self, bindings, host):
@ -2193,7 +2192,7 @@ class Ml2Plugin(db_base_plugin_v2.NeutronDbPluginV2,
original_binding = self._get_binding_for_host(
port_db.port_bindings, host)
if not original_binding:
raise n_exc.PortBindingNotFound(port_id=port_id, host=host)
raise exc.PortBindingNotFound(port_id=port_id, host=host)
is_active_binding = (original_binding.status == const.ACTIVE)
network = self.get_network(context, port_db['network_id'])
port_dict = self._make_port_dict(port_db)
@ -2210,7 +2209,7 @@ class Ml2Plugin(db_base_plugin_v2.NeutronDbPluginV2,
mech_context, allow_commit=is_active_binding)
if (bind_context._binding.vif_type ==
portbindings.VIF_TYPE_BINDING_FAILED):
raise n_exc.PortBindingError(port_id=port_id, host=host)
raise exc.PortBindingError(port_id=port_id, host=host)
if not is_active_binding:
with db_api.CONTEXT_WRITER.using(context):
bind_context._binding.persist_state_to_session(context.session)
@ -2230,12 +2229,12 @@ class Ml2Plugin(db_base_plugin_v2.NeutronDbPluginV2,
active_binding = p_utils.get_port_binding_by_status_and_host(
port_db.port_bindings, const.ACTIVE)
if host == (active_binding and active_binding.host):
raise n_exc.PortBindingAlreadyActive(port_id=port_id,
host=host)
raise exc.PortBindingAlreadyActive(port_id=port_id,
host=host)
inactive_binding = p_utils.get_port_binding_by_status_and_host(
port_db.port_bindings, const.INACTIVE, host=host)
if not inactive_binding or inactive_binding.host != host:
raise n_exc.PortBindingNotFound(port_id=port_id, host=host)
raise exc.PortBindingNotFound(port_id=port_id, host=host)
network = self.get_network(context, port_db['network_id'])
port_dict = self._make_port_dict(port_db)
levels = db.get_binding_level_objs(context, port_id,
@ -2267,7 +2266,7 @@ class Ml2Plugin(db_base_plugin_v2.NeutronDbPluginV2,
self.notifier.binding_activate(context, port_id,
inactive_binding.host)
return self._make_port_binding_dict(cur_context._binding)
raise n_exc.PortBindingError(port_id=port_id, host=host)
raise exc.PortBindingError(port_id=port_id, host=host)
@utils.transaction_guard
@db_api.retry_if_session_inactive()

2
neutron/privileged/agent/linux/netlink_lib.py
View File

@ -40,10 +40,10 @@ from ctypes import util
import re
from neutron_lib import constants
from neutron_lib import exceptions
from oslo_log import log as logging
from neutron._i18n import _
from neutron.common import exceptions
from neutron import privileged
from neutron.privileged.agent.linux import netlink_constants as nl_constants

5
neutron/quota/__init__.py
View File

@ -16,7 +16,7 @@
import sys
from neutron_lib import exceptions as lib_exc
from neutron_lib import exceptions
from oslo_config import cfg
from oslo_log import log as logging
from oslo_log import versionutils
@ -25,7 +25,6 @@ import six
import webob
from neutron._i18n import _
from neutron.common import exceptions
from neutron.conf import quota
from neutron.db.quota import api as quota_api
from neutron.quota import resource_registry
@ -96,7 +95,7 @@ class ConfDriver(object):
overs = [key for key, val in values.items()
if quotas[key] >= 0 and quotas[key] < val]
if overs:
raise lib_exc.OverQuota(overs=sorted(overs), quotas=quotas,
raise exceptions.OverQuota(overs=sorted(overs), quotas=quotas,
usages={})
@staticmethod

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

@ -31,7 +31,6 @@ from neutron_lib.plugins import utils as p_utils
from oslo_log import log as logging
from neutron._i18n import _
from neutron.common import exceptions as c_exc
from neutron.db import common_db_mixin
from neutron.objects import auto_allocate as auto_allocate_obj
from neutron.objects import base as base_obj
@ -295,7 +294,7 @@ class AutoAllocatedTopologyMixin(common_db_mixin.CommonDbMixin):
subnets.append(p_utils.create_subnet(
self.core_plugin, context, {'subnet': subnet_args}))
return subnets
except (c_exc.SubnetAllocationError, ValueError,
except (n_exc.SubnetAllocationError, ValueError,
n_exc.BadRequest, n_exc.NotFound) as e:
LOG.error("Unable to auto allocate topology for tenant "
"%(tenant_id)s due to missing or unmet "

2
neutron/services/logapi/drivers/manager.py
View File

@ -15,9 +15,9 @@
from neutron_lib.callbacks import events
from neutron_lib.callbacks import registry
from neutron_lib import exceptions
from oslo_log import log as logging
from neutron.common import exceptions
from neutron.services.logapi.common import constants as log_const
from neutron.services.logapi.common import exceptions as log_exc
from neutron.services.logapi.rpc import server as server_rpc

2
neutron/services/qos/drivers/manager.py
View File

@ -14,6 +14,7 @@ from neutron_lib.api.definitions import portbindings
from neutron_lib.callbacks import events
from neutron_lib.callbacks import registry
from neutron_lib import constants as lib_constants
from neutron_lib import exceptions
from neutron_lib.plugins import utils
from neutron_lib.services.qos import constants as qos_consts
from oslo_log import log as logging
@ -23,7 +24,6 @@ from neutron.api.rpc.callbacks.producer import registry as rpc_registry
from neutron.api.rpc.callbacks import resources
from neutron.api.rpc.handlers import resources_rpc
from neutron.common import constants
from neutron.common import exceptions
from neutron.objects.qos import policy as policy_object

10
neutron/services/qos/qos_plugin.py
View File

@ -26,11 +26,11 @@ from neutron_lib import context
from neutron_lib.db import api as db_api
from neutron_lib.db import resource_extend
from neutron_lib import exceptions as lib_exc
from neutron_lib.exceptions import qos as qos_exc
from neutron_lib.placement import constants as pl_constants
from neutron_lib.placement import utils as pl_utils
from neutron_lib.services.qos import constants as qos_consts
from neutron.common import exceptions as n_exc
from neutron.db import db_base_plugin_common
from neutron.extensions import qos
from neutron.objects import base as base_obj
@ -223,8 +223,8 @@ class QoSPlugin(qos.QoSPluginBase):
def validate_policy_for_port(self, policy, port):
for rule in policy.rules:
if not self.driver_manager.validate_rule_for_port(rule, port):
raise n_exc.QosRuleNotSupported(rule_type=rule.rule_type,
port_id=port['id'])
raise qos_exc.QosRuleNotSupported(rule_type=rule.rule_type,
port_id=port['id'])
@db_base_plugin_common.convert_result_to_dict
def create_policy(self, context, policy):
@ -472,14 +472,14 @@ class QoSPlugin(qos.QoSPluginBase):
:type policy_id: str uuid
:returns: a QoS policy rule object
:raises: n_exc.QosRuleNotFound
:raises: qos_exc.QosRuleNotFound
"""
with db_api.autonested_transaction(context.session):
# Ensure we have access to the policy.
policy_object.QosPolicy.get_policy_obj(context, policy_id)
rule = rule_cls.get_object(context, id=rule_id)
if not rule:
raise n_exc.QosRuleNotFound(policy_id=policy_id, rule_id=rule_id)
raise qos_exc.QosRuleNotFound(policy_id=policy_id, rule_id=rule_id)
return rule
# TODO(QoS): enforce rule types when accessing rule objects

2
neutron/tests/functional/agent/l3/extensions/qos/test_fip_qos_extension.py
View File

@ -15,12 +15,12 @@
import mock
from neutron_lib import constants
from neutron_lib import exceptions
from oslo_utils import uuidutils
from neutron.agent.l3 import agent as neutron_l3_agent
from neutron.agent.l3.extensions.qos import fip as fip_qos
from neutron.agent.linux import ip_lib
from neutron.common import exceptions
from neutron.common import utils as common_utils
from neutron.objects.qos import policy
from neutron.objects.qos import rule

2
neutron/tests/functional/agent/l3/extensions/test_gateway_ip_qos_extension.py
View File

@ -15,11 +15,11 @@
import mock
from neutron_lib import constants
from neutron_lib import exceptions
from oslo_utils import uuidutils
from neutron.agent.l3 import agent as neutron_l3_agent
from neutron.agent.l3.extensions.qos import gateway_ip as gateway_ip_qos
from neutron.common import exceptions
from neutron.common import utils as common_utils
from neutron.objects.qos import policy
from neutron.objects.qos import rule

8
neutron/tests/functional/agent/l3/test_dvr_router.py
View File

@ -20,6 +20,7 @@ import mock
import netaddr
from neutron_lib.api.definitions import portbindings
from neutron_lib import constants as lib_constants
from neutron_lib.exceptions import l3 as l3_exc
import six
import testtools
@ -33,7 +34,6 @@ from neutron.agent.l3 import namespaces
from neutron.agent.linux import ip_lib
from neutron.agent.linux import iptables_manager
from neutron.common import constants as n_const
from neutron.common import exceptions as n_exc
from neutron.common import utils
from neutron.tests.common import l3_test_common
from neutron.tests.common import machine_fixtures
@ -162,7 +162,7 @@ class TestDvrRouter(framework.L3AgentTestFramework):
for subnet in new_fg_port['subnets']:
subnet['gateway_ip'] = '19.4.4.2'
router.router[n_const.FLOATINGIP_AGENT_INTF_KEY] = [new_fg_port]
self.assertRaises(n_exc.FloatingIpSetupException,
self.assertRaises(l3_exc.FloatingIpSetupException,
self.agent._process_updated_router,
router.router)
self.agent._process_updated_router(router.router)
@ -216,7 +216,7 @@ class TestDvrRouter(framework.L3AgentTestFramework):
for subnet in new_fg_port['subnets']:
subnet['gateway_ip'] = '19.4.4.2'
router.router[n_const.FLOATINGIP_AGENT_INTF_KEY] = [new_fg_port]
self.assertRaises(n_exc.FloatingIpSetupException,
self.assertRaises(l3_exc.FloatingIpSetupException,
self.manage_router,
self.agent,
router.router)
@ -259,7 +259,7 @@ class TestDvrRouter(framework.L3AgentTestFramework):
router_info[n_const.FLOATINGIP_AGENT_INTF_KEY])
# This will raise the exception and will also clear
# subscription for the ext_net_id
self.assertRaises(n_exc.FloatingIpSetupException,
self.assertRaises(l3_exc.FloatingIpSetupException,
self.manage_router,
self.agent,
router_info)

2
neutron/tests/functional/agent/linux/test_interface.py
View File

@ -14,6 +14,7 @@
import functools
from neutron_lib import exceptions
from neutron_lib.utils import net
from oslo_config import cfg
from oslo_utils import uuidutils
@ -21,7 +22,6 @@ import testtools
from neutron.agent.linux import interface
from neutron.agent.linux import ip_lib
from neutron.common import exceptions
from neutron.common import utils
from neutron.conf.agent import common as config
from neutron.tests.common import net_helpers

2
neutron/tests/functional/agent/linux/test_l3_tc_lib.py
View File

@ -12,12 +12,12 @@
import mock
from neutron_lib import constants as common_constants
from neutron_lib import exceptions
from oslo_utils import uuidutils
from neutron.agent.l3 import namespaces
from neutron.agent.linux import ip_lib
from neutron.agent.linux import l3_tc_lib
from neutron.common import exceptions
from neutron.tests.functional import base as functional_base
RATE_LIMIT = 1024

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

@ -24,6 +24,7 @@ from neutron_lib.agent import constants as agent_consts
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.exceptions import l3 as l3_exc
from neutron_lib.plugins import constants as plugin_constants
from oslo_config import cfg
from oslo_log import log
@ -52,7 +53,6 @@ from neutron.agent.linux import ra
from neutron.agent.metadata import driver as metadata_driver
from neutron.agent import rpc as agent_rpc
from neutron.common import constants as n_const
from neutron.common import exceptions as n_exc
from neutron.conf.agent import common as agent_config
from neutron.conf.agent.l3 import config as l3_config
from neutron.conf.agent.l3 import ha as ha_conf
@ -2602,7 +2602,7 @@ class TestBasicRouterOperations(BasicRouterOperationsFramework):
with mock.patch.object(
agent, "_process_router_if_compatible",
side_effect=n_exc.RouterNotCompatibleWithAgent(
side_effect=l3_exc.RouterNotCompatibleWithAgent(
router_id=router['id'])
) as process_router_if_compatible, mock.patch.object(
agent, "_safe_router_removed"
@ -2718,7 +2718,7 @@ class TestBasicRouterOperations(BasicRouterOperationsFramework):
agent.router_info = {}
self.conf.set_override('gateway_external_network_id', 'aaa')
self.assertRaises(n_exc.RouterNotCompatibleWithAgent,
self.assertRaises(l3_exc.RouterNotCompatibleWithAgent,
agent._process_router_if_compatible,
router)
self.assertNotIn(router['id'], agent.router_info)

4
neutron/tests/unit/agent/l3/test_dvr_fip_ns.py
View File

@ -15,6 +15,7 @@
import copy
import mock
from neutron_lib.exceptions import l3 as l3_exc
from oslo_config import cfg
from oslo_utils import uuidutils
@ -24,7 +25,6 @@ from neutron.agent.l3 import link_local_allocator as lla
from neutron.agent.l3 import router_info
from neutron.agent.linux import ip_lib
from neutron.agent.linux import iptables_manager
from neutron.common import exceptions as n_exc
from neutron.common import utils as n_utils
from neutron.tests import base
@ -155,7 +155,7 @@ class TestDvrFipNs(base.BaseTestCase):
self.fip_ns._check_for_gateway_ip_change = mock.Mock(return_value=True)
self.fip_ns.agent_gateway_port = agent_gw_port
self.assertRaises(n_exc.FloatingIpSetupException,
self.assertRaises(l3_exc.FloatingIpSetupException,
self.fip_ns.create_or_update_gateway_port,
agent_gw_port)
self.assertTrue(fip_unsub.called)

6
neutron/tests/unit/agent/l3/test_router_info.py
View File

@ -12,11 +12,11 @@
import mock
from neutron_lib import constants as lib_constants
from neutron_lib.exceptions import l3 as l3_exc
from oslo_utils import uuidutils
from neutron.agent.l3 import router_info
from neutron.agent.linux import ip_lib
from neutron.common import exceptions as n_exc
from neutron.conf.agent import common as config
from neutron.conf.agent.l3 import config as l3_config
from neutron.tests import base
@ -352,7 +352,7 @@ class TestBasicRouterOperations(BasicRouterTestCaseFramework):
ri = self._create_router()
ri.process_floating_ip_nat_rules = mock.Mock(side_effect=Exception)
self.assertRaises(n_exc.FloatingIpSetupException,
self.assertRaises(l3_exc.FloatingIpSetupException,
ri.process_snat_dnat_for_fip)
ri.process_floating_ip_nat_rules.assert_called_once_with()
@ -374,7 +374,7 @@ class TestBasicRouterOperations(BasicRouterTestCaseFramework):
ri.process_floating_ip_addresses = mock.Mock(
side_effect=Exception)
self.assertRaises(n_exc.FloatingIpSetupException,
self.assertRaises(l3_exc.FloatingIpSetupException,
ri.configure_fip_addresses,
mock.sentinel.interface_name)

2
neutron/tests/unit/agent/linux/test_daemon.py
View File

@ -19,10 +19,10 @@ import os
import sys
import mock
from neutron_lib import exceptions
import testtools
from neutron.agent.linux import daemon
from neutron.common import exceptions
from neutron.tests import base
from neutron.tests import tools

2
neutron/tests/unit/agent/linux/test_interface.py
View File

@ -15,12 +15,12 @@
import mock
from neutron_lib import constants
from neutron_lib import exceptions
from oslo_utils import excutils
from neutron.agent.common import ovs_lib
from neutron.agent.linux import interface
from neutron.agent.linux import ip_lib
from neutron.common import exceptions
from neutron.conf.agent import common as config
from neutron.tests import base

5
neutron/tests/unit/agent/linux/test_ip_lib.py
View File

@ -31,7 +31,6 @@ import testtools
from neutron.agent.common import utils # noqa
from neutron.agent.linux import ip_lib
from neutron.common import exceptions as n_exc
from neutron.common import utils as common_utils
from neutron import privileged
from neutron.privileged.agent.linux import ip_lib as priv_lib
@ -389,7 +388,7 @@ class TestIpWrapper(base.BaseTestCase):
def test_add_vxlan_invalid_srcport_length(self):
wrapper = ip_lib.IPWrapper()
self.assertRaises(n_exc.NetworkVxlanPortRangeError,
self.assertRaises(exceptions.NetworkVxlanPortRangeError,
wrapper.add_vxlan, 'vxlan0', 'vni0', group='group0',
dev='dev0', ttl='ttl0', tos='tos0',
local='local0', proxy=True,
@ -397,7 +396,7 @@ class TestIpWrapper(base.BaseTestCase):
def test_add_vxlan_invalid_srcport_range(self):
wrapper = ip_lib.IPWrapper()
self.assertRaises(n_exc.NetworkVxlanPortRangeError,
self.assertRaises(exceptions.NetworkVxlanPortRangeError,
wrapper.add_vxlan, 'vxlan0', 'vni0', group='group0',
dev='dev0', ttl='ttl0', tos='tos0',
local='local0', proxy=True,

4
neutron/tests/unit/agent/linux/test_iptables_firewall.py
View File

@ -17,6 +17,7 @@ import copy
import mock
from neutron_lib import constants
from neutron_lib import exceptions
from oslo_config import cfg
import testtools
@ -25,7 +26,6 @@ from neutron.agent.linux import ip_conntrack
from neutron.agent.linux import ipset_manager
from neutron.agent.linux import iptables_comments as ic
from neutron.agent.linux import iptables_firewall
from neutron.common import exceptions as n_exc
from neutron.common import utils
from neutron.conf.agent import common as agent_config
from neutron.conf.agent import securitygroups_rpc as security_config
@ -2329,7 +2329,7 @@ class OVSHybridIptablesFirewallTestCase(BaseIptablesFirewallTestCase):
for i in range(ip_conntrack.ZONE_START,
ip_conntrack.MAX_CONNTRACK_ZONES):
self.firewall.ipconntrack._device_zone_map['dev-%s' % i] = i
with testtools.ExpectedException(n_exc.CTZoneExhaustedError):
with testtools.ExpectedException(exceptions.CTZoneExhaustedError):
self.firewall.ipconntrack._find_open_zone()
# with it full, try again, this should trigger a cleanup

9
neutron/tests/unit/agent/linux/test_iptables_manager.py
View File

@ -18,6 +18,8 @@ import sys
import fixtures
import mock
from neutron_lib import exceptions
from neutron_lib.exceptions import l3 as l3_exc
from oslo_config import cfg
import testtools
@ -26,7 +28,6 @@ from neutron.agent.linux import iptables_comments as ic
from neutron.agent.linux import iptables_manager
from neutron.agent.linux import utils as linux_utils
from neutron.common import constants
from neutron.common import exceptions as n_exc
from neutron.tests import base
from neutron.tests import tools
@ -439,7 +440,7 @@ class IptablesManagerStateFulTestCase(IptablesManagerBaseTestCase):
def test_defer_apply_with_exception(self):
self.iptables._apply = mock.Mock(side_effect=Exception)
with testtools.ExpectedException(n_exc.IpTablesApplyException):
with testtools.ExpectedException(l3_exc.IpTablesApplyException):
with self.iptables.defer_apply():
pass
@ -887,7 +888,7 @@ class IptablesManagerStateFulTestCase(IptablesManagerBaseTestCase):
# pretend line 11 failed
msg = ("Exit code: 1\nStdout: ''\n"
"Stderr: 'iptables-restore: line 11 failed\n'")
raise n_exc.ProcessExecutionError(
raise exceptions.ProcessExecutionError(
msg, iptables_manager.XTABLES_RESOURCE_PROBLEM_CODE)
return FILTER_DUMP
self.execute.side_effect = iptables_restore_failer
@ -926,7 +927,7 @@ class IptablesManagerStateFulTestCase(IptablesManagerBaseTestCase):
def test_iptables_use_table_lock(self):
# Under normal operation, if we do call iptables-restore with a -w
# and it succeeds, the next call will only use -w.
PE_error = n_exc.ProcessExecutionError(
PE_error = exceptions.ProcessExecutionError(
"", iptables_manager.XTABLES_RESOURCE_PROBLEM_CODE)
num_calls = 3

2
neutron/tests/unit/agent/linux/test_l3_tc_lib.py
View File

@ -12,9 +12,9 @@
import mock
from neutron_lib import constants
from neutron_lib import exceptions
from neutron.agent.linux import l3_tc_lib as tc_lib
from neutron.common import exceptions
from neutron.tests import base
FLOATING_IP_DEVICE_NAME = "qg-device_rfp"

18
neutron/tests/unit/agent/linux/test_utils.py
View File

@ -20,11 +20,11 @@ import mock
import six
import testtools
from neutron_lib import exceptions
from oslo_config import cfg
import oslo_i18n
from neutron.agent.linux import utils
from neutron.common import exceptions as n_exc
from neutron.tests import base
from neutron.tests.common import helpers
@ -147,7 +147,7 @@ class AgentUtilsExecuteTest(base.BaseTestCase):
self.mock_popen.return_value = ('', '')
self.process.return_value.returncode = 1
with mock.patch.object(utils, 'LOG') as log:
self.assertRaises(n_exc.ProcessExecutionError, utils.execute,
self.assertRaises(exceptions.ProcessExecutionError, utils.execute,
['ls'], log_fail_as_error=False)
self.assertFalse(log.error.called)
@ -201,8 +201,8 @@ class TestFindParentPid(base.BaseTestCase):
self.m_execute = mock.patch.object(utils, 'execute').start()
def test_returns_none_for_no_valid_pid(self):
self.m_execute.side_effect = n_exc.ProcessExecutionError('',
returncode=1)
self.m_execute.side_effect = exceptions.ProcessExecutionError(
'', returncode=1)
self.assertIsNone(utils.find_parent_pid(-1))
def test_returns_parent_id_for_good_ouput(self):
@ -210,9 +210,9 @@ class TestFindParentPid(base.BaseTestCase):
self.assertEqual(utils.find_parent_pid(-1), '123')
def test_raises_exception_returncode_0(self):
with testtools.ExpectedException(n_exc.ProcessExecutionError):
with testtools.ExpectedException(exceptions.ProcessExecutionError):
self.m_execute.side_effect = \
n_exc.ProcessExecutionError('', returncode=0)
exceptions.ProcessExecutionError('', returncode=0)
utils.find_parent_pid(-1)
def test_raises_unknown_exception(self):
@ -265,7 +265,7 @@ class TestKillProcess(base.BaseTestCase):
def _test_kill_process(self, pid, raise_exception=False,
kill_signal=signal.SIGKILL, pid_killed=True):
if raise_exception:
exc = n_exc.ProcessExecutionError('', returncode=0)
exc = exceptions.ProcessExecutionError('', returncode=0)
else:
exc = None
with mock.patch.object(utils, 'execute',
@ -284,7 +284,7 @@ class TestKillProcess(base.BaseTestCase):
self._test_kill_process('1', raise_exception=True)
def test_kill_process_raises_exception_for_execute_exception(self):
with testtools.ExpectedException(n_exc.ProcessExecutionError):
with testtools.ExpectedException(exceptions.ProcessExecutionError):
# Simulate that the process is running after trying to kill due to
# any reason such as, for example, Permission denied
self._test_kill_process('1', raise_exception=True,
@ -298,7 +298,7 @@ class TestFindChildPids(base.BaseTestCase):
def test_returns_empty_list_for_exit_code_1(self):
with mock.patch.object(utils, 'execute',
side_effect=n_exc.ProcessExecutionError(
side_effect=exceptions.ProcessExecutionError(
'', returncode=1)):
self.assertEqual([], utils.find_child_pids(-1))

2
neutron/tests/unit/agent/windows/test_utils.py
View File

@ -19,10 +19,10 @@ import ddt
import eventlet
from eventlet import tpool
import mock
from neutron_lib import exceptions
import six
from neutron.agent.windows import utils
from neutron.common import exceptions
from neutron.tests import base

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

@ -17,13 +17,12 @@ import mock
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 import exceptions
from neutron_lib.plugins import constants as plugin_constants
from neutron_lib.plugins import directory
from oslo_db import exception as db_exc
from neutron.api.rpc.handlers import dhcp_rpc
from neutron.common import exceptions
from neutron.common import utils
from neutron.db import provisioning_blocks
from neutron.tests import base
@ -153,19 +152,19 @@ class TestDhcpRpcCallback(base.BaseTestCase):
def test__port_action_bad_action(self):
self.assertRaises(
n_exc.Invalid,
exceptions.Invalid,
self._test__port_action_with_failures,
exc=None,
action='foo_action')
def test_create_port_catch_network_not_found(self):
self._test__port_action_with_failures(
exc=n_exc.NetworkNotFound(net_id='foo_network_id'),
exc=exceptions.NetworkNotFound(net_id='foo_network_id'),
action='create_port')
def test_create_port_catch_subnet_not_found(self):
self._test__port_action_with_failures(
exc=n_exc.SubnetNotFound(subnet_id='foo_subnet_id'),
exc=exceptions.SubnetNotFound(subnet_id='foo_subnet_id'),
action='create_port')
def test_create_port_catch_db_reference_error(self):
@ -175,23 +174,24 @@ class TestDhcpRpcCallback(base.BaseTestCase):
def test_create_port_catch_ip_generation_failure_reraise(self):
self.assertRaises(
n_exc.IpAddressGenerationFailure,
exceptions.IpAddressGenerationFailure,
self._test__port_action_with_failures,
exc=n_exc.IpAddressGenerationFailure(net_id='foo_network_id'),
exc=exceptions.IpAddressGenerationFailure(net_id='foo_network_id'),
action='create_port')
def test_create_port_catch_and_handle_ip_generation_failure(self):
self.plugin.get_subnet.side_effect = (
n_exc.SubnetNotFound(subnet_id='foo_subnet_id'))
exceptions.SubnetNotFound(subnet_id='foo_subnet_id'))
self._test__port_action_with_failures(
exc=n_exc.IpAddressGenerationFailure(net_id='foo_network_id'),
exc=exceptions.IpAddressGenerationFailure(net_id='foo_network_id'),
action='create_port')
self._test__port_action_with_failures(
exc=n_exc.InvalidInput(error_message='sorry'),
exc=exceptions.InvalidInput(error_message='sorry'),
action='create_port')
def test_update_port_missing_port_on_get(self):
self.plugin.get_port.side_effect = n_exc.PortNotFound(port_id='66')
self.plugin.get_port.side_effect = exceptions.PortNotFound(
port_id='66')
self.assertIsNone(self.callbacks.update_dhcp_port(
context='ctx', host='host', port_id='66',
port={'port': {'network_id': 'a'}}))
@ -199,13 +199,15 @@ class TestDhcpRpcCallback(base.BaseTestCase):
def test_update_port_missing_port_on_update(self):
self.plugin.get_port.return_value = {
'device_id': constants.DEVICE_ID_RESERVED_DHCP_PORT}
self.plugin.update_port.side_effect = n_exc.PortNotFound(port_id='66')
self.plugin.update_port.side_effect = exceptions.PortNotFound(
port_id='66')
self.assertIsNone(self.callbacks.update_dhcp_port(
context='ctx', host='host', port_id='66',
port={'port': {'network_id': 'a'}}))
def test_get_network_info_return_none_on_not_found(self):
self.plugin.get_network.side_effect = n_exc.NetworkNotFound(net_id='a')
self.plugin.get_network.side_effect = exceptions.NetworkNotFound(
net_id='a')
retval = self.callbacks.get_network_info(mock.Mock(), network_id='a')
self.assertIsNone(retval)

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

@ -17,6 +17,7 @@ import copy
import fixtures
import mock
from neutron_lib import exceptions
from neutron_lib.plugins import constants as lib_const
from neutron_lib.plugins import directory
from neutron_lib.services import base as service_base
@ -33,7 +34,6 @@ import webtest
import neutron
from neutron.api import extensions
from neutron.common import config
from neutron.common import exceptions
from neutron.plugins.common import constants
from neutron import quota
from neutron.tests import base

6
neutron/tests/unit/core_extensions/test_qos.py
View File

@ -15,11 +15,11 @@
import mock
from neutron_lib import context
from neutron_lib.exceptions import qos as qos_exc
from neutron_lib.plugins import constants as plugin_constants
from neutron_lib.services.qos import constants as qos_consts
from oslo_utils import uuidutils
from neutron.common import exceptions as n_exc
from neutron.core_extensions import base as base_core
from neutron.core_extensions import qos as qos_core
from neutron.objects.qos import policy
@ -154,7 +154,7 @@ class QosCoreResourceExtensionTestCase(base.BaseTestCase):
def test_process_resource_port_updated_remove_provided_policy(self):
self.policy_m.is_accessible.return_value = False
self.assertRaises(n_exc.PolicyRemoveAuthorizationError,
self.assertRaises(qos_exc.PolicyRemoveAuthorizationError,
self._process_port_updated_policy,
context=self.non_admin_context,
shared=False,
@ -249,7 +249,7 @@ class QosCoreResourceExtensionTestCase(base.BaseTestCase):
def test_process_fields_update_network_remove_provided_policy(self):
self.policy_m.is_accessible.return_value = False
self.assertRaises(n_exc.PolicyRemoveAuthorizationError,
self.assertRaises(qos_exc.PolicyRemoveAuthorizationError,
self._process_network_updated_policy,
context=self.non_admin_context,
shared=False,

7
neutron/tests/unit/db/quota/test_driver.py
View File

@ -14,9 +14,8 @@
# limitations under the License.
from neutron_lib import context
from neutron_lib import exceptions as lib_exc
from neutron_lib import exceptions
from neutron.common import exceptions
from neutron.db import db_base_plugin_v2 as base_plugin
from neutron.db.quota import api as quota_api
from neutron.db.quota import driver
@ -183,7 +182,7 @@ class TestDbQuotaDriver(testlib_api.SqlTestCase,
self.plugin.update_quota_limit(self.context, PROJECT, RESOURCE, 2)
self.assertRaises(lib_exc.OverQuota, self.plugin.limit_check,
self.assertRaises(exceptions.OverQuota, self.plugin.limit_check,
context.get_admin_context(), PROJECT, resources,
values)
@ -255,7 +254,7 @@ class TestDbQuotaDriver(testlib_api.SqlTestCase,
fake_count=2)}
deltas = {RESOURCE: 1}
self.plugin.update_quota_limit(self.context, PROJECT, RESOURCE, 2)
self.assertRaises(lib_exc.OverQuota,
self.assertRaises(exceptions.OverQuota,
quota_driver.make_reservation,
self.context,
self.context.tenant_id,

9
neutron/tests/unit/db/test_db_base_plugin_v2.py
View File

@ -46,7 +46,6 @@ import neutron
from neutron.api import api_common
from neutron.api import extensions
from neutron.api.v2 import router
from neutron.common import exceptions as n_exc
from neutron.common import ipv6_utils
from neutron.common import test_lib
from neutron.common import utils
@ -5618,12 +5617,12 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase):
def test_validate_subnet_dns_nameservers_exhausted(self):
self._helper_test_validate_subnet(
'max_dns_nameservers',
n_exc.DNSNameServersExhausted)
lib_exc.DNSNameServersExhausted)
def test_validate_subnet_host_routes_exhausted(self):
self._helper_test_validate_subnet(
'max_subnet_host_routes',
n_exc.HostRoutesExhausted)
lib_exc.HostRoutesExhausted)
def test_port_prevents_network_deletion(self):
with self.port() as p:
@ -6748,7 +6747,7 @@ class NeutronDbPluginV2AsMixinTestCase(NeutronDbPluginV2TestCase,
network.subnets = [models_v2.Subnet(subnetpool_id='test_id',
ip_version=constants.IP_VERSION_4)]
new_subnetpool_id = None
self.assertRaises(n_exc.NetworkSubnetPoolAffinityError,
self.assertRaises(lib_exc.NetworkSubnetPoolAffinityError,
self.plugin.ipam._validate_network_subnetpools,
network, new_subnetpool_id, 4)
@ -6802,7 +6801,7 @@ class TestNetworks(testlib_api.SqlTestCase):
plugin.update_network(ctx, net_id, network)
def test_update_shared_net_used_fails(self):
self._test_update_shared_net_used('', n_exc.InvalidSharedSetting)
self._test_update_shared_net_used('', lib_exc.InvalidSharedSetting)
def test_update_shared_net_used_as_router_gateway(self):
self._test_update_shared_net_used(

4
neutron/tests/unit/extensions/test_qos_fip.py
View File

@ -13,11 +13,11 @@
#
from neutron_lib import context
from neutron_lib.exceptions import qos as qos_exc
from neutron_lib.services.qos import constants as qos_consts
from oslo_config import cfg
from oslo_utils import uuidutils
from neutron.common import exceptions as n_exception
from neutron.conf.db import extraroute_db
from neutron.db import l3_fip_qos
from neutron.extensions import l3
@ -87,7 +87,7 @@ class FloatingIPQoSDBTestCaseBase(object):
qos_policy_id=policy_obj.id)
self.assertEqual(policy_obj.id,
fip['floatingip'][qos_consts.QOS_POLICY_ID])
self.assertRaises(n_exception.QosPolicyInUse, policy_obj.delete)
self.assertRaises(qos_exc.QosPolicyInUse, policy_obj.delete)
def test_floatingip_update_qos_policy_id(self):
ctx = context.get_admin_context()

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

@ -18,6 +18,7 @@ import sys
import mock
from neutron_lib import context
from neutron_lib.db import constants
from neutron_lib import exceptions
from neutron_lib import fixture
from oslo_config import cfg
import testtools
@ -27,7 +28,6 @@ import webtest
from neutron.api import extensions
from neutron.api.v2 import router
from neutron.common import config
from neutron.common import exceptions
from neutron.conf import quota as qconf
from neutron.db.quota import driver
from neutron import quota

4
neutron/tests/unit/extensions/test_securitygroup.py
View File

@ -22,13 +22,13 @@ from neutron_lib import constants as const
from neutron_lib import context
from neutron_lib.db import api as db_api
from neutron_lib.db import constants as db_const
from neutron_lib import exceptions
from neutron_lib.plugins import directory
from oslo_config import cfg
import oslo_db.exception as exc
import testtools
import webob.exc
from neutron.common import exceptions as n_exc
from neutron.db import db_base_plugin_v2
from neutron.db import securitygroups_db
from neutron.extensions import securitygroup as ext_sg
@ -1860,7 +1860,7 @@ class TestConvertIPPrefixToCIDR(base.BaseTestCase):
def test_convert_bad_ip_prefix_to_cidr(self):
for val in ['bad_ip', 256, "2001:db8:a::123/129"]:
self.assertRaises(n_exc.InvalidCIDR,
self.assertRaises(exceptions.InvalidCIDR,
ext_sg.convert_ip_prefix_to_cidr, val)
self.assertIsNone(ext_sg.convert_ip_prefix_to_cidr(None))

8
neutron/tests/unit/ipam/test_subnet_alloc.py
View File

@ -18,12 +18,12 @@ import netaddr
from neutron_lib import constants
from neutron_lib import context
from neutron_lib.db import api as db_api
from neutron_lib import exceptions
from neutron_lib.plugins import directory
from oslo_config import cfg
from oslo_db import exception as db_exc
from oslo_utils import uuidutils
from neutron.common import exceptions as n_exc
from neutron.ipam import requests as ipam_req
from neutron.ipam import subnet_alloc
from neutron.tests.unit.db import test_db_base_plugin_v2
@ -103,7 +103,7 @@ class TestSubnetAllocation(testlib_api.SqlTestCase):
uuidutils.generate_uuid(),
constants.IPv4,
21)
self.assertRaises(n_exc.SubnetAllocationError,
self.assertRaises(exceptions.SubnetAllocationError,
sa.allocate_subnet, req)
def test_insufficient_prefix_space_for_specific_allocation(self):
@ -115,7 +115,7 @@ class TestSubnetAllocation(testlib_api.SqlTestCase):
req = ipam_req.SpecificSubnetRequest(self._tenant_id,
uuidutils.generate_uuid(),
'10.1.0.0/21')
self.assertRaises(n_exc.SubnetAllocationError,
self.assertRaises(exceptions.SubnetAllocationError,
sa.allocate_subnet, req)
def test_allocate_any_subnet_gateway(self):
@ -183,7 +183,7 @@ class TestSubnetAllocation(testlib_api.SqlTestCase):
req = ipam_req.SpecificSubnetRequest(self._tenant_id,
uuidutils.generate_uuid(),
'fe80::/63')
self.assertRaises(n_exc.SubnetPoolQuotaExceeded,
self.assertRaises(exceptions.SubnetPoolQuotaExceeded,
sa.allocate_subnet,
req)

26
neutron/tests/unit/objects/qos/test_policy.py
View File

@ -14,12 +14,12 @@ import random
import mock
from neutron_lib import constants as n_const
from neutron_lib.exceptions import qos as qos_exc
from neutron_lib.services.qos import constants as qos_consts
from oslo_utils import uuidutils
from oslo_versionedobjects import exception
import testtools
from neutron.common import exceptions as n_exc
from neutron.objects.db import api as db_api
from neutron.objects import network as net_obj
from neutron.objects import ports as port_obj
@ -166,7 +166,7 @@ class QosPolicyObjectTestCase(test_base.BaseObjectIfaceTestCase):
def test_get_policy_obj_not_found(self):
context = self.context.elevated()
self.assertRaises(n_exc.QosPolicyNotFound,
self.assertRaises(qos_exc.QosPolicyNotFound,
policy.QosPolicy.get_policy_obj,
context, "fake_id")
@ -222,7 +222,7 @@ class QosPolicyDbObjectTestCase(test_base.BaseDbObjectTestCase,
def test_attach_network_nonexistent_network(self):
obj = self._create_test_policy()
self.assertRaises(n_exc.NetworkQosBindingError,
self.assertRaises(qos_exc.NetworkQosBindingError,
obj.attach_network, uuidutils.generate_uuid())
def test_attach_network_get_policy_network(self):
@ -254,19 +254,19 @@ class QosPolicyDbObjectTestCase(test_base.BaseDbObjectTestCase,
def test_attach_port_nonexistent_port(self):
obj = self._create_test_policy()
self.assertRaises(n_exc.PortQosBindingError,
self.assertRaises(qos_exc.PortQosBindingError,
obj.attach_port, uuidutils.generate_uuid())
def test_attach_network_nonexistent_policy(self):
policy_obj = self._make_object(self.obj_fields[0])
self.assertRaises(n_exc.NetworkQosBindingError,
self.assertRaises(qos_exc.NetworkQosBindingError,
policy_obj.attach_network, self._network_id)
def test_attach_port_nonexistent_policy(self):
policy_obj = self._make_object(self.obj_fields[0])
self.assertRaises(n_exc.PortQosBindingError,
self.assertRaises(qos_exc.PortQosBindingError,
policy_obj.attach_port, self._port['id'])
def test_attach_port_get_port_policy(self):
@ -337,22 +337,22 @@ class QosPolicyDbObjectTestCase(test_base.BaseDbObjectTestCase,
def test_detach_port_nonexistent_port(self):
obj = self._create_test_policy()
self.assertRaises(n_exc.PortQosBindingNotFound,
self.assertRaises(qos_exc.PortQosBindingNotFound,
obj.detach_port, 'non-existent-port')
def test_detach_network_nonexistent_network(self):
obj = self._create_test_policy()
self.assertRaises(n_exc.NetworkQosBindingNotFound,
self.assertRaises(qos_exc.NetworkQosBindingNotFound,
obj.detach_network, 'non-existent-port')
def test_detach_port_nonexistent_policy(self):
policy_obj = self._make_object(self.obj_fields[0])
self.assertRaises(n_exc.PortQosBindingNotFound,
self.assertRaises(qos_exc.PortQosBindingNotFound,
policy_obj.detach_port, self._port['id'])
def test_detach_network_nonexistent_policy(self):
policy_obj = self._make_object(self.obj_fields[0])
self.assertRaises(n_exc.NetworkQosBindingNotFound,
self.assertRaises(qos_exc.NetworkQosBindingNotFound,
policy_obj.detach_network, self._network_id)
@mock.patch.object(policy.QosPolicyDefault, 'create')
@ -365,7 +365,7 @@ class QosPolicyDbObjectTestCase(test_base.BaseDbObjectTestCase,
def test_set_default_default_policy_exists(self):
obj = self._create_test_policy()
with mock.patch.object(obj, 'get_default', return_value=mock.Mock()):
self.assertRaises(n_exc.QoSPolicyDefaultAlreadyExists,
self.assertRaises(qos_exc.QoSPolicyDefaultAlreadyExists,
obj.set_default)
def test_set_default_is_default_policy(self):
@ -432,7 +432,7 @@ class QosPolicyDbObjectTestCase(test_base.BaseDbObjectTestCase,
obj = self._create_test_policy()
obj.attach_port(self._port['id'])
self.assertRaises(n_exc.QosPolicyInUse, obj.delete)
self.assertRaises(qos_exc.QosPolicyInUse, obj.delete)
obj.detach_port(self._port['id'])
obj.delete()
@ -441,7 +441,7 @@ class QosPolicyDbObjectTestCase(test_base.BaseDbObjectTestCase,
obj = self._create_test_policy()
obj.attach_network(self._network_id)
self.assertRaises(n_exc.QosPolicyInUse, obj.delete)
self.assertRaises(qos_exc.QosPolicyInUse, obj.delete)
obj.detach_network(self._network_id)
obj.delete()

2
neutron/tests/unit/plugins/ml2/drivers/linuxbridge/agent/test_linuxbridge_neutron_agent.py
View File

@ -17,12 +17,12 @@ import sys
import mock
from neutron_lib import constants
from neutron_lib import exceptions
from oslo_config import cfg
from neutron.agent.linux import bridge_lib
from neutron.agent.linux import ip_lib
from neutron.agent.linux import utils
from neutron.common import exceptions
from neutron.plugins.ml2.drivers.agent import _agent_manager_base as amb
from neutron.plugins.ml2.drivers.linuxbridge.agent.common \
import constants as lconst

3
neutron/tests/unit/plugins/ml2/drivers/test_type_flat.py
View File

@ -19,7 +19,6 @@ from neutron_lib import exceptions as exc
from neutron_lib.plugins.ml2 import api
from oslo_config import cfg
from neutron.common import exceptions as n_exc
from neutron.objects.plugins.ml2 import flatallocation as flat_obj
from neutron.plugins.ml2.drivers import type_flat
from neutron.tests import base
@ -118,7 +117,7 @@ class FlatTypeTest(testlib_api.SqlTestCase):
segment = {api.NETWORK_TYPE: p_const.TYPE_FLAT,
api.PHYSICAL_NETWORK: 'flat_net1'}
self.driver.reserve_provider_segment(self.context, segment)
self.assertRaises(n_exc.FlatNetworkInUse,
self.assertRaises(exc.FlatNetworkInUse,
self.driver.reserve_provider_segment,
self.context, segment)

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

@ -18,13 +18,13 @@ from neutron_lib.api.definitions import portbindings
from neutron_lib.api.definitions import portbindings_extended as pbe_ext
from neutron_lib import constants as const
from neutron_lib import context
from neutron_lib import exceptions
from neutron_lib.plugins import directory
from neutron_lib.plugins import utils
from oslo_config import cfg
from oslo_serialization import jsonutils
import webob.exc
from neutron.common import exceptions
from neutron.conf.plugins.ml2 import config
from neutron.conf.plugins.ml2.drivers import driver_type
from neutron.plugins.ml2 import driver_context

2
neutron/tests/unit/privileged/agent/linux/test_netlink_lib.py
View File

@ -16,9 +16,9 @@
import mock
from neutron_lib import constants
from neutron_lib import exceptions
import testtools
from neutron.common import exceptions
from neutron.privileged.agent.linux import netlink_constants as nl_constants
from neutron.privileged.agent.linux import netlink_lib as nl_lib
from neutron.tests import base

3
neutron/tests/unit/services/auto_allocate/test_db.py
View File

@ -21,7 +21,6 @@ from neutron_lib import exceptions as n_exc
from oslo_db import exception as db_exc
from oslo_utils import uuidutils
from neutron.common import exceptions as c_exc
from neutron.services.auto_allocate import db
from neutron.services.auto_allocate import exceptions
from neutron.tests.unit import testlib_api
@ -216,7 +215,7 @@ class AutoAllocateTestCase(testlib_api.SqlTestCase):
self.mixin._core_plugin.create_network.return_value = (
{'id': network_id})
self.mixin._core_plugin.create_subnet.side_effect = (
c_exc.SubnetAllocationError(reason='disaster'))
n_exc.SubnetAllocationError(reason='disaster'))
with mock.patch.object(self.mixin, "_get_supported_subnetpools") as f,\
mock.patch.object(self.mixin, "_cleanup") as g:
f.return_value = (

2
neutron/tests/unit/services/logapi/drivers/test_manager.py
View File

@ -15,9 +15,9 @@
import mock
from neutron_lib.callbacks import events
from neutron_lib import exceptions
from neutron_lib import fixture
from neutron.common import exceptions
from neutron.services.logapi.common import constants as log_const
from neutron.services.logapi.common import exceptions as log_exc
from neutron.services.logapi.drivers import base as log_driver_base

2
neutron/tests/unit/services/qos/drivers/test_manager.py
View File

@ -14,12 +14,12 @@ import mock
from neutron_lib.api.definitions import portbindings
from neutron_lib import constants as lib_consts
from neutron_lib import context
from neutron_lib import exceptions
from neutron_lib.services.qos import base as qos_driver_base
from neutron_lib.services.qos import constants as qos_consts
from oslo_utils import uuidutils
from neutron.common import constants
from neutron.common import exceptions
from neutron.objects import ports as ports_object
from neutron.objects.qos import rule as rule_object
from neutron.services.qos.drivers import manager as driver_mgr

40
neutron/tests/unit/services/qos/test_qos_plugin.py
View File

@ -17,6 +17,7 @@ from neutron_lib.callbacks import events
from neutron_lib import constants as lib_constants
from neutron_lib import context
from neutron_lib import exceptions as lib_exc
from neutron_lib.exceptions import qos as qos_exc
from neutron_lib.objects import utils as obj_utils
from neutron_lib.placement import constants as pl_constants
from neutron_lib.plugins import constants as plugins_constants
@ -26,7 +27,6 @@ from oslo_config import cfg
from oslo_utils import uuidutils
from neutron.common import constants
from neutron.common import exceptions as n_exc
from neutron import manager
from neutron.objects.qos import policy as policy_object
from neutron.objects.qos import rule as rule_object
@ -421,7 +421,7 @@ class TestQosPlugin(base.BaseQosTestCase):
):
self.policy.rules = [self.rule]
self.assertRaises(
n_exc.QosRuleNotSupported,
qos_exc.QosRuleNotSupported,
self.qos_plugin.validate_policy_for_port,
self.policy, port)
@ -434,7 +434,7 @@ class TestQosPlugin(base.BaseQosTestCase):
self.policy.rules = [self.rule]
try:
self.qos_plugin.validate_policy_for_port(self.policy, port)
except n_exc.QosRuleNotSupported:
except qos_exc.QosRuleNotSupported:
self.fail("QosRuleNotSupported exception unexpectedly raised")
@mock.patch(
@ -587,7 +587,7 @@ class TestQosPlugin(base.BaseQosTestCase):
setattr(_policy, "rules", [self.min_rule])
with mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',
return_value=_policy) as mock_qos_get_obj:
self.assertRaises(n_exc.QoSRuleParameterConflict,
self.assertRaises(qos_exc.QoSRuleParameterConflict,
self.qos_plugin.create_policy_bandwidth_limit_rule,
self.ctxt, self.policy.id, self.rule_data)
mock_qos_get_obj.assert_called_once_with(self.ctxt, id=_policy.id)
@ -598,7 +598,7 @@ class TestQosPlugin(base.BaseQosTestCase):
setattr(_policy, "rules", [self.rule])
with mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',
return_value=_policy) as mock_qos_get_obj:
self.assertRaises(n_exc.QoSRuleParameterConflict,
self.assertRaises(qos_exc.QoSRuleParameterConflict,
self.qos_plugin.create_policy_minimum_bandwidth_rule,
self.ctxt, self.policy.id, self.rule_data)
mock_qos_get_obj.assert_called_once_with(self.ctxt, id=_policy.id)
@ -615,7 +615,7 @@ class TestQosPlugin(base.BaseQosTestCase):
with mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',
return_value=_policy) as mock_qos_get_obj:
self.assertRaises(
n_exc.QoSRulesConflict,
qos_exc.QoSRulesConflict,
self.qos_plugin.create_policy_bandwidth_limit_rule,
self.ctxt, _policy.id, new_rule_data)
mock_qos_get_obj.assert_called_once_with(self.ctxt, id=_policy.id)
@ -683,7 +683,7 @@ class TestQosPlugin(base.BaseQosTestCase):
with mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',
return_value=_policy):
self.assertRaises(
n_exc.QoSRuleParameterConflict,
qos_exc.QoSRuleParameterConflict,
self.qos_plugin.update_policy_minimum_bandwidth_rule,
self.ctxt, self.min_rule.id,
self.policy.id, self.rule_data)
@ -701,7 +701,7 @@ class TestQosPlugin(base.BaseQosTestCase):
with mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',
return_value=_policy):
self.assertRaises(
n_exc.QoSRuleParameterConflict,
qos_exc.QoSRuleParameterConflict,
self.qos_plugin.update_policy_bandwidth_limit_rule,
self.ctxt, self.rule.id,
self.policy.id, self.rule_data)
@ -717,7 +717,7 @@ class TestQosPlugin(base.BaseQosTestCase):
return_value=_policy):
setattr(_policy, "rules", [])
self.assertRaises(
n_exc.QosRuleNotFound,
qos_exc.QosRuleNotFound,
self.qos_plugin.update_policy_bandwidth_limit_rule,
self.ctxt, self.rule.id, self.policy.id,
self.rule_data)
@ -755,7 +755,7 @@ class TestQosPlugin(base.BaseQosTestCase):
return_value=_policy):
setattr(_policy, "rules", [])
self.assertRaises(
n_exc.QosRuleNotFound,
qos_exc.QosRuleNotFound,
self.qos_plugin.delete_policy_bandwidth_limit_rule,
self.ctxt, self.rule.id, _policy.id)
@ -800,7 +800,7 @@ class TestQosPlugin(base.BaseQosTestCase):
with mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',
return_value=None):
self.assertRaises(
n_exc.QosPolicyNotFound,
qos_exc.QosPolicyNotFound,
self.qos_plugin.get_policy,
self.ctxt, self.policy.id)
@ -808,7 +808,7 @@ class TestQosPlugin(base.BaseQosTestCase):
with mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',
return_value=None):
self.assertRaises(
n_exc.QosPolicyNotFound,
qos_exc.QosPolicyNotFound,
self.qos_plugin.get_policy_bandwidth_limit_rule,
self.ctxt, self.rule.id, self.policy.id)
@ -816,7 +816,7 @@ class TestQosPlugin(base.BaseQosTestCase):
with mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',
return_value=None):
self.assertRaises(
n_exc.QosPolicyNotFound,
qos_exc.QosPolicyNotFound,
self.qos_plugin.get_policy_bandwidth_limit_rules,
self.ctxt, self.policy.id)
@ -879,7 +879,7 @@ class TestQosPlugin(base.BaseQosTestCase):
with mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',
return_value=None):
self.assertRaises(
n_exc.QosPolicyNotFound,
qos_exc.QosPolicyNotFound,
self.qos_plugin.get_policy_dscp_marking_rule,
self.ctxt, self.dscp_rule.id, self.policy.id)
@ -887,7 +887,7 @@ class TestQosPlugin(base.BaseQosTestCase):
with mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',
return_value=None):
self.assertRaises(
n_exc.QosPolicyNotFound,
qos_exc.QosPolicyNotFound,
self.qos_plugin.get_policy_dscp_marking_rules,
self.ctxt, self.policy.id)
@ -932,7 +932,7 @@ class TestQosPlugin(base.BaseQosTestCase):
with mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',
return_value=None):
self.assertRaises(
n_exc.QosPolicyNotFound,
qos_exc.QosPolicyNotFound,
self.qos_plugin.get_policy_minimum_bandwidth_rule,
self.ctxt, self.rule.id, self.policy.id)
@ -940,7 +940,7 @@ class TestQosPlugin(base.BaseQosTestCase):
with mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',
return_value=None):
self.assertRaises(
n_exc.QosPolicyNotFound,
qos_exc.QosPolicyNotFound,
self.qos_plugin.get_policy_minimum_bandwidth_rules,
self.ctxt, self.policy.id)
@ -948,7 +948,7 @@ class TestQosPlugin(base.BaseQosTestCase):
with mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',
return_value=None):
self.assertRaises(
n_exc.QosPolicyNotFound,
qos_exc.QosPolicyNotFound,
self.qos_plugin.create_policy_bandwidth_limit_rule,
self.ctxt, self.policy.id, self.rule_data)
@ -956,7 +956,7 @@ class TestQosPlugin(base.BaseQosTestCase):
with mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',
return_value=None):
self.assertRaises(
n_exc.QosPolicyNotFound,
qos_exc.QosPolicyNotFound,
self.qos_plugin.update_policy_bandwidth_limit_rule,
self.ctxt, self.rule.id, self.policy.id, self.rule_data)
@ -964,7 +964,7 @@ class TestQosPlugin(base.BaseQosTestCase):
with mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',
return_value=None):
self.assertRaises(
n_exc.QosPolicyNotFound,
qos_exc.QosPolicyNotFound,
self.qos_plugin.delete_policy_bandwidth_limit_rule,
self.ctxt, self.rule.id, self.policy.id)

4
neutron/tests/unit/test_wsgi.py
View File

@ -26,7 +26,6 @@ import testtools
import webob
import webob.exc
from neutron.common import exceptions as n_exc
from neutron.common import ipv6_utils
from neutron.tests import base
from neutron.tests.common import helpers
@ -581,7 +580,8 @@ class JSONDeserializerTest(base.BaseTestCase):
deserializer = wsgi.JSONDeserializer()
self.assertRaises(
n_exc.MalformedRequestBody, deserializer.default, data_string)
exception.MalformedRequestBody,
deserializer.default, data_string)
def test_json_with_utf8(self):
data = b'{"a": "\xe7\xbd\x91\xe7\xbb\x9c"}'

5
neutron/wsgi.py
View File

@ -42,7 +42,6 @@ import webob.exc
from neutron._i18n import _
from neutron.common import config
from neutron.common import exceptions as n_exc
from neutron.conf import wsgi as wsgi_config
CONF = cfg.CONF
@ -385,7 +384,7 @@ class JSONDeserializer(TextDeserializer):
return jsonutils.loads(datastring)
except ValueError:
msg = _("Cannot understand JSON")
raise n_exc.MalformedRequestBody(reason=msg)
raise exception.MalformedRequestBody(reason=msg)
def default(self, datastring):
return {'body': self._from_json(datastring)}
@ -598,7 +597,7 @@ class Resource(Application):
msg = _("Unsupported Content-Type")
LOG.exception("InvalidContentType: %s", msg)
return Fault(webob.exc.HTTPBadRequest(explanation=msg))
except n_exc.MalformedRequestBody:
except exception.MalformedRequestBody:
msg = _("Malformed request body")
LOG.exception("MalformedRequestBody: %s", msg)
return Fault(webob.exc.HTTPBadRequest(explanation=msg))