openstack
/
nova
Code Issues Proposed changes

Merge "Stop neutron.api relying on base neutron package"

This commit is contained in:
Jenkins 2014-12-17 03:40:18 +00:00 committed by Gerrit Code Review
parent e956a257cd 4bab641e9a
commit 8c3271cbfc
7 changed files with 198 additions and 224 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

116
nova/network/neutronv2/__init__.py
View File

@ -1,116 +0,0 @@
# Copyright 2012 OpenStack Foundation
# 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 neutronclient.common import exceptions
from neutronclient.v2_0 import client as clientv20
from oslo.config import cfg
from oslo_concurrency import lockutils
from nova.openstack.common import log as logging
CONF = cfg.CONF
LOG = logging.getLogger(__name__)
class AdminTokenStore(object):
_instance = None
def __init__(self):
self.admin_auth_token = None
@classmethod
def get(cls):
if cls._instance is None:
cls._instance = cls()
return cls._instance
def _get_client(token=None, admin=False):
params = {
'endpoint_url': CONF.neutron.url,
'timeout': CONF.neutron.url_timeout,
'insecure': CONF.neutron.api_insecure,
'ca_cert': CONF.neutron.ca_certificates_file,
'auth_strategy': CONF.neutron.auth_strategy,
'token': token,
}
if admin:
if CONF.neutron.admin_user_id:
params['user_id'] = CONF.neutron.admin_user_id
else:
params['username'] = CONF.neutron.admin_username
if CONF.neutron.admin_tenant_id:
params['tenant_id'] = CONF.neutron.admin_tenant_id
else:
params['tenant_name'] = CONF.neutron.admin_tenant_name
params['password'] = CONF.neutron.admin_password
params['auth_url'] = CONF.neutron.admin_auth_url
return clientv20.Client(**params)
class ClientWrapper(clientv20.Client):
'''A neutron client wrapper class.
Wraps the callable methods, executes it and updates the token,
as it might change when expires.
'''
def __init__(self, base_client):
# Expose all attributes from the base_client instance
self.__dict__ = base_client.__dict__
self.base_client = base_client
def __getattribute__(self, name):
obj = object.__getattribute__(self, name)
if callable(obj):
obj = object.__getattribute__(self, 'proxy')(obj)
return obj
def proxy(self, obj):
def wrapper(*args, **kwargs):
ret = obj(*args, **kwargs)
new_token = self.base_client.get_auth_info()['auth_token']
_update_token(new_token)
return ret
return wrapper
def _update_token(new_token):
with lockutils.lock('neutron_admin_auth_token_lock'):
token_store = AdminTokenStore.get()
token_store.admin_auth_token = new_token
def get_client(context, admin=False):
# NOTE(dprince): In the case where no auth_token is present
# we allow use of neutron admin tenant credentials if
# it is an admin context.
# This is to support some services (metadata API) where
# an admin context is used without an auth token.
if admin or (context.is_admin and not context.auth_token):
with lockutils.lock('neutron_admin_auth_token_lock'):
orig_token = AdminTokenStore.get().admin_auth_token
client = _get_client(orig_token, admin=True)
return ClientWrapper(client)
# We got a user token that we can use that as-is
if context.auth_token:
token = context.auth_token
return _get_client(token=token)
# We did not get a user token and we should not be using
# an admin token so log an error
raise exceptions.Unauthorized()

160
nova/network/neutronv2/api.py
View File

