Commit 32bad8be authored by Jeremy Hylton's avatar Jeremy Hylton

Three new tests:

checkBuggyResolve2() -- _p_resolveConflict takes too few args
checkUndoConflictResolution() -- make sure conflict resolution is
     invoked properly during transactional undo
checkUndoUnresolvable() -- make sure transactional undo can cope with
     failed conflict resolution
parent 07cd63ab
"""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)
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment