openstack
/
shade
Code Issues Proposed changes

Process config options via os-client-config 

Assume that a cloud config object will be passed in. For compat,
create one and pass in kwargs if it's not - but based the internal
operation on the existence of a cloud_config object.

Change-Id: I31b2d851e0636c7742468aa62565295a2c5266e2
Depends-On: I8764ed68f8a38563c4242d4b50e2158e99ed4109
This commit is contained in:
Monty Taylor 2015-07-27 11:58:43 -04:00
parent 404675ef46
commit ad59fe3a72
17 changed files with 131 additions and 163 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
requirements.txt
View File

@ -3,7 +3,7 @@ pbr>=0.11,<2.0
bunch
decorator
jsonpatch
os-client-config>=1.3.0
os-client-config>=1.6.0
six
python-novaclient>=2.21.0

186
shade/__init__.py
View File

@ -126,38 +126,28 @@ def openstack_clouds(config=None, debug=False):
]
def openstack_cloud(debug=False, **kwargs):
config = kwargs.get('config')
if config is None:
def openstack_cloud(config=None, **kwargs):
if not config:
config = os_client_config.OpenStackConfig()
cloud_config = config.get_one_cloud(**kwargs)
return OpenStackCloud(
cloud=cloud_config.name,
cache_interval=config.get_cache_max_age(),
cache_class=config.get_cache_class(),
cache_arguments=config.get_cache_arguments(),
cloud_config=cloud_config,
debug=debug, **cloud_config.config)
cloud_config=cloud_config)
def operator_cloud(debug=False, **kwargs):
config = os_client_config.OpenStackConfig()
def operator_cloud(config=None, **kwargs):
if 'interface' not in kwargs:
kwargs['interface'] = 'admin'
if not config:
config = os_client_config.OpenStackConfig()
cloud_config = config.get_one_cloud(**kwargs)
return OperatorCloud(
cloud_config.name, debug=debug,
cache_interval=config.get_cache_max_age(),
cache_class=config.get_cache_class(),
cache_arguments=config.get_cache_arguments(),
cloud_config=cloud_config,
**cloud_config.config)
def _get_service_values(kwargs, service_key):
# get defauts returns a copy of the defaults dict
values = os_client_config.defaults.get_defaults()
values.update(kwargs)
return {k[:-(len(service_key) + 1)]: str(values[k])
for k in values.keys() if k.endswith(service_key)}
cloud_config=cloud_config)
def _cache_on_arguments(*cache_on_args, **cache_on_kwargs):
@ -204,32 +194,6 @@ class OpenStackCloud(object):
and that Floating IP will be actualized either via neutron or via nova
depending on how this particular cloud has decided to arrange itself.
:param string name: The name of the cloud
:param dict auth: Dictionary containing authentication information.
Depending on the value of auth_type, the contents
of this dict can vary wildly.
:param string region_name: The region of the cloud that all operations
should be performed against.
(optional, default '')
:param string auth_type: The name of the keystone auth_type to be used
:param string endpoint_type: The type of endpoint to get for services
from the service catalog. Valid types are
`public` ,`internal` or `admin`. (optional,
defaults to `public`)
:param bool private: Whether to return or use private IPs by default for
servers. (optional, defaults to False)
:param float api_timeout: A timeout to pass to REST client constructors
indicating network-level timeouts. (optional)
:param bool verify: The verification arguments to pass to requests. True
tells requests to verify SSL requests, False to not
verify. (optional, defaults to True)
:param string cacert: A path to a CA Cert bundle that can be used as part
of verifying SSL requests. If this is set, verify
is set to True. (optional)
:param string cert: A path to a client certificate to pass to requests.
(optional)
:param string key: A path to a client key to pass to requests. (optional)
:param bool debug: Deprecated and unused parameter.
:param int cache_interval: How long to cache items fetched from the cloud.
Value will be passed to dogpile.cache. None
means do not cache at all.
@ -242,64 +206,41 @@ class OpenStackCloud(object):
OpenStack API tasks. Unless you're doing
rate limiting client side, you almost
certainly don't need this. (optional)
:param bool image_api_use_tasks: Whether or not this cloud needs to
use the glance task-create interface for
image upload activities instead of direct
calls. (optional, defaults to False)
:param CloudConfig cloud_config: Cloud config object from os-client-config
In the future, this will be the only way
to pass in cloud configuration, but is
being phased in currently.
"""
def __init__(self, cloud, auth,
region_name='',
auth_type='password',
endpoint_type='public',
private=False,
verify=True, cacert=None, cert=None, key=None,
api_timeout=None,
debug=False, cache_interval=None,
cache_class='dogpile.cache.null',
cache_arguments=None,
manager=None,
image_api_use_tasks=False,
cloud_config=None,
**kwargs):
def __init__(
self,
cloud_config=None,
cache_interval=None,
cache_class='dogpile.cache.null',
cache_arguments=None,
manager=None, **kwargs):
self.log = logging.getLogger('shade')
if cloud_config is None:
if not cloud_config:
config = os_client_config.OpenStackConfig()
ssl_args = dict(
verify=verify, cacert=cacert, cert=cert, key=key,
)
if cloud in config.get_cloud_names():
cloud_config = config.get_one_cloud(cloud, **ssl_args)
else:
cloud_config = config.get_one_cloud(**ssl_args)
self._cloud_config = cloud_config
self.name = cloud
self.auth = auth
self.region_name = region_name
self.auth_type = auth_type
self.endpoint_type = endpoint_type
self.private = private
self.api_timeout = api_timeout
cloud_config = config.get_one_cloud(**kwargs)
self.name = cloud_config.name
self.auth = cloud_config.get_auth_args()
self.region_name = cloud_config.region_name
self.auth_type = cloud_config.config['auth_type']
self.default_interface = cloud_config.get_interface()
self.private = cloud_config.config.get('private', False)
self.api_timeout = cloud_config.config['api_timeout']
self.image_api_use_tasks = cloud_config.config['image_api_use_tasks']
self.secgroup_source = cloud_config.config['secgroup_source']
if manager is not None:
self.manager = manager
else:
self.manager = task_manager.TaskManager(
name=self.name, client=self)
self.service_types = _get_service_values(kwargs, 'service_type')
self.service_names = _get_service_values(kwargs, 'service_name')
self.endpoints = _get_service_values(kwargs, 'endpoint')
self.api_versions = _get_service_values(kwargs, 'api_version')
self.image_api_use_tasks = image_api_use_tasks
self.secgroup_source = kwargs.get('secgroup_source', None)
(self.verify, self.cert) = cloud_config.get_requests_verify_args()
self._cache = cache.make_region(
@ -324,6 +265,8 @@ class OpenStackCloud(object):
self._swift_service = None
self._trove_client = None
self.cloud_config = cloud_config
@contextlib.contextmanager
def _neutron_exceptions(self, error_message):
try:
@ -365,12 +308,6 @@ class OpenStackCloud(object):
return ans
return generate_key
def get_service_type(self, service):
return self.service_types.get(service, service)
def get_service_name(self, service):
return self.service_names.get(service, None)
@property
def nova_client(self):
if self._nova_client is None:
@ -380,9 +317,9 @@ class OpenStackCloud(object):
# trigger exception on lack of compute. (what?)
self.get_session_endpoint('compute')
self._nova_client = nova_client.Client(
self.api_versions['compute'],
self.cloud_config.get_api_version('compute'),
session=self.keystone_session,
service_name=self.get_service_name('compute'),
service_name=self.cloud_config.get_service_name('compute'),
region_name=self.region_name,
timeout=self.api_timeout)
except Exception:
@ -409,13 +346,13 @@ class OpenStackCloud(object):
plugin=self.auth_type, error=str(e)))
def _get_identity_client_class(self):
if self.api_versions['identity'] == '3':
if self.cloud_config.get_api_version('identity') == '3':
return k3_client.Client
elif self.api_versions['identity'] in ('2', '2.0'):
elif self.cloud_config.get_api_version('identity') in ('2', '2.0'):
return k2_client.Client
raise OpenStackCloudException(
"Unknown identity API version: {version}".format(
version=self.api_versions['identity']))
version=self.cloud_config.get_api_version('identity')))
@property
def keystone_session(self):
@ -488,7 +425,7 @@ class OpenStackCloud(object):
project_dict = dict()
if name_or_id:
project_id = self._get_project(name_or_id).id
if self.api_versions['identity'] == '3':
if self.cloud_config.get_api_version('identity') == '3':
project_dict['default_project'] = project_id
else:
project_dict['tenant_id'] = project_id
@ -501,7 +438,7 @@ class OpenStackCloud(object):
# not. However, keystone v2 does not allow user creation by non-admin
# users, so we can throw an error to the user that does not need to
# mention api versions
if self.api_versions['identity'] == '3':
if self.cloud_config.get_api_version('identity') == '3':
if not domain_id:
raise OpenStackCloudException(
"User creation requires an explicit domain_id argument.")
@ -667,7 +604,8 @@ class OpenStackCloud(object):
kwargs['timeout'] = self.api_timeout
try:
self._glance_client = glanceclient.Client(
self.api_versions['image'], endpoint, token=token,
self.cloud_config.get_api_version('image'),
endpoint, token=token,
session=self.keystone_session, insecure=not self.verify,
cacert=self.cert, **kwargs)
except Exception as e:
@ -689,7 +627,7 @@ class OpenStackCloud(object):
self._swift_client = swift_client.Connection(
preauthurl=endpoint,
preauthtoken=token,
auth_version=self.api_versions['identity'],
auth_version=self.cloud_config.get_api_version('identity'),
os_options=dict(
auth_token=token,
object_storage_url=endpoint,
@ -750,10 +688,10 @@ class OpenStackCloud(object):
# Make the connection - can't use keystone session until there
# is one
self._trove_client = trove_client.Client(
self.api_versions['database'],
self.cloud_config.get_api_version('database'),
session=self.keystone_session,
region_name=self.region_name,
service_type=self.get_service_type('database'),
service_type=self.cloud_config.get_service_type('database'),
timeout=self.api_timeout,
)
@ -780,7 +718,7 @@ class OpenStackCloud(object):
def designate_client(self):
if self._designate_client is None:
# get dns service type if defined in cloud config
dns_service_type = self.get_service_type('dns')
dns_service_type = self.cloud_config.get_service_type('dns')
# trigger exception on lack of designate
self.get_session_endpoint(dns_service_type)
@ -829,8 +767,9 @@ class OpenStackCloud(object):
ram=ram, include=include))
def get_session_endpoint(self, service_key):
if service_key in self.endpoints:
return self.endpoints[service_key]
override_endpoint = self.cloud_config.get_endpoint(service_key)
if override_endpoint:
return override_endpoint
try:
# keystone is a special case in keystone, because what?
if service_key == 'identity':
@ -838,9 +777,11 @@ class OpenStackCloud(object):
interface=ksc_auth.AUTH_INTERFACE)
else:
endpoint = self.keystone_session.get_endpoint(
service_type=self.get_service_type(service_key),
service_name=self.get_service_name(service_key),
interface=self.endpoint_type,
service_type=self.cloud_config.get_service_type(
service_key),
service_name=self.cloud_config.get_service_name(
service_key),
interface=self.cloud_config.get_interface(service_key),
region_name=self.region_name)
except keystone_exceptions.EndpointNotFound as e:
self.log.debug(
@ -1417,7 +1358,7 @@ class OpenStackCloud(object):
try:
# Note that in v1, the param name is image, but in v2,
# it's image_id
glance_api_version = self.api_versions['image']
glance_api_version = self.cloud_config.get_api_version('image')
if glance_api_version == '2':
self.manager.submitTask(
_tasks.ImageDelete(image_id=image.id))
@ -1455,7 +1396,7 @@ class OpenStackCloud(object):
kwargs[IMAGE_SHA256_KEY] = sha256
if disable_vendor_agent:
kwargs.update(self._cloud_config.config['disable_vendor_agent'])
kwargs.update(self.cloud_config.config['disable_vendor_agent'])
try:
# This makes me want to die inside
@ -1505,7 +1446,7 @@ class OpenStackCloud(object):
def _upload_image_put(self, name, filename, **image_kwargs):
image_data = open(filename, 'rb')
# Because reasons and crying bunnies
if self.api_versions['image'] == '2':
if self.cloud_config.get_api_version('image') == '2':
image = self._upload_image_put_v2(name, image_data, **image_kwargs)
else:
image = self._upload_image_put_v1(name, image_data, **image_kwargs)
@ -1586,7 +1527,7 @@ class OpenStackCloud(object):
img_props[k] = v
# This makes me want to die inside
if self.api_versions['image'] == '2':
if self.cloud_config.get_api_version('image') == '2':
return self._update_image_properties_v2(image, img_props)
else:
return self._update_image_properties_v1(image, img_props)
@ -3371,20 +3312,8 @@ class OperatorCloud(OpenStackCloud):
of which OpenStack service those operations are for.
See the :class:`OpenStackCloud` class for a description of most options.
`OperatorCloud` overrides the default value of `endpoint_type` from
`public` to `admin`.
:param string endpoint_type: The type of endpoint to get for services
from the service catalog. Valid types are
`public` ,`internal` or `admin`. (optional,
defaults to `admin`)
"""
def __init__(self, *args, **kwargs):
super(OperatorCloud, self).__init__(*args, **kwargs)
if 'endpoint_type' not in kwargs:
self.endpoint_type = 'admin'
@property
def auth_token(self):
if self.auth_type in (None, "None", ''):
@ -3425,7 +3354,8 @@ class OperatorCloud(OpenStackCloud):
endpoint = self.get_session_endpoint(service_key='baremetal')
try:
self._ironic_client = ironic_client.Client(
self.api_versions['baremetal'], endpoint, token=token,
self.cloud_config.get_api_version('baremetal'),
endpoint, token=token,
timeout=self.api_timeout,
os_ironic_api_version=ironic_api_microversion)
except Exception as e:

