Commit d922dd9c authored by Grégory Wisniewski's avatar Grégory Wisniewski

Add a functional test for storage verification step.

git-svn-id: https://svn.erp5.org/repos/neo/trunk@1989 71dcc9de-d417-0410-9af5-da40c76e7ee4
parent 297dd16e
......@@ -227,13 +227,16 @@ class NEOCluster(object):
self.uuid_set.add(uuid)
return uuid
def setupDB(self):
def getSqlConnection(self):
# Cleanup or bootstrap databases
connect_arg_dict = {'user': self.db_super_user}
password = self.db_super_password
if password is not None:
connect_arg_dict['passwd'] = password
sql_connection = MySQLdb.Connect(**connect_arg_dict)
return MySQLdb.Connect(**connect_arg_dict)
def setupDB(self):
sql_connection = self.getSqlConnection()
cursor = sql_connection.cursor()
for database in self.db_list:
cursor.execute('DROP DATABASE IF EXISTS `%s`' % (database, ))
......@@ -245,6 +248,16 @@ class NEOCluster(object):
sql_connection.commit()
sql_connection.close()
def switchTables(self, database):
sql_connection = self.getSqlConnection()
cursor = sql_connection.cursor()
cursor.execute('use %s' % (database, ))
cursor.execute('rename table obj to tmp')
cursor.execute('rename table tobj to obj')
cursor.execute('rename table tmp to tobj')
sql_connection.commit()
sql_connection.close()
def start(self, except_storages=()):
neoctl = self.neoctl
assert len(self.process_dict)
......
......@@ -16,6 +16,11 @@
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
import unittest
import transaction
from ZODB.tests.StorageTestBase import zodb_pickle
from Persistence import Persistent
from neo.tests.functional import NEOCluster, NEOFunctionalTest
class ClusterTests(NEOFunctionalTest):
......@@ -74,6 +79,26 @@ class ClusterTests(NEOFunctionalTest):
self.neo.expectAllMasters(MASTER_COUNT)
self.neo.expectOudatedCells(0)
def testVerificationCommitUnfinishedTransactions(self):
""" Verification step should commit unfinished transactions """
# XXX: this kind of definition should be defined in base test class
class PObject(Persistent):
pass
self.neo = NEOCluster(['test_neo1'], replicas=0,
temp_dir=self.getTempDirectory())
neoctl = self.neo.getNEOCTL()
self.neo.start()
db, conn = self.neo.getZODBConnection()
conn.root()[0] = 'ok'
transaction.commit()
self.neo.stop()
# XXX: (obj|trans) become t(obj|trans)
self.neo.switchTables('test_neo1')
self.neo.start()
db, conn = self.neo.getZODBConnection()
# transaction should be verified and commited
self.assertEqual(conn.root()[0], 'ok')
def test_suite():
return unittest.makeSuite(ClusterTests)
......
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