[z3-checkins] r22128 - z3/sqlos/branch/jinty-connection/src/sqlos/ftests

[jinty at codespeak.net]

Author: jinty
Date: Fri Jan 13 20:34:09 2006
New Revision: 22128

Modified:
   z3/sqlos/branch/jinty-connection/src/sqlos/ftests/test_transaction.py
Log:
Add a new test to check that one thread can immediately see the changes committed in another one

Modified: z3/sqlos/branch/jinty-connection/src/sqlos/ftests/test_transaction.py
==============================================================================
--- z3/sqlos/branch/jinty-connection/src/sqlos/ftests/test_transaction.py	(original)
+++ z3/sqlos/branch/jinty-connection/src/sqlos/ftests/test_transaction.py	Fri Jan 13 20:34:09 2006
@@ -10,7 +10,7 @@
 $Id$
 """
 import unittest
-from transaction import get
+from transaction import get, begin
 
 from zope.app.testing.functional import BrowserTestCase
 from zope.app import zapi
@@ -118,6 +118,43 @@
         self.assertEqual(person2.username, 'sidnei')
         self.assertEqual(person2.password, 'test')
 
+    def testCleanCacheOnOverTransaction(self):
+        ut = zapi.getUtility(IConnectionName)
+        adapter = zapi.queryUtility(IZopeDatabaseAdapter, ut.name)
+        if adapter.getDSN() == 'dbi://:memory:':
+            import warnings
+            warnings.warn('Warning, not testing Cache Isolation')
+            return # this test is NOT going to work against the default test
+                   # database
+        person = self.person
+        personid = person.id
+        # warm up the cache by getting an instance
+        person2 = SamplePerson.get(personid)
+        self.assertEqual(person2.fullname, 'Sidnei da Silva')
+        self.assertEqual(person2.username, 'sidnei')
+        self.assertEqual(person2.password, 'test')
+        # comit this transaction "just in case"
+        get().commit()
+        # get change and commit the person in another thread
+        def changePerson():
+            person1 = SamplePerson.get(personid)
+            person1.fullname = 'Another Person'
+            person1.username = 'notsidnei'
+            person1.password = 'nottest'
+            get().commit()
+        import threading
+        thread = threading.Thread(target=changePerson)
+        thread.start()
+        thread.join()
+        # start a new transaction in this one
+        get().commit()
+        begin()
+        # If we get a new person, we should see the changed person
+        person3 = SamplePerson.get(personid)
+        self.assertEqual(person3.fullname, 'Another Person')
+        self.assertEqual(person3.username, 'notsidnei')
+        self.assertEqual(person3.password, 'nottest')
+
     def tearDown(self):
         self.person.destroySelf()
         super(TestTransaction, self).tearDown()