[Zodb-checkins] CVS: ZODB3/ZEO/tests - ConnectionTests.py:1.40.2.6
testConnection.py:1.16.2.3
Jeremy Hylton
jeremy at zope.com
Thu Sep 11 00:27:52 EDT 2003
Update of /cvs-repository/ZODB3/ZEO/tests
In directory cvs.zope.org:/tmp/cvs-serv24879/ZEO/tests
Modified Files:
Tag: ZODB3-3_2-branch
ConnectionTests.py testConnection.py
Log Message:
Move to tests that never call startServer(keep=0) to a separate class.
If you don't pass keep=0, the server will never clean up its files.
This fixes the left-behind files for FileStorage. For some reason,
BDBStorage isn't cleaning up after itself.
=== ZODB3/ZEO/tests/ConnectionTests.py 1.40.2.5 => 1.40.2.6 ===
--- ZODB3/ZEO/tests/ConnectionTests.py:1.40.2.5 Sat Sep 6 22:26:14 2003
+++ ZODB3/ZEO/tests/ConnectionTests.py Wed Sep 10 23:27:51 2003
@@ -571,6 +571,70 @@
db2.close()
db1.close()
+class InvqTests(CommonSetupTearDown):
+ invq = 2
+
+ def checkQuickVerificationWith2Clients(self):
+ perstorage = self.openClientStorage(cache="test")
+ self.assertEqual(perstorage.verify_result, "full verification")
+
+ self._storage = self.openClientStorage()
+ oid = self._storage.new_oid()
+ # When we create a new storage, it should always do a full
+ # verification
+ self.assertEqual(self._storage.verify_result, "full verification")
+ # do two storages of the object to make sure an invalidation
+ # message is generated
+ revid = self._dostore(oid)
+ revid = self._dostore(oid, revid)
+
+ perstorage.load(oid, '')
+ perstorage.close()
+
+ revid = self._dostore(oid, revid)
+
+ perstorage = self.openClientStorage(cache="test")
+ self.assertEqual(perstorage.verify_result, "quick verification")
+
+ self.assertEqual(perstorage.load(oid, ''),
+ self._storage.load(oid, ''))
+ perstorage.close()
+
+ def checkVerificationWith2ClientsInvqOverflow(self):
+ perstorage = self.openClientStorage(cache="test")
+ self.assertEqual(perstorage.verify_result, "full verification")
+
+ self._storage = self.openClientStorage()
+ oid = self._storage.new_oid()
+ # When we create a new storage, it should always do a full
+ # verification
+ self.assertEqual(self._storage.verify_result, "full verification")
+ # do two storages of the object to make sure an invalidation
+ # message is generated
+ revid = self._dostore(oid)
+ revid = self._dostore(oid, revid)
+
+ perstorage.load(oid, '')
+ perstorage.close()
+
+ # the test code sets invq bound to 2
+ for i in range(5):
+ revid = self._dostore(oid, revid)
+
+ perstorage = self.openClientStorage(cache="test")
+ self.assertEqual(perstorage.verify_result, "full verification")
+ t = time.time() + 30
+ while not perstorage.end_verify.isSet():
+ perstorage.sync()
+ if time.time() > t:
+ self.fail("timed out waiting for endVerify")
+
+ self.assertEqual(self._storage.load(oid, '')[1], revid)
+ self.assertEqual(perstorage.load(oid, ''),
+ self._storage.load(oid, ''))
+
+ perstorage.close()
+
class ReconnectionTests(CommonSetupTearDown):
# The setUp() starts a server automatically. In order for its
# state to persist, we set the class variable keep to 1. In
@@ -760,69 +824,6 @@
self.assertEqual(perstorage.verify_result, "no verification")
perstorage.close()
self._storage.close()
-
- def checkQuickVerificationWith2Clients(self):
- perstorage = self.openClientStorage(cache="test")
- self.assertEqual(perstorage.verify_result, "full verification")
-
- self._storage = self.openClientStorage()
- oid = self._storage.new_oid()
- # When we create a new storage, it should always do a full
- # verification
- self.assertEqual(self._storage.verify_result, "full verification")
- # do two storages of the object to make sure an invalidation
- # message is generated
- revid = self._dostore(oid)
- revid = self._dostore(oid, revid)
-
- perstorage.load(oid, '')
- perstorage.close()
-
- revid = self._dostore(oid, revid)
-
- perstorage = self.openClientStorage(cache="test")
- self.assertEqual(perstorage.verify_result, "quick verification")
-
- self.assertEqual(perstorage.load(oid, ''),
- self._storage.load(oid, ''))
- perstorage.close()
-
-
-
- def checkVerificationWith2ClientsInvqOverflow(self):
- perstorage = self.openClientStorage(cache="test")
- self.assertEqual(perstorage.verify_result, "full verification")
-
- self._storage = self.openClientStorage()
- oid = self._storage.new_oid()
- # When we create a new storage, it should always do a full
- # verification
- self.assertEqual(self._storage.verify_result, "full verification")
- # do two storages of the object to make sure an invalidation
- # message is generated
- revid = self._dostore(oid)
- revid = self._dostore(oid, revid)
-
- perstorage.load(oid, '')
- perstorage.close()
-
- # the test code sets invq bound to 2
- for i in range(5):
- revid = self._dostore(oid, revid)
-
- perstorage = self.openClientStorage(cache="test")
- self.assertEqual(perstorage.verify_result, "full verification")
- t = time.time() + 30
- while not perstorage.end_verify.isSet():
- perstorage.sync()
- if time.time() > t:
- self.fail("timed out waiting for endVerify")
-
- self.assertEqual(self._storage.load(oid, '')[1], revid)
- self.assertEqual(perstorage.load(oid, ''),
- self._storage.load(oid, ''))
-
- perstorage.close()
class TimeoutTests(CommonSetupTearDown):
timeout = 1
=== ZODB3/ZEO/tests/testConnection.py 1.16.2.2 => 1.16.2.3 ===
--- ZODB3/ZEO/tests/testConnection.py:1.16.2.2 Wed Sep 10 22:48:07 2003
+++ ZODB3/ZEO/tests/testConnection.py Wed Sep 10 23:27:51 2003
@@ -57,19 +57,25 @@
class FileStorageReconnectionTests(
FileStorageConfig,
- ConnectionTests.ReconnectionTests
+ ConnectionTests.ReconnectionTests,
):
"""FileStorage-specific re-connection tests."""
# Run this at level 1 because MappingStorage can't do reconnection tests
level = 1
+class FileStorageInvqTests(
+ FileStorageConfig,
+ ConnectionTests.InvqTests
+ ):
+ """FileStorage-specific invalidation queue tests."""
+ level = 1
+
class FileStorageTimeoutTests(
FileStorageConfig,
ConnectionTests.TimeoutTests
):
level = 2
-
class BDBConnectionTests(
BerkeleyStorageConfig,
ConnectionTests.ConnectionTests,
@@ -80,11 +86,19 @@
class BDBReconnectionTests(
BerkeleyStorageConfig,
- ConnectionTests.ReconnectionTests
+ ConnectionTests.ReconnectionTests,
+ ConnectionTests.InvqTests
):
"""Berkeley storage re-connection tests."""
level = 2
+class BDBInvqTests(
+ BerkeleyStorageConfig,
+ ConnectionTests.InvqTests
+ ):
+ """Berkeley storage invalidation queue tests."""
+ level = 2
+
class BDBTimeoutTests(
BerkeleyStorageConfig,
ConnectionTests.TimeoutTests
@@ -112,16 +126,17 @@
test_classes = [FileStorageConnectionTests,
FileStorageReconnectionTests,
+ FileStorageInvqTests,
FileStorageTimeoutTests,
MappingStorageConnectionTests,
MappingStorageTimeoutTests]
import BDBStorage
if BDBStorage.is_available:
- test_classes.append(BDBConnectionTests)
- test_classes.append(BDBReconnectionTests)
- test_classes.append(BDBTimeoutTests)
-
+ test_classes += [BDBConnectionTests,
+ BDBReconnectionTests,
+ BDBInvqTests,
+ BDBTimeoutTests]
def test_suite():
suite = unittest.TestSuite()
More information about the Zodb-checkins
mailing list