openstack
/
tempest
Code Issues Proposed changes

Factor up config dependent credential classes 

Separate helpers methods to build credentials, credential providers and
client managers that depend on configuration. They are all moved to
the common.credentials module to achieve two objectives:
- prepare credential provider module to move to tempest-lib
- avoid circular dependencies

There are still dependencies to CONF in the credential providers,
removing all of them would make the patch too large; but there is
with this patch a single place where all relevant CONF parameters can
be looked up and passed into the credential providers.

Partially-implements: bp tempest-library
Change-Id: I351f86f0570070105bbf24f2d904e917feea6e57
This commit is contained in:
Andrea Frittoli (andreaf) 2015-10-08 10:25:02 +01:00
parent a1edb75d79
commit 290b3e1383
25 changed files with 429 additions and 347 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
doc/source/configuration.rst
View File

@ -127,16 +127,16 @@ to the tests using the credentials, and failure to do this will likely cause
unexpected failures in some tests.
Non-locking test accounts (aka credentials config options)
""""""""""""""""""""""""""""""""""""""""""""""""""""""""""
Legacy test accounts (aka credentials config options)
"""""""""""""""""""""""""""""""""""""""""""""""""""""
**Starting in the Liberty release this mechanism was deprecated and will be
removed in a future release**
When Tempest was refactored to allow for locking test accounts, the original
non-tenant isolated case was converted to internally work similarly to the
accounts.yaml file. This mechanism was then called the non-locking test accounts
provider. To use the non-locking test accounts provider you can specify the sets
of credentials in the configuration file like detailed above with following 9
accounts.yaml file. This mechanism was then called the legacy test accounts
provider. To use the legacy test accounts provider you can specify the sets of
credentials in the configuration file like detailed above with following 9
options in the identity section:
#. username

4
tempest/api/identity/admin/v3/test_trusts.py
View File