16
shade/tests/unit/test_caching.py
View File

@ -261,9 +261,10 @@ class TestMemoryCache(base.TestCase):
self.cloud.create_image(
name, imagefile.name, container=container, wait=True)
@mock.patch.object(occ.cloud_config.CloudConfig, 'get_api_version')
@mock.patch.object(shade.OpenStackCloud, 'glance_client')
def test_create_image_put_v1(self, glance_mock):
self.cloud.api_versions['image'] = '1'
def test_create_image_put_v1(self, glance_mock, mock_api_version):
mock_api_version.return_value = '1'
glance_mock.images.list.return_value = []
self.assertEqual([], self.cloud.list_images())
@ -280,9 +281,10 @@ class TestMemoryCache(base.TestCase):
fake_image_dict = meta.obj_to_dict(fake_image)
self.assertEqual([fake_image_dict], self.cloud.list_images())
@mock.patch.object(occ.cloud_config.CloudConfig, 'get_api_version')
@mock.patch.object(shade.OpenStackCloud, 'glance_client')
def test_create_image_put_v2(self, glance_mock):
self.cloud.api_versions['image'] = '2'
def test_create_image_put_v2(self, glance_mock, mock_api_version):
mock_api_version.return_value = '2'
self.cloud.image_api_use_tasks = False
glance_mock.images.list.return_value = []
@ -301,6 +303,7 @@ class TestMemoryCache(base.TestCase):
fake_image_dict = meta.obj_to_dict(fake_image)
self.assertEqual([fake_image_dict], self.cloud.list_images())
@mock.patch.object(occ.cloud_config.CloudConfig, 'get_api_version')
@mock.patch.object(shade.OpenStackCloud, '_get_file_hashes')
@mock.patch.object(shade.OpenStackCloud, 'glance_client')
@mock.patch.object(shade.OpenStackCloud, 'swift_client')
@ -309,8 +312,9 @@ class TestMemoryCache(base.TestCase):
swift_service_mock,
swift_mock,
glance_mock,
get_file_hashes):
self.cloud.api_versions['image'] = '2'
get_file_hashes,
mock_api_version):
mock_api_version.return_value = '2'
self.cloud.image_api_use_tasks = True
class Container(object):

