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

[jeremy@digicool.com]

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

Modified Files:
      Tag: ZEO-ZRPC-Dev
	testZEO.py 
Log Message:
Add new, specialized cache and conflict resolution test cases.

Add new OtherZEOTests TestCases class that doesn't use the standard
setUp() and tearDown().




--- Updated File testZEO.py in package Packages/ZEO --
--- testZEO.py	2001/05/07 13:12:54	1.1.2.11
+++ testZEO.py	2001/05/08 20:51:20	1.1.2.12
@@ -9,6 +9,7 @@
 
 import ZEO.ClientStorage, ZEO.StorageServer
 import ThreadedAsync, ZEO.trigger
+from ZODB.FileStorage import FileStorage
 
 from ZEO import zeolog
 from ZEO.tests import forker, Cache
@@ -30,6 +31,31 @@
     will get no later than the return value from vote.
     """
     
+    __super_setUp = StorageTestBase.StorageTestBase.setUp
+    __super_tearDown = StorageTestBase.StorageTestBase.tearDown
+
+    def setUp(self):
+        """Start a ZEO server using a Unix domain socket
+
+        The ZEO server uses the storage object returned by the
+        getStorage() method.
+        """
+        self.running = 1
+        client, exit, pid = forker.start_zeo(self.getStorage())
+        self._pid = pid
+        self._server = exit
+        self._storage = client
+        while not self._storage.is_connected():
+            time.sleep(0.1)
+        self.__super_setUp()
+
+    def tearDown(self):
+        """Try to cause the tests to halt"""
+        self.running = 0
+        self._server.close()
+        os.waitpid(self._pid, 0)
+        self.__super_tearDown()
+
     def _dostore(self, oid=None, revid=None, data=None, version=None,
                  already_pickled=0):
         """Do a complete storage transaction.
@@ -84,11 +110,13 @@
         
 class GenericTests(ZEOTestBase,
                    Cache.StorageWithCache,
+                   Cache.TransUndoStorageWithCache,
                    BasicStorage.BasicStorage,
                    VersionStorage.VersionStorage,
                    PackableStorage.PackableStorage,
                    Synchronization.SynchronizedStorage,
                    ConflictResolution.ConflictResolvingStorage,
+                   ConflictResolution.ConflictResolvingTransUndoStorage,
                    TransactionalUndoStorage.TransactionalUndoStorage,
       TransactionalUndoVersionStorage.TransactionalUndoVersionStorage,
                    ):
@@ -100,48 +128,43 @@
     returns a specific storage, e.g. FileStorage.
     """
 
-    __super_setUp = StorageTestBase.StorageTestBase.setUp
-    __super_tearDown = StorageTestBase.StorageTestBase.tearDown
-
-    def setUp(self):
-        """Start a ZEO server using a Unix domain socket
-
-        The ZEO server uses the storage object returned by the
-        getStorage() method.
-        """
-        self.running = 1
-        client, exit, pid = forker.start_zeo(self.getStorage())
-        self._pid = pid
-        self._server = exit
-        self._storage = client
-        while not self._storage.is_connected():
-            time.sleep(0.1)
-        self.__super_setUp()
-
-    def tearDown(self):
-        """Try to cause the tests to halt"""
-        self.running = 0
-        self._server.close()
-        os.waitpid(self._pid, 0)
-        self.__super_tearDown()
-
 class ZEOFileStorageTests(GenericTests):
+    """Tests of storage behavior using FileStorage underneath ZEO"""
     __super_setUp = GenericTests.setUp
     
-    from ZODB.FileStorage import FileStorage
-
     def setUp(self):
         self.__fs_base = tempfile.mktemp()
         self.__super_setUp()
 
     def getStorage(self):
-        return self.FileStorage(self.__fs_base, create=1)
+        return FileStorage(self.__fs_base, create=1)
 
     def delStorage(self):
         # file storage appears to create three files
         for ext in '', '.index', '.lock', '.tmp':
             path = self.__fs_base + ext
             os.unlink(path)
+
+class OtherZEOTests(unittest.TestCase):
+    """Tests of ZEO that are basically independent of the storage"""
+
+    def checkBadStorageID(self):
+        tmp = tempfile.mktemp()
+        fs = FileStorage(tmp, create=1)
+        try:
+            client, exit, pid = forker.start_zeo(fs, storage_id="gotcha")
+        except RuntimeError, msg:
+            err, exit, pid = msg
+            assert isinstance(err, ValueError)
+        else:
+            assert 0, "Established ZEO connection with invalid storage id"
+        exit.close()
+        os.waitpid(pid, 0)
+        fs.close()
+        for ext in '', '.index', '.lock', '.tmp':
+            path = tmp + ext
+            if os.path.exists(path):
+                os.unlink(path)
         
 def main():
     import sys, getopt
@@ -158,6 +181,7 @@
         return 0
     
     tests = unittest.makeSuite(ZEOFileStorageTests, 'check' + name_of_test)
+    tests.addTest(OtherZEOTests("checkBadStorageID"))
     runner = unittest.TextTestRunner()
     runner.run(tests)