openstack/python-keystoneclient
Code Issues Proposed changes

Merge "Discovery URL querying functions"

Browse Source
This commit is contained in:
Jenkins 2014-05-06 22:45:44 +00:00 committed by Gerrit Code Review
commit 7432db16b3
3 changed files with 566 additions and 314 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

252
keystoneclient/_discover.py Normal file
View File

@ -0,0 +1,252 @@
# 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.
"""The passive components to version discovery.
The Discover object in discover.py contains functions that can create objects
on your behalf. These functions are not usable from within the keystoneclient
library because you will get dependency resolution issues.
The Discover object in this file provides the querying components of Discovery.
This includes functions like url_for which allow you to retrieve URLs and the
raw data specified in version discovery responses.
"""
import logging
from keystoneclient import exceptions
_LOGGER = logging.getLogger(__name__)
def get_version_data(session, url):
"""Retrieve raw version data from a url."""
headers = {'Accept': 'application/json'}
resp = session.get(url, headers=headers)
try:
body_resp = resp.json()
except ValueError:
pass
else:
# In the event of querying a root URL we will get back a list of
# available versions.
try:
return body_resp['versions']['values']
except (KeyError, TypeError):
pass
# Most servers don't have a 'values' element so accept a simple
# versions dict if available.
try:
return body_resp['versions']
except KeyError:
pass
# Otherwise if we query an endpoint like /v2.0 then we will get back
# just the one available version.
try:
return [body_resp['version']]
except KeyError:
pass
err_text = resp.text[:50] + '...' if len(resp.text) > 50 else resp.text
msg = 'Invalid Response - Bad version data returned: %s' % err_text
raise exceptions.DiscoveryFailure(msg)
def normalize_version_number(version):
"""Turn a version representation into a tuple."""
# trim the v from a 'v2.0' or similar
try:
version = version.lstrip('v')
except AttributeError:
pass
# if it's an integer or a numeric as a string then normalize it
# to a string, this ensures 1 decimal point
try:
num = float(version)
except Exception:
pass
else:
version = str(num)
# if it's a string (or an integer) from above break it on .
try:
return tuple(map(int, version.split('.')))
except Exception:
pass
# last attempt, maybe it's a list or iterable.
try:
return tuple(map(int, version))
except Exception:
pass
raise TypeError('Invalid version specified: %s' % version)
def version_match(required, candidate):
"""Test that an available version is a suitable match for a required
version.
To be suitable a version must be of the same major version as required
and be at least a match in minor/patch level.
eg. 3.3 is a match for a required 3.1 but 4.1 is not.
:param tuple required: the version that must be met.
:param tuple candidate: the version to test against required.
:returns bool: True if candidate is suitable False otherwise.
"""
# major versions must be the same (e.g. even though v2 is a lower
# version than v3 we can't use it if v2 was requested)
if candidate[0] != required[0]:
return False
# prevent selecting a minor version less than what is required
if candidate < required:
return False
return True
class Discover(object):
CURRENT_STATUSES = ('stable', 'current', 'supported')
DEPRECATED_STATUSES = ('deprecated',)
EXPERIMENTAL_STATUSES = ('experimental',)
def __init__(self, session, url):
self._data = get_version_data(session, url)
def raw_version_data(self, allow_experimental=False,
allow_deprecated=True, allow_unknown=False):
"""Get raw version information from URL.
Raw data indicates that only minimal validation processing is performed
on the data, so what is returned here will be the data in the same
format it was received from the endpoint.
:param bool allow_experimental: Allow experimental version endpoints.
:param bool allow_deprecated: Allow deprecated version endpoints.
:param bool allow_unknown: Allow endpoints with an unrecognised status.
:returns list: The endpoints returned from the server that match the
criteria.
"""
versions = []
for v in self._data:
try:
status = v['status']
except KeyError:
_LOGGER.warning('Skipping over invalid version data. '
'No stability status in version.')
continue
status = status.lower()
if status in self.CURRENT_STATUSES:
versions.append(v)
elif status in self.DEPRECATED_STATUSES:
if allow_deprecated:
versions.append(v)
elif status in self.EXPERIMENTAL_STATUSES:
if allow_experimental:
versions.append(v)
elif allow_unknown:
versions.append(v)
return versions
def version_data(self, **kwargs):
"""Get normalized version data.
Return version data in a structured way.
:returns list(dict): A list of version data dictionaries sorted by
version number. Each data element in the returned
list is a dictionary consisting of at least:
:version tuple: The normalized version of the endpoint.
:url str: The url for the endpoint.
:raw_status str: The status as provided by the server
"""
data = self.raw_version_data(**kwargs)
versions = []
for v in data:
try:
version_str = v['id']
except KeyError:
_LOGGER.info('Skipping invalid version data. Missing ID.')
continue
try:
links = v['links']
except KeyError:
_LOGGER.info('Skipping invalid version data. Missing links')
continue
version_number = normalize_version_number(version_str)
for link in links:
try:
rel = link['rel']
url = link['href']
except (KeyError, TypeError):
_LOGGER.info('Skipping invalid version link. '
'Missing link URL or relationship.')
continue
if rel.lower() == 'self':
break
else:
_LOGGER.info('Skipping invalid version data. '
'Missing link to endpoint.')
continue
versions.append({'version': version_number,
'url': url,
'raw_status': v['status']})
versions.sort(key=lambda v: v['version'])
return versions
def data_for(self, version, **kwargs):
"""Return endpoint data for a specific version.
:returns dict: the endpoint data for a URL that matches the required
version (the format is described in version_data)
or None if no match.
"""
version = normalize_version_number(version)
version_data = self.version_data(**kwargs)
for data in reversed(version_data):
if version_match(version, data['version']):
return data
return None
def url_for(self, version, **kwargs):
"""Get the endpoint url for a required version.
:returns str: The url for the specified version or None if no match.
"""
data = self.data_for(version, **kwargs)
return data['url'] if data else None

