Commit e2e9e5ee authored by Jim Fulton's avatar Jim Fulton

Updated for transaction buffer changes

Don't test obsolete method ``invalidate``. It's not used anymore in
master and not present on this (asyncio) branch.

Removed the reuable tests because transaction buffers are no-longer reused.
parent cc333941
...@@ -14,6 +14,7 @@ ...@@ -14,6 +14,7 @@
import random import random
import unittest import unittest
from ZODB.ConflictResolution import ResolvedSerial
from ZEO.TransactionBuffer import TransactionBuffer from ZEO.TransactionBuffer import TransactionBuffer
def random_string(size): def random_string(size):
...@@ -25,46 +26,31 @@ def new_store_data(): ...@@ -25,46 +26,31 @@ def new_store_data():
"""Return arbitrary data to use as argument to store() method.""" """Return arbitrary data to use as argument to store() method."""
return random_string(8), random_string(random.randrange(1000)) return random_string(8), random_string(random.randrange(1000))
def new_invalidate_data(): def store(tbuf, serial=None):
"""Return arbitrary data to use as argument to invalidate() method.""" data = new_store_data()
return random_string(8) tbuf.store(*data)
tbuf.serial(data[0], serial)
return data
class TransBufTests(unittest.TestCase): class TransBufTests(unittest.TestCase):
def checkTypicalUsage(self): def checkTypicalUsage(self):
tbuf = TransactionBuffer() tbuf = TransactionBuffer(0)
tbuf.store(*new_store_data()) store(tbuf)
tbuf.invalidate(new_invalidate_data()) store(tbuf)
for o in tbuf: for o in tbuf:
pass pass
def doUpdates(self, tbuf): def checkOrderPreserved(self):
tbuf = TransactionBuffer(0)
data = [] data = []
for i in range(10): for i in range(10):
d = new_store_data() data.append((store(tbuf), False))
tbuf.store(*d) data.append((store(tbuf, ResolvedSerial), True))
data.append(d)
d = new_invalidate_data()
tbuf.invalidate(d)
data.append(d)
for i, x in enumerate(tbuf):
if x[1] is None:
# the tbuf add a dummy None to invalidates
x = x[0]
self.assertEqual(x, data[i])
def checkOrderPreserved(self):
tbuf = TransactionBuffer()
self.doUpdates(tbuf)
def checkReusable(self): for i, (oid, d, resolved) in enumerate(tbuf):
tbuf = TransactionBuffer() self.assertEqual((oid, d), data[i][0])
self.doUpdates(tbuf) self.assertEqual(resolved, data[i][1])
tbuf.clear()
self.doUpdates(tbuf)
tbuf.clear()
self.doUpdates(tbuf)
def test_suite(): def test_suite():
return unittest.makeSuite(TransBufTests, 'check') return unittest.makeSuite(TransBufTests, 'check')
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