# License for the specific language governing permissions and limitations
# under the License.
+from tempest_lib.common.utils import data_utils
+
from neutron.tests.tempest.api.network import base
-from neutron.tests.tempest.common.utils import data_utils
from neutron.tests.tempest import test
# License for the specific language governing permissions and limitations
# under the License.
+from tempest_lib.common.utils import data_utils
+
from neutron.tests.tempest.api.network import base
from neutron.tests.api.contrib import clients
-from neutron.tests.tempest.common.utils import data_utils
from neutron.tests.tempest import config
from neutron.tests.tempest import test
# License for the specific language governing permissions and limitations
# under the License.
+from tempest_lib.common.utils import data_utils
+
from neutron.tests.tempest.api.network import base
-from neutron.tests.tempest.common.utils import data_utils
from neutron.tests.tempest import test
# License for the specific language governing permissions and limitations
# under the License.
+from tempest_lib.common.utils import data_utils
+
from neutron.tests.tempest.api.network import base
-from neutron.tests.tempest.common.utils import data_utils
from neutron.tests.tempest import test
# License for the specific language governing permissions and limitations
# under the License.
+from tempest_lib.common.utils import data_utils
+
from neutron.tests.tempest.api.network import base
-from neutron.tests.tempest.common.utils import data_utils
from neutron.tests.tempest import test
# License for the specific language governing permissions and limitations
# under the License.
+from tempest_lib.common.utils import data_utils
from neutron.tests.tempest.api.network import base
-from neutron.tests.tempest.common.utils import data_utils
from neutron.tests.tempest import test
# License for the specific language governing permissions and limitations
# under the License.
+from tempest_lib.common.utils import data_utils
+
from neutron.tests.tempest.api.network import base_routers as base
-from neutron.tests.tempest.common.utils import data_utils
from neutron.tests.tempest import test
import netaddr
from oslo_log import log as logging
+from tempest_lib.common.utils import data_utils
from tempest_lib import exceptions as lib_exc
from neutron.tests.api.contrib import clients
-from neutron.tests.tempest.common.utils import data_utils
from neutron.tests.tempest import config
from neutron.tests.tempest import exceptions
import neutron.tests.tempest.test
# License for the specific language governing permissions and limitations
# under the License.
+from tempest_lib.common.utils import data_utils
+
from neutron.tests.tempest.api.network import base
-from neutron.tests.tempest.common.utils import data_utils
class BaseSecGroupTest(base.BaseNetworkTest):
import netaddr
import random
+from tempest_lib.common.utils import data_utils
from tempest_lib import exceptions as lib_exc
from neutron.tests.tempest.api.network import base
-from neutron.tests.tempest.common.utils import data_utils
from neutron.tests.tempest import config
from neutron.tests.tempest import test
# License for the specific language governing permissions and limitations
# under the License.
+from tempest_lib.common.utils import data_utils
+
from neutron.tests.tempest.api.network import base
-from neutron.tests.tempest.common.utils import data_utils
from neutron.tests.tempest import test
# under the License.
import netaddr
+from tempest_lib.common.utils import data_utils
from neutron.tests.tempest.api.network import base
-from neutron.tests.tempest.common.utils import data_utils
from neutron.tests.tempest import config
from neutron.tests.tempest import test
# License for the specific language governing permissions and limitations
# under the License.
+from tempest_lib.common.utils import data_utils
from tempest_lib import exceptions as lib_exc
from neutron.tests.tempest.api.network import base
-from neutron.tests.tempest.common.utils import data_utils
from neutron.tests.tempest import config
from neutron.tests.tempest import test
# License for the specific language governing permissions and limitations
# under the License.
+from tempest_lib.common.utils import data_utils
from tempest_lib import exceptions as lib_exc
from neutron.tests.tempest.api.network import base
-from neutron.tests.tempest.common.utils import data_utils
from neutron.tests.tempest import config
from neutron.tests.tempest import exceptions
from neutron.tests.tempest import test
# License for the specific language governing permissions and limitations
# under the License.
+from tempest_lib.common.utils import data_utils
from tempest_lib import decorators
from neutron.tests.tempest.api.network import base
-from neutron.tests.tempest.common.utils import data_utils
from neutron.tests.tempest import test
# under the License.
from oslo_log import log as logging
+from tempest_lib.common.utils import data_utils
from neutron.tests.tempest.api.network import base
-from neutron.tests.tempest.common.utils import data_utils
from neutron.tests.tempest import test
import itertools
import netaddr
+from tempest_lib.common.utils import data_utils
from tempest_lib import exceptions as lib_exc
from neutron.tests.tempest.api.network import base
from neutron.tests.tempest.common import custom_matchers
-from neutron.tests.tempest.common.utils import data_utils
from neutron.tests.tempest import config
from neutron.tests.tempest import test
# License for the specific language governing permissions and limitations
# under the License.
+from tempest_lib.common.utils import data_utils
from tempest_lib import exceptions as lib_exc
from neutron.tests.tempest.api.network import base
-from neutron.tests.tempest.common.utils import data_utils
from neutron.tests.tempest import test
# License for the specific language governing permissions and limitations
# under the License.
-import netaddr
import socket
+import netaddr
+from tempest_lib.common.utils import data_utils
+
from neutron.tests.tempest.api.network import base
from neutron.tests.tempest.api.network import base_security_groups as sec_base
from neutron.tests.tempest.common import custom_matchers
-from neutron.tests.tempest.common.utils import data_utils
from neutron.tests.tempest import config
from neutron.tests.tempest import test
# under the License.
import netaddr
+from tempest_lib.common.utils import data_utils
from neutron.tests.tempest.api.network import base_routers as base
from neutron.tests.api.contrib import clients
-from neutron.tests.tempest.common.utils import data_utils
from neutron.tests.tempest import config
from neutron.tests.tempest import test
# under the License.
import netaddr
+from tempest_lib.common.utils import data_utils
from tempest_lib import exceptions as lib_exc
from neutron.tests.tempest.api.network import base_routers as base
-from neutron.tests.tempest.common.utils import data_utils
from neutron.tests.tempest import config
from neutron.tests.tempest import test
# under the License.
import six
+from tempest_lib.common.utils import data_utils
from neutron.tests.tempest.api.network import base_security_groups as base
-from neutron.tests.tempest.common.utils import data_utils
from neutron.tests.tempest import config
from neutron.tests.tempest import test
# License for the specific language governing permissions and limitations
# under the License.
+from tempest_lib.common.utils import data_utils
from tempest_lib import exceptions as lib_exc
from neutron.tests.tempest.api.network import base
-from neutron.tests.tempest.common.utils import data_utils
from neutron.tests.tempest import config
from neutron.tests.tempest import test
def _auth_params(self):
return dict(
- user=self.credentials.username,
+ user_id=self.credentials.user_id,
+ username=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,
+ 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,
auth_data=True)
def _fill_credentials(self, auth_data_body):
return identity_version in IDENTITY_VERSION
-def get_credentials(auth_url, fill_in=True, identity_version='v2', **kwargs):
+def get_credentials(auth_url, fill_in=True, identity_version='v2',
+ disable_ssl_certificate_validation=None, ca_certs=None,
+ trace_requests=None, **kwargs):
"""
Builds a credentials object based on the configured auth_version
by invoking ``is_valid()``
:param identity_version (string): identity API version is used to
select the matching auth provider and credentials class
+ :param disable_ssl_certificate_validation: whether to enforce SSL
+ certificate validation in SSL API requests to the auth system
+ :param ca_certs: CA certificate bundle for validation of certificates
+ in SSL API requests to the auth system
+ :param trace_requests: trace in log API requests to the auth system
:param kwargs (dict): Dict of credential key/value pairs
Examples:
creds = credential_class(**kwargs)
# Fill in the credentials fields that were not specified
if fill_in:
- auth_provider = auth_provider_class(creds, auth_url)
+ dsvm = disable_ssl_certificate_validation
+ auth_provider = auth_provider_class(
+ creds, auth_url, disable_ssl_certificate_validation=dsvm,
+ ca_certs=ca_certs, trace_requests=trace_requests)
creds = auth_provider.fill_credentials()
return creds
Credentials suitable for the Keystone Identity V3 API
"""
- ATTRIBUTES = ['domain_name', 'password', 'tenant_name', 'username',
+ ATTRIBUTES = ['domain_id', 'domain_name', 'password', 'username',
'project_domain_id', 'project_domain_name', 'project_id',
'project_name', 'tenant_id', 'tenant_name', 'user_domain_id',
'user_domain_name', 'user_id']
- None
- Project id (optional domain)
- Project name and its domain id/name
+ - Domain id
+ - Domain name
"""
valid_user_domain = any(
[self.user_domain_id is not None,
valid_user = any(
[self.user_id is not None,
self.username is not None and valid_user_domain])
- valid_project = any(
+ valid_project_scope = 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])
+ valid_domain_scope = any(
+ [self.domain_id is None and self.domain_name is None,
+ self.domain_id or self.domain_name])
+ return all([self.password is not None,
+ valid_user,
+ valid_project_scope and valid_domain_scope])
IDENTITY_VERSION = {'v2': (KeystoneV2Credentials, KeystoneV2AuthProvider),
accounts = {}
self.use_default_creds = True
self.hash_dict = self.get_hash_dict(accounts)
- self.accounts_dir = os.path.join(CONF.oslo_concurrency.lock_path, 'test_accounts')
+ # FIXME(dhellmann): The configuration option is not part of
+ # the API of the library, because if we change the option name
+ # or group it will break this use. Tempest needs to set this
+ # value somewhere that it owns, and then use
+ # lockutils.set_defaults() to tell oslo.concurrency what value
+ # to use.
+ self.accounts_dir = os.path.join(CONF.oslo_concurrency.lock_path,
+ 'test_accounts')
self.isolated_creds = {}
@classmethod
'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
if identity_version == 'v3':
conf_attributes.append('domain_name')
# Read the parts of credentials from config
- params = {}
+ params = DEFAULT_PARAMS.copy()
section, prefix = CREDENTIAL_TYPES[credential_type]
for attr in conf_attributes:
_section = getattr(CONF, section)
# 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':
return auth.get_credentials(auth_url,
fill_in=fill_in,
identity_version=identity_version,
- **kwargs)
+ **params)
@six.add_metaclass(abc.ABCMeta)
is_admin = False
else:
try:
- cred_provider.get_configured_credentials('identity_admin')
+ cred_provider.get_configured_credentials('identity_admin',
+ fill_in=False)
except exceptions.InvalidConfiguration:
is_admin = False
return is_admin
import functools
import jsonschema
+
from oslo_log import log as logging
LOG = logging.getLogger(__name__)
import netaddr
from oslo_log import log as logging
+from tempest_lib.common.utils import data_utils
from tempest_lib import exceptions as lib_exc
from neutron.tests.api.contrib import clients
from neutron.tests.tempest.common import cred_provider
-from neutron.tests.tempest.common.utils import data_utils
from neutron.tests.tempest import config
from neutron.tests.tempest import exceptions
# License for the specific language governing permissions and limitations
# under the License.
-from neutron.openstack.common.fixture import lockutils
+from oslo_concurrency.fixture import lockutils
class LockFixture(lockutils.LockFixture):
import time
from oslo_log import log as logging
+from tempest_lib.common.utils import misc as misc_utils
-from neutron.tests.tempest.common.utils import misc as misc_utils
from neutron.tests.tempest import config
from neutron.tests.tempest import exceptions
import logging as std_logging
import os
-from oslo_concurrency import lockutils
from oslo_config import cfg
+
from oslo_log import log as logging
default='m1.micro',
help="Instance type for tests. Needs to be big enough for a "
"full OS plus the test workload"),
- cfg.StrOpt('image_ref',
- help="Name of heat-cfntools enabled image to use when "
- "launching test instances."),
cfg.StrOpt('keypair_name',
help="Name of existing keypair to launch servers with."),
cfg.IntOpt('max_template_size',
register_opt_group(cfg.CONF, g, o)
-# TODO(ihrachys): this function should probably be removed since it's not used
-# anywhere, and accesses internal implementation details of olso libraries
def list_opts():
"""Return a list of oslo.config options available.
The purpose of this is to allow tools like the Oslo sample config file
generator to discover the options exposed to users.
"""
- optlist = [(g.name, o) for g, o in _opts]
-
- # NOTE(jgrimm): Can be removed once oslo-incubator/oslo changes happen.
- optlist.append((None, lockutils.util_opts))
- optlist.append((None, logging.common_cli_opts))
- optlist.append((None, logging.logging_cli_opts))
- optlist.append((None, logging.generic_log_opts))
- optlist.append((None, logging.log_opts))
-
- return optlist
+ return [(g.name, o) for g, o in _opts]
# this should never be called outside of this class
# to remove an issue with the config file up to date checker.
if parse_conf:
config_files.append(path)
+ logging.register_options(cfg.CONF)
if os.path.isfile(path):
cfg.CONF([], project='tempest', default_config_files=config_files)
else:
_path = None
_extra_log_defaults = [
- 'keystoneclient.session=INFO',
- 'paramiko.transport=INFO',
- 'requests.packages.urllib3.connectionpool=WARN'
+ ('keystoneclient.session', std_logging.INFO),
+ ('paramiko.transport', std_logging.INFO),
+ ('requests.packages.urllib3.connectionpool', std_logging.WARN),
]
def _fix_log_levels(self):
"""Tweak the oslo log defaults."""
- # TODO(ihrachys): this code accesses internal details of oslo.log
- # library (and does it wrong), hence should be fixed
- for opt in logging.log_opts:
- if opt.dest == 'default_log_levels':
- opt.default.extend(self._extra_log_defaults)
+ for name, level in self._extra_log_defaults:
+ std_logging.getLogger(name).setLevel(level)
def __getattr__(self, attr):
if not self._config:
import urlparse
from neutron.tests.tempest import config
-from neutron.tests.tempest import exceptions
import boto
import boto.ec2
ALLOWED_METHODS = set()
- def __init__(self, username=None, password=None,
- auth_url=None, tenant_name=None,
- *args, **kwargs):
- # FIXME(andreaf) replace credentials and auth_url with auth_provider
+ def __init__(self, identity_client):
+ self.identity_client = identity_client
- insecure_ssl = CONF.identity.disable_ssl_certificate_validation
self.ca_cert = CONF.identity.ca_certificates_file
-
self.connection_timeout = str(CONF.boto.http_socket_timeout)
self.num_retries = str(CONF.boto.num_retries)
self.build_timeout = CONF.boto.build_timeout
- self.ks_cred = {"username": username,
- "password": password,
- "auth_url": auth_url,
- "tenant_name": tenant_name,
- "insecure": insecure_ssl,
- "cacert": self.ca_cert}
-
- def _keystone_aws_get(self):
- # FIXME(andreaf) Move EC2 credentials to AuthProvider
- import keystoneclient.v2_0.client
-
- keystone = keystoneclient.v2_0.client.Client(**self.ks_cred)
- ec2_cred_list = keystone.ec2.list(keystone.auth_user_id)
- ec2_cred = None
- for cred in ec2_cred_list:
- if cred.tenant_id == keystone.auth_tenant_id:
- ec2_cred = cred
- break
- else:
- ec2_cred = keystone.ec2.create(keystone.auth_user_id,
- keystone.auth_tenant_id)
- if not all((ec2_cred, ec2_cred.access, ec2_cred.secret)):
- raise lib_exc.NotFound("Unable to get access and secret keys")
- return ec2_cred
+
+ self.connection_data = {}
def _config_boto_timeout(self, timeout, retries):
try:
def get_connection(self):
self._config_boto_timeout(self.connection_timeout, self.num_retries)
self._config_boto_ca_certificates_file(self.ca_cert)
- if not all((self.connection_data["aws_access_key_id"],
- self.connection_data["aws_secret_access_key"])):
- if all([self.ks_cred.get('auth_url'),
- self.ks_cred.get('username'),
- self.ks_cred.get('tenant_name'),
- self.ks_cred.get('password')]):
- ec2_cred = self._keystone_aws_get()
- self.connection_data["aws_access_key_id"] = \
- ec2_cred.access
- self.connection_data["aws_secret_access_key"] = \
- ec2_cred.secret
- else:
- raise exceptions.InvalidConfiguration(
- "Unable to get access and secret keys")
+
+ ec2_client_args = {'aws_access_key_id': CONF.boto.aws_access,
+ 'aws_secret_access_key': CONF.boto.aws_secret}
+ if not all(ec2_client_args.values()):
+ ec2_client_args = self.get_aws_credentials(self.identity_client)
+
+ self.connection_data.update(ec2_client_args)
return self.connect_method(**self.connection_data)
+ def get_aws_credentials(self, identity_client):
+ """
+ Obtain existing, or create new AWS credentials
+ :param identity_client: identity client with embedded credentials
+ :return: EC2 credentials
+ """
+ ec2_cred_list = identity_client.list_user_ec2_credentials(
+ identity_client.user_id)
+ for cred in ec2_cred_list:
+ if cred['tenant_id'] == identity_client.tenant_id:
+ ec2_cred = cred
+ break
+ else:
+ ec2_cred = identity_client.create_user_ec2_credentials(
+ identity_client.user_id, identity_client.tenant_id)
+ if not all((ec2_cred, ec2_cred['access'], ec2_cred['secret'])):
+ raise lib_exc.NotFound("Unable to get access and secret keys")
+ else:
+ ec2_cred_aws = {}
+ ec2_cred_aws['aws_access_key_id'] = ec2_cred['access']
+ ec2_cred_aws['aws_secret_access_key'] = ec2_cred['secret']
+ return ec2_cred_aws
+
class APIClientEC2(BotoClientBase):
def connect_method(self, *args, **kwargs):
return boto.connect_ec2(*args, **kwargs)
- def __init__(self, *args, **kwargs):
- super(APIClientEC2, self).__init__(*args, **kwargs)
+ def __init__(self, identity_client):
+ super(APIClientEC2, self).__init__(identity_client)
insecure_ssl = CONF.identity.disable_ssl_certificate_validation
- aws_access = CONF.boto.aws_access
- aws_secret = CONF.boto.aws_secret
purl = urlparse.urlparse(CONF.boto.ec2_url)
region_name = CONF.compute.region
port = 443
else:
port = int(port)
- self.connection_data = {"aws_access_key_id": aws_access,
- "aws_secret_access_key": aws_secret,
- "is_secure": purl.scheme == "https",
- "validate_certs": not insecure_ssl,
- "region": region,
- "host": purl.hostname,
- "port": port,
- "path": purl.path}
+ self.connection_data.update({"is_secure": purl.scheme == "https",
+ "validate_certs": not insecure_ssl,
+ "region": region,
+ "host": purl.hostname,
+ "port": port,
+ "path": purl.path})
ALLOWED_METHODS = set(('create_key_pair', 'get_key_pair',
'delete_key_pair', 'import_key_pair',
def connect_method(self, *args, **kwargs):
return boto.connect_s3(*args, **kwargs)
- def __init__(self, *args, **kwargs):
- super(ObjectClientS3, self).__init__(*args, **kwargs)
+ def __init__(self, identity_client):
+ super(ObjectClientS3, self).__init__(identity_client)
insecure_ssl = CONF.identity.disable_ssl_certificate_validation
- aws_access = CONF.boto.aws_access
- aws_secret = CONF.boto.aws_secret
purl = urlparse.urlparse(CONF.boto.s3_url)
port = purl.port
if port is None:
port = 443
else:
port = int(port)
- self.connection_data = {"aws_access_key_id": aws_access,
- "aws_secret_access_key": aws_secret,
- "is_secure": purl.scheme == "https",
- "validate_certs": not insecure_ssl,
- "host": purl.hostname,
- "port": port,
- "calling_format": boto.s3.connection.
- OrdinaryCallingFormat()}
+ self.connection_data.update({"is_secure": purl.scheme == "https",
+ "validate_certs": not insecure_ssl,
+ "host": purl.hostname,
+ "port": port,
+ "calling_format": boto.s3.connection.
+ OrdinaryCallingFormat()})
ALLOWED_METHODS = set(('create_bucket', 'delete_bucket', 'generate_url',
'get_all_buckets', 'get_bucket', 'delete_key',
body = json.loads(body)
return service_client.ResponseBodyList(resp,
body['extensions']['values'])
+
+ def create_user_ec2_credentials(self, user_id, tenant_id):
+ post_body = json.dumps({'tenant_id': tenant_id})
+ resp, body = self.post('/users/%s/credentials/OS-EC2' % user_id,
+ post_body)
+ self.expected_success(200, resp.status)
+ return service_client.ResponseBody(resp, self._parse_resp(body))
+
+ def list_user_ec2_credentials(self, user_id):
+ resp, body = self.get('/users/%s/credentials/OS-EC2' % user_id)
+ self.expected_success(200, resp.status)
+ return service_client.ResponseBodyList(resp, self._parse_resp(body))
self.auth_url = auth_url
- def auth(self, user=None, password=None, project=None, user_type='id',
- user_domain=None, project_domain=None, token=None):
+ def auth(self, user_id=None, username=None, password=None, project_id=None,
+ project_name=None, user_domain_id=None, user_domain_name=None,
+ project_domain_id=None, project_domain_name=None, domain_id=None,
+ domain_name=None, token=None):
"""
- :param user: user id or name, as specified in user_type
- :param user_domain: the user domain
- :param project_domain: the project domain
+ :param user_id: user id
+ :param username: user name
+ :param user_domain_id: the user domain id
+ :param user_domain_name: the user domain name
+ :param project_domain_id: the project domain id
+ :param project_domain_name: the project domain name
+ :param domain_id: a domain id to scope to
+ :param domain_name: a domain name to scope to
+ :param project_id: a project id to scope to
+ :param project_name: a project name to scope to
:param token: a token to re-scope.
- Accepts different combinations of credentials. Restrictions:
- - project and domain are only name (no id)
+ Accepts different combinations of credentials.
Sample sample valid combinations:
- token
- - token, project, project_domain
+ - token, project_name, project_domain_id
- user_id, password
- - username, password, user_domain
- - username, password, project, user_domain, project_domain
+ - username, password, user_domain_id
+ - username, password, project_name, user_domain_id, project_domain_id
Validation is left to the server side.
"""
creds = {
id_obj['token'] = {
'id': token
}
- if user and password:
+
+ if (user_id or username) and password:
id_obj['methods'].append('password')
id_obj['password'] = {
'user': {
'password': password,
}
}
- if user_type == 'id':
- id_obj['password']['user']['id'] = user
+ if user_id:
+ id_obj['password']['user']['id'] = user_id
else:
- id_obj['password']['user']['name'] = user
- if user_domain is not None:
- _domain = dict(name=user_domain)
+ id_obj['password']['user']['name'] = username
+
+ _domain = None
+ if user_domain_id is not None:
+ _domain = dict(id=user_domain_id)
+ elif user_domain_name is not None:
+ _domain = dict(name=user_domain_name)
+ if _domain:
id_obj['password']['user']['domain'] = _domain
- if project is not None:
- _domain = dict(name=project_domain)
- _project = dict(name=project, domain=_domain)
- scope = dict(project=_project)
- creds['auth']['scope'] = scope
+
+ if (project_id or project_name):
+ _project = dict()
+
+ if project_id:
+ _project['id'] = project_id
+ elif project_name:
+ _project['name'] = project_name
+
+ if project_domain_id is not None:
+ _project['domain'] = {'id': project_domain_id}
+ elif project_domain_name is not None:
+ _project['domain'] = {'name': project_domain_name}
+
+ creds['auth']['scope'] = dict(project=_project)
+ elif domain_id:
+ creds['auth']['scope'] = dict(domain={'id': domain_id})
+ elif domain_name:
+ creds['auth']['scope'] = dict(domain={'name': domain_name})
body = json.dumps(creds)
resp, body = self.post(self.auth_url, body=body)
return resp, json.loads(resp_body)
- def get_token(self, user, password, project=None, project_domain='Default',
- user_domain='Default', auth_data=False):
+ def get_token(self, **kwargs):
"""
- :param user: username
Returns (token id, token data) for supplied credentials
"""
- body = self.auth(user, password, project, user_type='name',
- user_domain=user_domain,
- project_domain=project_domain)
+
+ auth_data = kwargs.pop('auth_data', False)
+
+ if not (kwargs.get('user_domain_id') or
+ kwargs.get('user_domain_name')):
+ kwargs['user_domain_name'] = 'Default'
+
+ if not (kwargs.get('project_domain_id') or
+ kwargs.get('project_domain_name')):
+ kwargs['project_domain_name'] = 'Default'
+
+ body = self.auth(**kwargs)
token = body.response.get('x-subject-token')
if auth_data:
import time
import urllib
+from tempest_lib.common.utils import misc
from tempest_lib import exceptions as lib_exc
from neutron.tests.tempest.common import service_client
-from neutron.tests.tempest.common.utils import misc
from neutron.tests.tempest import exceptions
'object_storage': CONF.service_available.swift,
'dashboard': CONF.service_available.horizon,
'telemetry': CONF.service_available.ceilometer,
- 'data_processing': CONF.service_available.sahara
+ 'data_processing': CONF.service_available.sahara,
+ 'database': CONF.service_available.trove
}
return service_list
def decorator(f):
services = ['compute', 'image', 'baremetal', 'volume', 'orchestration',
'network', 'identity', 'object_storage', 'dashboard',
- 'telemetry', 'data_processing']
+ 'telemetry', 'data_processing', 'database']
for service in args:
if service not in services:
raise exceptions.InvalidServiceTag('%s is not a valid '
Code Review / openstack-build / neutron-build.git / commitdiff