Mercurial > repos > yufei-luo > s_mart
view commons/core/sql/test/Test_DbMySql.py @ 31:0ab839023fe4
Uploaded
author | m-zytnicki |
---|---|
date | Tue, 30 Apr 2013 14:33:21 -0400 |
parents | 769e306b7933 |
children |
line wrap: on
line source
# 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 import os from MySQLdb import ProgrammingError from commons.core.sql.DbMySql import DbMySql from commons.core.sql.DbMySql import TABLE_SCHEMA_DESCRIPTOR from commons.core.sql.DbMySql import TABLE_TYPE_SYNONYMS from commons.core.utils.FileUtils import FileUtils from commons.core.coord.Path import Path class Test_DbMySql( unittest.TestCase ): def setUp( self ): self._iDb = DbMySql( ) self._uniqId = "%s" % time.strftime("%Y%m%d%H%M%S") def tearDown( self ): if self._iDb.db.open: self._iDb.close() self._iDb = None def test_execute_syntax_error(self): expErrorMsg = "You have an error in your SQL syntax; check the manual that corresponds to your MySQL server version for the right syntax to use near 'CHAUD TABLES' at line 1" obsErrorMsg = "" sqlCmd = "CHAUD TABLES" try: self._iDb.execute(sqlCmd) except ProgrammingError as excep: obsErrorMsg = excep.args[1] self.assertEquals(expErrorMsg, obsErrorMsg) def test_execute_with_1_retry(self): tableName = "dummyTable%s" % self._uniqId sqlCmd = "CREATE TABLE %s (dummyColumn varchar(255))" % tableName self._iDb.close() self._iDb.execute(sqlCmd) self.assertTrue(self._iDb.doesTableExist(tableName)) self._iDb.dropTable(tableName) def test_setAttributesFromConfigFile(self): expHost = "dummyHost" expUser = "dummyUser" expPw = "dummyPw" expDb = "dummyDb" expPort = 1000 configFileName = "dummyConfigFileName.cfg" f = open( configFileName, "w" ) f.write("[repet_env]\n") f.write("repet_host: " + expHost + "\n") f.write("repet_user: " + expUser + "\n") f.write("repet_pw: " + expPw + "\n") f.write("repet_db: " + expDb + "\n") f.write("repet_port: " + str(expPort) + "\n") f.close() self._iDb.setAttributesFromConfigFile(configFileName) obsHost = self._iDb.host obsUser = self._iDb.user obsPw = self._iDb.passwd obsDb = self._iDb.dbname obsPort = self._iDb.port os.remove(configFileName) self.assertEquals( expHost, obsHost ) self.assertEquals( expUser, obsUser ) self.assertEquals( expPw, obsPw ) self.assertEquals( expDb, obsDb ) self.assertEquals( expPort, obsPort ) def test_open_True(self): self._iDb.close() self.assertTrue(self._iDb.open()) self.assertEquals(1, self._iDb.db.open) self._iDb.close() self.assertEquals(0, self._iDb.db.open) def test_open_False(self): self._iDb.close() self._iDb.user = "dummyUser" self.assertFalse( self._iDb.open() ) 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) ) self._iDb.dropTable(tableName) def test_doesTableExist_False(self): tableName = "dummyTable" + self._uniqId self.assertFalse( self._iDb.doesTableExist(tableName) ) def test_dropTable(self): tableName = "dummyTable" + self._uniqId sqlCmd = "CREATE TABLE %s ( dummyColumn varchar(255) )" % ( tableName ) self._iDb.execute( sqlCmd ) self._iDb.dropTable(tableName) self.assertFalse( self._iDb.doesTableExist(tableName) ) def test_renameTable(self): tableName = "dummyTable" + self._uniqId sqlCmd = "CREATE TABLE %s ( dummyColumn varchar(255) )" % ( tableName ) self._iDb.execute( sqlCmd ) self._iDb.updateInfoTable( tableName, "" ) newTableName = "newDummyTable" self._iDb.renameTable(tableName, newTableName) self.assertFalse( self._iDb.doesTableExist(tableName) ) self.assertTrue( self._iDb.doesTableExist(newTableName) ) expTuple = (('newDummyTable', ''),) sqlCmd = 'SELECT * FROM info_tables WHERE name = "%s"' % ( newTableName ) self._iDb.execute( sqlCmd ) obsTuple = self._iDb.cursor.fetchall() self.assertEquals( expTuple, obsTuple) expTuple = () sqlCmd = 'SELECT * FROM info_tables WHERE name = "%s"' % ( tableName ) self._iDb.execute( sqlCmd ) obsTuple = self._iDb.cursor.fetchall() self.assertEquals( expTuple, obsTuple) self._iDb.dropTable(newTableName) def test_copyTable(self): tableName = "dummyTable" + self._uniqId sqlCmd = "CREATE TABLE %s ( dummyColumn varchar(255) );" % ( tableName ) self._iDb.execute( sqlCmd ) sqlCmd = "CREATE INDEX idummyColumn ON %s ( dummyColumn );" % (tableName) self._iDb.execute( sqlCmd ) newTableName = "newDummyTable" self._iDb.copyTable(tableName, newTableName) self.assertTrue( self._iDb.doesTableExist(tableName) ) self.assertTrue( self._iDb.doesTableExist(newTableName) ) expTuple = (('newDummyTable', ''),) sqlCmd = 'SELECT * FROM info_tables WHERE name = "%s";' % ( newTableName ) self._iDb.execute( sqlCmd ) obsTuple = self._iDb.cursor.fetchall() self.assertEquals( expTuple, obsTuple) expTuple = (('newDummyTable', 1L, 'idummyColumn', 1L, 'dummyColumn', 'A', None, None, None, 'YES', 'BTREE', ''),) sqlCmd = "SHOW INDEX FROM %s;" % ( newTableName ) self._iDb.execute( sqlCmd ) obsTuple = self._iDb.cursor.fetchall() self.assertEquals( expTuple, obsTuple) self._iDb.dropTable(tableName) self._iDb.dropTable(newTableName) def test_getTableType(self): lTypesToTest = TABLE_SCHEMA_DESCRIPTOR.keys() for tableType in lTypesToTest: tableName = "dummy%sTable%s" % (tableType, self._uniqId) self._iDb.createTable(tableName, tableType) obsType = self._iDb.getTableType(tableName) self.assertEquals(tableType, obsType) self._iDb.dropTable(tableName) def test_getSize_empty_table(self): tableName = "dummyPathTable" + self._uniqId sqlCmd = "CREATE TABLE %s ( path int unsigned, query_name varchar(255), query_start int , query_end int, subject_name varchar(255), subject_start int unsigned, subject_end int unsigned, E_value double, score int unsigned, identity float)" % ( tableName ) self._iDb.execute( sqlCmd ) pathFileName = "dummyPathFile.txt" pathF = open( pathFileName, "w" ) pathF.write( "") pathF.close() self._iDb.loadDataFromFile(tableName, pathFileName, False) expSize = 0 obsSize = self._iDb.getSize(tableName) self._iDb.dropTable(tableName) os.remove(pathFileName) self.assertEquals( expSize, obsSize ) def test_getSize_two_rows(self): tableName = "dummyPathTable" + self._uniqId sqlCmd = "CREATE TABLE %s ( path int unsigned, query_name varchar(255), query_start int , query_end int, subject_name varchar(255), subject_start int unsigned, subject_end int unsigned, E_value double, score int unsigned, identity float)" % ( tableName ) self._iDb.execute( sqlCmd ) pathFileName = "dummyPathFile.txt" pathF = open( pathFileName, "w" ) pathF.write( "1\tqry\t1\t100\tsbj\t1\t100\t1e-123\t136\t98.4\n" ) pathF.write( "2\tqry\t500\t401\tsbj\t1\t100\t1e-152\t161\t98.7\n" ) pathF.close() self._iDb.loadDataFromFile(tableName, pathFileName, False) expSize = 2 obsSize = self._iDb.getSize(tableName) self._iDb.dropTable(tableName) os.remove(pathFileName) 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 ) fileName = "dummyTableFile.txt" f = open( fileName, "w" ) f.write( "" ) f.close() self._iDb.loadDataFromFile(tableName, fileName, False) self.assertTrue( self._iDb.isEmpty(tableName) ) self._iDb.dropTable(tableName) os.remove(fileName) def test_isEmpty_False(self): tableName = "dummyTable" + self._uniqId sqlCmd = "CREATE TABLE %s ( dummyColumn varchar(255) )" % tableName self._iDb.execute( sqlCmd ) fileName = "dummyTableFile.txt" f = open( fileName, "w" ) f.write( "test" ) f.close() self._iDb.loadDataFromFile(tableName, fileName, False) self.assertFalse( self._iDb.isEmpty(tableName) ) self._iDb.dropTable(tableName) os.remove(fileName) 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.cursor.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_loadDataFromFile_with_empty_file(self): tableName = "dummyPathTable1" + self._uniqId sqlCmd = "CREATE TABLE %s ( path int unsigned, query_name varchar(255), query_start int , query_end int, subject_name varchar(255), subject_start int unsigned, subject_end int unsigned, E_value double, score int unsigned, identity float)" % ( tableName ) self._iDb.execute( sqlCmd ) pathFileName = "dummyPathFile.txt" pathF = open( pathFileName, "w" ) pathF.write( "" ) pathF.close() expTPathTuples = () self._iDb.loadDataFromFile(tableName, pathFileName, False) sqlCmd = "SELECT * FROM %s" % ( tableName ) self._iDb.execute( sqlCmd ) obsTPathTuples = self._iDb.cursor.fetchall() self._iDb.dropTable(tableName) os.remove(pathFileName) self.assertEquals( expTPathTuples, obsTPathTuples ) def test_loadDataFromFile_with_first_line(self): tableName = "dummyPathTable2" + self._uniqId sqlCmd = "CREATE TABLE %s ( path int unsigned, query_name varchar(255), query_start int , query_end int, subject_name varchar(255), subject_start int unsigned, subject_end int unsigned, E_value double, score int unsigned, identity float)" % ( tableName ) self._iDb.execute( sqlCmd ) pathFileName = "dummyPathFile.txt" pathF = open( pathFileName, "w" ) pathF.write( "1\tqry\t1\t100\tsbj\t1\t100\t1e-123\t136\t98.4\n" ) pathF.write( "2\tqry\t500\t401\tsbj\t1\t100\t1e-152\t161\t98.7\n" ) pathF.close() expPathTuple1 = (1L, 'qry', 1L, 100L, 'sbj', 1L, 100L, 1e-123, 136L, 98.4) expPathTuple2 = (2L, 'qry', 500L, 401L, 'sbj', 1L, 100L, 1e-152, 161L, 98.7) expTPathTuples = (expPathTuple1, expPathTuple2) self._iDb.loadDataFromFile(tableName, pathFileName, False) sqlCmd = "SELECT * FROM %s" % ( tableName ) self._iDb.execute( sqlCmd ) obsTPathTuples = self._iDb.cursor.fetchall() self._iDb.dropTable(tableName) os.remove(pathFileName) self.assertEquals( expTPathTuples, obsTPathTuples ) def test_loadDataFromFile_without_first_line(self): tableName = "dummyPathTable3" + self._uniqId sqlCmd = "CREATE TABLE %s ( path int unsigned, query_name varchar(255), query_start int , query_end int, subject_name varchar(255), subject_start int unsigned, subject_end int unsigned, E_value double, score int unsigned, identity float)" % ( tableName ) self._iDb.execute( sqlCmd ) pathFileName = "dummyPathFile.txt" pathF = open( pathFileName, "w" ) pathF.write( "1\tqry\t1\t100\tsbj\t1\t100\t1e-123\t136\t98.4\n" ) pathF.write( "2\tqry\t500\t401\tsbj\t1\t100\t1e-152\t161\t98.7\n" ) pathF.close() expPathTuple = (2L, 'qry', 500L, 401L, 'sbj', 1L, 100L, 1e-152, 161L, 98.7) expTPathTuples = (expPathTuple,) self._iDb.loadDataFromFile(tableName, pathFileName, True) sqlCmd = "SELECT * FROM %s" % ( tableName ) self._iDb.execute( sqlCmd ) obsTPathTuples = self._iDb.cursor.fetchall() self._iDb.dropTable(tableName) os.remove(pathFileName) self.assertEquals( expTPathTuples, obsTPathTuples ) def test_createIndex_Map(self): tableName = "dummyMapTable" + self._uniqId sqlCmd = "CREATE TABLE %s ( name varchar(255), chr varchar(255), start int, end int)" % ( tableName ) self._iDb.execute( sqlCmd ) expLIndex = ["iname", "ichr", "istart", "iend", "icoord", "icoord"] self._iDb.createIndex(tableName, "map") sqlCmd = "SHOW INDEX FROM %s" % ( tableName ) self._iDb.execute( sqlCmd ) results = self._iDb.cursor.fetchall() for index in expLIndex[:-1]: sqlCmd = "DROP INDEX %s ON %s" % ( index, tableName ) self._iDb.execute( sqlCmd ) self._iDb.dropTable(tableName) obsLIndex = [] for tuple in results: obsLIndex.append(tuple[2]) self.assertEquals( expLIndex, obsLIndex) def test_createIndex_Map_coord_index_already_exist(self): tableName = "dummyMapTable" + self._uniqId sqlCmd = "CREATE TABLE %s ( name varchar(255), chr varchar(255), start int, end int)" % ( tableName ) self._iDb.execute( sqlCmd ) sqlCmd = "CREATE INDEX icoord ON %s ( start,end );" % (tableName) self._iDb.execute( sqlCmd ) expLIndex = ["icoord", "icoord", "iname", "ichr", "istart", "iend"] self._iDb.createIndex(tableName, "map") sqlCmd = "SHOW INDEX FROM %s" % ( tableName ) self._iDb.execute( sqlCmd ) results = self._iDb.cursor.fetchall() for index in expLIndex[1:]: sqlCmd = "DROP INDEX %s ON %s" % ( index, tableName ) self._iDb.execute( sqlCmd ) self._iDb.dropTable(tableName) obsLIndex = [] for tuple in results: obsLIndex.append(tuple[2]) self.assertEquals( expLIndex, obsLIndex) def test_createTable_Map( self ): tableName = "dummyMapTable" + self._uniqId mapFileName = "dummyMapFile.txt" mapF = open( mapFileName, "w" ) mapF.write( "map1\tseq1\t20\t50\n" ) mapF.write( "map2\tseq2\t700\t760\n" ) mapF.close() expMapTuple1 = ("map1", "seq1", 20L, 50L) expMapTuple2 = ("map2", "seq2", 700L, 760L) expTMapTuples = (expMapTuple1, expMapTuple2) self._iDb.createTable(tableName, 'map', mapFileName) sqlCmd = "SELECT * FROM %s" % ( tableName ) self._iDb.execute( sqlCmd ) obsTMapTuples = self._iDb.cursor.fetchall() self._iDb.dropTable(tableName) os.remove(mapFileName) self.assertEquals( expTMapTuples, obsTMapTuples ) def test_createIndex_Match(self): tableName = "dummyMatchTable" + self._uniqId sqlCmd = "CREATE TABLE %s ( query_name varchar(255), query_start int, query_end int, query_length int unsigned, query_length_perc float, match_length_perc float, subject_name varchar(255), subject_start int unsigned, subject_end int unsigned, subject_length int unsigned, subject_length_perc float, E_value double, score int unsigned, identity float, path int unsigned)" % ( tableName ) self._iDb.execute( sqlCmd ) expLIndex = ["id", "qname", "qstart", "qend", "sname", "sstart", "send", "qcoord", "qcoord"] self._iDb.createIndex(tableName, "match") sqlCmd = "SHOW INDEX FROM %s" % ( tableName ) self._iDb.execute( sqlCmd ) results = self._iDb.cursor.fetchall() obsLIndex = [] for tuple in results: obsLIndex.append(tuple[2]) self._iDb.dropTable(tableName) self.assertEquals( expLIndex, obsLIndex) def test_createIndex_Match_all_index_already_exist(self): tableName = "dummyMatchTable" + self._uniqId sqlCmd = "CREATE TABLE %s ( query_name varchar(255), query_start int, query_end int, query_length int unsigned, query_length_perc float, match_length_perc float, subject_name varchar(255), subject_start int unsigned, subject_end int unsigned, subject_length int unsigned, subject_length_perc float, E_value double, score int unsigned, identity float, path int unsigned)" % ( tableName ) self._iDb.execute( sqlCmd ) sqlCmd = "CREATE UNIQUE INDEX id ON %s ( path );" % (tableName) self._iDb.execute( sqlCmd ) sqlCmd = "CREATE INDEX qname ON %s ( query_name(10) );" % (tableName) self._iDb.execute( sqlCmd ) sqlCmd = "CREATE INDEX qstart ON %s ( query_start );" % (tableName) self._iDb.execute( sqlCmd ) sqlCmd = "CREATE INDEX qend ON %s ( query_end );" % (tableName) self._iDb.execute( sqlCmd ) sqlCmd = "CREATE INDEX sname ON %s ( subject_name(10) );" % (tableName) self._iDb.execute( sqlCmd ) sqlCmd = "CREATE INDEX sstart ON %s ( subject_start );" % (tableName) self._iDb.execute( sqlCmd ) sqlCmd = "CREATE INDEX send ON %s ( subject_end );" % (tableName) self._iDb.execute( sqlCmd ) sqlCmd = "CREATE INDEX qcoord ON %s ( query_start,query_end );" % (tableName) self._iDb.execute( sqlCmd ) expLIndex = ["id", "qname", "qstart", "qend", "sname", "sstart", "send", "qcoord", "qcoord"] self._iDb.createIndex(tableName, "match") sqlCmd = "SHOW INDEX FROM %s" % ( tableName ) self._iDb.execute( sqlCmd ) results = self._iDb.cursor.fetchall() for index in expLIndex[:-1]: sqlCmd = "DROP INDEX %s ON %s" % ( index, tableName ) self._iDb.execute( sqlCmd ) self._iDb.dropTable(tableName) obsLIndex = [] for tuple in results: obsLIndex.append(tuple[2]) self.assertEquals( expLIndex, obsLIndex) def test_createTable_match( self ): tableName = "dummyMatchTable" + self._uniqId matchFileName = "dummyMatchFile.txt" matchF = open( matchFileName, "w" ) matchF.write( "qry1\t700\t760\t60\t100\t100\tsbj2\t500\t560\t60\t100\t1e-123\t136\t98.4\t2\n" ) matchF.write( "qry2\t700\t760\t60\t100\t100\tsbj2\t500\t560\t60\t100\t1e-123\t136\t98.4\t2\n" ) matchF.close() expMatchTuple = ("qry2", 700L, 760L, 60L, 100.0, 100.0, "sbj2", 500L, 560L, 60L, 100.0, 1e-123, 136L, 98.4, 2L) expTMatchTuples = (expMatchTuple,) self._iDb.createTable(tableName, "match", matchFileName) sqlCmd = "SELECT * FROM %s" % ( tableName ) self._iDb.execute( sqlCmd ) obsTMatchTuples = self._iDb.cursor.fetchall() self._iDb.dropTable(tableName) os.remove(matchFileName) self.assertEquals( expTMatchTuples, obsTMatchTuples ) def test_createIndex_Path(self): tableName = "dummyPathTable" + self._uniqId sqlCmd = "CREATE TABLE %s ( path int unsigned, query_name varchar(255), query_start int , query_end int, subject_name varchar(255), subject_start int unsigned, subject_end int unsigned, E_value double, score int unsigned, identity float)" % ( tableName ) self._iDb.execute( sqlCmd ) expLIndex = ["id", "qname", "qstart", "qend", "sname", "sstart", "send", "qcoord", "qcoord"] self._iDb.createIndex(tableName, "path") sqlCmd = "SHOW INDEX FROM %s" % ( tableName ) self._iDb.execute( sqlCmd ) results = self._iDb.cursor.fetchall() for index in expLIndex[:-1]: sqlCmd = "DROP INDEX %s ON %s" % ( index, tableName ) self._iDb.execute( sqlCmd ) self._iDb.dropTable(tableName) obsLIndex = [] for tuple in results: obsLIndex.append(tuple[2]) self.assertEquals( expLIndex, obsLIndex) def test_createIndex_Path_id_and_send_index_already_exist(self): tableName = "dummyPathTable" + self._uniqId sqlCmd = "CREATE TABLE %s ( path int unsigned, query_name varchar(255), query_start int , query_end int, subject_name varchar(255), subject_start int unsigned, subject_end int unsigned, E_value double, score int unsigned, identity float)" % ( tableName ) self._iDb.execute( sqlCmd ) sqlCmd = "CREATE INDEX id ON %s ( path );" % (tableName) self._iDb.execute( sqlCmd ) sqlCmd = "CREATE INDEX send ON %s ( subject_end );" % (tableName) self._iDb.execute( sqlCmd ) expLIndex = ["id", "send", "qname", "qstart", "qend", "sname", "sstart", "qcoord", "qcoord"] self._iDb.createIndex(tableName, "path") sqlCmd = "SHOW INDEX FROM %s" % ( tableName ) self._iDb.execute( sqlCmd ) results = self._iDb.cursor.fetchall() for index in expLIndex[:-1]: sqlCmd = "DROP INDEX %s ON %s" % ( index, tableName ) self._iDb.execute( sqlCmd ) self._iDb.dropTable(tableName) obsLIndex = [] for tuple in results: obsLIndex.append(tuple[2]) self.assertEquals( expLIndex, obsLIndex) def test_createTable_path( self ): tableName = "dummyPathTable" + self._uniqId pathFileName = "dummyPathFile.txt" pathF = open( pathFileName, "w" ) pathF.write( "1\tqry\t1\t100\tsbj\t1\t100\t1e-123\t136\t98.4\n" ) pathF.write( "2\tqry\t500\t401\tsbj\t1\t100\t1e-152\t161\t98.7\n" ) pathF.close() expPathTuple1 = (1L, "qry", 1L, 100L, "sbj", 1L, 100L, 1e-123, 136L, 98.4) expPathTuple2 = (2L, "qry", 401L, 500L, "sbj", 100L, 1L, 1e-152, 161L, 98.7) # change coordinates expTPathTuples = (expPathTuple1, expPathTuple2) self._iDb.createTable( tableName, "path", pathFileName) sqlCmd = "SELECT * FROM %s" % ( tableName ) self._iDb.execute( sqlCmd ) obsTPathTuples = self._iDb.cursor.fetchall() self._iDb.dropTable(tableName) os.remove(pathFileName) self.assertEquals( expTPathTuples, obsTPathTuples ) def test_createIndex_align(self): tableName = "dummyAlignTable" + self._uniqId sqlCmd = "CREATE TABLE %s ( query_name varchar(255), query_start int, query_end int,subject_name varchar(255), subject_start int unsigned, subject_end int unsigned,E_value double, score int unsigned, identity float)" % ( tableName ) self._iDb.execute( sqlCmd ) expLIndex = ["qname", "qstart", "qend", "sname", "sstart", "send", "qcoord", "qcoord"] self._iDb.createIndex(tableName, "align") sqlCmd = "SHOW INDEX FROM %s" % ( tableName ) self._iDb.execute( sqlCmd ) results = self._iDb.cursor.fetchall() for index in expLIndex[:-1]: sqlCmd = "DROP INDEX %s ON %s" % ( index, tableName ) self._iDb.execute( sqlCmd ) self._iDb.dropTable(tableName) obsLIndex = [] for tuple in results: obsLIndex.append(tuple[2]) self.assertEquals( expLIndex, obsLIndex) def test_createIndex_align_qstart_index_already_exist(self): tableName = "dummyAlignTable" + self._uniqId sqlCmd = "CREATE TABLE %s ( query_name varchar(255), query_start int, query_end int,subject_name varchar(255), subject_start int unsigned, subject_end int unsigned,E_value double, score int unsigned, identity float)" % ( tableName ) self._iDb.execute( sqlCmd ) sqlCmd = "CREATE INDEX qstart ON %s ( query_start );" % (tableName) self._iDb.execute( sqlCmd ) expLIndex = ["qstart", "qname", "qend", "sname", "sstart", "send", "qcoord", "qcoord"] self._iDb.createIndex(tableName, "align") sqlCmd = "SHOW INDEX FROM %s" % ( tableName ) self._iDb.execute( sqlCmd ) results = self._iDb.cursor.fetchall() for index in expLIndex[:-1]: sqlCmd = "DROP INDEX %s ON %s" % ( index, tableName ) self._iDb.execute( sqlCmd ) self._iDb.dropTable(tableName) obsLIndex = [] for tuple in results: obsLIndex.append(tuple[2]) self.assertEquals( expLIndex, obsLIndex) def test_createTable_align( self ): tableName = "dummyAlignTable" + self._uniqId alignFileName = "dummyAlignFile.txt" alignF = open( alignFileName, "w" ) alignF.write( "query1\t1\t100\tsubject1\t1\t150\t0.5\t15\t35\n" ) alignF.write( "query2\t1\t100\tsubject2\t1\t150\t0.5\t15\t35\n" ) alignF.close() expAlignTuple1 = ("query1", 1L, 100L, "subject1", 1L, 150L, 0.5, 15L, 35) expAlignTuple2 = ("query2", 1L, 100L, "subject2", 1L, 150L, 0.5, 15L, 35) expTAlignTuples = (expAlignTuple1, expAlignTuple2) self._iDb.createTable( tableName, "align", alignFileName ) sqlCmd = "SELECT * FROM %s" % ( tableName ) self._iDb.execute( sqlCmd ) obsTAlignTuples = self._iDb.cursor.fetchall() self._iDb.dropTable(tableName) os.remove(alignFileName) self.assertEquals( expTAlignTuples, obsTAlignTuples ) def test_createIndex_set(self): tableName = "dummySetTable" + self._uniqId sqlCmd = "CREATE TABLE %s ( path int unsigned, name varchar(255), chr varchar(255), start int, end int)" % ( tableName ) self._iDb.execute( sqlCmd ) expLIndex = ["id", "iname", "ichr", "istart", "iend", "icoord", "icoord"] self._iDb.createIndex(tableName, "set") sqlCmd = "SHOW INDEX FROM %s" % ( tableName ) self._iDb.execute( sqlCmd ) results = self._iDb.cursor.fetchall() for index in expLIndex[:-1]: sqlCmd = "DROP INDEX %s ON %s" % ( index, tableName ) self._iDb.execute( sqlCmd ) self._iDb.dropTable(tableName) obsLIndex = [] for tuple in results: obsLIndex.append(tuple[2]) self.assertEquals( expLIndex, obsLIndex) def test_createIndex_set_id_index_already_exist(self): tableName = "dummySetTable" + self._uniqId sqlCmd = "CREATE TABLE %s ( path int unsigned, name varchar(255), chr varchar(255), start int, end int)" % ( tableName ) self._iDb.execute( sqlCmd ) sqlCmd = "CREATE INDEX id ON %s ( path );" % (tableName) self._iDb.execute( sqlCmd ) expLIndex = ["id", "iname", "ichr", "istart", "iend", "icoord", "icoord"] self._iDb.createIndex(tableName, 'set') sqlCmd = "SHOW INDEX FROM %s" % ( tableName ) self._iDb.execute( sqlCmd ) results = self._iDb.cursor.fetchall() for index in expLIndex[:-1]: sqlCmd = "DROP INDEX %s ON %s" % ( index, tableName ) self._iDb.execute( sqlCmd ) self._iDb.dropTable(tableName) obsLIndex = [] for tuple in results: obsLIndex.append(tuple[2]) self.assertEquals( expLIndex, obsLIndex) def test_createTable_set( self ): tableName = "dummySetTable" + self._uniqId setFileName = "dummySetFile.txt" setF = open( setFileName, "w" ) setF.write( "15\tset1\tchr1\t1\t100\n" ) setF.write( "15\tset2\tchr2\t1\t100\n" ) setF.close() expSetTuple1 = (15L, "set1", "chr1", 1L, 100L) expSetTuple2 = (15L, "set2", "chr2", 1L, 100L) expTSetTuples = (expSetTuple1, expSetTuple2) self._iDb.createTable( tableName, 'set', setFileName ) sqlCmd = "SELECT * FROM %s" % ( tableName ) self._iDb.execute( sqlCmd ) obsTSetTuples = self._iDb.cursor.fetchall() self._iDb.dropTable(tableName) os.remove(setFileName) self.assertEquals( expTSetTuples, obsTSetTuples ) def test_convertMapTableIntoSetTable( self ): mapTableName = "dummyMapTable" + self._uniqId mapFileName = "dummyMapFile.txt" with open(mapFileName, "w") as mapFH: mapFH.write("map1\tchr1\t1\t100\n") mapFH.write("map2\tchr2\t1\t100\n") self._iDb.createTable(mapTableName, 'map', mapFileName) expSetTuple1 = (1, "map1", "chr1", 1, 100) expSetTuple2 = (2, "map2", "chr2", 1, 100) expTSetTuples = (expSetTuple1, expSetTuple2) setTableName = "dummySetTable" + self._uniqId self._iDb.convertMapTableIntoSetTable(mapTableName, setTableName) sqlCmd = "SELECT * FROM %s" % setTableName self._iDb.execute(sqlCmd) obsTSetTuples = self._iDb.cursor.fetchall() self._iDb.dropTable(mapTableName) self._iDb.dropTable(setTableName) os.remove(mapFileName) self.assertEquals( expTSetTuples, obsTSetTuples ) def test_createIndex_seq(self): tableName = "dummySeqTable" + self._uniqId sqlCmd = "CREATE TABLE %s ( accession varchar(255), sequence longtext, description varchar(255), length int unsigned)" % ( tableName ) self._iDb.execute( sqlCmd ) expLIndex = ["iacc", "idescr"] self._iDb.createIndex(tableName,'seq') sqlCmd = "SHOW INDEX FROM %s" % ( tableName ) self._iDb.execute( sqlCmd ) results = self._iDb.cursor.fetchall() for index in expLIndex: sqlCmd = "DROP INDEX %s ON %s" % ( index, tableName ) self._iDb.execute( sqlCmd ) self._iDb.dropTable(tableName) obsLIndex = [] for tuple in results: obsLIndex.append(tuple[2]) self.assertEquals(expLIndex, obsLIndex) def test_createIndex_seq_idescr_index_already_exist(self): tableName = "dummySeqTable" + self._uniqId sqlCmd = "CREATE TABLE %s ( accession varchar(255), sequence longtext, description varchar(255), length int unsigned);" % ( tableName ) self._iDb.execute( sqlCmd ) sqlCmd = "CREATE INDEX idescr ON %s ( description(10) );" % ( tableName ) self._iDb.execute( sqlCmd ) expLIndex = ["idescr", "iacc"] self._iDb.createIndex(tableName,'seq') sqlCmd = "SHOW INDEX FROM %s" % ( tableName ) self._iDb.execute( sqlCmd ) results = self._iDb.cursor.fetchall() for index in expLIndex: sqlCmd = "DROP INDEX %s ON %s" % ( index, tableName ) self._iDb.execute( sqlCmd ) self._iDb.dropTable(tableName) obsLIndex = [] for tuple in results: obsLIndex.append(tuple[2]) self.assertEquals(expLIndex, obsLIndex) def test_createTable_seq( self ): tableName = "dummySeqTable" + self._uniqId seqFileName = "dummySeqFile.txt" seqF = open( seqFileName, "w" ) seqF.write( ">acc1 seq1\n" ) seqF.write( "ATACTTCGCTAGCTCGC\n" ) seqF.write( ">acc2 seq2\n" ) seqF.write( "ATACTTCGCTAGCTCGCATACTTCGCTAGCTCGCATACTTCGCTAGCTCGCATACTTCGCTAGCTCGC\n" ) seqF.close() expSeqTuple1 = ("acc1", "ATACTTCGCTAGCTCGC", "acc1 seq1", 17L) expSeqTuple2 = ("acc2", "ATACTTCGCTAGCTCGCATACTTCGCTAGCTCGCATACTTCGCTAGCTCGCATACTTCGCTAGCTCGC", "acc2 seq2", 68L) expTSeqTuples = (expSeqTuple1, expSeqTuple2) self._iDb.createTable( tableName,'seq', seqFileName ) sqlCmd = "SELECT * FROM %s" % ( tableName ) self._iDb.execute( sqlCmd ) obsTSeqTuples = self._iDb.cursor.fetchall() self._iDb.dropTable(tableName) os.remove(seqFileName) self.assertEquals( expTSeqTuples, obsTSeqTuples ) def test_createIndex_job(self): tableName = "dummyTable%s" % self._uniqId sqlCmd = "CREATE TABLE %s" % tableName sqlCmd += " ( jobid INT UNSIGNED" sqlCmd += ", jobname VARCHAR(255)" sqlCmd += ", groupid VARCHAR(255)" sqlCmd += ", command TEXT" sqlCmd += ", launcher VARCHAR(1024)" sqlCmd += ", queue VARCHAR(255)" sqlCmd += ", status VARCHAR(255)" sqlCmd += ", time DATETIME" sqlCmd += ", node VARCHAR(255) )" self._iDb.execute(sqlCmd) expLIndex = ["ijobid", "ijobname", "igroupid", "istatus"] self._iDb.createIndex(tableName, 'jobs') sqlCmd = "SHOW INDEX FROM %s" % tableName self._iDb.execute(sqlCmd) results = self._iDb.cursor.fetchall() obsLIndex = [] for tuple in results: obsLIndex.append(tuple[2]) for index in obsLIndex: sqlCmd = "DROP INDEX %s ON %s" % (index, tableName) self._iDb.execute(sqlCmd) self._iDb.dropTable(tableName) self.assertEquals(expLIndex, obsLIndex) def test_createTable_job( self ): tableName = "dummyTable%s" % self._uniqId expTuples = () self._iDb.createTable(tableName,'jobs') sqlCmd = "SELECT * FROM %s" % tableName self._iDb.execute(sqlCmd) obsTuples = self._iDb.cursor.fetchall() self._iDb.dropTable(tableName) self.assertEquals(expTuples, obsTuples) def test_createIndex_length(self): tableName = "dummyTable%s" % self._uniqId sqlCmd = "CREATE TABLE %s (accession varchar(255), length int unsigned)" % tableName self._iDb.execute(sqlCmd) expLIndex = ["iacc", "ilength"] self._iDb.createIndex(tableName,'length') sqlCmd = "SHOW INDEX FROM %s" % tableName self._iDb.execute(sqlCmd) results = self._iDb.cursor.fetchall() obsLIndex = [] for tuple in results: obsLIndex.append(tuple[2]) for index in obsLIndex: sqlCmd = "DROP INDEX %s ON %s" % (index, tableName) self._iDb.execute(sqlCmd) self._iDb.dropTable(tableName) self.assertEquals(expLIndex, obsLIndex) def test_createTable_length( self ): tableName = "dummyLengthTable%s" % self._uniqId seqFileName = "dummyFile.fa" seqF = open( seqFileName, "w" ) seqF.write(">acc1 seq1\n") seqF.write("ATACTTCGCTAGCTCGC\n") seqF.write(">acc2 seq2\n") seqF.write("ATACTTCGCTAGCTCGCATACTTCGCTAGCTCGCATACTTCGCTAGCTCGCATACTTCGCTAGCTCGC\n") seqF.close() expTuple1 = ("acc1", 17) expTuple2 = ("acc2", 68) expTTuples = (expTuple1, expTuple2) self._iDb.createTable(tableName, "length", seqFileName) sqlCmd = "SELECT * FROM %s" % tableName self._iDb.execute(sqlCmd) obsTTuples = self._iDb.cursor.fetchall() self._iDb.dropTable(tableName) os.remove(seqFileName) self.assertEquals(expTTuples, obsTTuples) def test_createTable_with_overwrite_Map( self ): tableName = "dummyMapTable" + self._uniqId sqlCmd = "CREATE TABLE %s ( dummyColumn varchar(255) )" % ( tableName ) self._iDb.execute( sqlCmd ) fileName = "dummyMapFile.txt" mapF = open( fileName, "w" ) mapF.write( "map1\tseq1\t20\t50\n" ) mapF.write( "map2\tseq2\t700\t760\n" ) mapF.close() expMapTuple1 = ("map1", "seq1", 20L, 50L) expMapTuple2 = ("map2", "seq2", 700L, 760L) expTMapTuples = (expMapTuple1, expMapTuple2) self._iDb.createTable(tableName, "Map", fileName, True) sqlCmd = "SELECT * FROM %s" % ( tableName ) self._iDb.execute( sqlCmd ) obsTMapTuples = self._iDb.cursor.fetchall() self._iDb.dropTable(tableName) os.remove(fileName) self.assertEquals( expTMapTuples, obsTMapTuples ) def test_createTable_without_overwrite_Align( self ): tableName = "dummyAlignTable" + self._uniqId alignFileName = "dummyAlignFile.txt" alignF = open( alignFileName, "w" ) alignF.write( "query1\t1\t100\tsubject1\t1\t150\t0.5\t15\t35\n" ) alignF.write( "query2\t1\t100\tsubject2\t1\t150\t0.5\t15\t35\n" ) alignF.close() expAlignTuple1 = ("query1", 1L, 100L, "subject1", 1L, 150L, 0.5, 15L, 35) expAlignTuple2 = ("query2", 1L, 100L, "subject2", 1L, 150L, 0.5, 15L, 35) expTAlignTuples = (expAlignTuple1, expAlignTuple2) self._iDb.createTable(tableName, "align", alignFileName, False) sqlCmd = "SELECT * FROM %s" % ( tableName ) self._iDb.execute( sqlCmd ) obsTAlignTuples = self._iDb.cursor.fetchall() self._iDb.dropTable(tableName) os.remove(alignFileName) self.assertEquals( expTAlignTuples, obsTAlignTuples ) def test_createTable_without_overwrite_Match( self ): tableName = "dummyMatchTable" + self._uniqId matchFileName = "dummyMatchFile.txt" matchF = open( matchFileName, "w" ) matchF.write( "qry1\t700\t760\t60\t100\t100\tsbj2\t500\t560\t60\t100\t1e-123\t136\t98.4\t2\n" ) matchF.write( "qry2\t700\t760\t60\t100\t100\tsbj2\t500\t560\t60\t100\t1e-123\t136\t98.4\t2\n" ) matchF.close() expMatchTuple = ("qry2", 700L, 760L, 60L, 100.0, 100.0, "sbj2", 500L, 560L, 60L, 100.0, 1e-123, 136L, 98.4, 2L) expTMatchTuples = (expMatchTuple,) self._iDb.createTable(tableName, "tab", matchFileName, False) sqlCmd = "SELECT * FROM %s" % ( tableName ) self._iDb.execute( sqlCmd ) obsTMatchTuples = self._iDb.cursor.fetchall() self._iDb.dropTable(tableName) os.remove(matchFileName) self.assertEquals( expTMatchTuples, obsTMatchTuples ) def test_createTable_without_overwrite_Path( self ): tableName = "dummyPathTable" + self._uniqId pathFileName = "dummyPathFile.txt" pathF = open( pathFileName, "w" ) pathF.write( "1\tqry\t1\t100\tsbj\t1\t100\t1e-123\t136\t98.4\n" ) pathF.write( "2\tqry\t500\t401\tsbj\t1\t100\t1e-152\t161\t98.7\n" ) pathF.close() expPathTuple1 = (1L, "qry", 1L, 100L, "sbj", 1L, 100L, 1e-123, 136L, 98.4) expPathTuple2 = (2L, "qry", 401L, 500L, "sbj", 100L, 1L, 1e-152, 161L, 98.7) # change coordinates expTPathTuples = (expPathTuple1, expPathTuple2) self._iDb.createTable(tableName, "Path", pathFileName, False) sqlCmd = "SELECT * FROM %s" % ( tableName ) self._iDb.execute( sqlCmd ) obsTPathTuples = self._iDb.cursor.fetchall() self._iDb.dropTable(tableName) os.remove(pathFileName) self.assertEquals( expTPathTuples, obsTPathTuples ) def test_createTable_without_overwrite_Set( self ): tableName = "dummySetTable" + self._uniqId setFileName = "dummySetFile.txt" setF = open( setFileName, "w" ) setF.write( "15\tset1\tchr1\t1\t100\n" ) setF.write( "15\tset2\tchr2\t1\t100\n" ) setF.close() expSetTuple1 = (15L, "set1", "chr1", 1L, 100L) expSetTuple2 = (15L, "set2", "chr2", 1L, 100L) expTSetTuples = (expSetTuple1, expSetTuple2) self._iDb.createTable(tableName, "Set", setFileName, False) sqlCmd = "SELECT * FROM %s" % ( tableName ) self._iDb.execute( sqlCmd ) obsTSetTuples = self._iDb.cursor.fetchall() self._iDb.dropTable(tableName) os.remove(setFileName) self.assertEquals( expTSetTuples, obsTSetTuples ) def test_createTable_without_overwrite_Seq( self ): tableName = "dummySeqTable" + self._uniqId seqFileName = "dummySeqFile.txt" seqF = open( seqFileName, "w" ) seqF.write( ">acc1 seq1\n" ) seqF.write( "ATACTTCGCTAGCTCGC\n" ) seqF.write( ">acc2 seq2\n" ) seqF.write( "ATACTTCGCTAGCTCGCATACTTCGCTAGCTCGCATACTTCGCTAGCTCGCATACTTCGCTAGCTCGC\n" ) seqF.close() expSeqTuple1 = ("acc1", "ATACTTCGCTAGCTCGC", "acc1 seq1", 17L) expSeqTuple2 = ("acc2", "ATACTTCGCTAGCTCGCATACTTCGCTAGCTCGCATACTTCGCTAGCTCGCATACTTCGCTAGCTCGC", "acc2 seq2", 68L) expTSeqTuples = (expSeqTuple1, expSeqTuple2) self._iDb.createTable(tableName, "fasta", seqFileName, False) sqlCmd = "SELECT * FROM %s" % ( tableName ) self._iDb.execute( sqlCmd ) obsTSeqTuples = self._iDb.cursor.fetchall() self._iDb.dropTable(tableName) os.remove(seqFileName) self.assertEquals( expTSeqTuples, obsTSeqTuples ) def test_createTable_with_overwrite_Classif( self ): tableName = "dummyClassifTable" + self._uniqId classifFileName = "dummyClassifFile.txt" with open( classifFileName, "w" ) as f: f.write("RIX-incomp-chim_DmelCaf1_2_0-B-G1000-Map3\t3508\t-\tPotentialChimeric\tI\tLINE\tincomplete\tCI=36; coding=(TE_BLRtx: DMCR1A:ClassI:LINE:Jockey: 14.16%); struct=(TElength: >700bps)\n") f.write("RLX-incomp_DmelCaf1_2_0-B-G1019-Map3\t4131\t+\tok\tI\tLTR\tincomplete\tCI=28; coding=(TE_BLRtx: ROO_I:ClassI:LTR:Bel-Pao: 43.27%, ROO_LTR:ClassI:LTR:Bel-Pao: 100.00%; TE_BLRx: BEL-6_DWil-I_2p:ClassI:LTR:Bel-Pao: 69.84%); struct=(TElength: >4000bps); other=(HG_BLRn: FBtr0087866_Dmel_r4.3: 4.72%; SSRCoverage=0.15<0.75)\n") self._iDb.createTable(tableName, "Classif", classifFileName, True) self.assertTrue(self._iDb.getSize(tableName) == 2) self._iDb.dropTable(tableName) os.remove(classifFileName) def test_createTable_no_file( self ): lTypesToTest = TABLE_SCHEMA_DESCRIPTOR.keys() lTypesToTest.extend(TABLE_TYPE_SYNONYMS) for tableType in lTypesToTest: tableName = "dummy%sTable%s" % (tableType, self._uniqId) self._iDb.createTable(tableName, tableType) self.assertTrue(self._iDb.doesTableExist(tableName)) self.assertTrue(self._iDb.isEmpty(tableName)) self._iDb.dropTable(tableName) def test_changePathQueryCoordinatesToDirectStrand(self): tableName = "dummyPathTable" + self._uniqId pathFileName = "dummyPathFile.txt" pathF = open( pathFileName, "w" ) pathF.write( "1\tqry\t100\t1\tsbj\t1\t100\t1e-123\t136\t98.4\n" ) pathF.write( "2\tqry\t500\t401\tsbj\t1\t100\t1e-152\t161\t98.7\n" ) pathF.write( "3\tqry\t5\t401\tsbj\t1\t100\t1e-152\t161\t98.7\n" ) pathF.close() expPathTuple1 = (1L, "qry", 1L, 100L, "sbj", 100L, 1L, 1e-123, 136L, 98.4) expPathTuple2 = (2L, "qry", 401L, 500L, "sbj", 100L, 1L, 1e-152, 161L, 98.7) expPathTuple3 = (3L, "qry", 5L, 401L, "sbj", 1L, 100L, 1e-152, 161L, 98.7) expTPathTuples = (expPathTuple1, expPathTuple2, expPathTuple3) sqlCmd = "CREATE TABLE %s ( path int unsigned, query_name varchar(255), query_start int , query_end int, subject_name varchar(255), subject_start int unsigned, subject_end int unsigned, E_value double, score int unsigned, identity float)" % tableName self._iDb.execute( sqlCmd ) self._iDb.loadDataFromFile(tableName, pathFileName, False) self._iDb.changePathQueryCoordinatesToDirectStrand(tableName) sqlCmd = "SELECT * FROM %s" % ( tableName ) self._iDb.execute( sqlCmd ) obsTPathTuples = self._iDb.cursor.fetchall() self._iDb.dropTable(tableName) os.remove(pathFileName) self.assertEquals( expTPathTuples, obsTPathTuples ) def test_exportDataToFile(self): tableName = "dummyPathTable" + self._uniqId expFileName = "dummyPathFile.txt" pathF = open( expFileName, "w" ) pathF.write( "1\tqry\t1\t100\tsbj\t100\t1\t1e-123\t136\t98.4\n" ) pathF.write( "2\tqry\t401\t500\tsbj\t100\t1\t1e-152\t161\t98.7\n" ) pathF.write( "3\tqry\t5\t401\tsbj\t1\t100\t1e-152\t161\t98.7\n" ) pathF.close() self._iDb.createTable(tableName, "Path", expFileName, False) obsFileName = "DummyObsFileName" self._iDb.exportDataToFile(tableName, obsFileName) self.assertTrue(FileUtils.isRessourceExists(obsFileName)) self.assertTrue(FileUtils.are2FilesIdentical(expFileName, obsFileName)) self._iDb.dropTable(tableName) os.remove(expFileName) os.remove(obsFileName) def test_exportDataToFile_keepFirstLineTrue(self): tableName = "dummyPathTable" + self._uniqId pathFileName = "dummyPathFile.txt" pathF = open( pathFileName, "w" ) pathF.write( "1\tqry\t1\t100\tsbj\t100\t1\t1e-123\t136\t98.4\n" ) pathF.write( "2\tqry\t401\t500\tsbj\t100\t1\t1e-152\t161\t98.7\n" ) pathF.write( "3\tqry\t5\t401\tsbj\t1\t100\t1e-152\t161\t98.7\n" ) pathF.close() expFileName = "expPathFile.txt" pathF = open( expFileName, "w" ) pathF.write("path\tquery_name\tquery_start\tquery_end\tsubject_name\tsubject_start\tsubject_end\tE_value\tscore\tidentity\n") pathF.write( "1\tqry\t1\t100\tsbj\t100\t1\t1e-123\t136\t98.4\n" ) pathF.write( "2\tqry\t401\t500\tsbj\t100\t1\t1e-152\t161\t98.7\n" ) pathF.write( "3\tqry\t5\t401\tsbj\t1\t100\t1e-152\t161\t98.7\n" ) pathF.close() self._iDb.createTable(tableName, "Path", pathFileName, False) obsFileName = "DummyObsFileName" self._iDb.exportDataToFile(tableName, obsFileName, True) self.assertTrue(FileUtils.isRessourceExists(obsFileName)) self.assertTrue(FileUtils.are2FilesIdentical(expFileName, obsFileName)) self._iDb.dropTable(tableName) os.remove(expFileName) os.remove(obsFileName) os.remove(pathFileName) def test_exportDataToFile_with_keepFirstLineTrue_and_param(self): tableName = "dummyPathTable" + self._uniqId pathFileName = "dummyPathFile.txt" pathF = open( pathFileName, "w" ) pathF.write( "1\tqry\t1\t100\tsbj\t100\t1\t1e-123\t136\t98.4\n" ) pathF.write( "2\tqry2\t401\t500\tsbj\t100\t1\t1e-152\t161\t98.7\n" ) pathF.write( "3\tqry\t5\t401\tsbj\t1\t100\t1e-152\t161\t98.7\n" ) pathF.close() expFileName = "expPathFile.txt" pathF = open( expFileName, "w" ) pathF.write("path\tquery_name\tquery_start\tquery_end\tsubject_name\tsubject_start\tsubject_end\tE_value\tscore\tidentity\n") pathF.write( "2\tqry2\t401\t500\tsbj\t100\t1\t1e-152\t161\t98.7\n" ) pathF.close() self._iDb.createTable(tableName, "Path", pathFileName, False) obsFileName = "DummyObsFileName" self._iDb.exportDataToFile(tableName, obsFileName, True, "where query_name = 'qry2'") self.assertTrue(FileUtils.isRessourceExists(obsFileName)) self.assertTrue(FileUtils.are2FilesIdentical(expFileName, obsFileName)) self._iDb.dropTable(tableName) os.remove(expFileName) os.remove(obsFileName) os.remove(pathFileName) def test_convertPathTableIntoAlignTable( self ): inPathTable = "dummyInPathTable_%s" % ( self._uniqId ) inPathFile = "dummyInPathFile_%s" % ( self._uniqId ) inPathFileHandler = open( inPathFile, "w" ) inPathFileHandler.write( "1\tqry\t1\t100\tsbj\t100\t1\t1e-123\t136\t98.4\n" ) inPathFileHandler.write( "2\tqry2\t401\t500\tsbj\t100\t1\t1e-152\t161\t98.7\n" ) inPathFileHandler.write( "3\tqry\t5\t401\tsbj\t1\t100\t1e-152\t161\t98.7\n" ) inPathFileHandler.close() self._iDb.createTable( inPathTable, "path", inPathFile, True ) expAlignFile = "dummyExpAlignFile_%s" % ( self._uniqId ) expAlignFileHandler = open( expAlignFile, "w" ) expAlignFileHandler.write( "qry\t1\t100\tsbj\t100\t1\t1e-123\t136\t98.4\n" ) expAlignFileHandler.write( "qry2\t401\t500\tsbj\t100\t1\t1e-152\t161\t98.7\n" ) expAlignFileHandler.write( "qry\t5\t401\tsbj\t1\t100\t1e-152\t161\t98.7\n" ) expAlignFileHandler.close() obsAlignTable = "dummyObsAlignTable_%s" % ( self._uniqId ) self._iDb.convertPathTableIntoAlignTable( inPathTable, obsAlignTable ) obsAlignFile = "dummyObsAlignFile_%s" % ( self._uniqId ) self._iDb.exportDataToFile( obsAlignTable, obsAlignFile, False ) self.assertTrue( FileUtils.are2FilesIdentical( expAlignFile, obsAlignFile ) ) for f in [ inPathFile, expAlignFile, obsAlignFile ]: os.remove( f ) for t in [ inPathTable, obsAlignTable ]: self._iDb.dropTable( t ) def test_convertAlignTableIntoPathTable( self ): inAlignTable = "dummyInPathTable_%s" % ( self._uniqId ) inAlignFile = "dummyInPathFile_%s" % ( self._uniqId ) inAlignFileHandler = open( inAlignFile, "w" ) inAlignFileHandler.write( "qry\t1\t100\tsbj\t100\t1\t1e-123\t136\t98.4\n" ) inAlignFileHandler.write( "qry2\t401\t500\tsbj\t100\t1\t1e-152\t161\t98.7\n" ) inAlignFileHandler.write( "qry3\t5\t401\tsbj\t1\t100\t1e-152\t161\t98.7\n" ) inAlignFileHandler.close() self._iDb.createTable( inAlignTable, "align", inAlignFile, True ) expPathFile = "dummyExpPathFile_%s" % ( self._uniqId ) expPathFileHandler = open( expPathFile, "w" ) expPathFileHandler.write( "1\tqry\t1\t100\tsbj\t100\t1\t1e-123\t136\t98.4\n" ) expPathFileHandler.write( "2\tqry2\t401\t500\tsbj\t100\t1\t1e-152\t161\t98.7\n" ) expPathFileHandler.write( "3\tqry3\t5\t401\tsbj\t1\t100\t1e-152\t161\t98.7\n" ) expPathFileHandler.close() obsPathTable = "dummyObsPathTable_%s" % ( self._uniqId ) self._iDb.convertAlignTableIntoPathTable( inAlignTable, obsPathTable ) obsPathFile = "dummyObsAlignFile_%s" % ( self._uniqId ) self._iDb.exportDataToFile( obsPathTable, obsPathFile, False ) self.assertTrue( FileUtils.are2FilesIdentical( expPathFile, obsPathFile ) ) for f in [ inAlignFile, expPathFile, obsPathFile ]: os.remove( f ) for t in [ inAlignTable, obsPathTable ]: self._iDb.dropTable( t ) def test_convertAlignTableIntoPathTable_with_single_quote( self ): inAlignTable = "dummyInPathTable_%s" % ( self._uniqId ) inAlignFile = "dummyInPathFile_%s" % ( self._uniqId ) inAlignFileHandler = open( inAlignFile, "w" ) inAlignFileHandler.write( "qry\t1\t100\t'sbj\t100\t1\t1e-123\t136\t98.4\n" ) inAlignFileHandler.write( "qry2\t401\t500\tsbj'\t100\t1\t1e-152\t161\t98.7\n" ) inAlignFileHandler.write( "qry3\t5\t401\tsbj\t1\t100\t1e-152\t161\t98.7\n" ) inAlignFileHandler.close() self._iDb.createTable( inAlignTable, "align", inAlignFile, True ) expPathFile = "dummyExpPathFile_%s" % ( self._uniqId ) expPathFileHandler = open( expPathFile, "w" ) expPathFileHandler.write( "1\tqry\t1\t100\t'sbj\t100\t1\t1e-123\t136\t98.4\n" ) expPathFileHandler.write( "2\tqry2\t401\t500\tsbj'\t100\t1\t1e-152\t161\t98.7\n" ) expPathFileHandler.write( "3\tqry3\t5\t401\tsbj\t1\t100\t1e-152\t161\t98.7\n" ) expPathFileHandler.close() obsPathTable = "dummyObsPathTable_%s" % ( self._uniqId ) self._iDb.convertAlignTableIntoPathTable( inAlignTable, obsPathTable ) obsPathFile = "dummyObsAlignFile_%s" % ( self._uniqId ) self._iDb.exportDataToFile( obsPathTable, obsPathFile, False ) self.assertTrue( FileUtils.are2FilesIdentical( expPathFile, obsPathFile ) ) for f in [ inAlignFile, expPathFile, obsPathFile ]: os.remove( f ) for t in [ inAlignTable, obsPathTable ]: self._iDb.dropTable( t ) def test_getObjectListWithSQLCmd(self): inPathTable = "dummyInPathTable_%s" % ( self._uniqId ) inPathFile = "dummyInPathFile_%s" % ( self._uniqId ) inPathFileHandler = open( inPathFile, "w" ) inPathFileHandler.write( "1\tqry\t100\t1\tsbj\t1\t100\t1e-123\t136\t98.4\n" ) inPathFileHandler.write( "2\tqry\t500\t401\tsbj\t1\t100\t1e-152\t161\t98.7\n" ) inPathFileHandler.write( "3\tqry\t5\t401\tsbj\t1\t100\t1e-152\t161\t98.7\n" ) inPathFileHandler.close() self._iDb.createTable( inPathTable, "path", inPathFile, True ) path1 = Path() path1.setFromTuple((1, "qry", 1, 100, "sbj", 100, 1, 1e-123, 136, 98.4)) path2 = Path() path2.setFromTuple((2, "qry", 401, 500, "sbj", 100, 1, 1e-152, 161, 98.7)) path3 = Path() path3.setFromTuple((3, "qry", 5, 401, "sbj", 1, 100, 1e-152, 161, 98.7)) expLPath = [path1, path2, path3] sqlCmd = "SELECT * FROM %s;" % (inPathTable) obsLPath = self._iDb.getObjectListWithSQLCmd(sqlCmd, self._getInstanceToAdapt) os.remove( inPathFile ) self._iDb.dropTable( inPathTable ) self.assertEquals(expLPath, obsLPath) def test_getIntegerListWithSQLCmd(self): inPathTable = "dummyInPathTable_%s" % ( self._uniqId ) inPathFile = "dummyInPathFile_%s" % ( self._uniqId ) inPathFileHandler = open( inPathFile, "w" ) inPathFileHandler.write( "1\tqry\t100\t1\tsbj\t1\t100\t1e-123\t136\t98.4\n" ) inPathFileHandler.write( "2\tqry\t500\t401\tsbj\t1\t100\t1e-152\t161\t98.7\n" ) inPathFileHandler.write( "3\tqry\t5\t401\tsbj\t1\t100\t1e-152\t161\t98.7\n" ) inPathFileHandler.close() self._iDb.createTable( inPathTable, "path", inPathFile, True ) expLPath = [1, 2, 3] sqlCmd = "SELECT * FROM %s;" % (inPathTable) obsLPath = self._iDb.getIntegerListWithSQLCmd(sqlCmd) os.remove( inPathFile ) self._iDb.dropTable( inPathTable ) self.assertEquals(expLPath, obsLPath) def test_getIntegerWithSQLCmd(self): inPathTable = "dummyInPathTable_%s" % ( self._uniqId ) inPathFile = "dummyInPathFile_%s" % ( self._uniqId ) inPathFileHandler = open( inPathFile, "w" ) inPathFileHandler.write( "1\tqry\t100\t1\tsbj\t1\t100\t1e-123\t136\t98.4\n" ) inPathFileHandler.write( "2\tqry\t500\t401\tsbj\t1\t100\t1e-152\t161\t98.7\n" ) inPathFileHandler.write( "3\tqry\t5\t401\tsbj\t1\t100\t1e-152\t161\t98.7\n" ) inPathFileHandler.close() self._iDb.createTable( inPathTable, "path", inPathFile, True ) expId = 1 sqlCmd = "SELECT path FROM %s where path='%d';" % (inPathTable, 1) obsId = self._iDb.getIntegerWithSQLCmd(sqlCmd) os.remove( inPathFile ) self._iDb.dropTable( inPathTable ) self.assertEquals(expId, obsId) def test_getStringListWithSQLCmd(self): inPathTable = "dummyInPathTable_%s" % ( self._uniqId ) inPathFile = "dummyInPathFile_%s" % ( self._uniqId ) inPathFileHandler = open( inPathFile, "w" ) inPathFileHandler.write( "1\tqry\t100\t1\tsbj\t1\t100\t1e-123\t136\t98.4\n" ) inPathFileHandler.write( "2\tqry\t500\t401\tsbj\t1\t100\t1e-152\t161\t98.7\n" ) inPathFileHandler.write( "3\tqry\t5\t401\tsbj\t1\t100\t1e-152\t161\t98.7\n" ) inPathFileHandler.close() self._iDb.createTable( inPathTable, "path", inPathFile, True ) expLString = ["qry","qry","qry"] sqlCmd = "SELECT query_name FROM %s;" % (inPathTable) obsLString = self._iDb.getStringListWithSQLCmd(sqlCmd) os.remove( inPathFile ) self._iDb.dropTable( inPathTable ) self.assertEquals(expLString, obsLString) def test_removeDoublons( self ): inPathTable = "dummyInPathTable_%s" % ( self._uniqId ) inPathFile = "dummyInPathFile_%s" % ( self._uniqId ) inPathFileHandler = open( inPathFile, "w" ) inPathFileHandler.write( "1\tqry\t1\t100\tsbj\t1\t100\t1e-123\t136\t98.4\n" ) inPathFileHandler.write( "2\tqry\t401\t500\tsbj\t1\t100\t1e-152\t161\t98.7\n" ) inPathFileHandler.write( "2\tqry\t401\t500\tsbj\t1\t100\t1e-152\t161\t98.7\n" ) inPathFileHandler.close() self._iDb.createTable( inPathTable, "path", inPathFile, True ) expFile = "dummyExpFile_%s" % ( self._uniqId ) expFileHandler = open( expFile, "w" ) expFileHandler.write( "1\tqry\t1\t100\tsbj\t1\t100\t1e-123\t136\t98.4\n" ) expFileHandler.write( "2\tqry\t401\t500\tsbj\t1\t100\t1e-152\t161\t98.7\n" ) expFileHandler.close() self._iDb.removeDoublons( inPathTable ) obsFile = "dummyObsFile_%s" % ( self._uniqId ) self._iDb.exportDataToFile(inPathTable, obsFile) self.assertTrue( FileUtils.are2FilesIdentical( expFile, obsFile ) ) self._iDb.dropTable( inPathTable ) for f in [ inPathFile, expFile, obsFile ]: os.remove( f ) def test_getTableListFromPattern_oneTable( self ): inTable = "dummyInTable_%s" % ( self._uniqId ) self._iDb.createTable( inTable, "path", "", True ) exp = [ inTable ] obs = self._iDb.getTableListFromPattern( "%s%%" % inTable ) self.assertEqual( exp, obs ) self._iDb.dropTable( inTable ) def test_getTableListFromPattern_twoTables( self ): inTable1 = "dummyInTable1_%s" % ( self._uniqId ) inTable2 = "dummyInTable2_%s" % ( self._uniqId ) inTable3 = "dummyTotoTable3_%s" % ( self._uniqId ) for table in [ inTable1, inTable2, inTable3 ]: self._iDb.createTable( table, "path", "", True ) exp = [ inTable1, inTable2 ] obs = self._iDb.getTableListFromPattern( "dummyInTable%%_%s" % self._uniqId ) self.assertEqual( exp, obs ) for table in [ inTable1, inTable2, inTable3 ]: self._iDb.dropTable( table ) def test_createPathStatTable(self): statsFileName = "DmelCaf1_statsPerClassif.txt" f = open (statsFileName, "w") f.write("family\tmaxLength\tmeanLength\tcovg\tfrags\tfullLgthFrags\tcopies\tfullLgthCopies\tmeanId\tsdId\tminId\tq25Id\tmedId\tq75Id\tmaxId\tmeanLgth\tsdLgth\tminLgth\tq25Lgth\tmedLgth\tq75Lgth\tmaxLgth\tmeanLgthPerc\tsdLgthPerc\tminLgthPerc\tq25LgthPerc\tmedLgthPerc\tq75LgthPerc\tmaxLgthPerc\n") f.write("Helitron\t2367\t2367\t138367\t852\t0\t803\t0\t81.20\t4.24\t68.55\t78.32\t81.03\t83.49\t100.00\t172.46\t184.92\t21\t70.00\t129.00\t216.00\t2202\t7.29\t7.81\t0.89\t2.96\t5.45\t9.13\t93.03\n") f.write("LINE\t7688\t7688\t3769377\t8358\t10\t6329\t10\t85.52\t8.02\t62.80\t79.27\t83.33\t92.88\t100.00\t597.97\t980.29\t21\t117.00\t256.00\t537.00\t7726\t7.78\t12.75\t0.27\t1.52\t3.33\t6.98\t100.49\n") f.write("LTR\t13754\t13754\t9146587\t20749\t0\t17868\t1\t82.69\t7.39\t58.76\t77.81\t80.82\t85.67\t100.00\t519.75\t1217.12\t20\t105.00\t183.50\t336.00\t13738\t3.78\t8.85\t0.15\t0.76\t1.33\t2.44\t99.88\n") f.write("MITE\t378\t378\t2890\t10\t3\t9\t3\t98.78\t1.20\t95.80\t98.64\t99.18\t99.46\t99.73\t325.33\t47.86\t253\t290.00\t333.00\t362.00\t390\t86.07\t12.66\t66.93\t76.72\t88.10\t95.77\t103.17\n") f.write("NoCat\t9999\t9999\t384076\t1297\t1\t1219\t1\t82.60\t6.73\t61.20\t78.37\t81.41\t85.29\t100.00\t323.01\t686.85\t21\t64.00\t139.00\t280.00\t10000\t3.23\t6.87\t0.21\t0.64\t1.39\t2.80\t100.01\n") f.write("SSR\t680\t680\t325152\t2340\t24\t2290\t28\t79.07\t3.60\t69.19\t76.64\t79.02\t81.10\t97.83\t221.64\t139.84\t21\t121.00\t183.00\t285.00\t799\t32.59\t20.57\t3.09\t17.79\t26.91\t41.91\t117.50\n") f.write("TIR\t2532\t2532\t700173\t2503\t5\t2160\t5\t84.70\t7.43\t64.03\t79.46\t82.77\t90.09\t100.00\t326.54\t405.94\t21\t90.00\t187.00\t342.00\t2758\t12.90\t16.03\t0.83\t3.55\t7.39\t13.51\t108.93\n") f.write("confused\t19419\t19419\t1299224\t3903\t0\t3311\t0\t82.30\t6.34\t63.20\t78.17\t80.81\t84.58\t100.00\t408.22\t989.57\t21\t113.00\t207.00\t339.00\t17966\t2.10\t5.10\t0.11\t0.58\t1.07\t1.75\t92.52\n") f.close() tableName = "dummyDmelCaf1_chr_allTEs_nr_noSSR_join_path_statsPerClassif" self._iDb.createTable(tableName, "pathstat", statsFileName) self.assertTrue(self._iDb.doesTableExist(tableName)) expSize = 8 obsSize = self._iDb.getSize(tableName) self.assertEquals(expSize, obsSize) expColumnNb = 29 sqlCmd = "DESC %s;" % tableName self._iDb.execute(sqlCmd) res = self._iDb.fetchall() obsColumnNb = len(res) self.assertEquals(expColumnNb, obsColumnNb) self._iDb.dropTable(tableName) os.remove(statsFileName) def test_createJobTable_is_table_created(self): tableName = "dummyJobTable" + self._uniqId self._iDb.createTable(tableName, "jobs") self.assertTrue(self._iDb.doesTableExist(tableName)) self._iDb.dropTable(tableName) def test_createClassifTable(self): tableName = "dummyClassifTable" self._iDb.dropTable(tableName) fileName = "test.classif" with open(fileName, "w") as f: f.write("RIX-incomp-chim_DmelCaf1_2_0-B-G1000-Map3\t3508\t-\tPotentialChimeric\tI\tLINE\tincomplete\tCI=36; coding=(TE_BLRtx: DMCR1A:ClassI:LINE:Jockey: 14.16%, FW3_DM:ClassI:LINE:Jockey: 15.07%; TE_BLRx: CR1-1_DWil_2p:ClassI:LINE:Jockey: 18.98%, FW2_DM-ORF1p:ClassI:LINE:Jockey: 22.36%, Jockey-1_DYa_1p:ClassI:LINE:Jockey: 11.86%); struct=(TElength: >700bps); other=(TE_BLRx: Gypsy7-I_Dmoj_1p:ClassI:LTR:Gypsy: 12.58%; HG_BLRn: FBtr0089196_Dmel_r4.3: 11.74%; SSRCoverage=0.12<0.75)\n") f.write("RLX-incomp_DmelCaf1_2_0-B-G1019-Map3\t4131\t+\tok\tI\tLTR\tincomplete\tCI=28; coding=(TE_BLRtx: ROO_I:ClassI:LTR:Bel-Pao: 43.27%, ROO_LTR:ClassI:LTR:Bel-Pao: 100.00%; TE_BLRx: BEL-6_DWil-I_2p:ClassI:LTR:Bel-Pao: 69.84%); struct=(TElength: >4000bps); other=(HG_BLRn: FBtr0087866_Dmel_r4.3: 4.72%; SSRCoverage=0.15<0.75)\n") f.write("RLX-incomp_DmelCaf1_2_0-B-G1025-Map3\t6534\t-\tok\tI\tLTR\tincomplete\tCI=28; coding=(TE_BLRtx: Gypsy2-I_Dmoj:ClassI:LTR:Gypsy: 11.82%, MDG3_DM:ClassI:LTR:Gypsy: 17.43%, STALKER2_LTR:ClassI:LTR:Gypsy: 14.62%, STALKER4_LTR:ClassI:LTR:Gypsy: 57.21%; TE_BLRx: Gypsy-16_DWil-I_1p:ClassI:LTR:Gypsy: 32.19%; profiles: PF00665.18_rve_INT_32.0: 68.64%); struct=(TElength: >4000bps); other=(HG_BLRn: FBtr0070036_Dmel_r4.3: 3.73%; TermRepeats: non-termLTR: 1701; SSRCoverage=0.14<0.75)\n") self._iDb.createTable(tableName, "classif", fileName) self.assertTrue(self._iDb.doesTableExist(tableName)) expColumnNb = 8 sqlCmd = "DESC %s;" % tableName self._iDb.execute(sqlCmd) res = self._iDb.fetchall() obsColumnNb = len(res) self.assertEquals(expColumnNb, obsColumnNb) expSize = 3 obsSize = self._iDb.getSize(tableName) self.assertEquals(expSize, obsSize) expLIndex = ["iseq_name", "istatus", "iclass", "iorder", "icomp"] sqlCmd = "SHOW INDEX FROM %s" % tableName self._iDb.execute(sqlCmd) res = self._iDb.cursor.fetchall() obsLIndex = [] for tuple in res: obsLIndex.append(tuple[2]) self.assertEquals(expLIndex, obsLIndex) self._iDb.dropTable(tableName) os.remove(fileName) def test_createClassifIndex(self): tableName = "dummyclassifTable%s" % self._uniqId sqlCmd = "CREATE TABLE %s (seq_name varchar(255), length int unsigned, strand char, status varchar(255), class_classif varchar(255), order_classif varchar(255), completeness varchar(255), evidences text);" % tableName self._iDb.execute(sqlCmd) expLIndex = ["iseq_name", "istatus", "iclass", "iorder", "icomp"] self._iDb.createIndex(tableName, "classif") sqlCmd = "SHOW INDEX FROM %s" % tableName self._iDb.execute(sqlCmd) res = self._iDb.cursor.fetchall() obsLIndex = [] for tuple in res: obsLIndex.append(tuple[2]) self.assertEquals(expLIndex, obsLIndex) self._iDb.dropTable(tableName) def test_createBinPathTable(self): pathFileName = "dummy.path" with open(pathFileName, "w") as pathF: pathF.write("1\tqry\t1\t100\tsbj\t1\t100\t1e-123\t136\t98.4\n") pathF.write("2\tqry\t500\t401\tsbj\t1\t100\t1e-152\t161\t98.7\n") expPathTuple1 = (1, 1000000, "qry", 1, 100, 1) expPathTuple2 = (2, 1000000, "qry", 401, 500, 1) # change coordinates expTPathTuples = (expPathTuple1, expPathTuple2) pathTableName = "dummy_path" idxTableName = "dummy_path_idx" self._iDb.createTable(pathTableName, "path", pathFileName) self._iDb.createBinPathTable(pathTableName, True) sqlCmd = "SELECT * FROM %s" % idxTableName self._iDb.execute(sqlCmd) obsTPathTuples = self._iDb.fetchall() self._iDb.dropTable(pathTableName) self._iDb.dropTable(idxTableName) os.remove(pathFileName) self.assertEquals(expTPathTuples, obsTPathTuples) def test_createBinSetTable(self): setFileName = "dummy.set" with open(setFileName, "w") as setF: setF.write("1\tseq1\tchr1\t1900\t3900\n") setF.write("2\tseq2\tchr1\t2\t9\n") setF.write("3\tseq3\tchr1\t8\t13\n") expTuple = ((1L, 10000.0, 'chr1', 1900L, 3900L, 1L), (2L, 1000.0, 'chr1', 2L, 9L, 1L), (3L, 1000.0, 'chr1', 8L, 13L, 1L)) setTableName = "dummy_set" idxTableName = "dummy_set_idx" self._iDb.createTable(setTableName, "set", setFileName) self._iDb.createBinSetTable(setTableName, True) sqlCmd = "SELECT * FROM %s" % idxTableName self._iDb.execute(sqlCmd) obsTuple = self._iDb.fetchall() self._iDb.dropTable(setTableName) self._iDb.dropTable(idxTableName) os.remove(setFileName) self.assertEquals(expTuple, obsTuple) def _getInstanceToAdapt(self): iPath = Path() return iPath if __name__ == "__main__": unittest.main()