4
shade/tests/unit/test_create_server.py
View File

@ -20,6 +20,7 @@ Tests for the `create_server` command.
"""
from mock import patch, Mock
import os_client_config
from shade import meta
from shade import OpenStackCloud
from shade.exc import (OpenStackCloudException, OpenStackCloudTimeout)
@ -30,7 +31,8 @@ class TestCreateServer(base.TestCase):
def setUp(self):
super(TestCreateServer, self).setUp()
self.client = OpenStackCloud("cloud", {})
config = os_client_config.OpenStackConfig()
self.client = OpenStackCloud(cloud_config=config.get_one_cloud())
def test_create_server_with_create_exception(self):
"""

4
shade/tests/unit/test_delete_server.py
View File

@ -21,6 +21,7 @@ Tests for the `delete_server` command.
import mock
from novaclient import exceptions as nova_exc
import os_client_config
from shade import OpenStackCloud
from shade import exc as shade_exc
@ -40,7 +41,8 @@ class TestDeleteServer(base.TestCase):
def setUp(self):
super(TestDeleteServer, self).setUp()
self.cloud = OpenStackCloud("cloud", {})
config = os_client_config.OpenStackConfig()
self.cloud = OpenStackCloud(cloud_config=config.get_one_cloud())
@mock.patch('shade.OpenStackCloud.nova_client')
def test_delete_server(self, nova_mock):

20
shade/tests/unit/test_domain_params.py
View File

@ -13,6 +13,7 @@
# under the License.
import mock
import os_client_config as occ
import bunch
@ -27,11 +28,11 @@ class TestDomainParams(base.TestCase):
super(TestDomainParams, self).setUp()
self.cloud = shade.openstack_cloud()
@mock.patch.object(occ.cloud_config.CloudConfig, 'get_api_version')
@mock.patch.object(shade.OpenStackCloud, '_get_project')
def test_identity_params_v3(self, mock_get_project):
def test_identity_params_v3(self, mock_get_project, mock_api_version):
mock_get_project.return_value = bunch.Bunch(id=1234)
self.cloud.api_versions = dict(identity='3')
mock_api_version.return_value = '3'
ret = self.cloud._get_identity_params(domain_id='5678', project='bar')
self.assertIn('default_project', ret)
@ -39,22 +40,23 @@ class TestDomainParams(base.TestCase):
self.assertIn('domain', ret)
self.assertEqual(ret['domain'], '5678')
@mock.patch.object(occ.cloud_config.CloudConfig, 'get_api_version')
@mock.patch.object(shade.OpenStackCloud, '_get_project')
def test_identity_params_v3_no_domain(self, mock_get_project):
def test_identity_params_v3_no_domain(
self, mock_get_project, mock_api_version):
mock_get_project.return_value = bunch.Bunch(id=1234)
self.cloud.api_versions = dict(identity='3')
mock_api_version.return_value = '3'
self.assertRaises(
exc.OpenStackCloudException,
self.cloud._get_identity_params,
domain_id=None, project='bar')
@mock.patch.object(occ.cloud_config.CloudConfig, 'get_api_version')
@mock.patch.object(shade.OpenStackCloud, '_get_project')
def test_identity_params_v2(self, mock_get_project):
def test_identity_params_v2(self, mock_get_project, mock_api_version):
mock_get_project.return_value = bunch.Bunch(id=1234)
self.cloud.api_versions = dict(identity='2')
mock_api_version.return_value = '2'
ret = self.cloud._get_identity_params(domain_id='foo', project='bar')
self.assertIn('tenant_id', ret)

