[Erp5-report] r26973 - /erp5/trunk/products/ERP5Type/tests/testCacheTool.py
nobody at svn.erp5.org
nobody at svn.erp5.org
Thu May 14 11:47:39 CEST 2009
Author: nicolas
Date: Thu May 14 11:47:38 2009
New Revision: 26973
URL: http://svn.erp5.org?rev=26973&view=rev
Log:
Cleanup test
* harmonize indentation
* import transaction
* remove quiet and run_all_test variables
* cosmetic changes
Modified:
erp5/trunk/products/ERP5Type/tests/testCacheTool.py
Modified: erp5/trunk/products/ERP5Type/tests/testCacheTool.py
URL: http://svn.erp5.org/erp5/trunk/products/ERP5Type/tests/testCacheTool.py?rev=26973&r1=26972&r2=26973&view=diff
==============================================================================
--- erp5/trunk/products/ERP5Type/tests/testCacheTool.py [utf8] (original)
+++ erp5/trunk/products/ERP5Type/tests/testCacheTool.py [utf8] Thu May 14 11:47:38 2009
@@ -1,3 +1,4 @@
+# -*- coding: utf-8 -*-
##############################################################################
#
# Copyright (c) 2005 Nexedi SARL and Contributors. All Rights Reserved.
@@ -34,18 +35,14 @@
from Products.ERP5Type.CachePlugins.DummyCache import DummyCache
from AccessControl.SecurityManagement import newSecurityManager
from zLOG import LOG
-
-try:
- from transaction import get as get_transaction
-except ImportError:
- pass
+import transaction
class TestingCache(DummyCache):
"""A dummy cache that mark cache miss, so that you can later count access
using getCacheMisses() """
def __init__(self, params):
DummyCache.__init__(self, params)
-
+
def __call__(self, callable_object, cache_id, scope, cache_duration=None,
*args, **kwd):
self.markCacheMiss(1)
@@ -53,131 +50,102 @@
class TestCacheTool(ERP5TypeTestCase):
- run_all_test = 1
-
def getTitle(self):
return "Cache Tool"
-
+
def afterSetUp(self):
self.login()
-
- def login(self, quiet=0, run=run_all_test):
+
+ def login(self):
uf = self.getPortal().acl_users
- uf._doAddUser('seb', '', ['Manager'], [])
+ uf._doAddUser('admin', '', ['Manager'], [])
uf._doAddUser('ERP5TypeTestCase', '', ['Manager'], [])
- user = uf.getUserById('seb').__of__(uf)
+ user = uf.getUserById('admin').__of__(uf)
newSecurityManager(None, user)
- def test_01_CheckCacheTool(self, quiet=0, run=run_all_test):
- if not run:
- return
- if not quiet:
- message = '\nCheck CacheTool '
- ZopeTestCase._print(message)
- LOG('Testing... ',0,message)
-
- portal = self.getPortal()
- self.assertNotEqual(None,getattr(portal,'portal_caches',None))
- get_transaction().commit()
-
-
- def test_02_CheckPortalTypes(self, quiet=0, run=run_all_test):
- if not run:
- return
- if not quiet:
- message = '\nCheck Portal Types'
- ZopeTestCase._print(message)
- LOG('Testing... ',0,message)
-
- portal = self.getPortal()
- portal_types = portal.portal_types
- typeinfo_names = ("Cache Factory",
+ def test_01_CheckCacheTool(self):
+ portal = self.getPortal()
+ self.assertNotEqual(None,getattr(portal,'portal_caches',None))
+ transaction.commit()
+
+ def test_02_CheckPortalTypes(self):
+ portal = self.getPortal()
+ portal_types = portal.portal_types
+ typeinfo_names = ("Cache Factory",
"Ram Cache",
"Distributed Ram Cache",
"SQL Cache",
)
- for typeinfo_name in typeinfo_names:
- portal_type = getattr(portal_types,typeinfo_name,None)
- self.assertNotEqual(None,portal_type)
- get_transaction().commit()
-
- def test_03_CreateCacheFactories(self, quiet=0, run=run_all_test):
- if not run:
- return
- if not quiet:
- message = '\nCreate Cache Tool Factories'
- ZopeTestCase._print(message)
- LOG('Testing... ',0,message)
- portal = self.getPortal()
- portal_caches = portal.portal_caches
-
- # Cache plugins are organised into 'Cache factories' so we create
- # factories first ram_cache_factory (to test Ram Cache Plugin)
- ram_cache_factory = portal_caches.newContent(portal_type="Cache Factory",
- id = 'ram_cache_factory',
- container=portal_caches)
- ram_cache_plugin = ram_cache_factory.newContent(portal_type="Ram Cache",
- container=ram_cache_factory)
- ram_cache_plugin.setIntIndex(0)
-
-
- ## distributed_ram_cache_factory (to test Distributed Ram Cache Plugin)
- dram_cache_factory = portal_caches.newContent(portal_type="Cache Factory",
- id = 'distributed_ram_cache_factory',
- container=portal_caches)
- dram_cache_plugin = dram_cache_factory.newContent(
- portal_type="Distributed Ram Cache", container=dram_cache_factory)
- dram_cache_plugin.setIntIndex(0)
-
- ## sql_cache_factory (to test SQL Cache Plugin)
- sql_cache_factory = portal_caches.newContent(portal_type="Cache Factory",
- id = 'sql_cache_factory',
- container=portal_caches)
- sql_cache_plugin = sql_cache_factory.newContent(
- portal_type="SQL Cache", container=sql_cache_factory)
- sql_cache_plugin.setIntIndex(0)
-
- ## erp5_user_factory (to test a combination of all cache plugins)
- erp5_user_factory = portal_caches.newContent(portal_type="Cache Factory",
- id = "erp5_user_factory",
- container=portal_caches)
-
- ram_cache_plugin = erp5_user_factory.newContent(
- portal_type="Ram Cache", container=erp5_user_factory)
- ram_cache_plugin.setIntIndex(0)
- dram_cache_plugin = erp5_user_factory.newContent(
- portal_type="Distributed Ram Cache", container=erp5_user_factory)
- dram_cache_plugin.setIntIndex(1)
- sql_cache_plugin = erp5_user_factory.newContent(
- portal_type="SQL Cache", container=erp5_user_factory)
- sql_cache_plugin.setIntIndex(2)
-
- ##
- get_transaction().commit()
-
- ## update Ram Cache structure
- portal_caches.updateCache()
- from Products.ERP5Type.Cache import CachingMethod
-
- ## do we have the same structure we created above?
- self.assert_('ram_cache_factory' in CachingMethod.factories)
- self.assert_('distributed_ram_cache_factory' in CachingMethod.factories)
- self.assert_('sql_cache_factory' in CachingMethod.factories)
- self.assert_('erp5_user_factory' in CachingMethod.factories)
-
- def test_04_CreateCachedMethod(self, quiet=0, run=run_all_test):
- if not run:
- return
- if not quiet:
- message = '\nCreate Cache Method (Python Script)'
- ZopeTestCase._print(message)
- LOG('Testing... ',0,message)
- portal = self.getPortal()
-
- ## add test cached method
- py_script_id = "testCachedMethod"
- py_script_params = "value=10000, portal_path=('','erp5')"
- py_script_body = """
+ for typeinfo_name in typeinfo_names:
+ portal_type = getattr(portal_types,typeinfo_name,None)
+ self.assertNotEqual(None,portal_type)
+ transaction.commit()
+
+ def test_03_CreateCacheFactories(self):
+ portal = self.getPortal()
+ portal_caches = portal.portal_caches
+
+ # Cache plugins are organised into 'Cache factories' so we create
+ # factories first ram_cache_factory (to test Ram Cache Plugin)
+ ram_cache_factory = portal_caches.newContent(portal_type="Cache Factory",
+ id = 'ram_cache_factory',
+ container=portal_caches)
+ ram_cache_plugin = ram_cache_factory.newContent(portal_type="Ram Cache",
+ container=ram_cache_factory)
+ ram_cache_plugin.setIntIndex(0)
+
+
+ ## distributed_ram_cache_factory (to test Distributed Ram Cache Plugin)
+ dram_cache_factory = portal_caches.newContent(portal_type="Cache Factory",
+ id = 'distributed_ram_cache_factory',
+ container=portal_caches)
+ dram_cache_plugin = dram_cache_factory.newContent(
+ portal_type="Distributed Ram Cache", container=dram_cache_factory)
+ dram_cache_plugin.setIntIndex(0)
+
+ ## sql_cache_factory (to test SQL Cache Plugin)
+ sql_cache_factory = portal_caches.newContent(portal_type="Cache Factory",
+ id = 'sql_cache_factory',
+ container=portal_caches)
+ sql_cache_plugin = sql_cache_factory.newContent(
+ portal_type="SQL Cache", container=sql_cache_factory)
+ sql_cache_plugin.setIntIndex(0)
+
+ ## erp5_user_factory (to test a combination of all cache plugins)
+ erp5_user_factory = portal_caches.newContent(portal_type="Cache Factory",
+ id = "erp5_user_factory",
+ container=portal_caches)
+
+ ram_cache_plugin = erp5_user_factory.newContent(
+ portal_type="Ram Cache", container=erp5_user_factory)
+ ram_cache_plugin.setIntIndex(0)
+ dram_cache_plugin = erp5_user_factory.newContent(
+ portal_type="Distributed Ram Cache", container=erp5_user_factory)
+ dram_cache_plugin.setIntIndex(1)
+ sql_cache_plugin = erp5_user_factory.newContent(
+ portal_type="SQL Cache", container=erp5_user_factory)
+ sql_cache_plugin.setIntIndex(2)
+
+ ##
+ transaction.commit()
+
+ ## update Ram Cache structure
+ portal_caches.updateCache()
+ from Products.ERP5Type.Cache import CachingMethod
+
+ ## do we have the same structure we created above?
+ self.assert_('ram_cache_factory' in CachingMethod.factories)
+ self.assert_('distributed_ram_cache_factory' in CachingMethod.factories)
+ self.assert_('sql_cache_factory' in CachingMethod.factories)
+ self.assert_('erp5_user_factory' in CachingMethod.factories)
+
+ def test_04_CreateCachedMethod(self):
+ portal = self.getPortal()
+
+ ## add test cached method
+ py_script_id = "testCachedMethod"
+ py_script_params = "value=10000, portal_path=('','erp5')"
+ py_script_body = """
def veryExpensiveMethod(value):
## do something expensive for some time
## no 'time.sleep()' available in Zope
@@ -190,96 +158,83 @@
result = veryExpensiveMethod(value)
return result
"""
- portal.manage_addProduct['PythonScripts'].manage_addPythonScript(
- id=py_script_id)
- py_script_obj = getattr(portal, py_script_id)
- py_script_obj.ZPythonScript_edit(py_script_params, py_script_body)
- get_transaction().commit()
-
- def test_05_CacheFactoryOnePlugin(self, quiet=0, run=run_all_test):
+ portal.manage_addProduct['PythonScripts'].manage_addPythonScript(
+ id=py_script_id)
+ py_script_obj = getattr(portal, py_script_id)
+ py_script_obj.ZPythonScript_edit(py_script_params, py_script_body)
+ transaction.commit()
+
+ def test_05_CacheFactoryOnePlugin(self):
""" Test cache factory containing only one cache plugin. """
- if not run:
- return
- if not quiet:
- message = '\nTest each type of cache plugin individually.'
- ZopeTestCase._print(message)
- LOG('Testing... ',0,message)
- portal = self.getPortal()
- from Products.ERP5Type.Cache import CachingMethod
- py_script_id = "testCachedMethod"
- py_script_obj = getattr(portal, py_script_id)
- for cf_name in ('ram_cache_factory',
- 'distributed_ram_cache_factory',
- 'sql_cache_factory'):
- my_cache = CachingMethod(py_script_obj,
- 'py_script_obj',
- cache_factory=cf_name)
- self._cacheFactoryInstanceTest(my_cache, cf_name)
-
- def test_06_CacheFactoryMultiPlugins(self, quiet=0, run=run_all_test):
- """ Test a cache factory containing multiple cache plugins. """
- if not run:
- return
- if not quiet:
- message = '\nTest combination of available cache plugins under a cache'\
- ' factory'
- ZopeTestCase._print(message)
- LOG('Testing... ',0,message)
- portal = self.getPortal()
- from Products.ERP5Type.Cache import CachingMethod
- py_script_id = "testCachedMethod"
- py_script_obj = getattr(portal, py_script_id)
- cf_name = 'erp5_user_factory'
+ portal = self.getPortal()
+ from Products.ERP5Type.Cache import CachingMethod
+ py_script_id = "testCachedMethod"
+ py_script_obj = getattr(portal, py_script_id)
+ for cf_name in ('ram_cache_factory',
+ 'distributed_ram_cache_factory',
+ 'sql_cache_factory'):
my_cache = CachingMethod(py_script_obj,
'py_script_obj',
cache_factory=cf_name)
self._cacheFactoryInstanceTest(my_cache, cf_name)
-
+
+ def test_06_CacheFactoryMultiPlugins(self):
+ """ Test a cache factory containing multiple cache plugins. """
+ portal = self.getPortal()
+ from Products.ERP5Type.Cache import CachingMethod
+ py_script_id = "testCachedMethod"
+ py_script_obj = getattr(portal, py_script_id)
+ cf_name = 'erp5_user_factory'
+ my_cache = CachingMethod(py_script_obj,
+ 'py_script_obj',
+ cache_factory=cf_name)
+ self._cacheFactoryInstanceTest(my_cache, cf_name)
+
def _cacheFactoryInstanceTest(self, my_cache, cf_name):
- portal = self.getPortal()
- print
- print "="*40
- print "TESTING:", cf_name
-
- # if the test fails because your machine is too fast, increase this value.
- nb_iterations = 30000
-
- portal.portal_caches.clearCacheFactory(cf_name)
- ## 1st call
- start = time.time()
- original = my_cache(nb_iterations, portal_path=('', portal.getId()))
- end = time.time()
- calculation_time = end-start
- print "\n\tCalculation time (1st call)", calculation_time
-
- ## 2nd call - should be cached now
- start = time.time()
- cached = my_cache(nb_iterations, portal_path=('', portal.getId()))
- end = time.time()
- calculation_time = end-start
- print "\n\tCalculation time (2nd call)", calculation_time
-
- # check if cache works by getting calculation_time for last cache
- # operation even remote cache must have access time less than a second.
- # if it's greater than method wasn't previously cached and was calculated
- # instead
- self.assert_(1.0 > calculation_time)
-
- ## check if equal.
- self.assertEquals(original, cached)
-
- ## OK so far let's clear cache
- portal.portal_caches.clearCacheFactory(cf_name)
-
- ## 1st call
- start = time.time()
- original = my_cache(nb_iterations, portal_path=('', portal.getId()))
- end = time.time()
- calculation_time = end-start
- print "\n\tCalculation time (after cache clear)", calculation_time
-
- ## Cache cleared shouldn't be previously cached
- self.assert_(1.0 < calculation_time)
+ portal = self.getPortal()
+ print
+ print "="*40
+ print "TESTING:", cf_name
+
+ # if the test fails because your machine is too fast, increase this value.
+ nb_iterations = 30000
+
+ portal.portal_caches.clearCacheFactory(cf_name)
+ ## 1st call
+ start = time.time()
+ original = my_cache(nb_iterations, portal_path=('', portal.getId()))
+ end = time.time()
+ calculation_time = end-start
+ print "\n\tCalculation time (1st call)", calculation_time
+
+ ## 2nd call - should be cached now
+ start = time.time()
+ cached = my_cache(nb_iterations, portal_path=('', portal.getId()))
+ end = time.time()
+ calculation_time = end-start
+ print "\n\tCalculation time (2nd call)", calculation_time
+
+ # check if cache works by getting calculation_time for last cache
+ # operation even remote cache must have access time less than a second.
+ # if it's greater than method wasn't previously cached and was calculated
+ # instead
+ self.assert_(1.0 > calculation_time)
+
+ ## check if equal.
+ self.assertEquals(original, cached)
+
+ ## OK so far let's clear cache
+ portal.portal_caches.clearCacheFactory(cf_name)
+
+ ## 1st call
+ start = time.time()
+ original = my_cache(nb_iterations, portal_path=('', portal.getId()))
+ end = time.time()
+ calculation_time = end-start
+ print "\n\tCalculation time (after cache clear)", calculation_time
+
+ ## Cache cleared shouldn't be previously cached
+ self.assert_(1.0 < calculation_time)
def test_CachePersistentObjects(self):
# storing persistent objects in cache is not allowed, but this check is
@@ -296,7 +251,6 @@
return self.portal.getTitle
cached_func = CachingMethod(func, 'cache_bound_method')
self.assertRaises(TypeError, cached_func)
-
def test_suite():
suite = unittest.TestSuite()
More information about the Erp5-report
mailing list