# -*- coding: utf-8 -*-
# Copyright 2012 OpenStack Foundation
# Copyright 2013 IBM Corp.
#
# 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 ldappool
import mock
from oslo_config import cfg
from oslotest import mockpatch
from keystone.common.ldap import core as ldap_core
from keystone.identity.backends import ldap
from keystone.tests import unit
from keystone.tests.unit import fakeldap
from keystone.tests.unit import test_backend_ldap
CONF = cfg.CONF
[docs]class LdapPoolCommonTestMixin(object):
"""LDAP pool specific common tests used here and in live tests."""
[docs] def cleanup_pools(self):
ldap_core.PooledLDAPHandler.connection_pools.clear()
[docs] def test_handler_with_use_pool_enabled(self):
# by default use_pool and use_auth_pool is enabled in test pool config
user_ref = self.identity_api.get_user(self.user_foo['id'])
self.user_foo.pop('password')
self.assertDictEqual(self.user_foo, user_ref)
handler = ldap_core._get_connection(CONF.ldap.url, use_pool=True)
self.assertIsInstance(handler, ldap_core.PooledLDAPHandler)
@mock.patch.object(ldap_core.KeystoneLDAPHandler, 'connect')
@mock.patch.object(ldap_core.KeystoneLDAPHandler, 'simple_bind_s')
[docs] def test_handler_with_use_pool_not_enabled(self, bind_method,
connect_method):
self.config_fixture.config(group='ldap', use_pool=False)
self.config_fixture.config(group='ldap', use_auth_pool=True)
self.cleanup_pools()
user_api = ldap.UserApi(CONF)
handler = user_api.get_connection(user=None, password=None,
end_user_auth=True)
# use_auth_pool flag does not matter when use_pool is False
# still handler is non pool version
self.assertIsInstance(handler.conn, ldap_core.PythonLDAPHandler)
@mock.patch.object(ldap_core.KeystoneLDAPHandler, 'connect')
@mock.patch.object(ldap_core.KeystoneLDAPHandler, 'simple_bind_s')
[docs] def test_handler_with_end_user_auth_use_pool_not_enabled(self, bind_method,
connect_method):
# by default use_pool is enabled in test pool config
# now disabling use_auth_pool flag to test handler instance
self.config_fixture.config(group='ldap', use_auth_pool=False)
self.cleanup_pools()
user_api = ldap.UserApi(CONF)
handler = user_api.get_connection(user=None, password=None,
end_user_auth=True)
self.assertIsInstance(handler.conn, ldap_core.PythonLDAPHandler)
# For end_user_auth case, flag should not be false otherwise
# it will use, admin connections ldap pool
handler = user_api.get_connection(user=None, password=None,
end_user_auth=False)
self.assertIsInstance(handler.conn, ldap_core.PooledLDAPHandler)
[docs] def test_pool_size_set(self):
# get related connection manager instance
ldappool_cm = self.conn_pools[CONF.ldap.url]
self.assertEqual(CONF.ldap.pool_size, ldappool_cm.size)
[docs] def test_pool_retry_max_set(self):
# get related connection manager instance
ldappool_cm = self.conn_pools[CONF.ldap.url]
self.assertEqual(CONF.ldap.pool_retry_max, ldappool_cm.retry_max)
[docs] def test_pool_retry_delay_set(self):
# just make one identity call to initiate ldap connection if not there
self.identity_api.get_user(self.user_foo['id'])
# get related connection manager instance
ldappool_cm = self.conn_pools[CONF.ldap.url]
self.assertEqual(CONF.ldap.pool_retry_delay, ldappool_cm.retry_delay)
[docs] def test_pool_use_tls_set(self):
# get related connection manager instance
ldappool_cm = self.conn_pools[CONF.ldap.url]
self.assertEqual(CONF.ldap.use_tls, ldappool_cm.use_tls)
[docs] def test_pool_timeout_set(self):
# get related connection manager instance
ldappool_cm = self.conn_pools[CONF.ldap.url]
self.assertEqual(CONF.ldap.pool_connection_timeout,
ldappool_cm.timeout)
[docs] def test_pool_use_pool_set(self):
# get related connection manager instance
ldappool_cm = self.conn_pools[CONF.ldap.url]
self.assertEqual(CONF.ldap.use_pool, ldappool_cm.use_pool)
[docs] def test_pool_connection_lifetime_set(self):
# get related connection manager instance
ldappool_cm = self.conn_pools[CONF.ldap.url]
self.assertEqual(CONF.ldap.pool_connection_lifetime,
ldappool_cm.max_lifetime)
[docs] def test_max_connection_error_raised(self):
who = CONF.ldap.user
cred = CONF.ldap.password
# get related connection manager instance
ldappool_cm = self.conn_pools[CONF.ldap.url]
ldappool_cm.size = 2
# 3rd connection attempt should raise Max connection error
with ldappool_cm.connection(who, cred) as _: # conn1
with ldappool_cm.connection(who, cred) as _: # conn2
try:
with ldappool_cm.connection(who, cred) as _: # conn3
_.unbind_s()
self.fail()
except Exception as ex:
self.assertIsInstance(ex,
ldappool.MaxConnectionReachedError)
ldappool_cm.size = CONF.ldap.pool_size
[docs] def test_pool_size_expands_correctly(self):
who = CONF.ldap.user
cred = CONF.ldap.password
# get related connection manager instance
ldappool_cm = self.conn_pools[CONF.ldap.url]
ldappool_cm.size = 3
def _get_conn():
return ldappool_cm.connection(who, cred)
# Open 3 connections first
with _get_conn() as _: # conn1
self.assertEqual(1, len(ldappool_cm))
with _get_conn() as _: # conn2
self.assertEqual(2, len(ldappool_cm))
with _get_conn() as _: # conn2
_.unbind_ext_s()
self.assertEqual(3, len(ldappool_cm))
# Then open 3 connections again and make sure size does not grow
# over 3
with _get_conn() as _: # conn1
self.assertEqual(1, len(ldappool_cm))
with _get_conn() as _: # conn2
self.assertEqual(2, len(ldappool_cm))
with _get_conn() as _: # conn3
_.unbind_ext_s()
self.assertEqual(3, len(ldappool_cm))
[docs] def test_password_change_with_pool(self):
old_password = self.user_sna['password']
self.cleanup_pools()
# authenticate so that connection is added to pool before password
# change
user_ref = self.identity_api.authenticate(
context={},
user_id=self.user_sna['id'],
password=self.user_sna['password'])
self.user_sna.pop('password')
self.user_sna['enabled'] = True
self.assertDictEqual(self.user_sna, user_ref)
new_password = 'new_password'
user_ref['password'] = new_password
self.identity_api.update_user(user_ref['id'], user_ref)
# now authenticate again to make sure new password works with
# connection pool
user_ref2 = self.identity_api.authenticate(
context={},
user_id=self.user_sna['id'],
password=new_password)
user_ref.pop('password')
self.assertDictEqual(user_ref, user_ref2)
# Authentication with old password would not work here as there
# is only one connection in pool which get bind again with updated
# password..so no old bind is maintained in this case.
self.assertRaises(AssertionError,
self.identity_api.authenticate,
context={},
user_id=self.user_sna['id'],
password=old_password)
[docs]class LDAPIdentity(LdapPoolCommonTestMixin,
test_backend_ldap.LDAPIdentity,
unit.TestCase):
"""Executes tests in existing base class with pooled LDAP handler."""
[docs] def setUp(self):
self.useFixture(mockpatch.PatchObject(
ldap_core.PooledLDAPHandler, 'Connector', fakeldap.FakeLdapPool))
super(LDAPIdentity, self).setUp()
self.addCleanup(self.cleanup_pools)
# storing to local variable to avoid long references
self.conn_pools = ldap_core.PooledLDAPHandler.connection_pools
# super class loads db fixtures which establishes ldap connection
# so adding dummy call to highlight connection pool initialization
# as its not that obvious though its not needed here
self.identity_api.get_user(self.user_foo['id'])
[docs] def config_files(self):
config_files = super(LDAPIdentity, self).config_files()
config_files.append(unit.dirs.tests_conf('backend_ldap_pool.conf'))
return config_files
@mock.patch.object(ldap_core, 'utf8_encode')
[docs] def test_utf8_encoded_is_used_in_pool(self, mocked_method):
def side_effect(arg):
return arg
mocked_method.side_effect = side_effect
# invalidate the cache to get utf8_encode function called.
self.identity_api.get_user.invalidate(self.identity_api,
self.user_foo['id'])
self.identity_api.get_user(self.user_foo['id'])
mocked_method.assert_any_call(CONF.ldap.user)
mocked_method.assert_any_call(CONF.ldap.password)