Create Keystone session manager class.

Change-Id: I1dd581f02ebe6dd4b4dabb81b8b854ef5b00ef1e
changes/39/644539/3
Federico Ressi 4 years ago
parent b9247585aa
commit a29d930a84

@ -22,3 +22,4 @@ KeystoneCredentials = credentials.KeystoneCredentials
InvalidKeystoneCredentials = credentials.InvalidKeystoneCredentials
KeystoneSessionFixture = session.KeystoneSessionFixture
KeystoneSessionManager = session.KeystoneSessionManager

@ -57,3 +57,33 @@ class KeystoneSessionFixture(tobiko.SharedFixture):
del params['api_version'] # parameter not required
auth = loader.load_from_options(**params)
self.session = _session.Session(auth=auth, verify=False)
class KeystoneSessionManager(object):
def __init__(self):
self._sessions = {}
def get_session(self, credentials=None, shared=True, init_session=None):
if shared:
key = credentials
if credentials:
if tobiko.is_fixture(credentials):
key = tobiko.get_fixture_name(credentials)
session = self._sessions.get(key)
if session:
return session
init_session = init_session or KeystoneSessionFixture
assert callable(init_session)
LOG.debug('Initialize Keystone session: %r(credentials=%r)',
init_session, credentials)
session = init_session(credentials=credentials)
if shared:
self._sessions[key] = session
return session
SESSIONS = KeystoneSessionManager()

@ -16,8 +16,9 @@ from __future__ import absolute_import
import inspect
import tobiko
import mock
import tobiko
from tobiko.openstack import keystone
from tobiko.tests import unit
@ -56,7 +57,23 @@ DEFAULT_CREDENTIALS = keystone.keystone_credentials(
password='this is a secret')
class KeystoneSessionFixtureTest(unit.TobikoUnitTest):
class CheckSessionCredentialsMixin(object):
def check_session_credentials(self, session, credentials):
if credentials:
if tobiko.is_fixture(credentials):
self.assertIsNone(session.credentials)
self.assertIs(credentials, session.credentials_fixture)
else:
self.assertIs(credentials, session.credentials)
self.assertIsNone(session.credentials_fixture)
else:
self.assertIsNone(session.credentials)
self.assertIsNone(session.credentials_fixture)
class KeystoneSessionFixtureTest(CheckSessionCredentialsMixin,
unit.TobikoUnitTest):
default_credentials_fixture = (
'tobiko.openstack.keystone.credentials.'
@ -70,17 +87,8 @@ class KeystoneSessionFixtureTest(unit.TobikoUnitTest):
def test_init(self, credentials=None):
session = keystone.KeystoneSessionFixture(credentials=credentials)
if credentials:
if tobiko.is_fixture(credentials):
self.assertIsNone(session.credentials)
self.assertIs(credentials, session.credentials_fixture)
else:
self.assertIs(credentials, session.credentials)
self.assertIsNone(session.credentials_fixture)
else:
self.assertIsNone(session.credentials)
self.assertIsNone(session.credentials_fixture)
return session
self.check_session_credentials(session=session,
credentials=credentials)
def test_init_with_credentials(self):
self.test_init(credentials=CREDENTIALS)
@ -112,3 +120,47 @@ class KeystoneSessionFixtureTest(unit.TobikoUnitTest):
def test_setup_with_credentials_fixture_type(self):
self.test_setup(credentials=CredentialsFixture)
class KeystoneSessionManagerTest(CheckSessionCredentialsMixin,
unit.TobikoUnitTest):
def test_init(self):
manager = keystone.KeystoneSessionManager()
self.assertTrue(manager)
def test_get_session(self, credentials=None, shared=True):
manager = keystone.KeystoneSessionManager()
session = manager.get_session(credentials=credentials,
shared=shared)
self.assertIsInstance(session, keystone.KeystoneSessionFixture)
self.check_session_credentials(session=session,
credentials=credentials)
if shared:
self.assertIs(session, manager.get_session(
credentials=credentials))
else:
self.assertIsNot(session, manager.get_session(
credentials=credentials))
def test_get_session_with_credentials(self):
self.test_get_session(credentials=CREDENTIALS)
def test_get_session_with_not_shared(self):
self.test_get_session(shared=False)
def test_get_session_with_credentials_fixture(self):
self.test_get_session(credentials=CredentialsFixture())
def test_get_session_with_credentials_fixture_type(self):
self.test_get_session(credentials=CredentialsFixture)
def test_get_session_with_init_session(self):
mock_session = mock.MagicMock(specs=keystone.KeystoneSessionFixture)
init_session = mock.MagicMock(return_value=mock_session)
manager = keystone.KeystoneSessionManager()
session = manager.get_session(credentials=CREDENTIALS,
init_session=init_session)
self.assertIs(mock_session, session)
init_session.assert_called_once_with(credentials=CREDENTIALS)

Loading…
Cancel
Save