Merge "Introduce scope in the auth API"
This commit is contained in:
commit
a3c7945a5f
11
doc/source/library/auth.rst
Normal file
11
doc/source/library/auth.rst
Normal file
@ -0,0 +1,11 @@
|
||||
.. _auth:
|
||||
|
||||
Authentication Framework Usage
|
||||
==============================
|
||||
|
||||
---------------
|
||||
The auth module
|
||||
---------------
|
||||
|
||||
.. automodule:: tempest.lib.auth
|
||||
:members:
|
@ -0,0 +1,7 @@
|
||||
---
|
||||
features:
|
||||
- Tempest library auth interface now supports scope. Scope allows to control
|
||||
the scope of tokens requested via the identity API. Identity V2 supports
|
||||
unscoped and project scoped tokens, but only the latter are implemented.
|
||||
Identity V3 supports unscoped, project and domain scoped token, all three
|
||||
are available.
|
@ -197,14 +197,15 @@ class Manager(manager.Manager):
|
||||
}
|
||||
default_params_with_timeout_values.update(default_params)
|
||||
|
||||
def __init__(self, credentials, service=None):
|
||||
def __init__(self, credentials, service=None, scope='project'):
|
||||
"""Initialization of Manager class.
|
||||
|
||||
Setup all services clients and make them available for tests cases.
|
||||
:param credentials: type Credentials or TestResources
|
||||
:param service: Service name
|
||||
:param scope: default scope for tokens produced by the auth provider
|
||||
"""
|
||||
super(Manager, self).__init__(credentials=credentials)
|
||||
super(Manager, self).__init__(credentials=credentials, scope=scope)
|
||||
self._set_compute_clients()
|
||||
self._set_database_clients()
|
||||
self._set_identity_clients()
|
||||
|
@ -68,10 +68,16 @@ def apply_url_filters(url, filters):
|
||||
class AuthProvider(object):
|
||||
"""Provide authentication"""
|
||||
|
||||
def __init__(self, credentials):
|
||||
SCOPES = set(['project'])
|
||||
|
||||
def __init__(self, credentials, scope='project'):
|
||||
"""Auth provider __init__
|
||||
|
||||
:param credentials: credentials for authentication
|
||||
:param scope: the default scope to be used by the credential providers
|
||||
when requesting a token. Valid values depend on the
|
||||
AuthProvider class implementation, and are defined in
|
||||
the set SCOPES. Default value is 'project'.
|
||||
"""
|
||||
if self.check_credentials(credentials):
|
||||
self.credentials = credentials
|
||||
@ -88,6 +94,8 @@ class AuthProvider(object):
|
||||
raise TypeError("credentials object is of type %s, which is"
|
||||
" not a valid Credentials object type." %
|
||||
credentials.__class__.__name__)
|
||||
self._scope = None
|
||||
self.scope = scope
|
||||
self.cache = None
|
||||
self.alt_auth_data = None
|
||||
self.alt_part = None
|
||||
@ -123,8 +131,14 @@ class AuthProvider(object):
|
||||
|
||||
@property
|
||||
def auth_data(self):
|
||||
"""Auth data for set scope"""
|
||||
return self.get_auth()
|
||||
|
||||
@property
|
||||
def scope(self):
|
||||
"""Scope used in auth requests"""
|
||||
return self._scope
|
||||
|
||||
@auth_data.deleter
|
||||
def auth_data(self):
|
||||
self.clear_auth()
|
||||
@ -139,7 +153,7 @@ class AuthProvider(object):
|
||||
"""Forces setting auth.
|
||||
|
||||
Forces setting auth, ignores cache if it exists.
|
||||
Refills credentials
|
||||
Refills credentials.
|
||||
"""
|
||||
self.cache = self._get_auth()
|
||||
self._fill_credentials(self.cache[1])
|
||||
@ -222,6 +236,19 @@ class AuthProvider(object):
|
||||
"""Extracts the base_url based on provided filters"""
|
||||
return
|
||||
|
||||
@scope.setter
|
||||
def scope(self, value):
|
||||
"""Set the scope to be used in token requests
|
||||
|
||||
:param scope: scope to be used. If the scope is different, clear caches
|
||||
"""
|
||||
if value not in self.SCOPES:
|
||||
raise exceptions.InvalidScope(
|
||||
scope=value, auth_provider=self.__class__.__name__)
|
||||
if value != self.scope:
|
||||
self.clear_auth()
|
||||
self._scope = value
|
||||
|
||||
|
||||
class KeystoneAuthProvider(AuthProvider):
|
||||
|
||||
@ -231,17 +258,18 @@ class KeystoneAuthProvider(AuthProvider):
|
||||
|
||||
def __init__(self, credentials, auth_url,
|
||||
disable_ssl_certificate_validation=None,
|
||||
ca_certs=None, trace_requests=None):
|
||||
super(KeystoneAuthProvider, self).__init__(credentials)
|
||||
ca_certs=None, trace_requests=None, scope='project'):
|
||||
super(KeystoneAuthProvider, self).__init__(credentials, scope)
|
||||
self.dsvm = disable_ssl_certificate_validation
|
||||
self.ca_certs = ca_certs
|
||||
self.trace_requests = trace_requests
|
||||
self.auth_url = auth_url
|
||||
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
|
||||
auth_data = self.get_auth()
|
||||
token, _ = auth_data
|
||||
base_url = self.base_url(filters=filters, auth_data=auth_data)
|
||||
# build authenticated request
|
||||
@ -265,6 +293,11 @@ class KeystoneAuthProvider(AuthProvider):
|
||||
|
||||
@abc.abstractmethod
|
||||
def _auth_params(self):
|
||||
"""Auth parameters to be passed to the token request
|
||||
|
||||
By default all fields available in Credentials are passed to the
|
||||
token request. Scope may affect this.
|
||||
"""
|
||||
return
|
||||
|
||||
def _get_auth(self):
|
||||
@ -292,10 +325,17 @@ class KeystoneAuthProvider(AuthProvider):
|
||||
return expiry
|
||||
|
||||
def get_token(self):
|
||||
return self.auth_data[0]
|
||||
return self.get_auth()[0]
|
||||
|
||||
|
||||
class KeystoneV2AuthProvider(KeystoneAuthProvider):
|
||||
"""Provides authentication based on the Identity V2 API
|
||||
|
||||
The Keystone Identity V2 API defines both unscoped and project scoped
|
||||
tokens. This auth provider only implements 'project'.
|
||||
"""
|
||||
|
||||
SCOPES = set(['project'])
|
||||
|
||||
def _auth_client(self, auth_url):
|
||||
return json_v2id.TokenClient(
|
||||
@ -303,6 +343,10 @@ class KeystoneV2AuthProvider(KeystoneAuthProvider):
|
||||
ca_certs=self.ca_certs, trace_requests=self.trace_requests)
|
||||
|
||||
def _auth_params(self):
|
||||
"""Auth parameters to be passed to the token request
|
||||
|
||||
All fields available in Credentials are passed to the token request.
|
||||
"""
|
||||
return dict(
|
||||
user=self.credentials.username,
|
||||
password=self.credentials.password,
|
||||
@ -332,7 +376,7 @@ class KeystoneV2AuthProvider(KeystoneAuthProvider):
|
||||
- skip_path: take just the base URL
|
||||
"""
|
||||
if auth_data is None:
|
||||
auth_data = self.auth_data
|
||||
auth_data = self.get_auth()
|
||||
token, _auth_data = auth_data
|
||||
service = filters.get('service')
|
||||
region = filters.get('region')
|
||||
@ -365,6 +409,9 @@ class KeystoneV2AuthProvider(KeystoneAuthProvider):
|
||||
|
||||
|
||||
class KeystoneV3AuthProvider(KeystoneAuthProvider):
|
||||
"""Provides authentication based on the Identity V3 API"""
|
||||
|
||||
SCOPES = set(['project', 'domain', 'unscoped', None])
|
||||
|
||||
def _auth_client(self, auth_url):
|
||||
return json_v3id.V3TokenClient(
|
||||
@ -372,20 +419,36 @@ class KeystoneV3AuthProvider(KeystoneAuthProvider):
|
||||
ca_certs=self.ca_certs, trace_requests=self.trace_requests)
|
||||
|
||||
def _auth_params(self):
|
||||
return dict(
|
||||
"""Auth parameters to be passed to the token request
|
||||
|
||||
Fields available in Credentials are passed to the token request,
|
||||
depending on the value of scope. Valid values for scope are: "project",
|
||||
"domain". Any other string (e.g. "unscoped") or None will lead to an
|
||||
unscoped token request.
|
||||
"""
|
||||
|
||||
auth_params = dict(
|
||||
user_id=self.credentials.user_id,
|
||||
username=self.credentials.username,
|
||||
password=self.credentials.password,
|
||||
project_id=self.credentials.project_id,
|
||||
project_name=self.credentials.project_name,
|
||||
user_domain_id=self.credentials.user_domain_id,
|
||||
user_domain_name=self.credentials.user_domain_name,
|
||||
project_domain_id=self.credentials.project_domain_id,
|
||||
project_domain_name=self.credentials.project_domain_name,
|
||||
domain_id=self.credentials.domain_id,
|
||||
domain_name=self.credentials.domain_name,
|
||||
password=self.credentials.password,
|
||||
auth_data=True)
|
||||
|
||||
if self.scope == 'project':
|
||||
auth_params.update(
|
||||
project_domain_id=self.credentials.project_domain_id,
|
||||
project_domain_name=self.credentials.project_domain_name,
|
||||
project_id=self.credentials.project_id,
|
||||
project_name=self.credentials.project_name)
|
||||
|
||||
if self.scope == 'domain':
|
||||
auth_params.update(
|
||||
domain_id=self.credentials.domain_id,
|
||||
domain_name=self.credentials.domain_name)
|
||||
|
||||
return auth_params
|
||||
|
||||
def _fill_credentials(self, auth_data_body):
|
||||
# project or domain, depending on the scope
|
||||
project = auth_data_body.get('project', None)
|
||||
@ -422,6 +485,10 @@ class KeystoneV3AuthProvider(KeystoneAuthProvider):
|
||||
def base_url(self, filters, auth_data=None):
|
||||
"""Base URL from catalog
|
||||
|
||||
If scope is not 'project', it may be that there is not catalog in
|
||||
the auth_data. In such case, as long as the requested service is
|
||||
'identity', we can use the original auth URL to build the base_url.
|
||||
|
||||
Filters can be:
|
||||
- service: compute, image, etc
|
||||
- region: the service region
|
||||
@ -430,7 +497,7 @@ class KeystoneV3AuthProvider(KeystoneAuthProvider):
|
||||
- skip_path: take just the base URL
|
||||
"""
|
||||
if auth_data is None:
|
||||
auth_data = self.auth_data
|
||||
auth_data = self.get_auth()
|
||||
token, _auth_data = auth_data
|
||||
service = filters.get('service')
|
||||
region = filters.get('region')
|
||||
@ -442,14 +509,20 @@ class KeystoneV3AuthProvider(KeystoneAuthProvider):
|
||||
if 'URL' in endpoint_type:
|
||||
endpoint_type = endpoint_type.replace('URL', '')
|
||||
_base_url = None
|
||||
catalog = _auth_data['catalog']
|
||||
catalog = _auth_data.get('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)
|
||||
if len(catalog) == 0 and service == 'identity':
|
||||
# NOTE(andreaf) If there's no catalog at all and the service
|
||||
# is identity, it's a valid use case. Having a non-empty
|
||||
# catalog with no identity in it is not valid instead.
|
||||
return apply_url_filters(self.auth_url, filters)
|
||||
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]
|
||||
@ -465,7 +538,7 @@ class KeystoneV3AuthProvider(KeystoneAuthProvider):
|
||||
# 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)
|
||||
raise exceptions.EndpointNotFound(service)
|
||||
return apply_url_filters(_base_url, filters)
|
||||
|
||||
def is_expired(self, auth_data):
|
||||
@ -669,7 +742,7 @@ class KeystoneV3Credentials(Credentials):
|
||||
def is_valid(self):
|
||||
"""Check of credentials (no API call)
|
||||
|
||||
Valid combinations of v3 credentials (excluding token, scope)
|
||||
Valid combinations of v3 credentials (excluding token)
|
||||
- User id, password (optional domain)
|
||||
- User name, password and its domain id/name
|
||||
For the scope, valid combinations are:
|
||||
|
@ -207,6 +207,10 @@ class InvalidCredentials(TempestException):
|
||||
message = "Invalid Credentials"
|
||||
|
||||
|
||||
class InvalidScope(TempestException):
|
||||
message = "Invalid Scope %(scope)s for %(auth_provider)s"
|
||||
|
||||
|
||||
class SSHTimeout(TempestException):
|
||||
message = ("Connection to the %(host)s via SSH timed out.\n"
|
||||
"User: %(user)s, Password: %(password)s")
|
||||
|
@ -28,13 +28,14 @@ class Manager(object):
|
||||
and a client object for a test case to use in performing actions.
|
||||
"""
|
||||
|
||||
def __init__(self, credentials):
|
||||
def __init__(self, credentials, scope='project'):
|
||||
"""Initialization of base manager class
|
||||
|
||||
Credentials to be used within the various client classes managed by the
|
||||
Manager object must be defined.
|
||||
|
||||
:param credentials: type Credentials or TestResources
|
||||
:param scope: default scope for tokens produced by the auth provider
|
||||
"""
|
||||
self.credentials = credentials
|
||||
# Check if passed or default credentials are valid
|
||||
@ -48,7 +49,8 @@ class Manager(object):
|
||||
else:
|
||||
creds = self.credentials
|
||||
# Creates an auth provider for the credentials
|
||||
self.auth_provider = get_auth_provider(creds, pre_auth=True)
|
||||
self.auth_provider = get_auth_provider(creds, pre_auth=True,
|
||||
scope=scope)
|
||||
|
||||
|
||||
def get_auth_provider_class(credentials):
|
||||
@ -58,7 +60,7 @@ def get_auth_provider_class(credentials):
|
||||
return auth.KeystoneV2AuthProvider, CONF.identity.uri
|
||||
|
||||
|
||||
def get_auth_provider(credentials, pre_auth=False):
|
||||
def get_auth_provider(credentials, pre_auth=False, scope='project'):
|
||||
default_params = {
|
||||
'disable_ssl_certificate_validation':
|
||||
CONF.identity.disable_ssl_certificate_validation,
|
||||
@ -71,6 +73,7 @@ def get_auth_provider(credentials, pre_auth=False):
|
||||
auth_provider_class, auth_url = get_auth_provider_class(
|
||||
credentials)
|
||||
_auth_provider = auth_provider_class(credentials, auth_url,
|
||||
scope=scope,
|
||||
**default_params)
|
||||
if pre_auth:
|
||||
_auth_provider.set_auth()
|
||||
|
@ -57,3 +57,18 @@ class FakeKeystoneV3DomainCredentials(auth.KeystoneV3Credentials):
|
||||
user_domain_name='fake_domain_name'
|
||||
)
|
||||
super(FakeKeystoneV3DomainCredentials, self).__init__(**creds)
|
||||
|
||||
|
||||
class FakeKeystoneV3AllCredentials(auth.KeystoneV3Credentials):
|
||||
"""Fake credentials for the Keystone Identity V3 API, with no scope"""
|
||||
|
||||
def __init__(self):
|
||||
creds = dict(
|
||||
username='fake_username',
|
||||
password='fake_password',
|
||||
user_domain_name='fake_domain_name',
|
||||
project_name='fake_tenant_name',
|
||||
project_domain_name='fake_domain_name',
|
||||
domain_name='fake_domain_name'
|
||||
)
|
||||
super(FakeKeystoneV3AllCredentials, self).__init__(**creds)
|
||||
|
@ -133,6 +133,49 @@ IDENTITY_V3_RESPONSE = {
|
||||
}
|
||||
}
|
||||
|
||||
IDENTITY_V3_RESPONSE_DOMAIN_SCOPE = {
|
||||
"token": {
|
||||
"methods": [
|
||||
"token",
|
||||
"password"
|
||||
],
|
||||
"expires_at": "2020-01-01T00:00:10.000123Z",
|
||||
"domain": {
|
||||
"id": "fake_domain_id",
|
||||
"name": "domain_name"
|
||||
},
|
||||
"user": {
|
||||
"domain": {
|
||||
"id": "fake_domain_id",
|
||||
"name": "domain_name"
|
||||
},
|
||||
"id": "fake_user_id",
|
||||
"name": "username"
|
||||
},
|
||||
"issued_at": "2013-05-29T16:55:21.468960Z",
|
||||
"catalog": CATALOG_V3
|
||||
}
|
||||
}
|
||||
|
||||
IDENTITY_V3_RESPONSE_NO_SCOPE = {
|
||||
"token": {
|
||||
"methods": [
|
||||
"token",
|
||||
"password"
|
||||
],
|
||||
"expires_at": "2020-01-01T00:00:10.000123Z",
|
||||
"user": {
|
||||
"domain": {
|
||||
"id": "fake_domain_id",
|
||||
"name": "domain_name"
|
||||
},
|
||||
"id": "fake_user_id",
|
||||
"name": "username"
|
||||
},
|
||||
"issued_at": "2013-05-29T16:55:21.468960Z",
|
||||
}
|
||||
}
|
||||
|
||||
ALT_IDENTITY_V3 = IDENTITY_V3_RESPONSE
|
||||
|
||||
|
||||
@ -145,6 +188,28 @@ def _fake_v3_response(self, uri, method="GET", body=None, headers=None,
|
||||
json.dumps(IDENTITY_V3_RESPONSE))
|
||||
|
||||
|
||||
def _fake_v3_response_domain_scope(self, uri, method="GET", body=None,
|
||||
headers=None, redirections=5,
|
||||
connection_type=None):
|
||||
fake_headers = {
|
||||
"status": "201",
|
||||
"x-subject-token": TOKEN
|
||||
}
|
||||
return (fake_http.fake_http_response(fake_headers, status=201),
|
||||
json.dumps(IDENTITY_V3_RESPONSE_DOMAIN_SCOPE))
|
||||
|
||||
|
||||
def _fake_v3_response_no_scope(self, uri, method="GET", body=None,
|
||||
headers=None, redirections=5,
|
||||
connection_type=None):
|
||||
fake_headers = {
|
||||
"status": "201",
|
||||
"x-subject-token": TOKEN
|
||||
}
|
||||
return (fake_http.fake_http_response(fake_headers, status=201),
|
||||
json.dumps(IDENTITY_V3_RESPONSE_NO_SCOPE))
|
||||
|
||||
|
||||
def _fake_v2_response(self, uri, method="GET", body=None, headers=None,
|
||||
redirections=5, connection_type=None):
|
||||
return (fake_http.fake_http_response({}, status=200),
|
||||
|
@ -15,6 +15,7 @@
|
||||
|
||||
import copy
|
||||
import datetime
|
||||
import testtools
|
||||
|
||||
from oslotest import mockpatch
|
||||
|
||||
@ -425,6 +426,16 @@ class TestKeystoneV2AuthProvider(BaseAuthTestsSetUp):
|
||||
self.assertEqual(self.auth_provider.is_expired(auth_data),
|
||||
should_be_expired)
|
||||
|
||||
def test_set_scope_all_valid(self):
|
||||
for scope in self.auth_provider.SCOPES:
|
||||
self.auth_provider.scope = scope
|
||||
self.assertEqual(scope, self.auth_provider.scope)
|
||||
|
||||
def test_set_scope_invalid(self):
|
||||
with testtools.ExpectedException(exceptions.InvalidScope,
|
||||
'.* invalid_scope .*'):
|
||||
self.auth_provider.scope = 'invalid_scope'
|
||||
|
||||
|
||||
class TestKeystoneV3AuthProvider(TestKeystoneV2AuthProvider):
|
||||
_endpoints = fake_identity.IDENTITY_V3_RESPONSE['token']['catalog']
|
||||
@ -529,6 +540,98 @@ class TestKeystoneV3AuthProvider(TestKeystoneV2AuthProvider):
|
||||
expected = 'http://fake_url/v3'
|
||||
self._test_base_url_helper(expected, filters, ('token', auth_data))
|
||||
|
||||
# Base URL test with scope only for V3
|
||||
def test_base_url_scope_project(self):
|
||||
self.auth_provider.scope = 'project'
|
||||
self.filters = {
|
||||
'service': 'compute',
|
||||
'endpoint_type': 'publicURL',
|
||||
'region': 'FakeRegion'
|
||||
}
|
||||
expected = self._get_result_url_from_endpoint(
|
||||
self._endpoints[0]['endpoints'][1])
|
||||
self._test_base_url_helper(expected, self.filters)
|
||||
|
||||
# Base URL test with scope only for V3
|
||||
def test_base_url_unscoped_identity(self):
|
||||
self.auth_provider.scope = 'unscoped'
|
||||
self.patchobject(v3_client.V3TokenClient, 'raw_request',
|
||||
fake_identity._fake_v3_response_no_scope)
|
||||
self.filters = {
|
||||
'service': 'identity',
|
||||
'endpoint_type': 'publicURL',
|
||||
'region': 'FakeRegion'
|
||||
}
|
||||
expected = fake_identity.FAKE_AUTH_URL
|
||||
self._test_base_url_helper(expected, self.filters)
|
||||
|
||||
# Base URL test with scope only for V3
|
||||
def test_base_url_unscoped_other(self):
|
||||
self.auth_provider.scope = 'unscoped'
|
||||
self.patchobject(v3_client.V3TokenClient, 'raw_request',
|
||||
fake_identity._fake_v3_response_no_scope)
|
||||
self.filters = {
|
||||
'service': 'compute',
|
||||
'endpoint_type': 'publicURL',
|
||||
'region': 'FakeRegion'
|
||||
}
|
||||
self.assertRaises(exceptions.EndpointNotFound,
|
||||
self.auth_provider.base_url,
|
||||
auth_data=self.auth_provider.auth_data,
|
||||
filters=self.filters)
|
||||
|
||||
def test_auth_parameters_with_scope_unset(self):
|
||||
# No scope defaults to 'project'
|
||||
all_creds = fake_credentials.FakeKeystoneV3AllCredentials()
|
||||
self.auth_provider.credentials = all_creds
|
||||
auth_params = self.auth_provider._auth_params()
|
||||
self.assertNotIn('scope', auth_params.keys())
|
||||
for attr in all_creds.get_init_attributes():
|
||||
if attr.startswith('domain_'):
|
||||
self.assertNotIn(attr, auth_params.keys())
|
||||
else:
|
||||
self.assertIn(attr, auth_params.keys())
|
||||
self.assertEqual(getattr(all_creds, attr), auth_params[attr])
|
||||
|
||||
def test_auth_parameters_with_project_scope(self):
|
||||
all_creds = fake_credentials.FakeKeystoneV3AllCredentials()
|
||||
self.auth_provider.credentials = all_creds
|
||||
self.auth_provider.scope = 'project'
|
||||
auth_params = self.auth_provider._auth_params()
|
||||
self.assertNotIn('scope', auth_params.keys())
|
||||
for attr in all_creds.get_init_attributes():
|
||||
if attr.startswith('domain_'):
|
||||
self.assertNotIn(attr, auth_params.keys())
|
||||
else:
|
||||
self.assertIn(attr, auth_params.keys())
|
||||
self.assertEqual(getattr(all_creds, attr), auth_params[attr])
|
||||
|
||||
def test_auth_parameters_with_domain_scope(self):
|
||||
all_creds = fake_credentials.FakeKeystoneV3AllCredentials()
|
||||
self.auth_provider.credentials = all_creds
|
||||
self.auth_provider.scope = 'domain'
|
||||
auth_params = self.auth_provider._auth_params()
|
||||
self.assertNotIn('scope', auth_params.keys())
|
||||
for attr in all_creds.get_init_attributes():
|
||||
if attr.startswith('project_'):
|
||||
self.assertNotIn(attr, auth_params.keys())
|
||||
else:
|
||||
self.assertIn(attr, auth_params.keys())
|
||||
self.assertEqual(getattr(all_creds, attr), auth_params[attr])
|
||||
|
||||
def test_auth_parameters_unscoped(self):
|
||||
all_creds = fake_credentials.FakeKeystoneV3AllCredentials()
|
||||
self.auth_provider.credentials = all_creds
|
||||
self.auth_provider.scope = 'unscoped'
|
||||
auth_params = self.auth_provider._auth_params()
|
||||
self.assertNotIn('scope', auth_params.keys())
|
||||
for attr in all_creds.get_init_attributes():
|
||||
if attr.startswith('project_') or attr.startswith('domain_'):
|
||||
self.assertNotIn(attr, auth_params.keys())
|
||||
else:
|
||||
self.assertIn(attr, auth_params.keys())
|
||||
self.assertEqual(getattr(all_creds, attr), auth_params[attr])
|
||||
|
||||
|
||||
class TestKeystoneV3Credentials(base.TestCase):
|
||||
def testSetAttrUserDomain(self):
|
||||
@ -630,3 +733,20 @@ class TestReplaceVersion(base.TestCase):
|
||||
self.assertEqual(
|
||||
'http://localhost/identity/v2.0/uuid/',
|
||||
auth.replace_version('http://localhost/identity/v3/uuid/', 'v2.0'))
|
||||
|
||||
|
||||
class TestKeystoneV3AuthProvider_DomainScope(BaseAuthTestsSetUp):
|
||||
_endpoints = fake_identity.IDENTITY_V3_RESPONSE['token']['catalog']
|
||||
_auth_provider_class = auth.KeystoneV3AuthProvider
|
||||
credentials = fake_credentials.FakeKeystoneV3Credentials()
|
||||
|
||||
def setUp(self):
|
||||
super(TestKeystoneV3AuthProvider_DomainScope, self).setUp()
|
||||
self.patchobject(v3_client.V3TokenClient, 'raw_request',
|
||||
fake_identity._fake_v3_response_domain_scope)
|
||||
|
||||
def test_get_auth_with_domain_scope(self):
|
||||
self.auth_provider.scope = 'domain'
|
||||
_, auth_data = self.auth_provider.get_auth()
|
||||
self.assertIn('domain', auth_data)
|
||||
self.assertNotIn('project', auth_data)
|
||||
|
Loading…
Reference in New Issue
Block a user