375
keystoneclient/discover.py
View File

@ -14,6 +14,7 @@ import logging
import six
from keystoneclient import _discover
from keystoneclient import exceptions
from keystoneclient import session as client_session
from keystoneclient.v2_0 import client as v2_client
@ -23,155 +24,19 @@ from keystoneclient.v3 import client as v3_client
_logger = logging.getLogger(__name__)
class _KeystoneVersion(object):
"""A factory object that holds all the information to create a client.
Instances of this class are callable objects that hold all the kwargs that
were passed to discovery so that a user may simply call it to create a new
client object. Additional arguments passed to the call will override or
add to those provided to the object.
"""
_CLIENT_VERSIONS = {2: v2_client.Client,
3: v3_client.Client}
def __init__(self, version, status, client_class=None, **kwargs):
"""Create a new discovered version object.
:param tuple version: the version of the available API.
:param string status: the stability of the API.
:param Class client_class: the client class that should be used to
instantiate against this version of the API.
(optional, will be matched against known)
:param dict **kwargs: Additional arguments that should be passed on to
the client when it is constructed.
"""
self.version = version
self.status = status
self.client_class = client_class
self.client_kwargs = kwargs
if not self.client_class:
try:
self.client_class = self._CLIENT_VERSIONS[self.version[0]]
except KeyError:
raise exceptions.DiscoveryFailure("No client available "
"for version: %s" %
self.version)
def __lt__(self, other):
"""Version Ordering.
Versions are ordered by major, then minor version number, then
'stable' is deemed the highest possible status, then they are just
treated alphabetically (alpha < beta etc)
"""
if self.version == other.version:
if self.status == 'stable':
return False
elif other.status == 'stable':
return True
else:
return self.status < other.status
return self.version < other.version
def __eq__(self, other):
return self.version == other.version and self.status == other.status
def create_client(self, **kwargs):
if kwargs:
client_kwargs = self.client_kwargs.copy()
client_kwargs.update(kwargs)
else:
client_kwargs = self.client_kwargs
return self.client_class(**client_kwargs)
def __call__(self, **kwargs):
return self.create_client(**kwargs)
@property
def _str_ver(self):
ver = ".".join([str(v) for v in self.version])
if self.status != 'stable':
ver = "%s-%s" % (ver, self.status)
return ver
def _normalize_version_number(version):
"""Turn a version representation into a tuple."""
# trim the v from a 'v2.0' or similar
try:
version = version.lstrip('v')
except AttributeError:
pass
# if it's an integer or a numeric as a string then normalize it
# to a string, this ensures 1 decimal point
try:
num = float(version)
except Exception:
pass
else:
version = str(num)
# if it's a string (or an integer) from above break it on .
try:
return tuple(map(int, version.split(".")))
except Exception:
pass
# last attempt, maybe it's a list or iterable.
try:
return tuple(map(int, version))
except Exception:
pass
raise TypeError("Invalid version specified: %s" % version)
_CLIENT_VERSIONS = {2: v2_client.Client,
3: v3_client.Client}
def available_versions(url, session=None, **kwargs):
headers = {'Accept': 'application/json'}
"""Retrieve raw version data from a url."""
if not session:
session = client_session.Session.construct(kwargs)
resp = session.get(url, headers=headers)
try:
body_resp = resp.json()
except ValueError:
pass
else:
# In the event of querying a root URL we will get back a list of
# available versions.
try:
return body_resp['versions']['values']
except (KeyError, TypeError):
pass
# Most servers don't have a 'values' element so accept a simple
# versions dict if available.
try:
return body_resp['versions']
except KeyError:
pass
# Otherwise if we query an endpoint like /v2.0 then we will get back
# just the one available version.
try:
return [body_resp['version']]
except KeyError:
pass
raise exceptions.DiscoveryFailure("Invalid Response - Bad version"
" data returned: %s" % resp.text)
return _discover.get_version_data(session, url)
class Discover(object):
class Discover(_discover.Discover):
"""A means to discover and create clients depending on the supported API
versions on the server.
@ -239,39 +104,66 @@ class Discover(object):
session = client_session.Session.construct(kwargs)
kwargs['session'] = session
url = kwargs.get('endpoint') or kwargs.get('auth_url')
url = None
endpoint = kwargs.pop('endpoint', None)
auth_url = kwargs.pop('auth_url', None)
if endpoint:
self._use_endpoint = True
url = endpoint
elif auth_url:
self._use_endpoint = False
url = auth_url
if not url:
raise exceptions.DiscoveryFailure('Not enough information to '
'determine URL. Provide either '
'auth_url or endpoint')
self._client_kwargs = kwargs
self._available_versions = available_versions(url, session=session)
super(Discover, self).__init__(session, url)
def _get_client_constructor_kwargs(self, kwargs_dict={}, **kwargs):
client_kwargs = self._client_kwargs.copy()
client_kwargs.update(kwargs_dict)
client_kwargs.update(**kwargs)
return client_kwargs
def available_versions(self, unstable=False):
def available_versions(self, **kwargs):
"""Return a list of identity APIs available on the server and the data
associated with them.
DEPRECATED: use raw_version_data()
:param bool unstable: Accept endpoints not marked 'stable'. (optional)
DEPRECTED. Equates to setting allow_experimental
and allow_unknown to True.
:param bool allow_experimental: Allow experimental version endpoints.
:param bool allow_deprecated: Allow deprecated version endpoints.
:param bool allow_unknown: Allow endpoints with an unrecognised status.
:returns: A List of dictionaries as presented by the server. Each dict
will contain the version and the URL to use for the version.
It is a direct representation of the layout presented by the
identity API.
"""
return self.raw_version_data(**kwargs)
def raw_version_data(self, unstable=False, **kwargs):
"""Get raw version information from URL.
Raw data indicates that only minimal validation processing is performed
on the data, so what is returned here will be the data in the same
format it was received from the endpoint.
:param bool unstable: (deprecated) equates to setting
allow_experimental and allow_unknown.
:param bool allow_experimental: Allow experimental version endpoints.
:param bool allow_deprecated: Allow deprecated version endpoints.
:param bool allow_unknown: Allow endpoints with an unrecognised status.
:returns list: The endpoints returned from the server that match the
criteria.
Example::
>>> from keystoneclient import discover
>>> disc = discover.Discovery(auth_url='http://localhost:5000')
>>> disc.available_versions()
>>> disc.raw_version_data()
[{'id': 'v3.0',
'links': [{'href': u'http://127.0.0.1:5000/v3/',
'rel': u'self'}],
@ -297,110 +189,57 @@ class Discover(object):
'updated': '2013-03-06T00:00:00Z'}]
"""
if unstable:
# no need to determine the stable endpoints, just return everything
return self._available_versions
kwargs.setdefault('allow_experimental', True)
kwargs.setdefault('allow_unknown', True)
versions = []
return super(Discover, self).raw_version_data(**kwargs)
for v in self._available_versions:
try:
status = v['status']
except KeyError:
_logger.warning("Skipping over invalid version data. "
"No stability status in version.")
else:
if status == 'stable':
versions.append(v)
def _calculate_version(self, version, unstable):
version_data = None
return versions
if version:
version_data = self.data_for(version)
else:
# if no version specified pick the latest one
all_versions = self.version_data(unstable=unstable)
if all_versions:
version_data = all_versions[-1]
def _get_factory_from_response_entry(self, version_data, **kwargs):
"""Create a _KeystoneVersion factory object from a version response
entry returned from a server.
"""
if not version_data:
msg = 'Could not find a suitable endpoint'
if version:
msg += ' for client version: %s' % str(version)
raise exceptions.VersionNotAvailable(msg)
return version_data
def _create_client(self, version_data, **kwargs):
# Get the client for the version requested that was returned
try:
version_str = version_data['id']
status = version_data['status']
client_class = _CLIENT_VERSIONS[version_data['version'][0]]
except KeyError:
version = '.'.join(str(v) for v in version_data['version'])
msg = 'No client available for version: %s' % version
raise exceptions.DiscoveryFailure(msg)
if not version_str.startswith('v'):
raise exceptions.DiscoveryFailure('Skipping over invalid '
'version string: %s. It '
'should start with a v.' %
version_str)
# kwargs should take priority over stored kwargs.
for k, v in six.iteritems(self._client_kwargs):
kwargs.setdefault(k, v)
for link in version_data['links']:
# NOTE(jamielennox): there are plenty of links like with
# documentation and such, we only care about the self
# which is a link to the URL we should use.
if link['rel'].lower() == 'self':
version_number = _normalize_version_number(version_str)
version_url = link['href']
break
else:
raise exceptions.DiscoveryFailure("Didn't find any links "
"in version data.")
# restore the url to either auth_url or endpoint depending on what
# was initially given
if self._use_endpoint:
kwargs['auth_url'] = None
kwargs['endpoint'] = version_data['url']
else:
kwargs['auth_url'] = version_data['url']
kwargs['endpoint'] = None
except (KeyError, TypeError, ValueError):
raise exceptions.DiscoveryFailure('Skipping over invalid '
'version data.')
return client_class(**kwargs)
# NOTE(jamielennox): the url might be the auth_url or the endpoint
# depending on what was passed initially. Order is important, endpoint
# needs to override auth_url.
for url_type in ('auth_url', 'endpoint'):
if self._client_kwargs.get(url_type, False):
kwargs[url_type] = version_url
else:
kwargs[url_type] = None
return _KeystoneVersion(status=status,
version=version_number,
**kwargs)
def _available_clients(self, unstable=False, **kwargs):
"""Return a dictionary of factory functions for available API versions.
:returns: A dictionary of available API endpoints with the version
number as a tuple as the key, and a factory object that can
be used to create an appropriate client as the value.
To use the returned factory you simply call it with the parameters to
pass to keystoneclient. These parameters will override those saved in
the factory.
Example::
>>> from keystoneclient import client
>>> available_clients = client._available_clients(auth_url=url,
... **kwargs)
>>> try:
... v2_factory = available_clients[(2, 0)]
... except KeyError:
... print "Version 2.0 unavailable"
... else:
... v2_client = v2_factory(token='abcdef')
... v2_client.tenants.list()
:raises: DiscoveryFailure if the response is invalid
:raises: VersionNotAvailable if a suitable client cannot be found.
"""
versions = dict()
response_values = self.available_versions(unstable=unstable)
client_kwargs = self._get_client_constructor_kwargs(kwargs_dict=kwargs)
for version_data in response_values:
try:
v = self._get_factory_from_response_entry(version_data,
**client_kwargs)
except exceptions.DiscoveryFailure as e:
_logger.warning("Invalid entry: %s", e, exc_info=True)
else:
versions[v.version] = v
return versions
def create_client(self, version=None, **kwargs):
def create_client(self, version=None, unstable=False, **kwargs):
"""Factory function to create a new identity service client.
:param tuple version: The required version of the identity API. If
@ -418,41 +257,5 @@ class Discover(object):
:raises: DiscoveryFailure if the server response is invalid
:raises: VersionNotAvailable if a suitable client cannot be found.
"""
versions = self._available_clients(**kwargs)
chosen = None
if version:
version = _normalize_version_number(version)
for keystone_version in six.itervalues(versions):
# major versions must be the same (e.g. even though v2
# is a lower version than v3 we can't use it if v2 was
# requested)
if version[0] != keystone_version.version[0]:
continue
# prevent selecting a minor version less than what is required
if version <= keystone_version.version:
chosen = keystone_version
break
elif versions:
# if no version specified pick the latest one
chosen = max(six.iteritems(versions))[1]
if not chosen:
msg = "Could not find a suitable endpoint"
if version:
msg = "%s for client version: %s" % (msg, version)
if versions:
available = ", ".join([v._str_ver
for v in six.itervalues(versions)])
msg = "%s. Available_versions are: %s" % (msg, available)
else:
msg = "%s. No versions reported available" % msg
raise exceptions.VersionNotAvailable(msg)
return chosen.create_client()
version_data = self._calculate_version(version, unstable)
return self._create_client(version_data, **kwargs)

253
keystoneclient/tests/test_discovery.py
View File

@ -14,6 +14,7 @@ import httpretty
import six
from testtools import matchers
from keystoneclient import _discover
from keystoneclient import client
from keystoneclient import discover
from keystoneclient import exceptions
@ -464,12 +465,7 @@ class ClientDiscoveryTests(utils.TestCase):
self.assertDiscoveryFailure(auth_url=BASE_URL)
def test_discovery_ignore_invalid(self):
resp = [{'id': '3.99', # without a leading v
'links': [{'href': V3_URL, 'rel': 'self'}],
'media-types': V3_MEDIA_TYPES,
'status': 'stable',
'updated': UPDATED},
{'id': 'v3.0',
resp = [{'id': 'v3.0',
'links': [1, 2, 3, 4], # invalid links
'media-types': V3_MEDIA_TYPES,
'status': 'stable',
@ -546,16 +542,237 @@ class ClientDiscoveryTests(utils.TestCase):
self.assertEqual(client.username, 'foo')
self.assertEqual(client.password, 'bar')
def test_available_versions(self):
httpretty.register_uri(httpretty.GET, BASE_URL, status=300,
body=V3_VERSION_ENTRY)
disc = discover.Discover(auth_url=BASE_URL)
versions = disc.available_versions()
self.assertEqual(1, len(versions))
self.assertEqual(V3_VERSION, versions[0])
def test_unknown_client_version(self):
V4_VERSION = {'id': 'v4.0',
'links': [{'href': 'http://url', 'rel': 'self'}],
'media-types': V3_MEDIA_TYPES,
'status': 'stable',
'updated': UPDATED}
body = _create_version_list([V4_VERSION, V3_VERSION, V2_VERSION])
httpretty.register_uri(httpretty.GET, BASE_URL, status=300, body=body)
disc = discover.Discover(auth_url=BASE_URL)
self.assertRaises(exceptions.DiscoveryFailure,
disc.create_client, version=4)
@httpretty.activate
class DiscoverQueryTests(utils.TestCase):
def test_available_keystone_data(self):
httpretty.register_uri(httpretty.GET, BASE_URL, status=300,
body=V3_VERSION_LIST)
disc = discover.Discover(auth_url=BASE_URL)
versions = disc.version_data()
self.assertEqual((2, 0), versions[0]['version'])
self.assertEqual('stable', versions[0]['raw_status'])
self.assertEqual(V2_URL, versions[0]['url'])
self.assertEqual((3, 0), versions[1]['version'])
self.assertEqual('stable', versions[1]['raw_status'])
self.assertEqual(V3_URL, versions[1]['url'])
version = disc.data_for('v3.0')
self.assertEqual((3, 0), version['version'])
self.assertEqual('stable', version['raw_status'])
self.assertEqual(V3_URL, version['url'])
version = disc.data_for(2)
self.assertEqual((2, 0), version['version'])
self.assertEqual('stable', version['raw_status'])
self.assertEqual(V2_URL, version['url'])
self.assertIsNone(disc.url_for('v4'))
self.assertEqual(V3_URL, disc.url_for('v3'))
self.assertEqual(V2_URL, disc.url_for('v2'))
def test_available_cinder_data(self):
body = jsonutils.dumps(CINDER_EXAMPLES)
httpretty.register_uri(httpretty.GET, BASE_URL, status=300, body=body)
v1_url = "%sv1/" % BASE_URL
v2_url = "%sv2/" % BASE_URL
disc = discover.Discover(auth_url=BASE_URL)
versions = disc.version_data()
self.assertEqual((1, 0), versions[0]['version'])
self.assertEqual('CURRENT', versions[0]['raw_status'])
self.assertEqual(v1_url, versions[0]['url'])
self.assertEqual((2, 0), versions[1]['version'])
self.assertEqual('CURRENT', versions[1]['raw_status'])
self.assertEqual(v2_url, versions[1]['url'])
version = disc.data_for('v2.0')
self.assertEqual((2, 0), version['version'])
self.assertEqual('CURRENT', version['raw_status'])
self.assertEqual(v2_url, version['url'])
version = disc.data_for(1)
self.assertEqual((1, 0), version['version'])
self.assertEqual('CURRENT', version['raw_status'])
self.assertEqual(v1_url, version['url'])
self.assertIsNone(disc.url_for('v3'))
self.assertEqual(v2_url, disc.url_for('v2'))
self.assertEqual(v1_url, disc.url_for('v1'))
def test_available_glance_data(self):
body = jsonutils.dumps(GLANCE_EXAMPLES)
httpretty.register_uri(httpretty.GET, BASE_URL, status=200, body=body)
v1_url = "%sv1/" % BASE_URL
v2_url = "%sv2/" % BASE_URL
disc = discover.Discover(auth_url=BASE_URL)
versions = disc.version_data()
self.assertEqual((1, 0), versions[0]['version'])
self.assertEqual('SUPPORTED', versions[0]['raw_status'])
self.assertEqual(v1_url, versions[0]['url'])
self.assertEqual((1, 1), versions[1]['version'])
self.assertEqual('CURRENT', versions[1]['raw_status'])
self.assertEqual(v1_url, versions[1]['url'])
self.assertEqual((2, 0), versions[2]['version'])
self.assertEqual('SUPPORTED', versions[2]['raw_status'])
self.assertEqual(v2_url, versions[2]['url'])
self.assertEqual((2, 1), versions[3]['version'])
self.assertEqual('SUPPORTED', versions[3]['raw_status'])
self.assertEqual(v2_url, versions[3]['url'])
self.assertEqual((2, 2), versions[4]['version'])
self.assertEqual('CURRENT', versions[4]['raw_status'])
self.assertEqual(v2_url, versions[4]['url'])
for ver in (2, 2.1, 2.2):
version = disc.data_for(ver)
self.assertEqual((2, 2), version['version'])
self.assertEqual('CURRENT', version['raw_status'])
self.assertEqual(v2_url, version['url'])
self.assertEqual(v2_url, disc.url_for(ver))
for ver in (1, 1.1):
version = disc.data_for(ver)
self.assertEqual((1, 1), version['version'])
self.assertEqual('CURRENT', version['raw_status'])
self.assertEqual(v1_url, version['url'])
self.assertEqual(v1_url, disc.url_for(ver))
self.assertIsNone(disc.url_for('v3'))
self.assertIsNone(disc.url_for('v2.3'))
def test_allow_deprecated(self):
status = 'deprecated'
version_list = [{'id': 'v3.0',
'links': [{'href': V3_URL, 'rel': 'self'}],
'media-types': V3_MEDIA_TYPES,
'status': status,
'updated': UPDATED}]
body = jsonutils.dumps({'versions': version_list})
httpretty.register_uri(httpretty.GET, BASE_URL, status=200, body=body)
disc = discover.Discover(auth_url=BASE_URL)
# deprecated is allowed by default
versions = disc.version_data(allow_deprecated=False)
self.assertEqual(0, len(versions))
versions = disc.version_data(allow_deprecated=True)
self.assertEqual(1, len(versions))
self.assertEqual(status, versions[0]['raw_status'])
self.assertEqual(V3_URL, versions[0]['url'])
self.assertEqual((3, 0), versions[0]['version'])
def test_allow_experimental(self):
status = 'experimental'
version_list = [{'id': 'v3.0',
'links': [{'href': V3_URL, 'rel': 'self'}],
'media-types': V3_MEDIA_TYPES,
'status': status,
'updated': UPDATED}]
body = jsonutils.dumps({'versions': version_list})
httpretty.register_uri(httpretty.GET, BASE_URL, status=200, body=body)
disc = discover.Discover(auth_url=BASE_URL)
versions = disc.version_data()
self.assertEqual(0, len(versions))
versions = disc.version_data(allow_experimental=True)
self.assertEqual(1, len(versions))
self.assertEqual(status, versions[0]['raw_status'])
self.assertEqual(V3_URL, versions[0]['url'])
self.assertEqual((3, 0), versions[0]['version'])
def test_allow_unknown(self):
status = 'abcdef'
version_list = [{'id': 'v3.0',
'links': [{'href': V3_URL, 'rel': 'self'}],
'media-types': V3_MEDIA_TYPES,
'status': status,
'updated': UPDATED}]
body = jsonutils.dumps({'versions': version_list})
httpretty.register_uri(httpretty.GET, BASE_URL, status=200, body=body)
disc = discover.Discover(auth_url=BASE_URL)
versions = disc.version_data()
self.assertEqual(0, len(versions))
versions = disc.version_data(allow_unknown=True)
self.assertEqual(1, len(versions))
self.assertEqual(status, versions[0]['raw_status'])
self.assertEqual(V3_URL, versions[0]['url'])
self.assertEqual((3, 0), versions[0]['version'])
def test_ignoring_invalid_lnks(self):
version_list = [{'id': 'v3.0',
'links': [{'href': V3_URL, 'rel': 'self'}],
'media-types': V3_MEDIA_TYPES,
'status': 'stable',
'updated': UPDATED},
{'id': 'v3.1',
'media-types': V3_MEDIA_TYPES,
'status': 'stable',
'updated': UPDATED},
{'media-types': V3_MEDIA_TYPES,
'status': 'stable',
'updated': UPDATED,
'links': [{'href': V3_URL, 'rel': 'self'}],
}]
body = jsonutils.dumps({'versions': version_list})
httpretty.register_uri(httpretty.GET, BASE_URL, status=200, body=body)
disc = discover.Discover(auth_url=BASE_URL)
# raw_version_data will return all choices, even invalid ones
versions = disc.raw_version_data()
self.assertEqual(3, len(versions))
# only the version with both id and links will be actually returned
versions = disc.version_data()
self.assertEqual(1, len(versions))
class DiscoverUtils(utils.TestCase):
def test_version_number(self):
def assertVersion(inp, out):
self.assertEqual(discover._normalize_version_number(inp), out)
self.assertEqual(out, _discover.normalize_version_number(inp))
def versionRaises(inp):
self.assertRaises(TypeError,
discover._normalize_version_number,
_discover.normalize_version_number,
inp)
assertVersion('v1.2', (1, 2))
@ -571,23 +788,3 @@ class DiscoverUtils(utils.TestCase):
versionRaises('hello')
versionRaises('1.a')
versionRaises('vacuum')
def test_keystone_version_objects(self):
v31s = discover._KeystoneVersion((3, 1), 'stable')
v20s = discover._KeystoneVersion((2, 0), 'stable')
v30s = discover._KeystoneVersion((3, 0), 'stable')
v31a = discover._KeystoneVersion((3, 1), 'alpha')
v31b = discover._KeystoneVersion((3, 1), 'beta')
self.assertTrue(v31s > v30s)
self.assertTrue(v30s > v20s)
self.assertTrue(v31s > v31a)
self.assertFalse(v31s < v31a)
self.assertTrue(v31b > v31a)
self.assertTrue(v31a < v31b)
self.assertTrue(v31b > v30s)
self.assertNotEqual(v31s, v31b)
self.assertEqual(v31s, discover._KeystoneVersion((3, 1), 'stable'))