@ -18,7 +18,7 @@ from tempest_lib import exceptions as lib_exc
from tempest.api.identity import base
from tempest import clients
from tempest.common import cred_provider
from tempest.common import credentials_factory as common_creds
from tempest.common.utils import data_utils
from tempest import config
from tempest import test
@ -89,7 +89,7 @@ class BaseTrustsV3Test(base.BaseIdentityV3AdminTest):
self.assertIsNotNone(self.trustee_user_id)
# Initialize a new client with the trustor credentials
creds = cred_provider.get_credentials(
creds = common_creds.get_credentials(
identity_version='v3',
username=self.trustor_username,
password=self.trustor_password,

12
tempest/api/identity/base.py
View File

@ -16,7 +16,7 @@
from oslo_log import log as logging
from tempest_lib import exceptions as lib_exc
from tempest.common import cred_provider
from tempest.common import credentials_factory as common_creds
from tempest.common.utils import data_utils
from tempest import config
import tempest.test
@ -195,11 +195,11 @@ class DataGenerator(object):
@property
def test_credentials(self):
return cred_provider.get_credentials(username=self.test_user,
user_id=self.user['id'],
password=self.test_password,
tenant_name=self.test_tenant,
tenant_id=self.tenant['id'])
return common_creds.get_credentials(username=self.test_user,
user_id=self.user['id'],
password=self.test_password,
tenant_name=self.test_tenant,
tenant_id=self.tenant['id'])
def setup_test_user(self):
"""Set up a test user."""

4
tempest/api/orchestration/stacks/test_neutron_resources.py
View File

@ -16,7 +16,7 @@ import logging
import netaddr
from tempest.api.orchestration import base
from tempest import clients
from tempest.common import credentials_factory as credentials
from tempest.common.utils import data_utils
from tempest import config
from tempest import exceptions
@ -38,7 +38,7 @@ class NeutronResourcesTestJSON(base.BaseOrchestrationTest):
@classmethod
def setup_credentials(cls):
super(NeutronResourcesTestJSON, cls).setup_credentials()
cls.os = clients.Manager()
cls.os = credentials.ConfiguredUserManager()
@classmethod
def setup_clients(cls):

17
tempest/clients.py
View File

@ -38,7 +38,6 @@ from tempest_lib.services.compute.hypervisor_client import \
from tempest_lib.services.identity.v2.token_client import TokenClient
from tempest_lib.services.identity.v3.token_client import V3TokenClient
from tempest.common import cred_provider
from tempest.common import negative_rest_client
from tempest import config
from tempest import exceptions
@ -174,7 +173,7 @@ class Manager(manager.Manager):
}
default_params_with_timeout_values.update(default_params)
def __init__(self, credentials=None, service=None):
def __init__(self, credentials, service=None):
super(Manager, self).__init__(credentials=credentials)
self._set_compute_clients()
@ -475,17 +474,3 @@ class Manager(manager.Manager):
self.account_client = AccountClient(self.auth_provider, **params)
self.container_client = ContainerClient(self.auth_provider, **params)
self.object_client = ObjectClient(self.auth_provider, **params)
class AdminManager(Manager):
"""
Manager object that uses the admin credentials for its
managed client objects
"""
def __init__(self, service=None):
super(AdminManager, self).__init__(
credentials=cred_provider.get_configured_credentials(
'identity_admin'),
service=service)

8
tempest/cmd/cleanup.py
View File

@ -58,7 +58,7 @@ from oslo_serialization import jsonutils as json
from tempest import clients
from tempest.cmd import cleanup_service
from tempest.common import cred_provider
from tempest.common import credentials_factory as credentials
from tempest import config
SAVED_STATE_JSON = "saved_state.json"
@ -75,7 +75,7 @@ class TempestCleanup(command.Command):
def take_action(self, parsed_args):
cleanup_service.init_conf()
self.options = parsed_args
self.admin_mgr = clients.AdminManager()
self.admin_mgr = credentials.AdminManager()
self.dry_run_data = {}
self.json_data = {}
@ -162,7 +162,7 @@ class TempestCleanup(command.Command):
kwargs = {"username": CONF.auth.admin_username,
"password": CONF.auth.admin_password,
"tenant_name": tenant['name']}
mgr = clients.Manager(credentials=cred_provider.get_credentials(
mgr = clients.Manager(credentials=credentials.get_credentials(
**kwargs))
kwargs = {'data': tenant_data,
'is_dry_run': is_dry_run,
@ -235,7 +235,7 @@ class TempestCleanup(command.Command):
LOG.debug("Remove admin user role for tenant: %s" % tenant_id)
# Must initialize AdminManager for each user role
# Otherwise authentication exception is thrown, weird
id_cl = clients.AdminManager().identity_client
id_cl = credentials.AdminManager().identity_client
if (self._tenant_exists(tenant_id)):
try:
id_cl.remove_user_role(tenant_id, self.admin_id,

4
tempest/cmd/cleanup_service.py
View File

@ -16,7 +16,7 @@
from oslo_log import log as logging
from tempest import clients
from tempest.common import credentials_factory as credentials
from tempest import config
from tempest import test
@ -82,7 +82,7 @@ def init_conf():
def _get_network_id(net_name, tenant_name):
am = clients.AdminManager()
am = credentials.AdminManager()
net_cl = am.networks_client
id_cl = am.identity_client

2
tempest/cmd/verify_tempest_config.py
View File

@ -24,7 +24,7 @@ from six import moves
from six.moves.urllib import parse as urlparse
from tempest import clients
from tempest.common import credentials
from tempest.common import credentials_factory as credentials
from tempest import config

73
tempest/common/cred_provider.py
View File

@ -18,83 +18,10 @@ from oslo_log import log as logging
import six
from tempest_lib import auth
from tempest import config
from tempest import exceptions
CONF = config.CONF
LOG = logging.getLogger(__name__)
# Type of credentials available from configuration
CREDENTIAL_TYPES = {
'identity_admin': ('auth', 'admin'),
'user': ('identity', None),
'alt_user': ('identity', 'alt')
}
DEFAULT_PARAMS = {
'disable_ssl_certificate_validation':
CONF.identity.disable_ssl_certificate_validation,
'ca_certs': CONF.identity.ca_certificates_file,
'trace_requests': CONF.debug.trace_requests
}
# Read credentials from configuration, builds a Credentials object
# based on the specified or configured version
def get_configured_credentials(credential_type, fill_in=True,
identity_version=None):
identity_version = identity_version or CONF.identity.auth_version
if identity_version not in ('v2', 'v3'):
raise exceptions.InvalidConfiguration(
'Unsupported auth version: %s' % identity_version)
if credential_type not in CREDENTIAL_TYPES:
raise exceptions.InvalidCredentials()
conf_attributes = ['username', 'password', 'tenant_name']
if identity_version == 'v3':
conf_attributes.append('domain_name')
# Read the parts of credentials from config
params = DEFAULT_PARAMS.copy()
section, prefix = CREDENTIAL_TYPES[credential_type]
for attr in conf_attributes:
_section = getattr(CONF, section)
if prefix is None:
params[attr] = getattr(_section, attr)
else:
params[attr] = getattr(_section, prefix + "_" + attr)
# Build and validate credentials. We are reading configured credentials,
# so validate them even if fill_in is False
credentials = get_credentials(fill_in=fill_in,
identity_version=identity_version, **params)
if not fill_in:
if not credentials.is_valid():
msg = ("The %s credentials are incorrectly set in the config file."
" Double check that all required values are assigned" %
credential_type)
raise exceptions.InvalidConfiguration(msg)
return credentials
# Wrapper around auth.get_credentials to use the configured identity version
# is none is specified
def get_credentials(fill_in=True, identity_version=None, **kwargs):
params = dict(DEFAULT_PARAMS, **kwargs)
identity_version = identity_version or CONF.identity.auth_version
# In case of "v3" add the domain from config if not specified
if identity_version == 'v3':
domain_fields = set(x for x in auth.KeystoneV3Credentials.ATTRIBUTES
if 'domain' in x)
if not domain_fields.intersection(kwargs.keys()):
domain_name = CONF.auth.default_credentials_domain_name
params['user_domain_name'] = domain_name
auth_url = CONF.identity.uri_v3
else:
auth_url = CONF.identity.uri
return auth.get_credentials(auth_url,
fill_in=fill_in,
identity_version=identity_version,
**params)
@six.add_metaclass(abc.ABCMeta)
class CredentialProvider(object):

110
tempest/common/credentials.py
View File

@ -1,110 +0,0 @@
# Copyright (c) 2014 Hewlett-Packard Development Company, L.P.
# 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 os
from tempest.common import cred_provider
from tempest.common import dynamic_creds
from tempest.common import preprov_creds
from tempest import config
from tempest import exceptions
CONF = config.CONF
# Return the right implementation of CredentialProvider based on config
# Dropping interface and password, as they are never used anyways
# TODO(andreaf) Drop them from the CredentialsProvider interface completely
def get_credentials_provider(name, network_resources=None,
force_tenant_isolation=False,
identity_version=None):
# If a test requires a new account to work, it can have it via forcing
# dynamic credentials. A new account will be produced only for that test.
# In case admin credentials are not available for the account creation,
# the test should be skipped else it would fail.
identity_version = identity_version or CONF.identity.auth_version
if CONF.auth.use_dynamic_credentials or force_tenant_isolation:
return dynamic_creds.DynamicCredentialProvider(
name=name,
network_resources=network_resources,
identity_version=identity_version,
credentials_domain=CONF.auth.default_credentials_domain_name,
admin_role=CONF.identity.admin_role)
else:
if (CONF.auth.test_accounts_file and
os.path.isfile(CONF.auth.test_accounts_file)):
# Most params are not relevant for pre-created accounts
return preprov_creds.PreProvisionedCredentialProvider(
name=name, identity_version=identity_version,
credentials_domain=CONF.auth.default_credentials_domain_name,
admin_role=CONF.identity.admin_role)
else:
return preprov_creds.NonLockingCredentialProvider(
name=name, identity_version=identity_version,
admin_role=CONF.identity.admin_role)
# We want a helper function here to check and see if admin credentials
# are available so we can do a single call from skip_checks if admin
# creds area available.
# This depends on identity_version as there may be admin credentials
# available for v2 but not for v3.
def is_admin_available(identity_version):
is_admin = True
# If dynamic credentials is enabled admin will be available
if CONF.auth.use_dynamic_credentials:
return is_admin
# Check whether test accounts file has the admin specified or not
elif (CONF.auth.test_accounts_file and
os.path.isfile(CONF.auth.test_accounts_file)):
check_accounts = preprov_creds.PreProvisionedCredentialProvider(
identity_version=identity_version, name='check_admin',
admin_role=CONF.identity.admin_role)
if not check_accounts.admin_available():
is_admin = False
else:
try:
cred_provider.get_configured_credentials(
'identity_admin', fill_in=False,
identity_version=identity_version)
except exceptions.InvalidConfiguration:
is_admin = False
return is_admin
# We want a helper function here to check and see if alt credentials
# are available so we can do a single call from skip_checks if alt
# creds area available.
# This depends on identity_version as there may be alt credentials
# available for v2 but not for v3.
def is_alt_available(identity_version):
# If dynamic credentials is enabled alt will be available
if CONF.auth.use_dynamic_credentials:
return True
# Check whether test accounts file has the admin specified or not
if (CONF.auth.test_accounts_file and
os.path.isfile(CONF.auth.test_accounts_file)):
check_accounts = preprov_creds.PreProvisionedCredentialProvider(
identity_version=identity_version, name='check_alt',
admin_role=CONF.identity.admin_role)
else:
check_accounts = preprov_creds.NonLockingCredentialProvider(
identity_version=identity_version, name='check_alt',
admin_role=CONF.identity.admin_role)
try:
if not check_accounts.is_multi_user():
return False
else:
return True
except exceptions.InvalidConfiguration:
return False

316
tempest/common/credentials_factory.py Normal file
View File

@ -0,0 +1,316 @@
# Copyright (c) 2014 Hewlett-Packard Development Company, L.P.
# 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 os
from oslo_log import log as logging
from tempest_lib import auth
from tempest import clients
from tempest.common import cred_provider
from tempest.common import dynamic_creds
from tempest.common import preprov_creds
from tempest import config
from tempest import exceptions
CONF = config.CONF
LOG = logging.getLogger(__name__)
"""This module provides factories of credential and credential providers
Credentials providers and clients are (going to be) part of tempest-lib,
and so they may not hold any dependency to tempest configuration.
Methods in this module collect the relevant configuration details and pass
them to credentials providers and clients, so that test can have easy
access to these features.
Client managers with hard-coded configured credentials are also moved here,
to avoid circular dependencies."""
# === Credential Providers
class LegacyCredentialProvider(cred_provider.CredentialProvider):
def __init__(self, identity_version):
"""Credentials provider which returns credentials from tempest.conf
Credentials provider which always returns the first and second
configured accounts as primary and alt users.
Credentials from tempest.conf are deprecated, and this credential
provider is also accordingly.
This credential provider can be used in case of serial test execution
to preserve the current behaviour of the serial tempest run.
:param identity_version: Version of the identity API
:return: CredentialProvider
"""
super(LegacyCredentialProvider, self).__init__(
identity_version=identity_version)
self._creds = {}
def _unique_creds(self, cred_arg=None):
"""Verify that the configured credentials are valid and distinct """
try:
user = self.get_primary_creds()
alt_user = self.get_alt_creds()
return getattr(user, cred_arg) != getattr(alt_user, cred_arg)
except exceptions.InvalidCredentials as ic:
msg = "At least one of the configured credentials is " \
"not valid: %s" % ic.message
raise exceptions.InvalidConfiguration(msg)
def is_multi_user(self):
return self._unique_creds('username')
def is_multi_tenant(self):
return self._unique_creds('tenant_id')
def get_primary_creds(self):
if self._creds.get('primary'):
return self._creds.get('primary')
primary_credential = get_configured_credentials(
credential_type='user', fill_in=False,
identity_version=self.identity_version)
self._creds['primary'] = cred_provider.TestResources(
primary_credential)
return self._creds['primary']
def get_alt_creds(self):
if self._creds.get('alt'):
return self._creds.get('alt')
alt_credential = get_configured_credentials(
credential_type='alt_user', fill_in=False,
identity_version=self.identity_version)
self._creds['alt'] = cred_provider.TestResources(
alt_credential)
return self._creds['alt']
def clear_creds(self):
self._creds = {}
def get_admin_creds(self):
if self._creds.get('admin'):
return self._creds.get('admin')
creds = get_configured_credentials(
"identity_admin", fill_in=False)
self._creds['admin'] = cred_provider.TestResources(creds)
return self._creds['admin']
def get_creds_by_roles(self, roles, force_new=False):
msg = "Credentials being specified through the config file can not be"\
" used with tests that specify using credentials by roles. "\
"Either exclude/skip the tests doing this or use either an "\
"test_accounts_file or dynamic credentials."
raise exceptions.InvalidConfiguration(msg)
def is_role_available(self, role):
msg = "Credentials being specified through the config file can not be"\
" used with tests that specify using credentials by roles. "\
"Either exclude/skip the tests doing this or use either an "\
"test_accounts_file or dynamic credentials."
raise exceptions.InvalidConfiguration(msg)
# Return the right implementation of CredentialProvider based on config
# Dropping interface and password, as they are never used anyways
# TODO(andreaf) Drop them from the CredentialsProvider interface completely
def get_credentials_provider(name, network_resources=None,
force_tenant_isolation=False,
identity_version=None):
# If a test requires a new account to work, it can have it via forcing
# dynamic credentials. A new account will be produced only for that test.
# In case admin credentials are not available for the account creation,
# the test should be skipped else it would fail.
identity_version = identity_version or CONF.identity.auth_version
if CONF.auth.use_dynamic_credentials or force_tenant_isolation:
admin_creds = get_configured_credentials(
'identity_admin', fill_in=True, identity_version=identity_version)
return dynamic_creds.DynamicCredentialProvider(
name=name,
network_resources=network_resources,
identity_version=identity_version,
credentials_domain=CONF.auth.default_credentials_domain_name,
admin_role=CONF.identity.admin_role,
admin_creds=admin_creds)
else:
if (CONF.auth.test_accounts_file and
os.path.isfile(CONF.auth.test_accounts_file)):
# Most params are not relevant for pre-created accounts
return preprov_creds.PreProvisionedCredentialProvider(
name=name, identity_version=identity_version,
credentials_domain=CONF.auth.default_credentials_domain_name,
admin_role=CONF.identity.admin_role)
else:
# Dynamic credentials are disabled, and the account file is not
# defined - we fall back on credentials configured in tempest.conf
return LegacyCredentialProvider(identity_version=identity_version)
# We want a helper function here to check and see if admin credentials
# are available so we can do a single call from skip_checks if admin
# creds area available.
# This depends on identity_version as there may be admin credentials
# available for v2 but not for v3.
def is_admin_available(identity_version):
is_admin = True
# If dynamic credentials is enabled admin will be available
if CONF.auth.use_dynamic_credentials:
return is_admin
# Check whether test accounts file has the admin specified or not
elif (CONF.auth.test_accounts_file and
os.path.isfile(CONF.auth.test_accounts_file)):
check_accounts = preprov_creds.PreProvisionedCredentialProvider(
identity_version=identity_version, name='check_admin',
admin_role=CONF.identity.admin_role)
if not check_accounts.admin_available():
is_admin = False
else:
try:
get_configured_credentials('identity_admin', fill_in=False,
identity_version=identity_version)
except exceptions.InvalidConfiguration:
is_admin = False
return is_admin
# We want a helper function here to check and see if alt credentials
# are available so we can do a single call from skip_checks if alt
# creds area available.
# This depends on identity_version as there may be alt credentials
# available for v2 but not for v3.
def is_alt_available(identity_version):
# If dynamic credentials is enabled alt will be available
if CONF.auth.use_dynamic_credentials:
return True
# Check whether test accounts file has the admin specified or not
if (CONF.auth.test_accounts_file and
os.path.isfile(CONF.auth.test_accounts_file)):
check_accounts = preprov_creds.PreProvisionedCredentialProvider(
identity_version=identity_version, name='check_alt',
admin_role=CONF.identity.admin_role)
else:
check_accounts = LegacyCredentialProvider(identity_version)
try:
if not check_accounts.is_multi_user():
return False
else:
return True
except exceptions.InvalidConfiguration:
return False
# === Credentials
# Type of credentials available from configuration
CREDENTIAL_TYPES = {
'identity_admin': ('auth', 'admin'),
'user': ('identity', None),
'alt_user': ('identity', 'alt')
}
DEFAULT_PARAMS = {
'disable_ssl_certificate_validation':
CONF.identity.disable_ssl_certificate_validation,
'ca_certs': CONF.identity.ca_certificates_file,
'trace_requests': CONF.debug.trace_requests
}
# Read credentials from configuration, builds a Credentials object
# based on the specified or configured version
def get_configured_credentials(credential_type, fill_in=True,
identity_version=None):
identity_version = identity_version or CONF.identity.auth_version
if identity_version not in ('v2', 'v3'):
raise exceptions.InvalidConfiguration(
'Unsupported auth version: %s' % identity_version)
if credential_type not in CREDENTIAL_TYPES:
raise exceptions.InvalidCredentials()
conf_attributes = ['username', 'password', 'tenant_name']
if identity_version == 'v3':
conf_attributes.append('domain_name')
# Read the parts of credentials from config
params = DEFAULT_PARAMS.copy()
section, prefix = CREDENTIAL_TYPES[credential_type]
for attr in conf_attributes:
_section = getattr(CONF, section)
if prefix is None:
params[attr] = getattr(_section, attr)
else:
params[attr] = getattr(_section, prefix + "_" + attr)
# Build and validate credentials. We are reading configured credentials,
# so validate them even if fill_in is False
credentials = get_credentials(fill_in=fill_in,
identity_version=identity_version, **params)
if not fill_in:
if not credentials.is_valid():
msg = ("The %s credentials are incorrectly set in the config file."
" Double check that all required values are assigned" %
credential_type)
raise exceptions.InvalidConfiguration(msg)
return credentials
# Wrapper around auth.get_credentials to use the configured identity version
# is none is specified
def get_credentials(fill_in=True, identity_version=None, **kwargs):
params = dict(DEFAULT_PARAMS, **kwargs)
identity_version = identity_version or CONF.identity.auth_version
# In case of "v3" add the domain from config if not specified
if identity_version == 'v3':
domain_fields = set(x for x in auth.KeystoneV3Credentials.ATTRIBUTES
if 'domain' in x)
if not domain_fields.intersection(kwargs.keys()):
domain_name = CONF.auth.default_credentials_domain_name
params['user_domain_name'] = domain_name
auth_url = CONF.identity.uri_v3
else:
auth_url = CONF.identity.uri
return auth.get_credentials(auth_url,
fill_in=fill_in,
identity_version=identity_version,
**params)
# === Credential / client managers
class ConfiguredUserManager(clients.Manager):
"""
Manager object that uses the `user` credentials for its
managed client objects
"""
def __init__(self, service=None):
super(ConfiguredUserManager, self).__init__(
credentials=get_configured_credentials('user'),
service=service)
class AdminManager(clients.Manager):
"""
Manager object that uses the admin credentials for its
managed client objects
"""
def __init__(self, service=None):
super(AdminManager, self).__init__(
credentials=get_configured_credentials('identity_admin'),
service=service)

30
tempest/common/dynamic_creds.py
View File

@ -31,16 +31,32 @@ LOG = logging.getLogger(__name__)
class DynamicCredentialProvider(cred_provider.CredentialProvider):
def __init__(self, identity_version, name=None, network_resources=None,
credentials_domain=None, admin_role=None):
credentials_domain=None, admin_role=None, admin_creds=None):
"""Creates credentials dynamically for tests
A credential provider that, based on an initial set of
admin credentials, creates new credentials on the fly for
tests to use and then discard.
:param str identity_version: identity API version to use `v2` or `v3`
:param str admin_role: name of the admin role added to admin users
:param str name: names of dynamic resources include this parameter
when specified
:param str credentials_domain: name of the domain where the users
are created. If not defined, the project
domain from admin_credentials is used
:param dict network_resources: network resources to be created for
the created credentials
:param Credentials admin_creds: initial admin credentials
"""
super(DynamicCredentialProvider, self).__init__(
identity_version=identity_version, name=name,
network_resources=network_resources,
credentials_domain=credentials_domain, admin_role=admin_role)
identity_version=identity_version, admin_role=admin_role,
name=name, credentials_domain=credentials_domain,
network_resources=network_resources)
self.network_resources = network_resources
self._creds = {}
self.ports = []
self.default_admin_creds = cred_provider.get_configured_credentials(
'identity_admin', fill_in=True,
identity_version=self.identity_version)
self.default_admin_creds = admin_creds
(self.identity_admin_client, self.network_admin_client,
self.networks_admin_client,
self.subnets_admin_client) = self._get_admin_clients()

61
tempest/common/preprov_creds.py
View File

@ -310,64 +310,3 @@ class PreProvisionedCredentialProvider(cred_provider.CredentialProvider):
if not user_domain_fields.intersection(set(creds_dict.keys())):
creds_dict['user_domain_name'] = self.credentials_domain
return creds_dict
class NonLockingCredentialProvider(PreProvisionedCredentialProvider):
"""Credentials provider which always returns the first and second
configured accounts as primary and alt users.
This credential provider can be used in case of serial test execution
to preserve the current behaviour of the serial tempest run.
"""
def _unique_creds(self, cred_arg=None):
"""Verify that the configured credentials are valid and distinct """
try:
user = self.get_primary_creds()
alt_user = self.get_alt_creds()
return getattr(user, cred_arg) != getattr(alt_user, cred_arg)
except exceptions.InvalidCredentials as ic:
msg = "At least one of the configured credentials is " \
"not valid: %s" % ic.message
raise exceptions.InvalidConfiguration(msg)
def is_multi_user(self):
return self._unique_creds('username')
def is_multi_tenant(self):
return self._unique_creds('tenant_id')
def get_primary_creds(self):
if self._creds.get('primary'):
return self._creds.get('primary')
primary_credential = cred_provider.get_configured_credentials(
fill_in=False, credential_type='user',
identity_version=self.identity_version)
self._creds['primary'] = cred_provider.TestResources(
primary_credential)
return self._creds['primary']
def get_alt_creds(self):
if self._creds.get('alt'):
return self._creds.get('alt')
alt_credential = cred_provider.get_configured_credentials(
fill_in=False, credential_type='alt_user',
identity_version=self.identity_version)
self._creds['alt'] = cred_provider.TestResources(
alt_credential)
return self._creds['alt']
def clear_creds(self):
self._creds = {}
def get_admin_creds(self):
creds = cred_provider.get_configured_credentials(
"identity_admin", fill_in=False)
self._creds['admin'] = cred_provider.TestResources(creds)
return self._creds['admin']
def get_creds_by_roles(self, roles, force_new=False):
msg = "Credentials being specified through the config file can not be"\
" used with tests that specify using credentials by roles. "\
"Either exclude/skip the tests doing this or use either an "\
"test_accounts_file or dynamic credentials."
raise exceptions.InvalidConfiguration(msg)

16
tempest/manager.py
View File

@ -31,22 +31,18 @@ class Manager(object):
and a client object for a test case to use in performing actions.
"""
def __init__(self, credentials=None):
def __init__(self, credentials):
"""
We allow overriding of the credentials used within the various
client classes managed by the Manager object. Left as None, the
standard username/password/tenant_name[/domain_name] is used.
Credentials to be used within the various client classes managed by the
Manager object must be defined.
:param credentials: Override of the credentials
:param credentials: type Credentials or TestResources
"""
self.auth_version = CONF.identity.auth_version
if credentials is None:
self.credentials = cred_provider.get_configured_credentials('user')
else:
self.credentials = credentials
self.credentials = credentials
# Check if passed or default credentials are valid
if not self.credentials.is_valid():
raise exceptions.InvalidCredentials()
self.auth_version = CONF.identity.auth_version
# Tenant isolation creates TestResources, but
# PreProvisionedCredentialProvider and some tests create Credentials
if isinstance(credentials, cred_provider.TestResources):

2
tempest/scenario/utils.py
View File

@ -24,7 +24,7 @@ import testscenarios
import testtools
from tempest import clients
from tempest.common import credentials
from tempest.common import credentials_factory as credentials
from tempest import config
CONF = config.CONF

4
tempest/stress/cleanup.py
View File

@ -16,14 +16,14 @@
from oslo_log import log as logging
from tempest import clients
from tempest.common import credentials_factory as credentials
from tempest.common import waiters
LOG = logging.getLogger(__name__)
def cleanup():
admin_manager = clients.AdminManager()
admin_manager = credentials.AdminManager()
body = admin_manager.servers_client.list_servers(all_tenants=True)
LOG.info("Cleanup::remove %s servers" % len(body['servers']))

5
tempest/stress/driver.py
View File

@ -26,6 +26,7 @@ from tempest_lib.common import ssh
from tempest import clients
from tempest.common import cred_client
from tempest.common import credentials_factory as credentials
from tempest.common.utils import data_utils
from tempest import config
from tempest import exceptions
@ -122,7 +123,7 @@ def stress_openstack(tests, duration, max_runs=None, stop_on_error=False):
"""
Workload driver. Executes an action function against a nova-cluster.
"""
admin_manager = clients.AdminManager()
admin_manager = credentials.AdminManager()
ssh_user = CONF.stress.target_ssh_user
ssh_key = CONF.stress.target_private_key_path
@ -145,7 +146,7 @@ def stress_openstack(tests, duration, max_runs=None, stop_on_error=False):
if test.get('use_admin', False):
manager = admin_manager
else:
manager = clients.Manager()
manager = credentials.ConfiguredUserManager()
for p_number in moves.xrange(test.get('threads', default_thread_num)):
if test.get('use_isolated_tenants', False):
username = data_utils.rand_name("stress_user")

2
tempest/test.py
View File

@ -32,7 +32,7 @@ import testtools
from tempest import clients
from tempest.common import cred_client
from tempest.common import credentials
from tempest.common import credentials_factory as credentials
from tempest.common import fixed_network
import tempest.common.generator.valid_generator as valid
import tempest.common.validation_resources as vresources

2
tempest/tests/common/test_admin_available.py
View File

@ -15,7 +15,7 @@
from oslo_config import cfg
from oslotest import mockpatch
from tempest.common import credentials
from tempest.common import credentials_factory as credentials
from tempest import config
from tempest.tests import base
from tempest.tests import fake_config

2
tempest/tests/common/test_alt_available.py
View File

@ -15,7 +15,7 @@
from oslo_config import cfg
from oslotest import mockpatch
from tempest.common import credentials
from tempest.common import credentials_factory as credentials
from tempest import config
from tempest.tests import base
from tempest.tests import fake_config

9
tempest/tests/common/test_cred_provider.py → tempest/tests/common/test_configured_creds.py
View File

@ -18,8 +18,7 @@ from tempest_lib import exceptions as lib_exc
from tempest_lib.services.identity.v2 import token_client as v2_client
from tempest_lib.services.identity.v3 import token_client as v3_client
from tempest.common import cred_provider
from tempest.common import credentials_factory as common_creds
from tempest.common import tempest_fixtures as fixtures
from tempest import config
from tempest.tests import base
@ -65,12 +64,12 @@ class ConfiguredV2CredentialsTests(base.TestCase):
def _verify_credentials(self, credentials_class, filled=True,
identity_version=None):
for ctype in cred_provider.CREDENTIAL_TYPES:
for ctype in common_creds.CREDENTIAL_TYPES:
if identity_version is None:
creds = cred_provider.get_configured_credentials(
creds = common_creds.get_configured_credentials(
credential_type=ctype, fill_in=filled)
else:
creds = cred_provider.get_configured_credentials(
creds = common_creds.get_configured_credentials(
credential_type=ctype, fill_in=filled,
identity_version=identity_version)
self._check(creds, credentials_class, filled)

36
tempest/tests/common/test_credentials.py Normal file
View File

@ -0,0 +1,36 @@
# Copyright 2015 Hewlett-Packard Development Company, L.P.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
from tempest.common import credentials_factory as credentials
from tempest import config
from tempest import exceptions
from tempest.tests import base
from tempest.tests import fake_config
class TestLegacyCredentialsProvider(base.TestCase):
fixed_params = {'identity_version': 'v2'}
def setUp(self):
super(TestLegacyCredentialsProvider, self).setUp()
self.useFixture(fake_config.ConfigFixture())
self.stubs.Set(config, 'TempestConfigPrivate', fake_config.FakePrivate)
def test_get_creds_roles_legacy_invalid(self):
test_accounts_class = credentials.LegacyCredentialProvider(
**self.fixed_params)
self.assertRaises(exceptions.InvalidConfiguration,
test_accounts_class.get_creds_by_roles,
['fake_role'])

10
tempest/tests/common/test_dynamic_creds.py
View File

@ -17,6 +17,7 @@ from oslo_config import cfg
from oslotest import mockpatch
from tempest_lib.services.identity.v2 import token_client as json_token_client
from tempest.common import credentials_factory as credentials
from tempest.common import dynamic_creds
from tempest.common import service_client
from tempest import config
@ -46,6 +47,8 @@ class TestDynamicCredentialProvider(base.TestCase):
cfg.CONF.set_default('operator_role', 'FakeRole',
group='object-storage')
self._mock_list_ec2_credentials('fake_user_id', 'fake_tenant_id')
self.fixed_params.update(
admin_creds=self._get_fake_admin_creds())
def test_tempest_client(self):
creds = dynamic_creds.DynamicCredentialProvider(**self.fixed_params)
@ -54,6 +57,13 @@ class TestDynamicCredentialProvider(base.TestCase):
self.assertTrue(isinstance(creds.network_admin_client,
json_network_client.NetworkClient))
def _get_fake_admin_creds(self):
return credentials.get_credentials(
fill_in=False,
identity_version=self.fixed_params['identity_version'],
username='fake_username', password='fake_password',
tenant_name='fake_tenant')
def _mock_user_create(self, id, name):
user_fix = self.useFixture(mockpatch.PatchObject(
json_iden_client.IdentityClient,

33
tempest/tests/common/test_preprov_creds.py
View File

@ -329,36 +329,3 @@ class TestPreProvisionedCredentials(base.TestCase):
self.assertIn('id', network)
self.assertEqual('fake-id', network['id'])
self.assertEqual('network-2', network['name'])
class TestNotLockingAccount(base.TestCase):
fixed_params = {'name': 'test class',
'identity_version': 'v2',
'admin_role': 'admin'}
def setUp(self):
super(TestNotLockingAccount, self).setUp()
self.useFixture(fake_config.ConfigFixture())
self.stubs.Set(config, 'TempestConfigPrivate', fake_config.FakePrivate)
self.useFixture(lockutils_fixtures.ExternalLockFixture())
self.test_accounts = [
{'username': 'test_user1', 'tenant_name': 'test_tenant1',
'password': 'p'},
{'username': 'test_user2', 'tenant_name': 'test_tenant2',
'password': 'p'},
{'username': 'test_user3', 'tenant_name': 'test_tenant3',
'password': 'p'},
]
self.useFixture(mockpatch.Patch(
'tempest.common.preprov_creds.read_accounts_yaml',
return_value=self.test_accounts))
cfg.CONF.set_default('test_accounts_file', '', group='auth')
self.useFixture(mockpatch.Patch('os.path.isfile', return_value=True))
def test_get_creds_roles_nonlocking_invalid(self):
test_accounts_class = preprov_creds.NonLockingCredentialProvider(
**self.fixed_params)
self.assertRaises(exceptions.InvalidConfiguration,
test_accounts_class.get_creds_by_roles,
['fake_role'])

4
tempest/thirdparty/boto/test.py vendored
View File

@ -27,7 +27,7 @@ import six
from six.moves.urllib import parse as urlparse
from tempest_lib import exceptions as lib_exc
import tempest.clients
from tempest.common import credentials_factory as credentials
from tempest.common.utils import file_utils
from tempest import config
from tempest import exceptions
@ -67,7 +67,7 @@ def decision_maker():
raise Exception("Unknown (Authentication?) Error")
# NOTE(andreaf) Setting up an extra manager here is redundant,
# and should be removed.
openstack = tempest.clients.Manager()
openstack = credentials.ConfiguredUserManager()
try:
if urlparse.urlparse(CONF.boto.ec2_url).hostname is None:
raise Exception("Failed to get hostname from the ec2_url")