diff commons/core/sql/test/Test_DbSQLite.py @ 6:769e306b7933

Change the repository level.
author yufei-luo
date Fri, 18 Jan 2013 04:54:14 -0500
parents
children
line wrap: on
line diff
--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/commons/core/sql/test/Test_DbSQLite.py	Fri Jan 18 04:54:14 2013 -0500
@@ -0,0 +1,162 @@
+# Copyright INRA (Institut National de la Recherche Agronomique)
+# http://www.inra.fr
+# http://urgi.versailles.inra.fr
+#
+# This software is governed by the CeCILL license under French law and
+# abiding by the rules of distribution of free software.  You can  use, 
+# modify and/ or redistribute the software under the terms of the CeCILL
+# license as circulated by CEA, CNRS and INRIA at the following URL
+# "http://www.cecill.info". 
+#
+# As a counterpart to the access to the source code and  rights to copy,
+# modify and redistribute granted by the license, users are provided only
+# with a limited warranty  and the software's author,  the holder of the
+# economic rights,  and the successive licensors  have only  limited
+# liability. 
+#
+# In this respect, the user's attention is drawn to the risks associated
+# with loading,  using,  modifying and/or developing or reproducing the
+# software by the user in light of its specific status of free software,
+# that may mean  that it is complicated to manipulate,  and  that  also
+# therefore means  that it is reserved for developers  and  experienced
+# professionals having in-depth computer knowledge. Users are therefore
+# encouraged to load and test the software's suitability as regards their
+# requirements in conditions enabling the security of their systems and/or 
+# data to be ensured and,  more generally, to use and operate it in the 
+# same conditions as regards security. 
+#
+# The fact that you are presently reading this means that you have had
+# knowledge of the CeCILL license and that you accept its terms.
+
+
+import unittest
+import time
+from commons.core.sql.DbSQLite import DbSQLite
+
+class Test_DbSQLite(unittest.TestCase):
+
+    def setUp( self ):
+        self._iDb = DbSQLite("test.db")
+        self._uniqId = "%s" % time.strftime("%Y%m%d%H%M%S")
+        
+    def tearDown( self ):
+        if self._iDb.open():
+            self._iDb.close()
+        self._iDb.delete()
+        self._iDb = None
+        
+    def test_open_True(self):
+        self._iDb.close()
+        self.assertTrue( self._iDb.open(1) )
+
+    def test_open_False(self):
+        self._iDb.close()
+        self._iDb.host = "/toto/toto.db"
+        self.assertFalse( self._iDb.open(1) )
+        self._iDb.host = "test.db"
+
+    def test_updateInfoTable(self):
+        tableName = "dummyTable" + self._uniqId
+        info = "Table_for_test"
+        
+        self._iDb.updateInfoTable(tableName, info)
+        
+        sqlCmd = 'SELECT file FROM info_tables WHERE name = "%s"' % ( tableName )
+        self._iDb.execute( sqlCmd )
+        results = self._iDb.fetchall()
+        obsResult = False
+        if (info,) in results:
+            obsResult = True
+            sqlCmd = 'DELETE FROM info_tables WHERE name = "%s"' % ( tableName )
+            self._iDb.execute( sqlCmd )
+            
+        self.assertTrue( obsResult )
+        
+    def test_doesTableExist_True(self):
+        tableName = "dummyTable" + self._uniqId
+        sqlCmd = "CREATE TABLE %s ( dummyColumn varchar(255) )" % ( tableName )
+        self._iDb.execute( sqlCmd )
+        self.assertTrue( self._iDb.doesTableExist(tableName) )
+
+    def test_dropTable(self):
+        tableName = "dummyTable" + self._uniqId
+        sqlCmd = "CREATE TABLE %s ( dummyColumn varchar(255) )" % tableName
+        self._iDb.execute( sqlCmd )
+        sqlCmd = "CREATE TABLE info_tables ( name varchar(255), file varchar(255) )"
+        self._iDb.execute( sqlCmd )
+        sqlCmd = 'INSERT INTO info_tables VALUES ("%s","")' % tableName
+        self._iDb.execute( sqlCmd )
+        
+        self._iDb.dropTable(tableName)
+        self.assertFalse( self._iDb.doesTableExist(tableName) )
+        
+    def test_doesTableExist_False(self):
+        tableName = "dummyTable" + self._uniqId
+        self.assertFalse( self._iDb.doesTableExist(tableName) )
+        
+    def test_createJobTable_is_table_created(self):
+        self._iDb.createTable("dummyJobTable", "jobs")
+        isTableCreated = self._iDb.doesTableExist("dummyJobTable")
+        self.assertTrue(isTableCreated)
+    
+    def test_createJobTable_field_list(self):
+        self._iDb.createTable("dummyJobTable", "jobs")
+        obsLFiled = self._iDb.getFieldList("dummyJobTable")
+        expLField = ["jobid", "jobname", "groupid", "command", "launcher", "queue", "status", "time", "node"]
+        self.assertEquals(expLField, obsLFiled)
+        
+    def test_createTable(self):
+        tableName = "dummyJobTable" + self._uniqId
+        self._iDb.createTable(tableName, "job")
+        obsLFiled = self._iDb.getFieldList(tableName)
+        expLField = ["jobid", "jobname", "groupid", "command", "launcher", "queue", "status", "time", "node"]
+        self.assertEquals(expLField, obsLFiled)
+        
+    def test_createTable_with_overwrite_Job(self):
+        tableName = "dummyJobTable" + self._uniqId
+        sqlCmd = "CREATE TABLE %s ( dummyColumn varchar(255) )" % tableName
+        self._iDb.execute( sqlCmd )
+        sqlCmd = "CREATE TABLE info_tables ( name varchar(255), file varchar(255) )"
+        self._iDb.execute( sqlCmd )
+        sqlCmd = 'INSERT INTO info_tables VALUES ("%s","")' % tableName
+        self._iDb.execute( sqlCmd )
+        
+        self._iDb.createTable(tableName, "job", True)
+        obsLFiled = self._iDb.getFieldList(tableName)
+        expLField = ["jobid", "jobname", "groupid", "command", "launcher", "queue", "status", "time", "node"]
+        self.assertEquals(expLField, obsLFiled)
+        
+    def test_getSize_empty_table(self):
+        tableName = "dummyJobTable" + self._uniqId
+        sqlCmd = "CREATE TABLE %s ( dummyColumn varchar(255) )" % ( tableName )
+        self._iDb.execute( sqlCmd )
+        expSize = 0
+        obsSize = self._iDb.getSize(tableName)
+        self.assertEquals( expSize, obsSize )
+        
+    def test_getSize_one_rows(self):
+        tableName = "dummyJobTable" + self._uniqId
+        sqlCmd = "CREATE TABLE %s ( dummyColumn varchar(255) )" % ( tableName )
+        self._iDb.execute( sqlCmd )
+        sqlCmd = "INSERT INTO %s (dummyColumn) VALUES ('toto')" % tableName
+        self._iDb.execute( sqlCmd )
+        expSize = 1
+        obsSize = self._iDb.getSize(tableName)
+        self.assertEquals( expSize, obsSize )
+        
+    def test_isEmpty_True(self):
+        tableName = "dummyTable" + self._uniqId
+        sqlCmd = "CREATE TABLE %s ( dummyColumn varchar(255) )" % ( tableName )
+        self._iDb.execute( sqlCmd )
+        self.assertTrue(self._iDb.isEmpty(tableName))
+        
+    def test_isEmpty_False(self):
+        tableName = "dummyTable" + self._uniqId
+        sqlCmd = "CREATE TABLE %s ( dummyColumn varchar(255) )" % (tableName)
+        self._iDb.execute(sqlCmd)
+        sqlCmd = "INSERT INTO %s (dummyColumn) VALUES ('toto')" % tableName
+        self._iDb.execute(sqlCmd)
+        self.assertFalse(self._iDb.isEmpty(tableName))
+        
+if __name__ == "__main__":
+    unittest.main()
\ No newline at end of file