[Neo-report] r1989 gregory - in /trunk/neo/tests/functional: __init__.py testCluster.py
nobody at svn.erp5.org
nobody at svn.erp5.org
Thu Apr 1 10:54:54 CEST 2010
Author: gregory
Date: Thu Apr 1 10:54:54 2010
New Revision: 1989
Log:
Add a functional test for storage verification step.
Modified:
trunk/neo/tests/functional/__init__.py
trunk/neo/tests/functional/testCluster.py
Modified: trunk/neo/tests/functional/__init__.py
==============================================================================
--- trunk/neo/tests/functional/__init__.py [iso-8859-1] (original)
+++ trunk/neo/tests/functional/__init__.py [iso-8859-1] Thu Apr 1 10:54:54 2010
@@ -227,13 +227,16 @@
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, ))
@@ -242,6 +245,16 @@
'IDENTIFIED BY "%s"' % (database, self.db_user,
self.db_password))
cursor.close()
+ 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()
Modified: trunk/neo/tests/functional/testCluster.py
==============================================================================
--- trunk/neo/tests/functional/testCluster.py [iso-8859-1] (original)
+++ trunk/neo/tests/functional/testCluster.py [iso-8859-1] Thu Apr 1 10:54:54 2010
@@ -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 @@
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)
More information about the Neo-report
mailing list