4
shade/tests/unit/test_endpoints.py
View File

@ -20,6 +20,7 @@ Tests Keystone endpoints commands.
"""
from mock import patch
import os_client_config
from shade import OperatorCloud
from shade.tests.fakes import FakeEndpoint
from shade.tests.unit import base
@ -38,7 +39,8 @@ class TestCloudEndpoints(base.TestCase):
def setUp(self):
super(TestCloudEndpoints, self).setUp()
self.client = OperatorCloud("op_cloud", {})
config = os_client_config.OpenStackConfig()
self.client = OperatorCloud(cloud_config=config.get_one_cloud())
self.mock_ks_endpoints = \
[FakeEndpoint(**kwa) for kwa in self.mock_endpoints]

4
shade/tests/unit/test_floating_ip_common.py
View File

@ -20,6 +20,7 @@ Tests floating IP resource methods for Neutron and Nova-network.
"""
from mock import patch
import os_client_config
from shade import meta
from shade import OpenStackCloud
from shade.tests.fakes import FakeServer
@ -29,7 +30,8 @@ from shade.tests.unit import base
class TestFloatingIP(base.TestCase):
def setUp(self):
super(TestFloatingIP, self).setUp()
self.client = OpenStackCloud("cloud", {})
config = os_client_config.OpenStackConfig()
self.client = OpenStackCloud(cloud_config=config.get_one_cloud())
@patch.object(OpenStackCloud, 'get_floating_ip')
@patch.object(OpenStackCloud, 'attach_ip_to_server')

