Create fixtures for default Keystone credentials

Create fixtures to solve the problem of getting credentials
from 'OS_*' environment variables first and from tobiko.conf
later.

Implement method to lazily read credentials the first time
are required.

Change-Id: Ie65be72d597ce4c2624aacc1b0b0dc7a8ffcc0e7
changes/20/643020/10
Federico Ressi 4 years ago
parent 7808957452
commit 3c6c1bde6e

@ -16,5 +16,6 @@ from __future__ import absolute_import
from tobiko.openstack.keystone import credentials
keystone_credentials = credentials.keystone_credentials
default_keystone_credentials = credentials.default_keystone_credentials
KeystoneCredentials = credentials.KeystoneCredentials
InvalidKeystoneCredentials = credentials.InvalidKeystoneCredentials

@ -15,9 +15,20 @@ from __future__ import absolute_import
import collections
from oslo_log import log
import tobiko
from tobiko.common import exceptions
LOG = log.getLogger(__name__)
def default_keystone_credentials():
return tobiko.setup_fixture(
DefaultKeystoneCredentialsFixture).credentials
class KeystoneCredentials(collections.namedtuple(
'KeystoneCredentials', ['auth_url',
'username',
@ -66,3 +77,116 @@ def keystone_credentials(api_version=None, auth_url=None,
class InvalidKeystoneCredentials(exceptions.TobikoException):
message = "Invalid Keystone credentials (%(credentials)r): %(reason)s."
class EnvironKeystoneCredentialsFixture(tobiko.SharedFixture):
credentials = None
def setup_fixture(self):
from tobiko import config
auth_url = config.get_env('OS_AUTH_URL')
if not auth_url:
LOG.debug("OS_AUTH_URL environment variable not defined")
return
api_version = (config.get_int_env('OS_IDENTITY_API_VERSION') or
api_version_from_url(auth_url))
if api_version == 2:
credentials = keystone_credentials(
api_version=api_version, auth_url=auth_url,
username=config.get_env('OS_USERNAME'),
password=config.get_env('OS_PASSWORD'),
project_name=(config.get_env('OS_PROJECT_NAME') or
config.get_env('OS_TENANT_NAME')))
else:
credentials = keystone_credentials(
api_version=api_version, auth_url=auth_url,
username=config.get_env('OS_USERNAME'),
password=config.get_env('OS_PASSWORD'),
project_name=(config.get_env('OS_PROJECT_NAME') or
config.get_env('OS_TENANT_NAME')),
user_domain_name=config.get_env('OS_USER_DOMAIN_NAME'),
project_domain_name=config.get_env('OS_PROJECT_DOMAIN_NAME'))
try:
credentials.validate()
except InvalidKeystoneCredentials as ex:
LOG.info("No such valid credentials from environment: %r", ex)
else:
self.credentials = credentials
class ConfigKeystoneCredentialsFixture(tobiko.SharedFixture):
credentials = None
def setup_fixture(self):
from tobiko import config
conf = config.CONF.tobiko.keystone
auth_url = conf.auth_url
if not auth_url:
LOG.debug("auth_url option not defined in 'keystone' section of "
"tobiko.conf")
return
api_version = (conf.api_version or
api_version_from_url(auth_url))
if api_version == 2:
credentials = keystone_credentials(
api_version=api_version, auth_url=auth_url,
username=conf.username, password=conf.password,
project_name=conf.project_name)
else:
credentials = keystone_credentials(
api_version=api_version, auth_url=auth_url,
username=conf.username, password=conf.password,
project_name=conf.project_name,
user_domain_name=conf.user_domain_name,
project_domain_name=conf.project_domain_name)
try:
credentials.validate()
except InvalidKeystoneCredentials as ex:
LOG.info("No such valid credentials from tobiko.conf: %r", ex)
else:
self.credentials = credentials
DEFAULT_KEYSTONE_CREDENTIALS_FIXTURES = [
EnvironKeystoneCredentialsFixture,
ConfigKeystoneCredentialsFixture]
class DefaultKeystoneCredentialsFixture(tobiko.SharedFixture):
fixtures = DEFAULT_KEYSTONE_CREDENTIALS_FIXTURES
credentials = None
def setup_fixture(self):
for fixture in self.fixtures:
try:
credentials = tobiko.setup_fixture(fixture).credentials
except Exception:
LOG.exception("Error setting up fixture %r", fixture)
else:
if credentials:
LOG.info("Got default credentials from %r: %r",
fixture, credentials)
self.credentials = credentials
return credentials
raise RuntimeError('Unable to found any valid credentials')
def api_version_from_url(auth_url):
if auth_url.endswith('/v2.0'):
LOG.info('Got Keystone API version 2 from auth_url: %r', auth_url)
return 2
elif auth_url.endswith('/v3'):
LOG.info('Got Keystone API version 3 from auth_url: %r', auth_url)
return 3
else:
LOG.warning('Unable to get Keystone API version from auth_url: %r',
auth_url)
return None

@ -14,8 +14,11 @@
# under the License.
from __future__ import absolute_import
import tobiko
from tobiko import config
from tobiko.tests import unit
from tobiko.openstack import keystone
from tobiko.openstack.keystone import credentials as _credentials
V2_PARAMS = {
@ -31,6 +34,12 @@ V2_ENVIRON = {
'OS_PASSWORD': 'super-secret',
'OS_AUTH_URL': 'http://10.0.0.1:5678/v2.0'}
V2_ENVIRON_WITH_TENANT_NAME = {
'OS_TENANT_NAME': 'demo',
'OS_USERNAME': 'demo',
'OS_PASSWORD': 'super-secret',
'OS_AUTH_URL': 'http://10.0.0.1:5678/v2.0'}
V2_ENVIRON_WITH_VERSION = dict(V2_ENVIRON, OS_IDENTITY_API_VERSION='2')
@ -102,3 +111,124 @@ class KeystoneCredentialsTest(unit.TobikoUnitTest):
credentials = make_credentials(V2_PARAMS, password=None)
self.assertRaises(keystone.InvalidKeystoneCredentials,
credentials.validate)
class EnvironKeystoneCredentialsFixtureTest(unit.TobikoUnitTest):
def test_init(self):
fixture = _credentials.EnvironKeystoneCredentialsFixture()
self.assertIsNone(fixture.credentials)
def test_setup_v2(self):
self.patch('os.environ', V2_ENVIRON)
fixture = _credentials.EnvironKeystoneCredentialsFixture()
fixture.setUp()
fixture.credentials.validate()
self.assertEqual(V2_PARAMS, fixture.credentials.to_dict())
def test_setup_v2_with_tenant_name(self):
self.patch('os.environ', V2_ENVIRON_WITH_TENANT_NAME)
fixture = _credentials.EnvironKeystoneCredentialsFixture()
fixture.setUp()
fixture.credentials.validate()
self.assertEqual(V2_PARAMS, fixture.credentials.to_dict())
def test_setup_v2_with_api_version(self):
self.patch('os.environ', V2_ENVIRON_WITH_VERSION)
fixture = _credentials.EnvironKeystoneCredentialsFixture()
fixture.setUp()
fixture.credentials.validate()
self.assertEqual(V2_PARAMS, fixture.credentials.to_dict())
def test_setup_v3(self):
self.patch('os.environ', V3_ENVIRON)
fixture = _credentials.EnvironKeystoneCredentialsFixture()
fixture.setUp()
fixture.credentials.validate()
self.assertEqual(V3_PARAMS, fixture.credentials.to_dict())
def test_setup_v3_without_api_version(self):
self.patch('os.environ', V3_ENVIRON_WITH_VERSION)
fixture = _credentials.EnvironKeystoneCredentialsFixture()
fixture.setUp()
fixture.credentials.validate()
self.assertEqual(V3_PARAMS, fixture.credentials.to_dict())
class ConfigKeystoneCredentialsFixtureTest(unit.TobikoUnitTest):
def patch_config(self, params, **kwargs):
credentials = make_credentials(params, **kwargs)
return self.patch_object(config.CONF.tobiko, 'keystone', credentials)
def test_init(self):
fixture = _credentials.ConfigKeystoneCredentialsFixture()
self.assertIsNone(fixture.credentials)
def test_setup_v2(self):
self.patch_config(V2_PARAMS, api_version=None)
fixture = _credentials.ConfigKeystoneCredentialsFixture()
fixture.setUp()
fixture.credentials.validate()
self.assertEqual(V2_PARAMS, fixture.credentials.to_dict())
def test_setup_v2_with_api_version(self):
self.patch_config(V2_PARAMS, api_version=2)
fixture = _credentials.ConfigKeystoneCredentialsFixture()
fixture.setUp()
fixture.credentials.validate()
self.assertEqual(V2_PARAMS, fixture.credentials.to_dict())
def test_setup_v3(self):
self.patch_config(V3_PARAMS, api_version=None)
fixture = _credentials.ConfigKeystoneCredentialsFixture()
fixture.setUp()
fixture.credentials.validate()
self.assertEqual(V3_PARAMS, fixture.credentials.to_dict())
def test_setup_v3_with_api_version(self):
self.patch_config(V3_PARAMS, api_version=3)
fixture = _credentials.ConfigKeystoneCredentialsFixture()
fixture.setUp()
fixture.credentials.validate()
self.assertEqual(V3_PARAMS, fixture.credentials.to_dict())
class DefaultKeystoneCredentialsFixtureTest(unit.TobikoUnitTest):
def setUp(self):
super(DefaultKeystoneCredentialsFixtureTest, self).setUp()
self.patch_config({})
self.patch('os.environ', {})
tobiko.remove_fixture(_credentials.ConfigKeystoneCredentialsFixture)
tobiko.remove_fixture(_credentials.EnvironKeystoneCredentialsFixture)
def patch_config(self, params, **kwargs):
credentials = make_credentials(params, **kwargs)
return self.patch_object(config.CONF.tobiko, 'keystone', credentials)
def test_init(self):
fixture = _credentials.DefaultKeystoneCredentialsFixture()
self.assertIsNone(fixture.credentials)
def test_setup_from_environ(self):
self.patch('os.environ', V2_ENVIRON)
fixture = _credentials.DefaultKeystoneCredentialsFixture()
fixture.setUp()
fixture.credentials.validate()
self.assertEqual(V2_PARAMS, fixture.credentials.to_dict())
def test_setup_from_config(self):
self.patch_config(V2_PARAMS)
fixture = _credentials.DefaultKeystoneCredentialsFixture()
fixture.setUp()
fixture.credentials.validate()
self.assertEqual(V2_PARAMS, fixture.credentials.to_dict())
def test_setup_from_environ_and_confif(self):
self.patch('os.environ', V3_ENVIRON)
self.patch_config(V2_PARAMS)
fixture = _credentials.DefaultKeystoneCredentialsFixture()
fixture.setUp()
fixture.credentials.validate()
self.assertEqual(V3_PARAMS, fixture.credentials.to_dict())

Loading…
Cancel
Save