From adb57282c2c64e7155420d4ac906d885bc73c5b7 Mon Sep 17 00:00:00 2001 From: Tres Seaver Date: Sun, 2 May 2010 18:49:54 +0000 Subject: [PATCH] Split out the ZODB protocol tests from the tests specific to the module. Make the local tests use "normal" unittest conventions. --- src/tempstorage/tests/testTemporaryStorage.py | 126 +++++++++++++----- 1 file changed, 89 insertions(+), 37 deletions(-) diff --git a/src/tempstorage/tests/testTemporaryStorage.py b/src/tempstorage/tests/testTemporaryStorage.py index 322768a..24502d0 100644 --- a/src/tempstorage/tests/testTemporaryStorage.py +++ b/src/tempstorage/tests/testTemporaryStorage.py @@ -17,29 +17,77 @@ from ZODB.tests import BasicStorage from ZODB.tests import Synchronization from ZODB.tests import ConflictResolution -from ZODB.tests import Corruption from ZODB.tests import MTStorage +class ZODBProtocolTests(StorageTestBase.StorageTestBase, + # not a revision storage, but passes + #RevisionStorage.RevisionStorage, + BasicStorage.BasicStorage, + Synchronization.SynchronizedStorage, + ConflictResolution.ConflictResolvingStorage, + MTStorage.MTStorage, + ): -class TemporaryStorageTests(StorageTestBase.StorageTestBase, - # not a revision storage, but passes - #RevisionStorage.RevisionStorage, - BasicStorage.BasicStorage, - Synchronization.SynchronizedStorage, - ConflictResolution.ConflictResolvingStorage, - MTStorage.MTStorage, - ): + def setUp(self): + StorageTestBase.StorageTestBase.setUp(self) + self.open() def open(self, **kwargs): from tempstorage.TemporaryStorage import TemporaryStorage self._storage = TemporaryStorage('foo') - def setUp(self): - StorageTestBase.StorageTestBase.setUp(self) - self.open() - def tearDown(self): - StorageTestBase.StorageTestBase.tearDown(self) +class TemporaryStorageTests(unittest.TestCase): + + def _getTargetClass(self): + from tempstorage.TemporaryStorage import TemporaryStorage + return TemporaryStorage + + def _makeOne(self, name='foo'): + return self._getTargetClass()(name) + + def _dostore(self, storage, oid=None, revid=None, data=None, + already_pickled=0, user=None, description=None): + # Borrowed from StorageTestBase, to allow passing storage. + """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) + + Returns the object's new revision id. + """ + import transaction + from ZODB.tests.MinPO import MinPO + + if oid is None: + oid = storage.new_oid() + if revid is None: + revid = StorageTestBase.ZERO + if data is None: + data = MinPO(7) + if type(data) == int: + data = MinPO(data) + if not already_pickled: + data = StorageTestBase.zodb_pickle(data) + # Begin the transaction + t = transaction.Transaction() + if user is not None: + t.user = user + if description is not None: + t.description = description + try: + storage.tpc_begin(t) + # Store an object + r1 = storage.store(oid, revid, data, '', t) + # Finish the transaction + r2 = storage.tpc_vote(t) + revid = StorageTestBase.handle_serials(oid, r1, r2) + storage.tpc_finish(t) + except: + storage.tpc_abort(t) + raise + return revid def _do_read_conflict(self, db, mvcc): import transaction @@ -75,55 +123,59 @@ def _do_read_conflict(self, db, mvcc): obj.child1 return obj - def checkConflictCacheIsCleared(self): + def test_conflict_cache_clears_over_time(self): import time from ZODB.tests.MinPO import MinPO - self._storage._conflict_cache_gcevery = 1 # second - self._storage._conflict_cache_maxage = 1 # second + storage = self._makeOne() + storage._conflict_cache_gcevery = 1 # second + storage._conflict_cache_maxage = 1 # second - oid = self._storage.new_oid() - self._dostore(oid, data=MinPO(5)) + oid = storage.new_oid() + self._dostore(storage,oid, data=MinPO(5)) time.sleep(2) - oid2 = self._storage.new_oid() - self._dostore(oid2, data=MinPO(10)) + oid2 = storage.new_oid() + self._dostore(storage,oid2, data=MinPO(10)) - oid3 = self._storage.new_oid() - self._dostore(oid3, data=MinPO(9)) + oid3 = storage.new_oid() + self._dostore(storage,oid3, data=MinPO(9)) - self.assertEqual(len(self._storage._conflict_cache), 2) + self.assertEqual(len(storage._conflict_cache), 2) time.sleep(2) - oid4 = self._storage.new_oid() - self._dostore(oid4, data=MinPO(11)) + oid4 = storage.new_oid() + self._dostore(storage,oid4, data=MinPO(11)) - self.assertEqual(len(self._storage._conflict_cache), 1) + self.assertEqual(len(storage._conflict_cache), 1) - def checkWithMVCCDoesntRaiseReadConflict(self): + def test_have_MVCC_ergo_no_ReadConflict(self): from ZODB.DB import DB from ZODB.tests.MinPO import MinPO - db = DB(self._storage) + storage = self._makeOne() + db = DB(storage) ob = self._do_read_conflict(db, True) self.assertEquals(ob.__class__, MinPO) self.assertEquals(getattr(ob, 'child1', MinPO()).value, 'child1') self.failIf(getattr(ob, 'child2', None)) - def checkLoadEx(self): + def test_load_ex_matches_load(self): from ZODB.tests.MinPO import MinPO - oid = self._storage.new_oid() - self._dostore(oid, data=MinPO(1)) - loadp, loads = self._storage.load(oid, 'whatever') - exp, exs, exv = self._storage.loadEx(oid, 'whatever') + storage = self._makeOne() + oid = storage.new_oid() + self._dostore(storage,oid, data=MinPO(1)) + loadp, loads = storage.load(oid, 'whatever') + exp, exs, exv = storage.loadEx(oid, 'whatever') self.assertEqual(loadp, exp) self.assertEqual(loads, exs) self.assertEqual(exv, '') def test_suite(): - # Note: we follow the ZODB 'check' pattern here so that the base - # class tests are picked up. return unittest.TestSuite(( - unittest.makeSuite(TemporaryStorageTests, 'check'), + unittest.makeSuite(TemporaryStorageTests), + # Note: we follow the ZODB 'check' pattern here so that the base + # class tests are picked up. + unittest.makeSuite(ZODBProtocolTests, 'check'), ))