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

[jinty at codespeak.net]

Author: jinty
Date: Fri Jan 13 20:04:10 2006
New Revision: 22124

Modified:
   z3/sqlos/branch/jinty-connection/src/sqlos/ftests/test_transaction.py
Log:
Make a test that shows what happens when you use a global cache shared between threads.

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:04:10 2006
@@ -13,7 +13,10 @@
 from transaction import get
 
 from zope.app.testing.functional import BrowserTestCase
+from zope.app import zapi
+from zope.app.rdb.interfaces import IZopeDatabaseAdapter
 
+from sqlos.interfaces import IConnectionName
 from sqlos.testing.sampleperson import SamplePerson
 
 __metaclass__ = type
@@ -82,6 +85,39 @@
             self.assertEqual(person.username, 'dreamcatcher')
             self.assertEqual(person.password, 'pass')
 
+    def testCacheThreadIsolation(self):
+        """Tests that the changes we make in one thread don't appear in another.
+
+        This is a regression test for if the cache makes breaks the isolation
+        between threads.
+        """
+        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')
+        def changePerson():
+            person1 = SamplePerson.get(personid)
+            person1.fullname = 'Another Person'
+            person1.username = 'notsidnei'
+            person1.password = 'nottest'
+        import threading
+        thread = threading.Thread(target=changePerson)
+        thread.start()
+        thread.join()
+        self.assertEqual(person2.fullname, 'Sidnei da Silva')
+        self.assertEqual(person2.username, 'sidnei')
+        self.assertEqual(person2.password, 'test')
+
     def tearDown(self):
         self.person.destroySelf()
         super(TestTransaction, self).tearDown()