[Zodb-checkins] CVS: ZODB3/ZEO/tests - ConnectionTests.py:1.21
Jeremy Hylton
jeremy@zope.com
Wed, 9 Apr 2003 17:26:53 -0400
Update of /cvs-repository/ZODB3/ZEO/tests
In directory cvs.zope.org:/tmp/cvs-serv9868/ZEO/tests
Modified Files:
ConnectionTests.py
Log Message:
Fix ZEO to work correctly with atomic invalidations.
=== ZODB3/ZEO/tests/ConnectionTests.py 1.20 => 1.21 ===
--- ZODB3/ZEO/tests/ConnectionTests.py:1.20 Tue Mar 4 14:56:46 2003
+++ ZODB3/ZEO/tests/ConnectionTests.py Wed Apr 9 17:26:52 2003
@@ -31,6 +31,7 @@
from ZEO.zrpc.marshal import Marshaller
from ZEO.tests import forker
+from ZODB.DB import DB
from ZODB.Transaction import get_transaction, Transaction
from ZODB.POSException import ReadOnlyError
from ZODB.tests.StorageTestBase import StorageTestBase
@@ -61,6 +62,7 @@
invq = None
timeout = None
monitor = 0
+ db_class = DummyDB
def setUp(self):
"""Test setup for connection tests.
@@ -472,6 +474,36 @@
for t in threads:
t.closeclients()
+ def checkCrossDBInvalidations(self):
+ db1 = DB(self.openClientStorage())
+ c1 = db1.open()
+ r1 = c1.root()
+
+ r1["a"] = MinPO("a")
+ get_transaction().commit()
+
+ db2 = DB(self.openClientStorage())
+ r2 = db2.open().root()
+
+ self.assertEqual(r2["a"].value, "a")
+
+ r2["b"] = MinPO("b")
+ get_transaction().commit()
+
+ # make sure the invalidation is received in the other client
+ c1._storage.sync()
+ self.assert_(r1._p_oid in c1._invalidated)
+
+ # force the invalidations to be applied...
+ c1.setLocalTransaction()
+ c1.getTransaction().register(c1)
+ c1.getTransaction().abort()
+ r1.keys() # unghostify
+ self.assertEqual(r1._p_serial, r2._p_serial)
+
+ db2.close()
+ db1.close()
+
class ReconnectionTests(CommonSetupTearDown):
keep = 1
invq = 2