Commit f8ce322b by Julien Muchembled

Fix read of empty transactions

Since transactions have metadata like a description, it may not be useless
to allow them. But the behaviour of FileStorage is to silently drop them,
so we may have to do the same in the future.

An application that is not supposed to commit empty transactions should write
its own unit test to prevent this.
1 parent c05e65ac
......@@ -36,6 +36,12 @@ def fallback(func):
return func
return lazymethod(wraps(func)(warn))
def splitOIDField(tid, oids):
if len(oids) % 8:
raise DatabaseFailure('invalid oids length for tid %s: %s'
% (tid, len(oids)))
return [oids[i:i+8] for i in xrange(0, len(oids), 8)]
class CreationUndone(Exception):
pass
......
......@@ -27,20 +27,11 @@ import struct
import time
from . import DatabaseManager, LOG_QUERIES
from .manager import CreationUndone
from .manager import CreationUndone, splitOIDField
from neo.lib import logging, util
from neo.lib.exception import DatabaseFailure
from neo.lib.protocol import CellStates, ZERO_OID, ZERO_TID, ZERO_HASH
def splitOIDField(tid, oids):
if (len(oids) % 8) != 0 or len(oids) == 0:
raise DatabaseFailure('invalid oids length for tid %d: %d' % (tid,
len(oids)))
oid_list = []
append = oid_list.append
for i in xrange(0, len(oids), 8):
append(oids[i:i+8])
return oid_list
class MySQLDatabaseManager(DatabaseManager):
"""This class manages a database on MySQL."""
......
......@@ -20,7 +20,7 @@ import string
import traceback
from . import DatabaseManager, LOG_QUERIES
from .manager import CreationUndone
from .manager import CreationUndone, splitOIDField
from neo.lib import logging, util
from neo.lib.exception import DatabaseFailure
from neo.lib.protocol import CellStates, ZERO_OID, ZERO_TID, ZERO_HASH
......@@ -38,16 +38,6 @@ def unique_constraint_message(table, *columns):
return e.args[0]
assert False
def splitOIDField(tid, oids):
if (len(oids) % 8) != 0 or len(oids) == 0:
raise DatabaseFailure('invalid oids length for tid %d: %d' % (tid,
len(oids)))
oid_list = []
append = oid_list.append
for i in xrange(0, len(oids), 8):
append(oids[i:i+8])
return oid_list
def retry_if_locked(f, *args):
try:
return f(*args)
......
......@@ -785,5 +785,20 @@ class Test(NEOThreadedTest):
finally:
cluster.stop()
def testEmptyTransaction(self):
cluster = NEOCluster()
try:
cluster.start()
txn = transaction.Transaction()
storage = cluster.getZODBStorage()
storage.tpc_begin(txn)
storage.tpc_vote(txn)
serial = storage.tpc_finish(txn)
t, = storage.iterator()
self.assertEqual(t.tid, serial)
self.assertFalse(t.oid_list)
finally:
cluster.stop()
if __name__ == "__main__":
unittest.main()
Styling with Markdown is supported
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!