[Zodb-checkins] CVS: Packages/ZEO - testZEO.py:1.1.2.3

[jeremy@digicool.com]

Update of /cvs-repository/Packages/ZEO/tests
In directory korak:/tmp/cvs-serv16022/tests

Modified Files:
      Tag: ZEO-ZRPC-Dev
	testZEO.py 
Log Message:
Add ZEOTestBase that enables many tests to succeed with current ZEO

ZEOTestBase modifies _dostore() to account for the different way that
ZEO currently handles serialno/ConflictError returns.






--- Updated File testZEO.py in package Packages/ZEO --
--- testZEO.py	2001/04/18 21:02:39	1.1.2.2
+++ testZEO.py	2001/04/19 16:59:53	1.1.2.3
@@ -5,6 +5,7 @@
 import sys
 import tempfile
 import time
+import types
 import unittest
 
 import ZEO.ClientStorage, ZEO.StorageServer
@@ -29,8 +30,66 @@
     def shutdown(self):
         self._StorageProxy__storage.close()
         os._exit(0)
+
+ZERO = '\0'*8
+import pickle
+
+class ZEOTestBase(StorageTestBase.StorageTestBase):
+    """Version of the storage test class that supports ZEO.
+    
+    For ZEO, we don't always get the serialno/exception for a
+    particular store as the return value from the store.   But we
+    will get no later than the return value from vote.
+    """
+    
+    def _dostore(self, oid=None, revid=None, data=None, version=None):
+        """Do a complete storage transaction.
+
+        The defaults are:
+         - oid=None, ask the storage for a new oid
+         - revid=None, use a revid of ZERO
+         - data=None, pickle up some arbitrary data (the integer 7)
+         - version=None, use the empty string version
+        
+        Returns the object's new revision id.
+        """
+        if oid is None:
+            oid = self._storage.new_oid()
+        if revid is None:
+            revid = ZERO
+        if data is None:
+            data = pickle.dumps(7)
+        else:
+            data = pickle.dumps(data)
+        if version is None:
+            version = ''
+        # Begin the transaction
+        self._storage.tpc_begin(self._transaction)
+        # Store an object
+        r1 = self._storage.store(oid, revid, data, version,
+                                 self._transaction)
+        # Finish the transaction
+        r2 = self._storage.tpc_vote(self._transaction)
+        self._storage.tpc_finish(self._transaction)
+        return self._get_serial([r1, r2])[oid]
+
+    def _get_serial(self, resps):
+        """Return oid -> serialno dict from sequence of ZEO replies."""
+        d = {}
+        for r in resps:
+            if not r:
+                continue
+            if type(r) == types.StringType:
+                raise RuntimeError, "unexpected ZEO response: no oid"
+            else:
+                for oid, serial in r:
+                    if type(serial) != types.StringType:
+                        raise s
+                    else:
+                        d[oid] = serial
+        return d
         
-class GenericTests(StorageTestBase.StorageTestBase,
+class GenericTests(ZEOTestBase,
                    BasicStorage.BasicStorage,
                    VersionStorage.VersionStorage
                    ):