[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