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

Change the repository level.

# 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()