Allow caching to be disabled and tests still pass

Skip cache specific tests if caching is disabled.  Cache should only
be disabled to eliminate caching as an issue with tests, but in all
cases tests should pass in either configuration.  The default is
to use caching for all tests.

The caching options can be changed for testing purposes in the
keystone/tests/test_overrides.conf file.  The specific options
and sections that support caching can be located in the
configuration document in the section covering the caching layer.

Change-Id: I7848128aeaf3f578b89ab0b9201c45e00efecf11
This commit is contained in:
Morgan Fainberg 2013-10-25 15:05:52 -07:00
parent fe7653679c
commit 99fafca3ba
4 changed files with 106 additions and 32 deletions

View File

@ -17,6 +17,7 @@
from __future__ import absolute_import
import errno
import functools
import os
import re
import shutil
@ -31,6 +32,7 @@ import logging
from lxml import etree
from paste import deploy
import testtools
from testtools import testcase
from keystone.openstack.common import gettextutils
@ -174,6 +176,41 @@ def teardown_test_database():
sql.core.set_global_engine(None)
def skip_if_cache_disabled(*sections):
"""This decorator is used to skip a test if caching is disabled either
globally or for the specific section.
In the code fragment:
@skip_if_cache_is_disabled('assignment', 'token')
def test_method(*args):
...
The method test_method would be skipped if caching is disabled globally via
the `enabled` option in the `cache` section of the configuration or if
the `caching` option is set to false in either `assignment` or `token`
sections of the configuration. This decorator can be used with no
arguments to only check global caching.
If a specified configuration section does not define the `caching` option,
this decorator makes the same assumption as the `should_cache_fn` in
keystone.common.cache that caching should be enabled.
"""
def wrapper(f):
@functools.wraps(f)
def inner(*args, **kwargs):
if not CONF.cache.enabled:
raise testcase.TestSkipped('Cache globally disabled.')
for s in sections:
conf_sec = getattr(CONF, s, None)
if conf_sec is not None:
if not getattr(conf_sec, 'caching', True):
raise testcase.TestSkipped('%s caching disabled.' % s)
return f(*args, **kwargs)
return inner
return wrapper
class TestClient(object):
def __init__(self, app=None, token=None):
self.app = app

View File

@ -2571,6 +2571,7 @@ class IdentityTests(object):
user_projects = self.assignment_api.list_projects_for_user(user1['id'])
self.assertEqual(len(user_projects), 3)
@tests.skip_if_cache_disabled('assignment')
def test_cache_layer_domain_crud(self):
domain = {'id': uuid.uuid4().hex, 'name': uuid.uuid4().hex,
'enabled': True}
@ -2617,6 +2618,7 @@ class IdentityTests(object):
self.assignment_api.get_domain,
domain_id)
@tests.skip_if_cache_disabled('assignment')
def test_cache_layer_project_crud(self):
domain = {'id': uuid.uuid4().hex, 'name': uuid.uuid4().hex,
'enabled': True}
@ -2669,6 +2671,7 @@ class IdentityTests(object):
self.assignment_api.get_project,
project_id)
@tests.skip_if_cache_disabled('assignment')
def test_cache_layer_role_crud(self):
role = {'id': uuid.uuid4().hex, 'name': uuid.uuid4().hex}
role_id = role['id']
@ -2996,6 +2999,7 @@ class TokenTests(object):
self.assertEqual(len(tokens), 1)
self.assertIn(token_id, tokens)
@tests.skip_if_cache_disabled('token')
def test_revocation_list_cache(self):
expire_time = timeutils.utcnow() + datetime.timedelta(minutes=10)
token_id = uuid.uuid4().hex

View File

@ -796,6 +796,7 @@ class LDAPIdentity(tests.TestCase, BaseLDAPIdentity):
self.assignment_api.get_project,
project['id'])
@tests.skip_if_cache_disabled('assignment')
def test_cache_layer_project_crud(self):
# NOTE(morganfainberg): LDAP implementation does not currently support
# updating project names. This method override provides a different

View File