4
shade/tests/unit/test_floating_ip_neutron.py
View File

@ -20,6 +20,7 @@ Tests Floating IP resource methods for Neutron
"""
from mock import patch
import os_client_config
from neutronclient.common import exceptions as n_exc
@ -121,7 +122,8 @@ class TestFloatingIP(base.TestCase):
def setUp(self):
super(TestFloatingIP, self).setUp()
# floating_ip_source='neutron' is default for OpenStackCloud()
self.client = OpenStackCloud("cloud", {})
config = os_client_config.OpenStackConfig()
self.client = OpenStackCloud(cloud_config=config.get_one_cloud())
@patch.object(OpenStackCloud, 'neutron_client')
@patch.object(OpenStackCloud, 'has_service')

4
shade/tests/unit/test_floating_ip_nova.py
View File

@ -20,6 +20,7 @@ Tests Floating IP resource methods for nova-network
"""
from mock import patch
import os_client_config
from novaclient import exceptions as n_exc
@ -69,7 +70,8 @@ class TestFloatingIP(base.TestCase):
def setUp(self):
super(TestFloatingIP, self).setUp()
self.client = OpenStackCloud("cloud", {})
config = os_client_config.OpenStackConfig()
self.client = OpenStackCloud(cloud_config=config.get_one_cloud())
@patch.object(OpenStackCloud, 'nova_client')
@patch.object(OpenStackCloud, 'has_service')

4
shade/tests/unit/test_floating_ip_pool.py
View File

@ -20,6 +20,7 @@ Test floating IP pool resource (managed by nova)
"""
from mock import patch
import os_client_config
from shade import OpenStackCloud
from shade import OpenStackCloudException
from shade.tests.unit import base
@ -33,7 +34,8 @@ class TestFloatingIPPool(base.TestCase):
def setUp(self):
super(TestFloatingIPPool, self).setUp()
self.client = OpenStackCloud('cloud', {})
config = os_client_config.OpenStackConfig()
self.client = OpenStackCloud(cloud_config=config.get_one_cloud())
@patch.object(OpenStackCloud, '_has_nova_extension')
@patch.object(OpenStackCloud, 'nova_client')

4
shade/tests/unit/test_object.py
View File

@ -14,6 +14,7 @@
# under the License.
import mock
import os_client_config
from swiftclient import client as swift_client
from swiftclient import service as swift_service
from swiftclient import exceptions as swift_exc
@ -28,7 +29,8 @@ class TestObject(base.TestCase):
def setUp(self):
super(TestObject, self).setUp()
self.cloud = OpenStackCloud('cloud', {})
config = os_client_config.OpenStackConfig()
self.cloud = OpenStackCloud(cloud_config=config.get_one_cloud())
@mock.patch.object(swift_client, 'Connection')
@mock.patch.object(shade.OpenStackCloud, 'auth_token',

4
shade/tests/unit/test_port.py
View File

@ -20,6 +20,7 @@ Test port resource (managed by neutron)
"""
from mock import patch
import os_client_config
from shade import OpenStackCloud
from shade.exc import OpenStackCloudException
from shade.tests.unit import base
@ -143,7 +144,8 @@ class TestPort(base.TestCase):
def setUp(self):
super(TestPort, self).setUp()
self.client = OpenStackCloud('cloud', {})
config = os_client_config.OpenStackConfig()
self.client = OpenStackCloud(cloud_config=config.get_one_cloud())
@patch.object(OpenStackCloud, 'neutron_client')
def test_create_port(self, mock_neutron_client):

