Commit f726e40a authored by Julien Muchembled's avatar Julien Muchembled

DemoStorage: fix history()

- It failed with oids that already exist in the base storage.
- MappingStorage returned timestamps in wrong format.
parent 9d13627c
......@@ -113,7 +113,7 @@ class DemoStorage(ConflictResolvingStorage):
def _copy_methods_from_changes(self, changes):
for meth in (
'_lock_acquire', '_lock_release',
'getSize', 'history', 'isReadOnly',
'getSize', 'isReadOnly',
'sortKey', 'tpc_transaction', 'tpc_vote',
):
setattr(self, meth, getattr(changes, meth))
......@@ -138,6 +138,20 @@ class DemoStorage(ConflictResolvingStorage):
except ZODB.POSException.POSKeyError:
return self.base.getTid(oid)
def history(self, oid, size=1):
try:
r = self.changes.history(oid, size)
except ZODB.POSException.POSKeyError:
r = []
size -= len(r)
if size:
try:
r += self.base.history(oid, size)
except ZODB.POSException.POSKeyError:
if not r:
raise
return r
def iterator(self, start=None, end=None):
for t in self.base.iterator(start, end):
yield t
......
......@@ -106,7 +106,7 @@ class MappingStorage(object):
tids.reverse()
return [
dict(
time = ZODB.TimeStamp.TimeStamp(tid),
time = ZODB.TimeStamp.TimeStamp(tid).timeTime(),
tid = tid,
serial = tid,
user_name = self._transactions[tid].user,
......
......@@ -17,45 +17,31 @@ Any storage that supports the history() method should be able to pass
all these tests.
"""
from time import time
from ZODB.tests.MinPO import MinPO
class HistoryStorage:
def checkSimpleHistory(self):
eq = self.assertEqual
self._checkHistory((11, 12, 13))
def _checkHistory(self, data):
start = time()
# Store a couple of revisions of the object
oid = self._storage.new_oid()
self.assertRaises(KeyError,self._storage.history,oid)
revid1 = self._dostore(oid, data=MinPO(11))
revid2 = self._dostore(oid, revid=revid1, data=MinPO(12))
revid3 = self._dostore(oid, revid=revid2, data=MinPO(13))
revids = [None]
for data in data:
revids.append(self._dostore(oid, revids[-1], MinPO(data)))
revids.reverse()
del revids[-1]
# Now get various snapshots of the object's history
h = self._storage.history(oid, size=1)
eq(len(h), 1)
d = h[0]
eq(d['tid'], revid3)
# Try to get 2 historical revisions
h = self._storage.history(oid, size=2)
eq(len(h), 2)
d = h[0]
eq(d['tid'], revid3)
d = h[1]
eq(d['tid'], revid2)
# Try to get all 3 historical revisions
h = self._storage.history(oid, size=3)
eq(len(h), 3)
d = h[0]
eq(d['tid'], revid3)
d = h[1]
eq(d['tid'], revid2)
d = h[2]
eq(d['tid'], revid1)
# There should be no more than 3 revisions
h = self._storage.history(oid, size=4)
eq(len(h), 3)
d = h[0]
eq(d['tid'], revid3)
d = h[1]
eq(d['tid'], revid2)
d = h[2]
eq(d['tid'], revid1)
for i in range(1, 1 + len(revids)):
h = self._storage.history(oid, size=i)
self.assertEqual([d['tid'] for d in h], revids[:i])
# Check results are sorted by timestamp, in descending order.
a = time()
for d in h:
b = a
a = d['time']
self.assertLess(a, b)
self.assertLess(start, a)
......@@ -87,6 +87,23 @@ class DemoStorageTests(
pass # we don't support undo yet
checkUndoZombie = checkLoadBeforeUndo
def checkBaseHistory(self):
def base_only():
yield 11
yield 12
yield 13
self._storage = self._storage.push()
self._checkHistory(base_only())
self._storage = self._storage.pop()
def base_and_changes():
yield 11
yield 12
self._storage = self._storage.push()
yield 13
yield 14
self._checkHistory(base_and_changes())
self._storage = self._storage.pop()
class DemoStorageHexTests(DemoStorageTests):
......@@ -149,7 +166,7 @@ def testSomeDelegation():
... six.print_(self.name, 'size')
... def close(self):
... six.print_(self.name, 'closed')
... sortKey = __len__ = history = getTid = None
... sortKey = __len__ = getTid = None
... tpc_finish = tpc_vote = tpc_transaction = None
... _lock_acquire = _lock_release = lambda self: None
... getName = lambda self: 'S'
......
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