diff --git a/src/ZODB/tests/ConflictResolution.py b/src/ZODB/tests/ConflictResolution.py index 1aa5b42117916a7ecfe3adae4ef99b11d2f5fe90..22551bd8e03d62b2c6aa1699bf4b4125eec9f780 100644 --- a/src/ZODB/tests/ConflictResolution.py +++ b/src/ZODB/tests/ConflictResolution.py @@ -1,7 +1,7 @@ """Tests for application-level conflict resolution.""" from ZODB.Transaction import Transaction -from ZODB.POSException import ConflictError +from ZODB.POSException import ConflictError, UndoError from Persistence import Persistent from ZODB.tests.StorageTestBase import zodb_unpickle, zodb_pickle @@ -41,6 +41,10 @@ class PCounter3(PCounter): def _p_resolveConflict(self, oldState, savedState, newState): raise AttributeError, "no attribute" +class PCounter4(PCounter): + def _p_resolveConflict(self, oldState, savedState): + raise RuntimeError, "Can't get here; not enough args" + class ConflictResolvingStorage: def checkResolve(self): @@ -82,7 +86,7 @@ class ConflictResolvingStorage: oid, revid=revid1, data=zodb_pickle(obj)) - def checkBuggyResolve(self): + def checkBuggyResolve1(self): obj = PCounter3() obj.inc() @@ -100,4 +104,63 @@ class ConflictResolvingStorage: self._dostoreNP, oid, revid=revid1, data=zodb_pickle(obj)) - # XXX test conflict error raised during undo + def checkBuggyResolve2(self): + obj = PCounter4() + obj.inc() + + oid = self._storage.new_oid() + + revid1 = self._dostoreNP(oid, data=zodb_pickle(obj)) + + obj.inc() + obj.inc() + # The effect of committing two transactions with the same + # pickle is to commit two different transactions relative to + # revid1 that add two to _value. + revid2 = self._dostoreNP(oid, revid=revid1, data=zodb_pickle(obj)) + self.assertRaises(TypeError, + self._dostoreNP, + oid, revid=revid1, data=zodb_pickle(obj)) + + def checkUndoConflictResolution(self): + # This test is based on checkNotUndoable in the + # TransactionalUndoStorage test suite. Except here, conflict + # resolution should allow us to undo the transaction anyway. + + obj = PCounter() + obj.inc() + oid = self._storage.new_oid() + revid_a = self._dostore(oid, data=obj) + obj.inc() + revid_b = self._dostore(oid, revid=revid_a, data=obj) + obj.inc() + revid_c = self._dostore(oid, revid=revid_b, data=obj) + # Start the undo + info = self._storage.undoInfo() + tid = info[1]['id'] + self._storage.tpc_begin(self._transaction) + self._storage.transactionalUndo(tid, self._transaction) + self._storage.tpc_finish(self._transaction) + + def checkUndoUnresolvable(self): + # This test is based on checkNotUndoable in the + # TransactionalUndoStorage test suite. Except here, conflict + # resolution should allow us to undo the transaction anyway. + + obj = PCounter2() + obj.inc() + oid = self._storage.new_oid() + revid_a = self._dostore(oid, data=obj) + obj.inc() + revid_b = self._dostore(oid, revid=revid_a, data=obj) + obj.inc() + revid_c = self._dostore(oid, revid=revid_b, data=obj) + # Start the undo + info = self._storage.undoInfo() + tid = info[1]['id'] + self._storage.tpc_begin(self._transaction) + self.assertRaises(UndoError, + self._storage.transactionalUndo, + tid, self._transaction) + self._storage.tpc_abort(self._transaction) +