[Zodb-checkins] CVS: StandaloneZODB/ZEO/tests - testZEO.py:1.9.2.7

[Jeremy Hylton]

Update of /cvs-repository/StandaloneZODB/ZEO/tests
In directory cvs.zope.org:/tmp/cvs-serv7058

Modified Files:
      Tag: zeo-1_0-branch
	testZEO.py 
Log Message:
Add two new tests of the distributed commit lock.

The tests cover the normal case (a waiting transaction begins) and the
specific error case that the client disconnects before its transaction
gets a chance to run.  There is no test of non-disconnect errors.


=== StandaloneZODB/ZEO/tests/testZEO.py 1.9.2.6 => 1.9.2.7 ===
 import ThreadedAsync, ZEO.trigger
 from ZODB.FileStorage import FileStorage
+from ZODB.TimeStamp import TimeStamp
+from ZODB.Transaction import Transaction
 import thread
 
 from ZEO.tests import forker, Cache
@@ -72,15 +74,15 @@
         if version is None:
             version = ''
         # Begin the transaction
-        self._storage.tpc_begin(self._transaction)
+        t = Transaction()
+        self._storage.tpc_begin(t)
         # Store an object
-        r1 = self._storage.store(oid, revid, data, version,
-                                 self._transaction)
+        r1 = self._storage.store(oid, revid, data, version, t)
         s1 = self._get_serial(r1)
         # Finish the transaction
-        r2 = self._storage.tpc_vote(self._transaction)
+        r2 = self._storage.tpc_vote(t)
         s2 = self._get_serial(r2)
-        self._storage.tpc_finish(self._transaction)
+        self._storage.tpc_finish(t)
         # s1, s2 can be None or dict
         assert not (s1 and s2)
         return s1 and s1[oid] or s2 and s2[oid]
@@ -142,13 +144,13 @@
         The ZEO server uses the storage object returned by the
         getStorage() method.
         """
+        self.__super_setUp()
         self.running = 1
         client, exit, pid = forker.start_zeo(self.getStorage())
         self._pid = pid
         self._server = exit
         self._storage = PackWaitWrapper(client)
         client.registerDB(DummyDB(), None)
-        self.__super_setUp()
 
     def tearDown(self):
         """Try to cause the tests to halt"""
@@ -159,9 +161,66 @@
         self.delStorage()
         self.__super_tearDown()
 
+    def checkTwoArgBegin(self):
+        # XXX ZEO doesn't support 2-arg begin
+        pass
+
     def checkLargeUpdate(self):
         obj = MinPO("X" * (10 * 128 * 1024))
         self._dostore(data=obj)
+
+    def checkCommitLockOnCommit(self):
+        self._checkCommitLock("tpc_finish")
+
+    def checkCommitLockOnAbort(self):
+        self._checkCommitLock("tpc_abort")
+
+    def _checkCommitLock(self, method_name):
+        # check the commit lock when a client attemps a transaction,
+        # but fails/exits before finishing the commit.
+
+        # Start on transaction normally.
+        t = Transaction()
+        self._storage.tpc_begin(t)
+
+        # Start a second transaction on a different connection without
+        # blocking the test thread.
+        self._storages = []
+        for i in range(3):
+            storage2 = self._duplicate_client()
+            t2 = Transaction()
+            tid = self._get_timestamp()
+            storage2._call.sendMessage('tpc_begin_sync',
+                                       tid, t2.user, t2.description,
+                                       t2._extension)
+            if i == 0:
+                storage2.close()
+            else:
+                self._storages.append((storage2, t2))
+
+        oid = self._storage.new_oid()
+        self._storage.store(oid, None, '', '', t)
+        self._storage.tpc_vote(t)
+        self._storage.tpc_finish(t)
+
+        for store, trans in self._storages:
+            store.tpc_abort(trans)
+            store.close()
+
+        # Make sure the server is still responsive
+        self._dostore()
+
+    def _duplicate_client(self):
+        "Open another ClientStorage to the same server."
+        addr = self._storage._connection
+        new = ZEO.ClientStorage.ClientStorage(addr)
+        new.registerDB(DummyDB(), None)
+        return new
+
+    def _get_timestamp(self):
+        t = time.time()
+        t = apply(TimeStamp,(time.gmtime(t)[:5]+(t%60,)))
+        return 't'
 
 class ZEOFileStorageTests(GenericTests):
     __super_setUp = GenericTests.setUp