4
shade/tests/unit/test_rebuild_server.py
View File

@ -20,6 +20,7 @@ Tests for the `rebuild_server` command.
"""
from mock import patch, Mock
import os_client_config
from shade import meta
from shade import OpenStackCloud
from shade.exc import (OpenStackCloudException, OpenStackCloudTimeout)
@ -30,7 +31,8 @@ class TestRebuildServer(base.TestCase):
def setUp(self):
super(TestRebuildServer, self).setUp()
self.client = OpenStackCloud("cloud", {})
config = os_client_config.OpenStackConfig()
self.client = OpenStackCloud(cloud_config=config.get_one_cloud())
def test_rebuild_server_rebuild_exception(self):
"""

4
shade/tests/unit/test_services.py
View File

@ -20,6 +20,7 @@ Tests Keystone services commands.
"""
from mock import patch
import os_client_config
from shade import OpenStackCloudException
from shade import OperatorCloud
from shade.tests.fakes import FakeService
@ -40,7 +41,8 @@ class CloudServices(base.TestCase):
def setUp(self):
super(CloudServices, self).setUp()
self.client = OperatorCloud("op_cloud", {})
config = os_client_config.OpenStackConfig()
self.client = OperatorCloud(cloud_config=config.get_one_cloud())
self.mock_ks_services = [FakeService(**kwa) for kwa in
self.mock_services]

19
shade/tests/unit/test_shade.py
View File

@ -21,6 +21,7 @@ from keystoneclient.v2_0 import client as k2_client
from keystoneclient.v3 import client as k3_client
from neutronclient.common import exceptions as n_exc
import os_client_config.cloud_config
import shade
from shade import exc
from shade import meta
@ -55,22 +56,28 @@ class TestShade(base.TestCase):
plugin,
keystoneclient.auth.identity.generic.password.Password)
def test_get_client_v2(self):
self.cloud.api_versions['identity'] = '2'
@mock.patch.object(
os_client_config.cloud_config.CloudConfig, 'get_api_version')
def test_get_client_v2(self, mock_api_version):
mock_api_version.return_value = '2'
self.assertIs(
self.cloud._get_identity_client_class(),
k2_client.Client)
def test_get_client_v3(self):
self.cloud.api_versions['identity'] = '3'
@mock.patch.object(
os_client_config.cloud_config.CloudConfig, 'get_api_version')
def test_get_client_v3(self, mock_api_version):
mock_api_version.return_value = '3'
self.assertIs(
self.cloud._get_identity_client_class(),
k3_client.Client)
def test_get_client_v4(self):
self.cloud.api_versions['identity'] = '4'
@mock.patch.object(
os_client_config.cloud_config.CloudConfig, 'get_api_version')
def test_get_client_v4(self, mock_api_version):
mock_api_version.return_value = '4'
self.assertRaises(
exc.OpenStackCloudException,

7
shade/tests/unit/test_shade_operator.py
View File

@ -17,6 +17,7 @@ from keystoneclient import auth as ksc_auth
import mock
import testtools
import os_client_config.cloud_config
import shade
from shade import exc
from shade import meta
@ -548,8 +549,10 @@ class TestShadeOperator(base.TestCase):
self.assertEqual('22', self.cloud.get_image_id('22'))
self.assertEqual('22', self.cloud.get_image_id('22 name'))
def test_get_session_endpoint_provided(self):
self.cloud.endpoints['image'] = 'http://fake.url'
@mock.patch.object(
os_client_config.cloud_config.CloudConfig, 'get_endpoint')
def test_get_session_endpoint_provided(self, fake_get_endpoint):
fake_get_endpoint.return_value = 'http://fake.url'
self.assertEqual(
'http://fake.url', self.cloud.get_session_endpoint('image'))