Back to index

plone3  3.1.7
test_Caching.py
Go to the documentation of this file.
00001 ##############################################################################
00002 #
00003 # Copyright (c) 2001 Zope Corporation and Contributors. All Rights
00004 # Reserved.
00005 #
00006 # This software is subject to the provisions of the Zope Public License,
00007 # Version 2.1 (ZPL).  A copy of the ZPL should accompany this
00008 # distribution.
00009 # THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED
00010 # WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
00011 # WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS
00012 # FOR A PARTICULAR PURPOSE.
00013 #
00014 ##############################################################################
00015 import unittest
00016 
00017 from Acquisition import aq_base
00018 from OFS.Cache import isCacheable
00019 
00020 from Products.StandardCacheManagers.RAMCacheManager import RAMCacheManager
00021 
00022 class FauxRequest:
00023 
00024     def __init__( self, steps=(), **kw ):
00025 
00026         self.steps = steps
00027         self._dict = {}
00028         self._dict.update( kw )
00029 
00030     def get( self, key, default=None ):
00031 
00032         return self._dict.get( key, default )
00033 
00034     def _authUserPW( self ):
00035         form = self.get( 'form' )
00036         return ( form.get( 'login' ), form.get( 'password' ) )
00037 
00038     def __getitem__( self, key ):
00039 
00040         return self._dict[ key ]
00041 
00042     def __setitem__( self, key, value ):
00043 
00044         self._dict[ key ] = value
00045 
00046 class PluggableAuthServiceCachingTests( unittest.TestCase ):
00047 
00048     def tearDown( self ):
00049         pass
00050 
00051 
00052     def _getTargetClass( self ):
00053 
00054         from Products.PluggableAuthService.PluggableAuthService \
00055             import PluggableAuthService
00056 
00057         return PluggableAuthService
00058 
00059     def _makeOne( self, plugins=None, *args, **kw ):
00060 
00061         zcuf = self._getTargetClass()( *args, **kw )
00062 
00063         if plugins is not None:
00064             zcuf._setObject( 'plugins', plugins )
00065 
00066         rcm = RAMCacheManager('ramcache')
00067         zcuf._setObject('ramcache', rcm)
00068 
00069         return zcuf
00070 
00071     def _makePlugins( self, plugin_type_info=None ):
00072 
00073         from Products.PluggableAuthService.PluggableAuthService \
00074             import _PLUGIN_TYPE_INFO
00075         from Products.PluginRegistry.PluginRegistry import PluginRegistry
00076 
00077         if plugin_type_info is None:
00078             plugin_type_info = _PLUGIN_TYPE_INFO
00079 
00080         reg = PluginRegistry( plugin_type_info=plugin_type_info )
00081         reg._setId( 'plugins' )
00082         reg._plugins = {}
00083 
00084         return reg
00085 
00086     def _makeAndFill(self):
00087 
00088         from Products.PluggableAuthService.plugins import ZODBUserManager
00089         from Products.PluggableAuthService.plugins import ZODBRoleManager
00090 
00091         plugin_registry = self._makePlugins()
00092         user_source = ZODBUserManager.ZODBUserManager('zodb_users')
00093         roles_source = ZODBRoleManager.ZODBRoleManager('zodb_roles')
00094         pas_instance = self._makeOne(plugins=plugin_registry)
00095         pas_instance._setObject('zodb_users', user_source)
00096         pas_instance._setObject('zodb_roles', roles_source)
00097 
00098         return pas_instance
00099 
00100     def test_empty( self ):
00101         zcuf = self._makeOne()
00102         rcm = getattr(zcuf, 'ramcache')
00103 
00104         # This is needed because some underlying ZCacheable code wants to
00105         # use self.REQUEST :/
00106         setattr(rcm, 'REQUEST', FauxRequest())
00107 
00108         # Make sure the PAS instance itself is Cacheable
00109         self.assert_(isCacheable(zcuf))
00110 
00111         # Make sure the PAS instance is not associated with any cache manager
00112         # by default
00113         self.assert_(zcuf.ZCacheable_getManager() is None)
00114 
00115         # Make sure the RAMCacheManager is empty
00116         self.assert_(len(rcm.getCacheReport()) == 0)
00117 
00118     def test_caching_in_PAS(self):
00119         zcuf = self._makeAndFill()
00120         rcm = getattr(zcuf, 'ramcache')
00121         plugin_registry = getattr(zcuf, 'plugins')
00122         user_source = getattr(zcuf, 'zodb_users')
00123         roles_source = getattr(zcuf, 'zodb_roles')
00124 
00125         # This is needed because some underlying ZCacheable code wants to
00126         # use self.REQUEST :/
00127         setattr(zcuf, 'REQUEST', FauxRequest())
00128 
00129         # First, we register the ZODBUserManager as a plugin suitable
00130         # for storing and returning user objects and the ZODBRoleManager
00131         # for roles. Basic scaffolding to be able to store and retrieve users.
00132         from Products.PluggableAuthService.interfaces import plugins
00133 
00134         plugin_registry.activatePlugin( plugins.IUserEnumerationPlugin
00135                                       , user_source.getId()
00136                                       )
00137         plugin_registry.activatePlugin( plugins.IUserAdderPlugin
00138                                       , user_source.getId()
00139                                       )
00140         plugin_registry.activatePlugin( plugins.IRolesPlugin
00141                                       , roles_source.getId()
00142                                       )
00143         plugin_registry.activatePlugin( plugins.IRoleEnumerationPlugin
00144                                       , roles_source.getId()
00145                                       )
00146         plugin_registry.activatePlugin( plugins.IRoleAssignerPlugin
00147                                       , roles_source.getId()
00148                                       )
00149 
00150         # Now add a user and make sure it's there
00151         zcuf._doAddUser('testlogin', 'secret', ['Member', 'Anonymous'], [])
00152         self.failIf(zcuf.getUser('testlogin') is None)
00153 
00154         # Then we activate caching for the PAS instance itself
00155         zcuf.ZCacheable_setManagerId(rcm.getId())
00156 
00157         # Make sure the PAS instance is associated with the cache
00158         self.failUnless(aq_base(zcuf.ZCacheable_getManager()) is aq_base(rcm))
00159 
00160         # Now we can see if the cache is getting used. Test for emptiness
00161         # first, then retrieve a user, and the cache should have content.
00162         # Then test again to see if the cache entries are being used.
00163         # This is a bit nasty because I am relying on knowing the structure
00164         # of the cache report, which is really an internal implementation
00165         # detail.
00166 
00167         # First check: The cache must be empty
00168         report = rcm.getCacheReport()
00169         self.failUnless(len(report) == 0)
00170 
00171         # The user is being requested once. At this point there must be one
00172         # entry for the PAS instance. The number of "misses" must be >0 because
00173         # the first cache check will have failed. The number of cache hits must
00174         # be zero.
00175         zcuf.getUser('testlogin')
00176         report = rcm.getCacheReport()
00177         self.failUnless(len(report) == 1)
00178         report_item = report[0]
00179         firstpass_misses = report_item.get('misses')
00180         firstpass_hits = report_item.get('hits')
00181         firstpass_entries = report_item.get('entries')
00182         self.failUnless(firstpass_misses > 0)
00183         self.failUnless(firstpass_hits == 0)
00184 
00185         # The user is requested again. This request should produce a cache hit,
00186         # so the number of "misses" must have stayed the same as after the
00187         # first pass, but the number of hits must now be >0. Also, the number
00188         # of in-memory entries must have remained the same to prove that we are
00189         # reusing the same cache entries.
00190         zcuf.getUser('testlogin')
00191         report = rcm.getCacheReport()
00192         self.failUnless(len(report) == 1)
00193         report_item = report[0]
00194         self.failIf(report_item.get('misses') != firstpass_misses)
00195         self.failUnless(report_item.get('hits') > firstpass_hits)
00196         self.failIf(report_item.get('entries') != firstpass_entries)
00197 
00198 
00199 if __name__ == "__main__":
00200     unittest.main()
00201 
00202 def test_suite():
00203     return unittest.TestSuite((
00204         unittest.makeSuite( PluggableAuthServiceCachingTests ),
00205         ))