[Erp5-report] r34781 nicolas - /erp5/trunk/products/ERP5Type/tests/testMemcachedTool.py

[nobody at svn.erp5.org]

Author: nicolas
Date: Mon Apr 26 14:41:42 2010
New Revision: 34781

URL: http://svn.erp5.org?rev=34781&view=rev
Log:
Test new duration_time parameter of MemcachedTool

Modified:
    erp5/trunk/products/ERP5Type/tests/testMemcachedTool.py

Modified: erp5/trunk/products/ERP5Type/tests/testMemcachedTool.py
URL: http://svn.erp5.org/erp5/trunk/products/ERP5Type/tests/testMemcachedTool.py?rev=34781&r1=34780&r2=34781&view=diff
==============================================================================
--- erp5/trunk/products/ERP5Type/tests/testMemcachedTool.py [utf8] (original)
+++ erp5/trunk/products/ERP5Type/tests/testMemcachedTool.py [utf8] Mon Apr 26 14:41:42 2010
@@ -34,6 +34,7 @@
 from Products.ERP5Type.tests.ERP5TypeTestCase import ERP5TypeTestCase
 from AccessControl.SecurityManagement import newSecurityManager
 from Products.ERP5Type.tests.utils import installRealMemcachedTool
+import time
 
 class TestMemcachedTool(ERP5TypeTestCase):
   """
@@ -42,6 +43,8 @@
            positive or negative if the value already exists in an existing
            memcached server.
   """
+
+  expiration_time = 10 # second
 
   def getBusinessTemplateList(self):
     return tuple()
@@ -59,6 +62,13 @@
                                 portal_type='Memcached Plugin',
                                 int_index=0,
                                 url_string='127.0.0.1:11211')
+    if getattr(memcached_tool, 'memcached_plugin_with_expiration', None) is None:
+      memcached_tool.newContent(id='memcached_plugin_with_expiration',
+                                portal_type='Memcached Plugin',
+                                int_index=1,
+                                expiration_time=self.expiration_time,
+                                url_string='127.0.0.1:11211')
+
     transaction.commit()
     self.tic()
 
@@ -74,6 +84,12 @@
   def getMemcachedDict(self):
     return self.getPortal().portal_memcached.getMemcachedDict(key_prefix='unit_test',
                                                               plugin_path='portal_memcached/default_memcached_plugin')
+
+  def getMemcachedDictWithExpiration(self):
+    return self.getPortal().portal_memcached.getMemcachedDict(
+                                                        key_prefix='unit_test',
+               plugin_path='portal_memcached/memcached_plugin_with_expiration')
+
 
   def test_00_MemcachedToolIsEnabled(self):
     """
@@ -183,5 +199,21 @@
     else:
       self.fail('No error was raised when assigning a value to a non-string key.')
 
+  def test_07_checkExpirationTimeOfMemcachedDict(self):
+    """
+    Test that expiration time parameter is well handle by memcached tool
+    """
+    tested_dict = self.getMemcachedDictWithExpiration()
+    key = 'my_key'
+    value = 'a'*100
+    tested_dict[key] = value
+    transaction.commit()
+    self.assertEquals(tested_dict.get(key), value)
+    transaction.commit()
+    # Sleep 10s
+    time.sleep(self.expiration_time)
+    # now value should have expired
+    self.assertRaises(KeyError, tested_dict.__getitem__, key)
+
 if __name__ == '__main__':
   unittest.main()