Initial copy of api tests from tempest
This change is the result of running
tools/copy_api_tests_from_tempest.sh.
Change-Id: Ica02dbe1ed26f1bc9526ea9682756ebc5877cf4a
diff --git a/neutron/tests/tempest/auth.py b/neutron/tests/tempest/auth.py
new file mode 100644
index 0000000..ceee265
--- /dev/null
+++ b/neutron/tests/tempest/auth.py
@@ -0,0 +1,636 @@
+# Copyright 2014 Hewlett-Packard Development Company, L.P.
+# 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.
+
+import abc
+import copy
+import datetime
+import exceptions
+import re
+import urlparse
+
+import six
+
+from neutron.openstack.common import log as logging
+from neutron.tests.tempest.services.identity.v2.json import token_client as json_v2id
+from neutron.tests.tempest.services.identity.v3.json import token_client as json_v3id
+
+
+LOG = logging.getLogger(__name__)
+
+
+@six.add_metaclass(abc.ABCMeta)
+class AuthProvider(object):
+ """
+ Provide authentication
+ """
+
+ def __init__(self, credentials):
+ """
+ :param credentials: credentials for authentication
+ """
+ if self.check_credentials(credentials):
+ self.credentials = credentials
+ else:
+ raise TypeError("Invalid credentials")
+ self.cache = None
+ self.alt_auth_data = None
+ self.alt_part = None
+
+ def __str__(self):
+ return "Creds :{creds}, cached auth data: {cache}".format(
+ creds=self.credentials, cache=self.cache)
+
+ @abc.abstractmethod
+ def _decorate_request(self, filters, method, url, headers=None, body=None,
+ auth_data=None):
+ """
+ Decorate request with authentication data
+ """
+ return
+
+ @abc.abstractmethod
+ def _get_auth(self):
+ return
+
+ @abc.abstractmethod
+ def _fill_credentials(self, auth_data_body):
+ return
+
+ def fill_credentials(self):
+ """
+ Fill credentials object with data from auth
+ """
+ auth_data = self.get_auth()
+ self._fill_credentials(auth_data[1])
+ return self.credentials
+
+ @classmethod
+ def check_credentials(cls, credentials):
+ """
+ Verify credentials are valid.
+ """
+ return isinstance(credentials, Credentials) and credentials.is_valid()
+
+ @property
+ def auth_data(self):
+ return self.get_auth()
+
+ @auth_data.deleter
+ def auth_data(self):
+ self.clear_auth()
+
+ def get_auth(self):
+ """
+ Returns auth from cache if available, else auth first
+ """
+ if self.cache is None or self.is_expired(self.cache):
+ self.set_auth()
+ return self.cache
+
+ def set_auth(self):
+ """
+ Forces setting auth, ignores cache if it exists.
+ Refills credentials
+ """
+ self.cache = self._get_auth()
+ self._fill_credentials(self.cache[1])
+
+ def clear_auth(self):
+ """
+ Can be called to clear the access cache so that next request
+ will fetch a new token and base_url.
+ """
+ self.cache = None
+ self.credentials.reset()
+
+ @abc.abstractmethod
+ def is_expired(self, auth_data):
+ return
+
+ def auth_request(self, method, url, headers=None, body=None, filters=None):
+ """
+ Obtains auth data and decorates a request with that.
+ :param method: HTTP method of the request
+ :param url: relative URL of the request (path)
+ :param headers: HTTP headers of the request
+ :param body: HTTP body in case of POST / PUT
+ :param filters: select a base URL out of the catalog
+ :returns a Tuple (url, headers, body)
+ """
+ orig_req = dict(url=url, headers=headers, body=body)
+
+ auth_url, auth_headers, auth_body = self._decorate_request(
+ filters, method, url, headers, body)
+ auth_req = dict(url=auth_url, headers=auth_headers, body=auth_body)
+
+ # Overwrite part if the request if it has been requested
+ if self.alt_part is not None:
+ if self.alt_auth_data is not None:
+ alt_url, alt_headers, alt_body = self._decorate_request(
+ filters, method, url, headers, body,
+ auth_data=self.alt_auth_data)
+ alt_auth_req = dict(url=alt_url, headers=alt_headers,
+ body=alt_body)
+ auth_req[self.alt_part] = alt_auth_req[self.alt_part]
+
+ else:
+ # If alt auth data is None, skip auth in the requested part
+ auth_req[self.alt_part] = orig_req[self.alt_part]
+
+ # Next auth request will be normal, unless otherwise requested
+ self.reset_alt_auth_data()
+
+ return auth_req['url'], auth_req['headers'], auth_req['body']
+
+ def reset_alt_auth_data(self):
+ """
+ Configure auth provider to provide valid authentication data
+ """
+ self.alt_part = None
+ self.alt_auth_data = None
+
+ def set_alt_auth_data(self, request_part, auth_data):
+ """
+ Configure auth provider to provide alt authentication data
+ on a part of the *next* auth_request. If credentials are None,
+ set invalid data.
+ :param request_part: request part to contain invalid auth: url,
+ headers, body
+ :param auth_data: alternative auth_data from which to get the
+ invalid data to be injected
+ """
+ self.alt_part = request_part
+ self.alt_auth_data = auth_data
+
+ @abc.abstractmethod
+ def base_url(self, filters, auth_data=None):
+ """
+ Extracts the base_url based on provided filters
+ """
+ return
+
+
+class KeystoneAuthProvider(AuthProvider):
+
+ token_expiry_threshold = datetime.timedelta(seconds=60)
+
+ def __init__(self, credentials, auth_url,
+ disable_ssl_certificate_validation=None,
+ ca_certs=None, trace_requests=None):
+ super(KeystoneAuthProvider, self).__init__(credentials)
+ self.dsvm = disable_ssl_certificate_validation
+ self.ca_certs = ca_certs
+ self.trace_requests = trace_requests
+ self.auth_client = self._auth_client(auth_url)
+
+ def _decorate_request(self, filters, method, url, headers=None, body=None,
+ auth_data=None):
+ if auth_data is None:
+ auth_data = self.auth_data
+ token, _ = auth_data
+ base_url = self.base_url(filters=filters, auth_data=auth_data)
+ # build authenticated request
+ # returns new request, it does not touch the original values
+ _headers = copy.deepcopy(headers) if headers is not None else {}
+ _headers['X-Auth-Token'] = str(token)
+ if url is None or url == "":
+ _url = base_url
+ else:
+ # Join base URL and url, and remove multiple contiguous slashes
+ _url = "/".join([base_url, url])
+ parts = [x for x in urlparse.urlparse(_url)]
+ parts[2] = re.sub("/{2,}", "/", parts[2])
+ _url = urlparse.urlunparse(parts)
+ # no change to method or body
+ return str(_url), _headers, body
+
+ @abc.abstractmethod
+ def _auth_client(self):
+ return
+
+ @abc.abstractmethod
+ def _auth_params(self):
+ return
+
+ def _get_auth(self):
+ # Bypasses the cache
+ auth_func = getattr(self.auth_client, 'get_token')
+ auth_params = self._auth_params()
+
+ # returns token, auth_data
+ token, auth_data = auth_func(**auth_params)
+ return token, auth_data
+
+ def get_token(self):
+ return self.auth_data[0]
+
+
+class KeystoneV2AuthProvider(KeystoneAuthProvider):
+
+ EXPIRY_DATE_FORMAT = '%Y-%m-%dT%H:%M:%SZ'
+
+ def _auth_client(self, auth_url):
+ return json_v2id.TokenClientJSON(
+ auth_url, disable_ssl_certificate_validation=self.dsvm,
+ ca_certs=self.ca_certs, trace_requests=self.trace_requests)
+
+ def _auth_params(self):
+ return dict(
+ user=self.credentials.username,
+ password=self.credentials.password,
+ tenant=self.credentials.tenant_name,
+ auth_data=True)
+
+ def _fill_credentials(self, auth_data_body):
+ tenant = auth_data_body['token']['tenant']
+ user = auth_data_body['user']
+ if self.credentials.tenant_name is None:
+ self.credentials.tenant_name = tenant['name']
+ if self.credentials.tenant_id is None:
+ self.credentials.tenant_id = tenant['id']
+ if self.credentials.username is None:
+ self.credentials.username = user['name']
+ if self.credentials.user_id is None:
+ self.credentials.user_id = user['id']
+
+ def base_url(self, filters, auth_data=None):
+ """
+ Filters can be:
+ - service: compute, image, etc
+ - region: the service region
+ - endpoint_type: adminURL, publicURL, internalURL
+ - api_version: replace catalog version with this
+ - skip_path: take just the base URL
+ """
+ if auth_data is None:
+ auth_data = self.auth_data
+ token, _auth_data = auth_data
+ service = filters.get('service')
+ region = filters.get('region')
+ endpoint_type = filters.get('endpoint_type', 'publicURL')
+
+ if service is None:
+ raise exceptions.EndpointNotFound("No service provided")
+
+ _base_url = None
+ for ep in _auth_data['serviceCatalog']:
+ if ep["type"] == service:
+ for _ep in ep['endpoints']:
+ if region is not None and _ep['region'] == region:
+ _base_url = _ep.get(endpoint_type)
+ if not _base_url:
+ # No region matching, use the first
+ _base_url = ep['endpoints'][0].get(endpoint_type)
+ break
+ if _base_url is None:
+ raise exceptions.EndpointNotFound(service)
+
+ parts = urlparse.urlparse(_base_url)
+ if filters.get('api_version', None) is not None:
+ path = "/" + filters['api_version']
+ noversion_path = "/".join(parts.path.split("/")[2:])
+ if noversion_path != "":
+ path += "/" + noversion_path
+ _base_url = _base_url.replace(parts.path, path)
+ if filters.get('skip_path', None) is not None and parts.path != '':
+ _base_url = _base_url.replace(parts.path, "/")
+
+ return _base_url
+
+ def is_expired(self, auth_data):
+ _, access = auth_data
+ expiry = datetime.datetime.strptime(access['token']['expires'],
+ self.EXPIRY_DATE_FORMAT)
+ return expiry - self.token_expiry_threshold <= \
+ datetime.datetime.utcnow()
+
+
+class KeystoneV3AuthProvider(KeystoneAuthProvider):
+
+ EXPIRY_DATE_FORMAT = '%Y-%m-%dT%H:%M:%S.%fZ'
+
+ def _auth_client(self, auth_url):
+ return json_v3id.V3TokenClientJSON(
+ auth_url, disable_ssl_certificate_validation=self.dsvm,
+ ca_certs=self.ca_certs, trace_requests=self.trace_requests)
+
+ def _auth_params(self):
+ return dict(
+ user=self.credentials.username,
+ password=self.credentials.password,
+ project=self.credentials.tenant_name,
+ user_domain=self.credentials.user_domain_name,
+ project_domain=self.credentials.project_domain_name,
+ auth_data=True)
+
+ def _fill_credentials(self, auth_data_body):
+ # project or domain, depending on the scope
+ project = auth_data_body.get('project', None)
+ domain = auth_data_body.get('domain', None)
+ # user is always there
+ user = auth_data_body['user']
+ # Set project fields
+ if project is not None:
+ if self.credentials.project_name is None:
+ self.credentials.project_name = project['name']
+ if self.credentials.project_id is None:
+ self.credentials.project_id = project['id']
+ if self.credentials.project_domain_id is None:
+ self.credentials.project_domain_id = project['domain']['id']
+ if self.credentials.project_domain_name is None:
+ self.credentials.project_domain_name = \
+ project['domain']['name']
+ # Set domain fields
+ if domain is not None:
+ if self.credentials.domain_id is None:
+ self.credentials.domain_id = domain['id']
+ if self.credentials.domain_name is None:
+ self.credentials.domain_name = domain['name']
+ # Set user fields
+ if self.credentials.username is None:
+ self.credentials.username = user['name']
+ if self.credentials.user_id is None:
+ self.credentials.user_id = user['id']
+ if self.credentials.user_domain_id is None:
+ self.credentials.user_domain_id = user['domain']['id']
+ if self.credentials.user_domain_name is None:
+ self.credentials.user_domain_name = user['domain']['name']
+
+ def base_url(self, filters, auth_data=None):
+ """
+ Filters can be:
+ - service: compute, image, etc
+ - region: the service region
+ - endpoint_type: adminURL, publicURL, internalURL
+ - api_version: replace catalog version with this
+ - skip_path: take just the base URL
+ """
+ if auth_data is None:
+ auth_data = self.auth_data
+ token, _auth_data = auth_data
+ service = filters.get('service')
+ region = filters.get('region')
+ endpoint_type = filters.get('endpoint_type', 'public')
+
+ if service is None:
+ raise exceptions.EndpointNotFound("No service provided")
+
+ if 'URL' in endpoint_type:
+ endpoint_type = endpoint_type.replace('URL', '')
+ _base_url = None
+ catalog = _auth_data['catalog']
+ # Select entries with matching service type
+ service_catalog = [ep for ep in catalog if ep['type'] == service]
+ if len(service_catalog) > 0:
+ service_catalog = service_catalog[0]['endpoints']
+ else:
+ # No matching service
+ raise exceptions.EndpointNotFound(service)
+ # Filter by endpoint type (interface)
+ filtered_catalog = [ep for ep in service_catalog if
+ ep['interface'] == endpoint_type]
+ if len(filtered_catalog) == 0:
+ # No matching type, keep all and try matching by region at least
+ filtered_catalog = service_catalog
+ # Filter by region
+ filtered_catalog = [ep for ep in filtered_catalog if
+ ep['region'] == region]
+ if len(filtered_catalog) == 0:
+ # No matching region, take the first endpoint
+ filtered_catalog = [service_catalog[0]]
+ # There should be only one match. If not take the first.
+ _base_url = filtered_catalog[0].get('url', None)
+ if _base_url is None:
+ raise exceptions.EndpointNotFound(service)
+
+ parts = urlparse.urlparse(_base_url)
+ if filters.get('api_version', None) is not None:
+ path = "/" + filters['api_version']
+ noversion_path = "/".join(parts.path.split("/")[2:])
+ if noversion_path != "":
+ path += "/" + noversion_path
+ _base_url = _base_url.replace(parts.path, path)
+ if filters.get('skip_path', None) is not None:
+ _base_url = _base_url.replace(parts.path, "/")
+
+ return _base_url
+
+ def is_expired(self, auth_data):
+ _, access = auth_data
+ expiry = datetime.datetime.strptime(access['expires_at'],
+ self.EXPIRY_DATE_FORMAT)
+ return expiry - self.token_expiry_threshold <= \
+ datetime.datetime.utcnow()
+
+
+def is_identity_version_supported(identity_version):
+ return identity_version in IDENTITY_VERSION
+
+
+def get_credentials(auth_url, fill_in=True, identity_version='v2', **kwargs):
+ """
+ Builds a credentials object based on the configured auth_version
+
+ :param auth_url (string): Full URI of the OpenStack Identity API(Keystone)
+ which is used to fetch the token from Identity service.
+ :param fill_in (boolean): obtain a token and fill in all credential
+ details provided by the identity service. When fill_in is not
+ specified, credentials are not validated. Validation can be invoked
+ by invoking ``is_valid()``
+ :param identity_version (string): identity API version is used to
+ select the matching auth provider and credentials class
+ :param kwargs (dict): Dict of credential key/value pairs
+
+ Examples:
+
+ Returns credentials from the provided parameters:
+ >>> get_credentials(username='foo', password='bar')
+
+ Returns credentials including IDs:
+ >>> get_credentials(username='foo', password='bar', fill_in=True)
+ """
+ if not is_identity_version_supported(identity_version):
+ raise exceptions.InvalidIdentityVersion(
+ identity_version=identity_version)
+
+ credential_class, auth_provider_class = IDENTITY_VERSION.get(
+ identity_version)
+
+ creds = credential_class(**kwargs)
+ # Fill in the credentials fields that were not specified
+ if fill_in:
+ auth_provider = auth_provider_class(creds, auth_url)
+ creds = auth_provider.fill_credentials()
+ return creds
+
+
+class Credentials(object):
+ """
+ Set of credentials for accessing OpenStack services
+
+ ATTRIBUTES: list of valid class attributes representing credentials.
+ """
+
+ ATTRIBUTES = []
+
+ def __init__(self, **kwargs):
+ """
+ Enforce the available attributes at init time (only).
+ Additional attributes can still be set afterwards if tests need
+ to do so.
+ """
+ self._initial = kwargs
+ self._apply_credentials(kwargs)
+
+ def _apply_credentials(self, attr):
+ for key in attr.keys():
+ if key in self.ATTRIBUTES:
+ setattr(self, key, attr[key])
+ else:
+ raise exceptions.InvalidCredentials
+
+ def __str__(self):
+ """
+ Represent only attributes included in self.ATTRIBUTES
+ """
+ _repr = dict((k, getattr(self, k)) for k in self.ATTRIBUTES)
+ return str(_repr)
+
+ def __eq__(self, other):
+ """
+ Credentials are equal if attributes in self.ATTRIBUTES are equal
+ """
+ return str(self) == str(other)
+
+ def __getattr__(self, key):
+ # If an attribute is set, __getattr__ is not invoked
+ # If an attribute is not set, and it is a known one, return None
+ if key in self.ATTRIBUTES:
+ return None
+ else:
+ raise AttributeError
+
+ def __delitem__(self, key):
+ # For backwards compatibility, support dict behaviour
+ if key in self.ATTRIBUTES:
+ delattr(self, key)
+ else:
+ raise AttributeError
+
+ def get(self, item, default):
+ # In this patch act as dict for backward compatibility
+ try:
+ return getattr(self, item)
+ except AttributeError:
+ return default
+
+ def get_init_attributes(self):
+ return self._initial.keys()
+
+ def is_valid(self):
+ raise NotImplementedError
+
+ def reset(self):
+ # First delete all known attributes
+ for key in self.ATTRIBUTES:
+ if getattr(self, key) is not None:
+ delattr(self, key)
+ # Then re-apply initial setup
+ self._apply_credentials(self._initial)
+
+
+class KeystoneV2Credentials(Credentials):
+
+ ATTRIBUTES = ['username', 'password', 'tenant_name', 'user_id',
+ 'tenant_id']
+
+ def is_valid(self):
+ """
+ Minimum set of valid credentials, are username and password.
+ Tenant is optional.
+ """
+ return None not in (self.username, self.password)
+
+
+class KeystoneV3Credentials(Credentials):
+ """
+ Credentials suitable for the Keystone Identity V3 API
+ """
+
+ ATTRIBUTES = ['domain_name', 'password', 'tenant_name', 'username',
+ 'project_domain_id', 'project_domain_name', 'project_id',
+ 'project_name', 'tenant_id', 'tenant_name', 'user_domain_id',
+ 'user_domain_name', 'user_id']
+
+ def __setattr__(self, key, value):
+ parent = super(KeystoneV3Credentials, self)
+ # for tenant_* set both project and tenant
+ if key == 'tenant_id':
+ parent.__setattr__('project_id', value)
+ elif key == 'tenant_name':
+ parent.__setattr__('project_name', value)
+ # for project_* set both project and tenant
+ if key == 'project_id':
+ parent.__setattr__('tenant_id', value)
+ elif key == 'project_name':
+ parent.__setattr__('tenant_name', value)
+ # for *_domain_* set both user and project if not set yet
+ if key == 'user_domain_id':
+ if self.project_domain_id is None:
+ parent.__setattr__('project_domain_id', value)
+ if key == 'project_domain_id':
+ if self.user_domain_id is None:
+ parent.__setattr__('user_domain_id', value)
+ if key == 'user_domain_name':
+ if self.project_domain_name is None:
+ parent.__setattr__('project_domain_name', value)
+ if key == 'project_domain_name':
+ if self.user_domain_name is None:
+ parent.__setattr__('user_domain_name', value)
+ # support domain_name coming from config
+ if key == 'domain_name':
+ parent.__setattr__('user_domain_name', value)
+ parent.__setattr__('project_domain_name', value)
+ # finally trigger default behaviour for all attributes
+ parent.__setattr__(key, value)
+
+ def is_valid(self):
+ """
+ Valid combinations of v3 credentials (excluding token, scope)
+ - User id, password (optional domain)
+ - User name, password and its domain id/name
+ For the scope, valid combinations are:
+ - None
+ - Project id (optional domain)
+ - Project name and its domain id/name
+ """
+ valid_user_domain = any(
+ [self.user_domain_id is not None,
+ self.user_domain_name is not None])
+ valid_project_domain = any(
+ [self.project_domain_id is not None,
+ self.project_domain_name is not None])
+ valid_user = any(
+ [self.user_id is not None,
+ self.username is not None and valid_user_domain])
+ valid_project = any(
+ [self.project_name is None and self.project_id is None,
+ self.project_id is not None,
+ self.project_name is not None and valid_project_domain])
+ return all([self.password is not None, valid_user, valid_project])
+
+
+IDENTITY_VERSION = {'v2': (KeystoneV2Credentials, KeystoneV2AuthProvider),
+ 'v3': (KeystoneV3Credentials, KeystoneV3AuthProvider)}