@ -19,6 +19,7 @@ import time
import uuid
from neutronclient.common import exceptions as neutron_client_exc
from neutronclient.v2_0 import client as clientv20
from oslo.config import cfg
from oslo.utils import excutils
from oslo_concurrency import lockutils
@ -30,7 +31,6 @@ from nova import exception
from nova.i18n import _, _LE, _LI, _LW
from nova.network import base_api
from nova.network import model as network_model
from nova.network import neutronv2
from nova.network.neutronv2 import constants
from nova import objects
from nova.openstack.common import log as logging
@ -101,6 +101,98 @@ soft_external_network_attach_authorize = extensions.soft_core_authorizer(
'network', 'attach_external_network')
class AdminTokenStore(object):
_instance = None
def __init__(self):
self.admin_auth_token = None
@classmethod
def get(cls):
if cls._instance is None:
cls._instance = cls()
return cls._instance
def _get_client(token=None, admin=False):
params = {
'endpoint_url': CONF.neutron.url,
'timeout': CONF.neutron.url_timeout,
'insecure': CONF.neutron.api_insecure,
'ca_cert': CONF.neutron.ca_certificates_file,
'auth_strategy': CONF.neutron.auth_strategy,
'token': token,
}
if admin:
if CONF.neutron.admin_user_id:
params['user_id'] = CONF.neutron.admin_user_id
else:
params['username'] = CONF.neutron.admin_username
if CONF.neutron.admin_tenant_id:
params['tenant_id'] = CONF.neutron.admin_tenant_id
else:
params['tenant_name'] = CONF.neutron.admin_tenant_name
params['password'] = CONF.neutron.admin_password
params['auth_url'] = CONF.neutron.admin_auth_url
return clientv20.Client(**params)
class ClientWrapper(clientv20.Client):
'''A neutron client wrapper class.
Wraps the callable methods, executes it and updates the token,
as it might change when expires.
'''
def __init__(self, base_client):
# Expose all attributes from the base_client instance
self.__dict__ = base_client.__dict__
self.base_client = base_client
def __getattribute__(self, name):
obj = object.__getattribute__(self, name)
if callable(obj):
obj = object.__getattribute__(self, 'proxy')(obj)
return obj
def proxy(self, obj):
def wrapper(*args, **kwargs):
ret = obj(*args, **kwargs)
new_token = self.base_client.get_auth_info()['auth_token']
_update_token(new_token)
return ret
return wrapper
def _update_token(new_token):
with lockutils.lock('neutron_admin_auth_token_lock'):
token_store = AdminTokenStore.get()
token_store.admin_auth_token = new_token
def get_client(context, admin=False):
# NOTE(dprince): In the case where no auth_token is present
# we allow use of neutron admin tenant credentials if
# it is an admin context.
# This is to support some services (metadata API) where
# an admin context is used without an auth token.
if admin or (context.is_admin and not context.auth_token):
with lockutils.lock('neutron_admin_auth_token_lock'):
orig_token = AdminTokenStore.get().admin_auth_token
client = _get_client(orig_token, admin=True)
return ClientWrapper(client)
# We got a user token that we can use that as-is
if context.auth_token:
token = context.auth_token
return _get_client(token=token)
# We did not get a user token and we should not be using
# an admin token so log an error
raise neutron_client_exc.Unauthorized()
class API(base_api.NetworkAPI):
"""API for interacting with the neutron 2.x API."""
@ -120,7 +212,7 @@ class API(base_api.NetworkAPI):
If net_ids specified, it searches networks with requested IDs only.
"""
if not neutron:
neutron = neutronv2.get_client(context)
neutron = get_client(context)
if net_ids:
# If user has specified to attach instance only to specific
@ -254,12 +346,12 @@ class API(base_api.NetworkAPI):
# a number of different calls for the instance allocation.
# We do not want to create a new neutron session for each of these
# calls.
neutron = neutronv2.get_client(context)
neutron = get_client(context)
# Requires admin creds to set port bindings
port_client = (neutron if not
self._has_port_binding_extension(context,
refresh_cache=True, neutron=neutron) else
neutronv2.get_client(context, admin=True))
get_client(context, admin=True))
# Store the admin client - this is used later
admin_client = port_client if neutron != port_client else None
LOG.debug('allocate_for_instance()', instance=instance)
@ -441,7 +533,7 @@ class API(base_api.NetworkAPI):
((time.time() - self.last_neutron_extension_sync)
>= CONF.neutron.extension_sync_interval)):
if neutron is None:
neutron = neutronv2.get_client(context)
neutron = get_client(context)
extensions_list = neutron.list_extensions()['extensions']
self.last_neutron_extension_sync = time.time()
self.extensions.clear()
@ -496,7 +588,7 @@ class API(base_api.NetworkAPI):
for port in ports:
try:
neutron.delete_port(port)
except neutronv2.exceptions.NeutronClientException as e:
except neutron_client_exc.NeutronClientException as e:
if e.status_code == 404:
LOG.warning(_LW("Port %s does not exist"), port)
else:
@ -511,7 +603,7 @@ class API(base_api.NetworkAPI):
"""Deallocate all network resources related to the instance."""
LOG.debug('deallocate_for_instance()', instance=instance)
search_opts = {'device_id': instance.uuid}
neutron = neutronv2.get_client(context)
neutron = get_client(context)
data = neutron.list_ports(**search_opts)
ports = [port['id'] for port in data.get('ports', [])]
@ -526,7 +618,7 @@ class API(base_api.NetworkAPI):
for port in ports_to_skip:
port_req_body = {'port': {'device_id': '', 'device_owner': ''}}
try:
neutronv2.get_client(context).update_port(port,
get_client(context).update_port(port,
port_req_body)
except Exception:
LOG.info(_LI('Unable to reset device ID for port %s'), port,
@ -557,18 +649,18 @@ class API(base_api.NetworkAPI):
Return network information for the instance
"""
neutron = neutronv2.get_client(context)
neutron = get_client(context)
self._delete_ports(neutron, instance, [port_id], raise_if_fail=True)
return self.get_instance_nw_info(context, instance)
def list_ports(self, context, **search_opts):
"""List ports for the client based on search options."""
return neutronv2.get_client(context).list_ports(**search_opts)
return get_client(context).list_ports(**search_opts)
def show_port(self, context, port_id):
"""Return the port for the client given the port id."""
try:
return neutronv2.get_client(context).show_port(port_id)
return get_client(context).show_port(port_id)
except neutron_client_exc.PortNotFoundClient:
raise exception.PortNotFound(port_id=port_id)
except neutron_client_exc.Unauthorized:
@ -642,7 +734,7 @@ class API(base_api.NetworkAPI):
def add_fixed_ip_to_instance(self, context, instance, network_id):
"""Add a fixed ip to the instance from specified network."""
search_opts = {'network_id': network_id}
data = neutronv2.get_client(context).list_subnets(**search_opts)
data = get_client(context).list_subnets(**search_opts)
ipam_subnets = data.get('subnets', [])
if not ipam_subnets:
raise exception.NetworkNotFoundForInstance(
@ -652,7 +744,7 @@ class API(base_api.NetworkAPI):
search_opts = {'device_id': instance['uuid'],
'device_owner': zone,
'network_id': network_id}
data = neutronv2.get_client(context).list_ports(**search_opts)
data = get_client(context).list_ports(**search_opts)
ports = data['ports']
for p in ports:
for subnet in ipam_subnets:
@ -660,7 +752,7 @@ class API(base_api.NetworkAPI):
fixed_ips.append({'subnet_id': subnet['id']})
port_req_body = {'port': {'fixed_ips': fixed_ips}}
try:
neutronv2.get_client(context).update_port(p['id'],
get_client(context).update_port(p['id'],
port_req_body)
return self._get_instance_nw_info(context, instance)
except Exception as ex:
@ -680,7 +772,7 @@ class API(base_api.NetworkAPI):
search_opts = {'device_id': instance['uuid'],
'device_owner': zone,
'fixed_ips': 'ip_address=%s' % address}
data = neutronv2.get_client(context).list_ports(**search_opts)
data = get_client(context).list_ports(**search_opts)
ports = data['ports']
for p in ports:
fixed_ips = p['fixed_ips']
@ -690,7 +782,7 @@ class API(base_api.NetworkAPI):
new_fixed_ips.append(fixed_ip)
port_req_body = {'port': {'fixed_ips': new_fixed_ips}}
try:
neutronv2.get_client(context).update_port(p['id'],
get_client(context).update_port(p['id'],
port_req_body)
except Exception as ex:
msg = ("Unable to update port %(portid)s with"
@ -730,7 +822,7 @@ class API(base_api.NetworkAPI):
if not requested_networks:
return
neutron = neutronv2.get_client(context, admin=True)
neutron = get_client(context, admin=True)
for request_net in requested_networks:
phynet_name = None
vnic_type = network_model.VNIC_TYPE_NORMAL
@ -759,7 +851,7 @@ class API(base_api.NetworkAPI):
LOG.debug('validate_networks() for %s',
requested_networks)
neutron = neutronv2.get_client(context)
neutron = get_client(context)
ports_needed_per_instance = 0
if requested_networks is None or len(requested_networks) == 0:
@ -884,7 +976,7 @@ class API(base_api.NetworkAPI):
e.g. [{'instance_uuid': uuid}, ...]
"""
search_opts = {"fixed_ips": 'ip_address=%s' % address}
data = neutronv2.get_client(context).list_ports(**search_opts)
data = get_client(context).list_ports(**search_opts)
ports = data.get('ports', [])
return [{'instance_uuid': port['device_id']} for port in ports
if port['device_id']]
@ -931,7 +1023,7 @@ class API(base_api.NetworkAPI):
# since it is not used anywhere in nova code and I could
# find why this parameter exists.
client = neutronv2.get_client(context)
client = get_client(context)
port_id = self._get_port_id_by_fixed_address(client, instance,
fixed_address)
fip = self._get_floating_ip_by_address(client, floating_address)
@ -956,7 +1048,7 @@ class API(base_api.NetworkAPI):
def get_all(self, context):
"""Get all networks for client."""
client = neutronv2.get_client(context)
client = get_client(context)
networks = client.list_networks().get('networks')
for network in networks:
network['label'] = network['name']
@ -964,7 +1056,7 @@ class API(base_api.NetworkAPI):
def get(self, context, network_uuid):
"""Get specific network for client."""
client = neutronv2.get_client(context)
client = get_client(context)
try:
network = client.show_network(network_uuid).get('network') or {}
except neutron_client_exc.NetworkNotFoundClient:
@ -1023,7 +1115,7 @@ class API(base_api.NetworkAPI):
def get_floating_ip(self, context, id):
"""Return floating ip object given the floating ip id."""
client = neutronv2.get_client(context)
client = get_client(context)
try:
fip = client.show_floatingip(id)['floatingip']
except neutron_client_exc.NeutronClientException as e:
@ -1046,7 +1138,7 @@ class API(base_api.NetworkAPI):
def get_floating_ip_pools(self, context):
"""Return floating ip pool names."""
client = neutronv2.get_client(context)
client = get_client(context)
pools = self._get_floating_ip_pools(client)
# Note(salv-orlando): Return a list of names to be consistent with
# nova.network.api.get_floating_ip_pools
@ -1073,7 +1165,7 @@ class API(base_api.NetworkAPI):
def get_floating_ip_by_address(self, context, address):
"""Return a floating ip given an address."""
client = neutronv2.get_client(context)
client = get_client(context)
fip = self._get_floating_ip_by_address(client, address)
pool_dict = self._setup_net_dict(client,
fip['floating_network_id'])
@ -1081,7 +1173,7 @@ class API(base_api.NetworkAPI):
return self._format_floating_ip_model(fip, pool_dict, port_dict)
def get_floating_ips_by_project(self, context):
client = neutronv2.get_client(context)
client = get_client(context)
project_id = context.project_id
fips = client.list_floatingips(tenant_id=project_id)['floatingips']
pool_dict = self._setup_pools_dict(client)
@ -1091,7 +1183,7 @@ class API(base_api.NetworkAPI):
def get_instance_id_by_floating_address(self, context, address):
"""Return the instance id a floating ip's fixed ip is allocated to."""
client = neutronv2.get_client(context)
client = get_client(context)
fip = self._get_floating_ip_by_address(client, address)
if not fip['port_id']:
return None
@ -1124,7 +1216,7 @@ class API(base_api.NetworkAPI):
def allocate_floating_ip(self, context, pool=None):
"""Add a floating ip to a project from a pool."""
client = neutronv2.get_client(context)
client = get_client(context)
pool = pool or CONF.default_floating_pool
pool_id = self._get_floating_ip_pool_id_by_name_or_id(client, pool)
@ -1194,7 +1286,7 @@ class API(base_api.NetworkAPI):
def _release_floating_ip(self, context, address,
raise_if_associated=True):
client = neutronv2.get_client(context)
client = get_client(context)
fip = self._get_floating_ip_by_address(client, address)
if raise_if_associated and fip['port_id']:
@ -1210,7 +1302,7 @@ class API(base_api.NetworkAPI):
# since it is not used anywhere in nova code and I could
# find why this parameter exists.
client = neutronv2.get_client(context)
client = get_client(context)
fip = self._get_floating_ip_by_address(client, address)
client.update_floatingip(fip['id'], {'floatingip': {'port_id': None}})
@ -1224,7 +1316,7 @@ class API(base_api.NetworkAPI):
"""Finish migrating the network of an instance."""
if not self._has_port_binding_extension(context, refresh_cache=True):
return
neutron = neutronv2.get_client(context, admin=True)
neutron = get_client(context, admin=True)
search_opts = {'device_id': instance['uuid'],
'tenant_id': instance['project_id']}
data = neutron.list_ports(**search_opts)
@ -1338,7 +1430,7 @@ class API(base_api.NetworkAPI):
search_opts = {'tenant_id': instance['project_id'],
'device_id': instance['uuid'], }
if admin_client is None:
client = neutronv2.get_client(context, admin=True)
client = get_client(context, admin=True)
else:
client = admin_client
@ -1402,7 +1494,7 @@ class API(base_api.NetworkAPI):
if not fixed_ips:
return []
search_opts = {'id': [ip['subnet_id'] for ip in fixed_ips]}
data = neutronv2.get_client(context).list_subnets(**search_opts)
data = get_client(context).list_subnets(**search_opts)
ipam_subnets = data.get('subnets', [])
subnets = []
@ -1416,7 +1508,7 @@ class API(base_api.NetworkAPI):
# attempt to populate DHCP server field
search_opts = {'network_id': subnet['network_id'],
'device_owner': 'network:dhcp'}
data = neutronv2.get_client(context).list_ports(**search_opts)
data = get_client(context).list_ports(**search_opts)
dhcp_ports = data.get('ports', [])
for p in dhcp_ports:
for ip_pair in p['fixed_ips']:

24
nova/network/security_group/neutron_driver.py
View File

@ -25,7 +25,7 @@ from webob import exc
from nova.compute import api as compute_api
from nova import exception
from nova.i18n import _, _LE, _LI, _LW
from nova.network import neutronv2
from nova.network.neutronv2 import api as neutronapi
from nova.network.security_group import security_group_base
from nova import objects
from nova.openstack.common import log as logging
@ -47,7 +47,7 @@ class SecurityGroupAPI(security_group_base.SecurityGroupBase):
id_is_uuid = True
def create_security_group(self, context, name, description):
neutron = neutronv2.get_client(context)
neutron = neutronapi.get_client(context)
body = self._make_neutron_security_group_dict(name, description)
try:
security_group = neutron.create_security_group(
@ -68,7 +68,7 @@ class SecurityGroupAPI(security_group_base.SecurityGroupBase):
def update_security_group(self, context, security_group,
name, description):
neutron = neutronv2.get_client(context)
neutron = neutronapi.get_client(context)
body = self._make_neutron_security_group_dict(name, description)
try:
security_group = neutron.update_security_group(
@ -120,7 +120,7 @@ class SecurityGroupAPI(security_group_base.SecurityGroupBase):
return nova_rule
def get(self, context, name=None, id=None, map_exception=False):
neutron = neutronv2.get_client(context)
neutron = neutronapi.get_client(context)
try:
if not id and name:
# NOTE(flwang): The project id should be honoured so as to get
@ -146,7 +146,7 @@ class SecurityGroupAPI(security_group_base.SecurityGroupBase):
def list(self, context, names=None, ids=None, project=None,
search_opts=None):
"""Returns list of security group rules owned by tenant."""
neutron = neutronv2.get_client(context)
neutron = neutronapi.get_client(context)
search_opts = {}
if names:
search_opts['name'] = names
@ -175,7 +175,7 @@ class SecurityGroupAPI(security_group_base.SecurityGroupBase):
def destroy(self, context, security_group):
"""This function deletes a security group."""
neutron = neutronv2.get_client(context)
neutron = neutronapi.get_client(context)
try:
neutron.delete_security_group(security_group['id'])
except n_exc.NeutronClientException as e:
@ -197,7 +197,7 @@ class SecurityGroupAPI(security_group_base.SecurityGroupBase):
installed to a security group in neutron using bulk support.
"""
neutron = neutronv2.get_client(context)
neutron = neutronapi.get_client(context)
body = self._make_neutron_security_group_rules_list(vals)
try:
rules = neutron.create_security_group_rule(
@ -256,7 +256,7 @@ class SecurityGroupAPI(security_group_base.SecurityGroupBase):
return {'security_group_rules': new_rules}
def remove_rules(self, context, security_group, rule_ids):
neutron = neutronv2.get_client(context)
neutron = neutronapi.get_client(context)
rule_ids = set(rule_ids)
try:
# The ec2 api allows one to delete multiple security group rules
@ -271,7 +271,7 @@ class SecurityGroupAPI(security_group_base.SecurityGroupBase):
rule_ids)
def get_rule(self, context, id):
neutron = neutronv2.get_client(context)
neutron = neutronapi.get_client(context)
try:
rule = neutron.show_security_group_rule(
id).get('security_group_rule')
@ -342,7 +342,7 @@ class SecurityGroupAPI(security_group_base.SecurityGroupBase):
all of the instances and their security groups in one shot.
"""
neutron = neutronv2.get_client(context)
neutron = neutronapi.get_client(context)
ports = self._get_ports_from_server_list(servers, neutron)
@ -395,7 +395,7 @@ class SecurityGroupAPI(security_group_base.SecurityGroupBase):
def add_to_instance(self, context, instance, security_group_name):
"""Add security group to the instance."""
neutron = neutronv2.get_client(context)
neutron = neutronapi.get_client(context)
try:
security_group_id = neutronv20.find_resourceid_by_name_or_id(
neutron, 'security_group',
@ -452,7 +452,7 @@ class SecurityGroupAPI(security_group_base.SecurityGroupBase):
@compute_api.wrap_check_security_groups_policy
def remove_from_instance(self, context, instance, security_group_name):
"""Remove the security group associated with the instance."""
neutron = neutronv2.get_client(context)
neutron = neutronapi.get_client(context)
try:
security_group_id = neutronv20.find_resourceid_by_name_or_id(
neutron, 'security_group',

8
nova/tests/unit/api/ec2/test_cloud.py
View File

@ -47,7 +47,7 @@ from nova.image import s3
from nova.network import api as network_api
from nova.network import base_api as base_network_api
from nova.network import model
from nova.network import neutronv2
from nova.network.neutronv2 import api as neutronapi
from nova import objects
from nova.objects import base as obj_base
from nova.openstack.common import log as logging
@ -3150,8 +3150,8 @@ class CloudTestCaseNeutronProxy(test.NoDBTestCase):
super(CloudTestCaseNeutronProxy, self).setUp()
cfg.CONF.set_override('security_group_api', 'neutron')
self.cloud = cloud.CloudController()
self.original_client = neutronv2.get_client
neutronv2.get_client = test_neutron.get_client
self.original_client = neutronapi.get_client
neutronapi.get_client = test_neutron.get_client
self.user_id = 'fake'
self.project_id = 'fake'
self.context = context.RequestContext(self.user_id,
@ -3159,7 +3159,7 @@ class CloudTestCaseNeutronProxy(test.NoDBTestCase):
is_admin=True)
def tearDown(self):
neutronv2.get_client = self.original_client
neutronapi.get_client = self.original_client
test_neutron.get_client()._reset()
super(CloudTestCaseNeutronProxy, self).tearDown()

9
nova/tests/unit/api/openstack/compute/contrib/test_neutron_security_groups.py
View File

@ -29,7 +29,6 @@ from nova import context
import nova.db
from nova import exception
from nova.network import model
from nova.network import neutronv2
from nova.network.neutronv2 import api as neutron_api
from nova.network.security_group import neutron_driver
from nova.objects import instance as instance_obj
@ -42,11 +41,11 @@ class TestNeutronSecurityGroupsTestCase(test.TestCase):
def setUp(self):
super(TestNeutronSecurityGroupsTestCase, self).setUp()
cfg.CONF.set_override('security_group_api', 'neutron')
self.original_client = neutronv2.get_client
neutronv2.get_client = get_client
self.original_client = neutron_api.get_client
neutron_api.get_client = get_client
def tearDown(self):
neutronv2.get_client = self.original_client
neutron_api.get_client = self.original_client
get_client()._reset()
super(TestNeutronSecurityGroupsTestCase, self).tearDown()
@ -421,7 +420,7 @@ class TestNeutronSecurityGroupRulesTestCase(TestNeutronSecurityGroupsTestCase):
neutron._fake_security_groups[id2] = sg_template2
def tearDown(self):
neutronv2.get_client = self.original_client
neutron_api.get_client = self.original_client
get_client()._reset()
super(TestNeutronSecurityGroupsTestCase, self).tearDown()

6
nova/tests/unit/network/security_group/test_neutron_driver.py
View File

@ -19,7 +19,7 @@ from neutronclient.v2_0 import client
from nova import context
from nova import exception
from nova.network import neutronv2
from nova.network.neutronv2 import api as neutronapi
from nova.network.security_group import neutron_driver
from nova import test
@ -27,9 +27,9 @@ from nova import test
class TestNeutronDriver(test.NoDBTestCase):
def setUp(self):
super(TestNeutronDriver, self).setUp()
self.mox.StubOutWithMock(neutronv2, 'get_client')
self.mox.StubOutWithMock(neutronapi, 'get_client')
self.moxed_client = self.mox.CreateMock(client.Client)
neutronv2.get_client(mox.IgnoreArg()).MultipleTimes().AndReturn(
neutronapi.get_client(mox.IgnoreArg()).MultipleTimes().AndReturn(
self.moxed_client)
self.context = context.RequestContext('userid', 'my_tenantid')
setattr(self.context,

99
nova/tests/unit/network/test_neutronv2.py
View File

@ -31,7 +31,6 @@ from nova.compute import flavors
from nova import context
from nova import exception
from nova.network import model
from nova.network import neutronv2
from nova.network.neutronv2 import api as neutronapi
from nova.network.neutronv2 import constants
from nova import objects
@ -114,12 +113,12 @@ class TestNeutronClient(test.TestCase):
insecure=False,
ca_cert=None).AndReturn(None)
self.mox.ReplayAll()
neutronv2.get_client(my_context)
neutronapi.get_client(my_context)
def test_withouttoken(self):
my_context = context.RequestContext('userid', 'my_tenantid')
self.assertRaises(exceptions.Unauthorized,
neutronv2.get_client,
neutronapi.get_client,
my_context)
def test_withtoken_context_is_admin(self):
@ -141,20 +140,20 @@ class TestNeutronClient(test.TestCase):
# Note that although we have admin set in the context we
# are not asking for an admin client, and so we auth with
# our own token
neutronv2.get_client(my_context)
neutronapi.get_client(my_context)
def test_withouttoken_keystone_connection_error(self):
self.flags(auth_strategy='keystone', group='neutron')
self.flags(url='http://anyhost/', group='neutron')
my_context = context.RequestContext('userid', 'my_tenantid')
self.assertRaises(NEUTRON_CLIENT_EXCEPTION,
neutronv2.get_client,
neutronapi.get_client,
my_context)
def test_reuse_admin_token(self):
self.flags(url='http://anyhost/', group='neutron')
self.flags(url_timeout=30, group='neutron')
token_store = neutronv2.AdminTokenStore.get()
token_store = neutronapi.AdminTokenStore.get()
token_store.admin_auth_token = 'new_token'
my_context = context.RequestContext('userid', 'my_tenantid',
auth_token='token')
@ -164,17 +163,17 @@ class TestNeutronClient(test.TestCase):
mock.patch.object(client.Client, 'get_auth_info',
return_value={'auth_token': 'new_token1'}),
):
client1 = neutronv2.get_client(my_context, True)
client1 = neutronapi.get_client(my_context, True)
client1.list_networks(retrieve_all=False)
self.assertEqual('new_token1', token_store.admin_auth_token)
client1 = neutronv2.get_client(my_context, True)
client1 = neutronapi.get_client(my_context, True)
client1.list_networks(retrieve_all=False)
self.assertEqual('new_token1', token_store.admin_auth_token)
def test_admin_token_updated(self):
self.flags(url='http://anyhost/', group='neutron')
self.flags(url_timeout=30, group='neutron')
token_store = neutronv2.AdminTokenStore.get()
token_store = neutronapi.AdminTokenStore.get()
token_store.admin_auth_token = 'new_token'
tokens = [{'auth_token': 'new_token1'}, {'auth_token': 'new_token'}]
my_context = context.RequestContext('userid', 'my_tenantid',
@ -185,10 +184,10 @@ class TestNeutronClient(test.TestCase):
mock.patch.object(client.Client, 'get_auth_info',
side_effect=tokens.pop),
):
client1 = neutronv2.get_client(my_context, True)
client1 = neutronapi.get_client(my_context, True)
client1.list_networks(retrieve_all=False)
self.assertEqual('new_token', token_store.admin_auth_token)
client1 = neutronv2.get_client(my_context, True)
client1 = neutronapi.get_client(my_context, True)
client1.list_networks(retrieve_all=False)
self.assertEqual('new_token1', token_store.admin_auth_token)
@ -341,7 +340,7 @@ class TestNeutronv2Base(test.TestCase):
'fixed_ip_address': fixed_ip_address,
'router_id': 'router_id1'}
self._returned_nw_info = []
self.mox.StubOutWithMock(neutronv2, 'get_client')
self.mox.StubOutWithMock(neutronapi, 'get_client')
self.moxed_client = self.mox.CreateMock(client.Client)
self.addCleanup(CONF.reset)
self.addCleanup(self.mox.VerifyAll)
@ -368,9 +367,9 @@ class TestNeutronv2Base(test.TestCase):
has_portbinding = True
api.extensions[constants.PORTBINDING_EXT] = 1
self.mox.StubOutWithMock(api, '_refresh_neutron_extensions_cache')
neutronv2.get_client(mox.IgnoreArg()).AndReturn(
neutronapi.get_client(mox.IgnoreArg()).AndReturn(
self.moxed_client)
neutronv2.get_client(
neutronapi.get_client(
mox.IgnoreArg(), admin=True).AndReturn(
self.moxed_client)
api._refresh_neutron_extensions_cache(mox.IgnoreArg(),
@ -604,19 +603,19 @@ class TestNeutronv2(TestNeutronv2Base):
def setUp(self):
super(TestNeutronv2, self).setUp()
neutronv2.get_client(mox.IgnoreArg()).MultipleTimes().AndReturn(
neutronapi.get_client(mox.IgnoreArg()).MultipleTimes().AndReturn(
self.moxed_client)
def test_get_instance_nw_info_1(self):
# Test to get one port in one network and subnet.
neutronv2.get_client(mox.IgnoreArg(),
neutronapi.get_client(mox.IgnoreArg(),
admin=True).MultipleTimes().AndReturn(
self.moxed_client)
self._get_instance_nw_info(1)
def test_get_instance_nw_info_2(self):
# Test to get one port in each of two networks and subnets.
neutronv2.get_client(mox.IgnoreArg(),
neutronapi.get_client(mox.IgnoreArg(),
admin=True).MultipleTimes().AndReturn(
self.moxed_client)
self._get_instance_nw_info(2)
@ -710,7 +709,7 @@ class TestNeutronv2(TestNeutronv2Base):
api.db.instance_info_cache_update(
mox.IgnoreArg(),
self.instance['uuid'], mox.IgnoreArg())
neutronv2.get_client(mox.IgnoreArg(),
neutronapi.get_client(mox.IgnoreArg(),
admin=True).MultipleTimes().AndReturn(
self.moxed_client)
self.moxed_client.list_ports(
@ -791,7 +790,7 @@ class TestNeutronv2(TestNeutronv2Base):
self.moxed_client.list_networks(
id=[self.port_data1[0]['network_id']]).AndReturn(
{'networks': self.nets1})
neutronv2.get_client(mox.IgnoreArg(),
neutronapi.get_client(mox.IgnoreArg(),
admin=True).MultipleTimes().AndReturn(
self.moxed_client)
@ -821,7 +820,7 @@ class TestNeutronv2(TestNeutronv2Base):
# Note: Don't want the default get_client from setUp()
self.mox.ResetAll()
neutronv2.get_client(mox.IgnoreArg()).AndReturn(
neutronapi.get_client(mox.IgnoreArg()).AndReturn(
self.moxed_client)
self.moxed_client.list_extensions().AndReturn(
{'extensions': [{'name': constants.QOS_QUEUE}]})
@ -836,7 +835,7 @@ class TestNeutronv2(TestNeutronv2Base):
# Note: Don't want the default get_client from setUp()
self.mox.ResetAll()
neutronv2.get_client(mox.IgnoreArg()).AndReturn(
neutronapi.get_client(mox.IgnoreArg()).AndReturn(
self.moxed_client)
self.moxed_client.list_extensions().AndReturn(
{'extensions': [{'name': constants.QOS_QUEUE}]})
@ -1289,13 +1288,13 @@ class TestNeutronv2(TestNeutronv2Base):
six.text_type(
jsonutils.dumps(net_info_cache))}
api = neutronapi.API()
neutronv2.get_client(mox.IgnoreArg(), admin=True).AndReturn(
neutronapi.get_client(mox.IgnoreArg(), admin=True).AndReturn(
self.moxed_client)
self.moxed_client.list_ports(
tenant_id=self.instance['project_id'],
device_id=self.instance['uuid']).AndReturn(
{'ports': port_data[1:]})
neutronv2.get_client(mox.IgnoreArg()).MultipleTimes().AndReturn(
neutronapi.get_client(mox.IgnoreArg()).MultipleTimes().AndReturn(
self.moxed_client)
net_ids = [port['network_id'] for port in port_data]
self.moxed_client.list_networks(id=net_ids).AndReturn(
@ -1410,7 +1409,7 @@ class TestNeutronv2(TestNeutronv2Base):
('my_netid1', None, None, None)]
self.mox.ReplayAll()
# Expected call from setUp.
neutronv2.get_client(None)
neutronapi.get_client(None)
api = neutronapi.API()
self.assertRaises(exception.NetworkDuplicated,
api.validate_networks,
@ -1497,7 +1496,7 @@ class TestNeutronv2(TestNeutronv2Base):
NeutronNotFound)
self.mox.ReplayAll()
# Expected call from setUp.
neutronv2.get_client(None)
neutronapi.get_client(None)
api = neutronapi.API()
self.assertRaises(exception.PortNotFound,
api.validate_networks,
@ -1517,7 +1516,7 @@ class TestNeutronv2(TestNeutronv2Base):
NeutronNotFound)
self.mox.ReplayAll()
# Expected call from setUp.
neutronv2.get_client(None)
neutronapi.get_client(None)
api = neutronapi.API()
self.assertRaises(exceptions.NeutronClientException,
api.validate_networks,
@ -2266,7 +2265,7 @@ class TestNeutronv2(TestNeutronv2Base):
self.moxed_client.list_floatingips(
fixed_ip_address='1.1.1.1', port_id=1).AndRaise(NeutronNotFound)
self.mox.ReplayAll()
neutronv2.get_client('fake')
neutronapi.get_client('fake')
floatingips = api._get_floating_ips_by_fixed_and_port(
self.moxed_client, '1.1.1.1', 1)
self.assertEqual(floatingips, [])
@ -2283,7 +2282,7 @@ class TestNeutronv2(TestNeutronv2Base):
self.moxed_client, '1.1.1.1', 'port-id').AndReturn(
[{'floating_ip_address': '10.0.0.1'}])
self.mox.ReplayAll()
neutronv2.get_client('fake')
neutronapi.get_client('fake')
result = api._nw_info_get_ips(self.moxed_client, fake_port)
self.assertEqual(len(result), 1)
self.assertEqual(result[0]['address'], '1.1.1.1')
@ -2303,7 +2302,7 @@ class TestNeutronv2(TestNeutronv2Base):
api._get_subnets_from_port(self.context, fake_port).AndReturn(
[fake_subnet])
self.mox.ReplayAll()
neutronv2.get_client('fake')
neutronapi.get_client('fake')
subnets = api._nw_info_get_subnets(self.context, fake_port, fake_ips)
self.assertEqual(len(subnets), 1)
self.assertEqual(len(subnets[0]['ips']), 1)
@ -2320,7 +2319,7 @@ class TestNeutronv2(TestNeutronv2Base):
fake_nets = [{'id': 'net-id', 'name': 'foo', 'tenant_id': 'tenant'}]
api = neutronapi.API()
self.mox.ReplayAll()
neutronv2.get_client('fake')
neutronapi.get_client('fake')
net, iid = api._nw_info_build_network(fake_port, fake_nets,
fake_subnets)
self.assertEqual(net['subnets'], fake_subnets)
@ -2367,7 +2366,7 @@ class TestNeutronv2(TestNeutronv2Base):
fake_nets = [{'id': 'net-id2', 'name': 'foo', 'tenant_id': 'tenant'}]
api = neutronapi.API()
self.mox.ReplayAll()
neutronv2.get_client('fake')
neutronapi.get_client('fake')
net, iid = api._nw_info_build_network(fake_port, fake_nets,
fake_subnets)
self.assertEqual(fake_subnets, net['subnets'])
@ -2469,7 +2468,7 @@ class TestNeutronv2(TestNeutronv2Base):
'tenant_id': 'fake',
}
]
neutronv2.get_client(mox.IgnoreArg(), admin=True).MultipleTimes(
neutronapi.get_client(mox.IgnoreArg(), admin=True).MultipleTimes(
).AndReturn(self.moxed_client)
self.moxed_client.list_ports(
tenant_id='fake', device_id='uuid').AndReturn(
@ -2488,7 +2487,7 @@ class TestNeutronv2(TestNeutronv2Base):
).AndReturn(fake_subnets)
self.mox.ReplayAll()
neutronv2.get_client('fake')
neutronapi.get_client('fake')
nw_infos = api._build_network_info_model(self.context, fake_inst,
fake_nets,
[fake_ports[2]['id'],
@ -2564,7 +2563,7 @@ class TestNeutronv2(TestNeutronv2Base):
networks = api.get_all(self.context)
self.assertEqual(networks, [])
@mock.patch.object(neutronv2, 'get_client', return_value=mock.Mock())
@mock.patch.object(neutronapi, 'get_client', return_value=mock.Mock())
def test_get_port_vnic_info_1(self, mock_get_client):
api = neutronapi.API()
self.mox.ResetAll()
@ -2613,17 +2612,17 @@ class TestNeutronv2(TestNeutronv2Base):
self.assertEqual(model.VNIC_TYPE_NORMAL, vnic_type)
self.assertFalse(phynet_name)
@mock.patch.object(neutronv2, 'get_client', return_value=mock.Mock())
@mock.patch.object(neutronapi, 'get_client', return_value=mock.Mock())
def test_get_port_vnic_info_2(self, mock_get_client):
self._test_get_port_vnic_info(mock_get_client,
binding_vnic_type=model.VNIC_TYPE_NORMAL)
@mock.patch.object(neutronv2, 'get_client', return_value=mock.Mock())
@mock.patch.object(neutronapi, 'get_client', return_value=mock.Mock())
def test_get_port_vnic_info_3(self, mock_get_client):
self._test_get_port_vnic_info(mock_get_client)
@mock.patch.object(neutronapi.API, "_get_port_vnic_info")
@mock.patch.object(neutronv2, 'get_client', return_value=mock.Mock())
@mock.patch.object(neutronapi, 'get_client', return_value=mock.Mock())
def test_create_pci_requests_for_sriov_ports(self, mock_get_client,
mock_get_port_vnic_info):
api = neutronapi.API()
@ -2796,7 +2795,7 @@ class TestNeutronv2WithMock(test.TestCase):
'device_owner': zone}}
self.assertRaises(exception.NoMoreFixedIps,
self.api._create_port,
neutronv2.get_client(self.context),
neutronapi.get_client(self.context),
instance, net['id'], port_req_body)
create_port_mock.assert_called_once_with(port_req_body)
@ -2818,7 +2817,7 @@ class TestNeutronv2WithMock(test.TestCase):
# Run the code.
self.assertRaises(exception.PortInUse,
self.api._create_port,
neutronv2.get_client(self.context),
neutronapi.get_client(self.context),
instance, net['id'], port_req_body,
available_macs=available_macs)
# Assert the calls.
@ -2841,7 +2840,7 @@ class TestNeutronv2WithMock(test.TestCase):
# Run the code.
self.assertRaises(exception.FixedIpAlreadyInUse,
self.api._create_port,
neutronv2.get_client(self.context),
neutronapi.get_client(self.context),
instance, net['id'], port_req_body,
fixed_ip=fake_ip)
# Assert the calls.
@ -2870,7 +2869,7 @@ class TestNeutronv2WithMock(test.TestCase):
mock_client = mock.Mock()
mock_client.list_ports.return_value = port_data
with contextlib.nested(
mock.patch.object(neutronv2, 'get_client',
mock.patch.object(neutronapi, 'get_client',
return_value=mock_client),
mock.patch.object(api, '_delete_ports')
) as (
@ -2912,7 +2911,7 @@ class TestNeutronv2WithMock(test.TestCase):
mock_client = mock.Mock()
api = neutronapi.API()
with contextlib.nested(
mock.patch.object(neutronv2, 'get_client',
mock.patch.object(neutronapi, 'get_client',
return_value=mock_client),
mock.patch.object(api, '_delete_ports',
side_effect=exceptions.Unauthorized),
@ -2927,7 +2926,7 @@ class TestNeutronv2WithMock(test.TestCase):
# make sure that we didn't try to reload nw info
self.assertFalse(get_nw_info.called)
@mock.patch.object(neutronv2, 'get_client', return_value=mock.Mock())
@mock.patch.object(neutronapi, 'get_client', return_value=mock.Mock())
def _test_show_port_exceptions(self, client_exc, expected_nova_exc,
get_client_mock):
show_port_mock = mock.Mock(side_effect=client_exc)
@ -2997,7 +2996,7 @@ class TestNeutronv2Portbinding(TestNeutronv2Base):
def test_populate_neutron_extension_values_binding(self):
api = neutronapi.API()
neutronv2.get_client(mox.IgnoreArg()).AndReturn(
neutronapi.get_client(mox.IgnoreArg()).AndReturn(
self.moxed_client)
self.moxed_client.list_extensions().AndReturn(
{'extensions': [{'name': constants.PORTBINDING_EXT}]})
@ -3054,7 +3053,7 @@ class TestNeutronv2Portbinding(TestNeutronv2Base):
self.mox.StubOutWithMock(api, '_has_port_binding_extension')
api._has_port_binding_extension(mox.IgnoreArg(),
refresh_cache=True).AndReturn(True)
neutronv2.get_client(mox.IgnoreArg(), admin=True).AndReturn(
neutronapi.get_client(mox.IgnoreArg(), admin=True).AndReturn(
self.moxed_client)
search_opts = {'device_id': self.instance['uuid'],
'tenant_id': self.instance['project_id']}
@ -3074,7 +3073,7 @@ class TestNeutronv2Portbinding(TestNeutronv2Base):
self.mox.StubOutWithMock(api, '_has_port_binding_extension')
api._has_port_binding_extension(mox.IgnoreArg(),
refresh_cache=True).AndReturn(True)
neutronv2.get_client(mox.IgnoreArg(), admin=True).AndReturn(
neutronapi.get_client(mox.IgnoreArg(), admin=True).AndReturn(
self.moxed_client)
search_opts = {'device_id': self.instance['uuid'],
'tenant_id': self.instance['project_id']}
@ -3102,7 +3101,7 @@ class TestNeutronv2Portbinding(TestNeutronv2Base):
class TestNeutronv2ExtraDhcpOpts(TestNeutronv2Base):
def setUp(self):
super(TestNeutronv2ExtraDhcpOpts, self).setUp()
neutronv2.get_client(mox.IgnoreArg()).MultipleTimes().AndReturn(
neutronapi.get_client(mox.IgnoreArg()).MultipleTimes().AndReturn(
self.moxed_client)
def test_allocate_for_instance_1_with_extra_dhcp_opts_turned_off(self):
@ -3158,18 +3157,18 @@ class TestNeutronClientForAdminScenarios(test.TestCase):
self.mox.ReplayAll()
# clean global
token_store = neutronv2.AdminTokenStore.get()
token_store = neutronapi.AdminTokenStore.get()
token_store.admin_auth_token = None
if admin_context:
# Note that the context does not contain a token but is
# an admin context which will force an elevation to admin
# credentials.
neutronv2.get_client(my_context)
neutronapi.get_client(my_context)
else:
# Note that the context is not elevated, but the True is passed in
# which will force an elevation to admin credentials even though
# the context has an auth_token.
neutronv2.get_client(my_context, True)
neutronapi.get_client(my_context, True)
def test_get_client_for_admin(self):
self._test_get_client_for_admin()