@ -83,6 +83,16 @@ class CacheRegionTest(tests.TestCase):
self.region = cache.make_region()
cache.configure_cache_region(self.region)
self.region.wrap(TestProxy)
self.test_value = TestProxyValue('Decorator Test')
def _get_cacheable_function(self):
SHOULD_CACHE_FN = cache.should_cache_fn('cache')
@self.region.cache_on_arguments(should_cache_fn=SHOULD_CACHE_FN)
def cacheable_function(value):
return value
return cacheable_function
def test_region_built_with_proxy_direct_cache_test(self):
# Verify cache regions are properly built with proxies.
@ -96,44 +106,66 @@ class CacheRegionTest(tests.TestCase):
cache.configure_cache_region(self.region)
cache.configure_cache_region(self.region)
def test_should_cache_fn(self):
# Verify should_cache_fn generates a sane function for subsystem
# toggle.
SHOULD_CACHE = cache.should_cache_fn('cache')
test_value = TestProxyValue('Decorator Test')
def test_should_cache_fn_global_cache_enabled(self):
# Verify should_cache_fn generates a sane function for subsystem and
# functions as expected with caching globally enabled.
cacheable_function = self._get_cacheable_function()
@self.region.cache_on_arguments(should_cache_fn=SHOULD_CACHE)
def cacheable_function(value):
return value
setattr(CONF.cache, 'caching', False)
cacheable_function(test_value)
cached_value = cacheable_function(test_value)
self.assertFalse(cached_value.cached)
setattr(CONF.cache, 'caching', True)
cacheable_function(test_value)
cached_value = cacheable_function(test_value)
self.opt_in_group('cache', enabled=True)
cacheable_function(self.test_value)
cached_value = cacheable_function(self.test_value)
self.assertTrue(cached_value.cached)
def test_should_cache_fn_global(self):
# Verify should_cache_fn generates a sane function for global
# toggle.
SHOULD_CACHE = cache.should_cache_fn('cache')
test_value = TestProxyValue('Decorator Test')
def test_should_cache_fn_global_cache_disabled(self):
# Verify should_cache_fn generates a sane function for subsystem and
# functions as expected with caching globally disabled.
cacheable_function = self._get_cacheable_function()
@self.region.cache_on_arguments(should_cache_fn=SHOULD_CACHE)
def cacheable_function(value):
return value
setattr(CONF.cache, 'enabled', False)
cacheable_function(test_value)
cached_value = cacheable_function(test_value)
self.opt_in_group('cache', enabled=False)
cacheable_function(self.test_value)
cached_value = cacheable_function(self.test_value)
self.assertFalse(cached_value.cached)
setattr(CONF.cache, 'enabled', True)
cacheable_function(test_value)
cached_value = cacheable_function(test_value)
def test_should_cache_fn_global_cache_disabled_section_cache_enabled(self):
# Verify should_cache_fn generates a sane function for subsystem and
# functions as expected with caching globally disabled and the specific
# section caching enabled.
cacheable_function = self._get_cacheable_function()
self.opt_in_group('cache', enabled=False)
# NOTE(morganfainberg): cannot use opt_in_group because 'caching' is a
# fabricated setting for tests.
setattr(CONF.cache, 'caching', True)
cacheable_function(self.test_value)
cached_value = cacheable_function(self.test_value)
self.assertFalse(cached_value.cached)
def test_should_cache_fn_global_cache_enabled_section_cache_disabled(self):
# Verify should_cache_fn generates a sane function for subsystem and
# functions as expected with caching globally enabled and the specific
# section caching disabled.
cacheable_function = self._get_cacheable_function()
self.opt_in_group('cache', enabled=True)
# NOTE(morganfainberg): cannot use opt_in_group because 'caching' is a
# fabricated setting for tests.
setattr(CONF.cache, 'caching', False)
cacheable_function(self.test_value)
cached_value = cacheable_function(self.test_value)
self.assertFalse(cached_value.cached)
def test_should_cache_fn_global_cache_enabled_section_cache_enabled(self):
# Verify should_cache_fn generates a sane function for subsystem and
# functions as expected with caching globally enabled and the specific
# section caching enabled.
cacheable_function = self._get_cacheable_function()
self.opt_in_group('cache', enabled=True)
# NOTE(morganfainberg): cannot use opt_in_group because 'caching' is a
# fabricated setting for tests.
setattr(CONF.cache, 'caching', True)
cacheable_function(self.test_value)
cached_value = cacheable_function(self.test_value)
self.assertTrue(cached_value.cached)
def test_cache_dictionary_config_builder(self):