[Zope3-checkins] CVS: Zope3/lib/python/ZODB/tests - BasicStorage.py:1.20 ConflictResolution.py:1.9 HistoryStorage.py:1.9 IteratorStorage.py:1.14 MTStorage.py:1.8 PackableStorage.py:1.15 ReadOnlyStorage.py:1.5 StorageTestBase.py:1.17 Synchronization.py:1.7 TransactionalUndoStorage.py:1.23 TransactionalUndoVersionStorage.py:1.9 VersionStorage.py:1.16 testConnection.py:1.4 testFileStorage.py:1.19 testTransaction.py:1.10 testZODB.py:1.7
Jeremy Hylton
jeremy@zope.com
Wed, 24 Jul 2002 19:12:47 -0400
Update of /cvs-repository/Zope3/lib/python/ZODB/tests
In directory cvs.zope.org:/tmp/cvs-serv3960/ZODB/tests
Modified Files:
BasicStorage.py ConflictResolution.py HistoryStorage.py
IteratorStorage.py MTStorage.py PackableStorage.py
ReadOnlyStorage.py StorageTestBase.py Synchronization.py
TransactionalUndoStorage.py TransactionalUndoVersionStorage.py
VersionStorage.py testConnection.py testFileStorage.py
testTransaction.py testZODB.py
Log Message:
Implement new Transaction APIs.
Add ZODB.ZTransaction.Transaction, which extends the basic transaction
object with user, description, and extended metadata.
XXX The rollback() part of the savepoint() protocol doesn't work yet.
There's an XXX in the code that explains why, and the test of it is
not enabled.
Connect.py:
Also change the registerDB() protocol to omit the limit argument,
which was not used in any implementation.
Refactor the ExportImport hooks to use a single hook method --
importHook() to do all the work. Remove the onCommitActions()
framework.
Delete all the code related to reset for persistence-based classes,
since it was complex and unused by the new Persistence.Class. Also
remove exchange() method which was used for ZClass support.
Use Set(dict) to replace many of the small lists of modified objects.
Remove the cacheGC attribute on Connection.
DB.py:
Fixup AbortVersion, CommitVersion, and TransactionalUndo data
managers. Each is now a subclass of the base SimpleDataManager.
ExportImport.py:
Changes for importHook() as mentioned above.
Also, lots of little code cleanup.
TmpStore.py:
Refactor enough to support rollback() at the storage level.
=== Zope3/lib/python/ZODB/tests/BasicStorage.py 1.19 => 1.20 ===
$Id$
"""
-from Transaction import Transaction
+from ZODB.ZTransaction import Transaction
from ZODB import POSException
from ZODB.tests.MinPO import MinPO
=== Zope3/lib/python/ZODB/tests/ConflictResolution.py 1.8 => 1.9 ===
##############################################################################
"""Tests for application-level conflict resolution."""
-from Transaction import Transaction
+from ZODB.ZTransaction import Transaction
from ZODB.POSException import ConflictError, UndoError
from Persistence import Persistent
=== Zope3/lib/python/ZODB/tests/HistoryStorage.py 1.8 => 1.9 ===
all these tests.
"""
-from Transaction import Transaction
+from ZODB.ZTransaction import Transaction
from ZODB.tests.MinPO import MinPO
from ZODB.tests.StorageTestBase import zodb_unpickle
=== Zope3/lib/python/ZODB/tests/IteratorStorage.py 1.13 => 1.14 ===
from ZODB.tests.MinPO import MinPO
from ZODB.tests.StorageTestBase import zodb_unpickle
from ZODB.utils import U64, p64
-from Transaction import Transaction
+from ZODB.ZTransaction import Transaction
class IteratorCompare:
=== Zope3/lib/python/ZODB/tests/MTStorage.py 1.7 => 1.8 ===
import ZODB.DB
from Persistence.PersistentMapping import PersistentMapping
-from Transaction import Transaction, get_transaction
+from Transaction import get_transaction
+from ZODB.ZTransaction import Transaction
from ZODB.tests.StorageTestBase \
import StorageTestBase, zodb_pickle, zodb_unpickle, handle_serials
=== Zope3/lib/python/ZODB/tests/PackableStorage.py 1.14 => 1.15 ===
self._storage.load(ZERO, '')
except KeyError:
from Persistence.PersistentMapping import PersistentMapping
- from Transaction import Transaction
+ from ZODB.ZTransaction import Transaction
file = StringIO()
p = cPickle.Pickler(file, 1)
p.dump((PersistentMapping, None))
=== Zope3/lib/python/ZODB/tests/ReadOnlyStorage.py 1.4 => 1.5 ===
#
##############################################################################
from ZODB.POSException import ReadOnlyError
-from Transaction import Transaction
+from ZODB.ZTransaction import Transaction
class ReadOnlyStorage:
=== Zope3/lib/python/ZODB/tests/StorageTestBase.py 1.16 => 1.17 ===
from cPickle import Pickler, Unpickler
from cStringIO import StringIO
-from Transaction import Transaction
+from ZODB.ZTransaction import Transaction
from ZODB.tests.MinPO import MinPO
=== Zope3/lib/python/ZODB/tests/Synchronization.py 1.6 => 1.7 ===
"""
-from Transaction import Transaction
+from ZODB.ZTransaction import Transaction
from ZODB.POSException import StorageTransactionError
VERSION = "testversion"
=== Zope3/lib/python/ZODB/tests/TransactionalUndoStorage.py 1.22 => 1.23 ===
import types
from ZODB import POSException
-from Transaction import Transaction
+from ZODB.ZTransaction import Transaction
from ZODB.tests.MinPO import MinPO
from ZODB.tests.StorageTestBase import zodb_pickle, zodb_unpickle
=== Zope3/lib/python/ZODB/tests/TransactionalUndoVersionStorage.py 1.8 => 1.9 ===
# that supports both transactionalUndo() and versions must pass these tests.
from ZODB import POSException
-from Transaction import Transaction
+from ZODB.ZTransaction import Transaction
from ZODB.tests.MinPO import MinPO
from ZODB.tests.StorageTestBase import zodb_unpickle
=== Zope3/lib/python/ZODB/tests/VersionStorage.py 1.15 => 1.16 ===
# code. Barry and Jeremy didn't understand versions then.
from ZODB import POSException
-from Transaction import Transaction
+from ZODB.ZTransaction import Transaction
from ZODB.tests.MinPO import MinPO
from ZODB.tests.StorageTestBase import zodb_unpickle
=== Zope3/lib/python/ZODB/tests/testConnection.py 1.3 => 1.4 ===
from Transaction.tests.abstestIDataManager import IDataManagerTests
from ZODB.MappingStorage import DB
+from ZODB.ZTransaction import Transaction
class P(Persistent):
pass
@@ -27,7 +28,14 @@
self.db = DB()
self.datamgr = self.db.open()
self.obj = P()
+ self.txn_factory = Transaction
+ def get_transaction(self):
+ t = super(ConnectionTests, self).get_transaction()
+ t.setUser('IDataManagerTests')
+ t.note('dummy note')
+ return t
+
def tearDown(self):
self.datamgr.close()
self.db.close()
=== Zope3/lib/python/ZODB/tests/testFileStorage.py 1.18 => 1.19 ===
import ZODB.FileStorage
import sys, os, unittest
import errno
-from Transaction import Transaction
+from ZODB.ZTransaction import Transaction
from ZODB.tests import StorageTestBase, BasicStorage, \
TransactionalUndoStorage, VersionStorage, \
=== Zope3/lib/python/ZODB/tests/testTransaction.py 1.9 => 1.10 ===
def checkSingleCommit(self, subtrans=None):
self.root["a"] = MinPO("a")
- get_transaction().commit(subtrans)
- assert self.root["a"].value == "a"
+ if subtrans:
+ get_transaction().savepoint()
+ else:
+ get_transaction().commit()
+ self.assertEqual(self.root["a"].value, "a")
def checkMultipleCommits(self, subtrans=None):
a = self.root["a"] = MinPO("a")
- get_transaction().commit(subtrans)
+ get_transaction().commit()
a.extra_attr = MinPO("b")
- get_transaction().commit(subtrans)
+ if subtrans:
+ get_transaction().savepoint()
+ else:
+ get_transaction().commit()
del a
- assert self.root["a"].value == "a"
- assert self.root["a"].extra_attr == MinPO("b")
+ self.assertEqual(self.root["a"].value, "a")
+ self.assertEqual(self.root["a"].extra_attr, MinPO("b"))
def checkCommitAndAbort(self, subtrans=None):
a = self.root["a"] = MinPO("a")
- get_transaction().commit(subtrans)
+ if subtrans:
+ get_transaction().savepoint()
+ else:
+ get_transaction().commit()
a.extra_attr = MinPO("b")
get_transaction().abort()
del a
if subtrans:
- assert not self.root.has_key("a")
+ self.assert_("a" not in self.root)
else:
- assert self.root["a"].value == "a"
- assert not hasattr(self.root["a"], 'extra_attr')
+ self.assertEqual(self.root["a"].value, "a")
+ self.assert_(not hasattr(self.root["a"], 'extra_attr'))
-class SubtransTests:
+class SubtransTests(BasicTests):
def wrap_test(self, klass, meth_name):
- obj = klass()
- obj.root = self.root
- meth = getattr(obj, meth_name)
- meth(1)
+ unbound_method = getattr(klass, meth_name)
+ unbound_method(self, 1)
get_transaction().commit()
checkSubSingleCommit = lambda self:\
@@ -89,7 +96,21 @@
"checkCommitAndAbort")
class AllTests(TransactionTestBase, BasicTests, SubtransTests):
- pass
+
+ def XXXcheckSavepointAndRollback(self):
+ self.root["a"] = MinPO()
+ rb1 = get_transaction().savepoint()
+ self.root["b"] = MinPO()
+ rb2 = get_transaction().savepoint()
+ self.assertEqual(len(self.root), 2)
+ self.assert_("a" in self.root)
+ self.assert_("b" in self.root)
+
+ rb2.rollback()
+ self.assertEqual(len(self.root), 1)
+ self.assert_("a" in self.root)
+ self.assert_("b" not in self.root)
+
def test_suite():
return unittest.makeSuite(AllTests, 'check')
=== Zope3/lib/python/ZODB/tests/testZODB.py 1.6 => 1.7 ===
# FOR A PARTICULAR PURPOSE.
#
##############################################################################
-import sys, os
+import os
import unittest
import ZODB.DB, ZODB.FileStorage
from Persistence.PersistentMapping import PersistentMapping
from Transaction import get_transaction
+from ZODB.utils import u64
class ExportImportTests:
@@ -39,7 +40,8 @@
get_transaction().abort()
else:
get_transaction().commit()
- except:
+ except Exception, err:
+ print err
get_transaction().abort()
raise
finally:
@@ -62,22 +64,27 @@
raise
else:
if abort_it:
- assert 0, 'Did not abort duplication'
+ oid = ob2._p_oid
+ if oid is not None:
+ oid = u64(oid)
+ print oid, ob2.__class__, ob2._p_state
+ print ob2
+ self.fail("Did not abort duplication")
l1 = list(ob.items())
l1.sort()
l2 = list(ob2.items())
l2.sort()
- l1 = map(lambda (k, v): (k, v[0]), l1)
- l2 = map(lambda (k, v): (k, v[0]), l2)
- assert l1 == l2, 'Duplicate did not match'
- assert ob._p_oid != ob2._p_oid, 'Did not duplicate'
- assert ob._p_jar == ob2._p_jar, 'Not same connection'
+ l1 = [(k, v[0]) for k, v in l1]
+ l2 = [(k, v[0]) for k, v in l2]
+ self.assertEqual(l1, l2, 'Duplicate did not match')
+ self.assert_(ob._p_oid != ob2._p_oid, 'Did not duplicate')
+ self.assertEqual(ob._p_jar, ob2._p_jar, 'Not same connection')
oids = {}
for v in ob.values():
oids[v._p_oid] = 1
for v in ob2.values():
- assert not oids.has_key(v._p_oid), (
- 'Did not fully separate duplicate from original')
+ self.assert_(v._p_oid not in oids,
+ 'Did not fully separate duplicate from original')
get_transaction().commit()
finally:
conn.close()
@@ -112,6 +119,7 @@
def checkModifyGhost(self):
root = self._db.open().root()
o = root["test"][5]
+ o._p_activate()
o._p_deactivate()
o.anattr = "anattr"
self.assert_(o._p_changed)
@@ -130,13 +138,5 @@
def test_suite():
return unittest.makeSuite(ZODBTests, 'check')
-def main():
- alltests=test_suite()
- runner = unittest.TextTestRunner()
- runner.run(alltests)
-
if __name__=='__main__':
- if len(sys.argv) > 1:
- globals()[sys.argv[1]]()
- else:
- main()
+ unittest.TextTestRunner().run(test_suite())