|
18
|
1 import unittest
|
|
|
2 import os
|
|
|
3 import time
|
|
|
4 from commons.core.sql.DbMySql import DbMySql
|
|
|
5 from commons.tools.getCumulLengthFromTEannot import getCumulLengthFromTEannot
|
|
|
6 from commons.core.utils.FileUtils import FileUtils
|
|
|
7
|
|
|
8
|
|
|
9 class Test_getCumulLengthFromTEannot( unittest.TestCase ):
|
|
|
10
|
|
|
11 def setUp( self ):
|
|
|
12 self.fileUtils = FileUtils()
|
|
|
13 self._getCumulLengthFromTEannot = None
|
|
|
14 self._uniqId = "%s_%s" % ( time.strftime("%Y_%m_%d_%H_%M_%S") , os.getpid() )
|
|
|
15 self._configFileName = "dummyConfig_%s" % self._uniqId
|
|
|
16 configFile = open( self._configFileName, "w" )
|
|
|
17 configFile.write("[repet_env]\n")
|
|
|
18 configFile.write( "repet_host: %s\n" % ( os.environ["REPET_HOST"] ) )
|
|
|
19 configFile.write( "repet_user: %s\n" % ( os.environ["REPET_USER"] ) )
|
|
|
20 configFile.write( "repet_pw: %s\n" % ( os.environ["REPET_PW"] ) )
|
|
|
21 configFile.write( "repet_db: %s\n" % ( os.environ["REPET_DB"] ) )
|
|
|
22 configFile.write( "repet_port: %s\n" % ( os.environ["REPET_PORT"] ) )
|
|
|
23 configFile.close()
|
|
|
24
|
|
|
25
|
|
|
26 def tearDown( self ):
|
|
|
27 os.remove( self._configFileName )
|
|
|
28
|
|
|
29
|
|
|
30 def test_getAllSubjectsAsMapOfQueries( self ):
|
|
|
31 inFileName = "dummyInFile_%s" % self._uniqId
|
|
|
32 expFileName = "dummyExp_%s" % self._uniqId
|
|
|
33 for f in [inFileName,expFileName]:
|
|
|
34 if os.path.exists(f): os.remove(f)
|
|
|
35 inFile = open( inFileName, "w" )
|
|
|
36 inFile.write( "1\tchr1\t1501\t2500\tTE1\t1\t500\t0.0\t880\t95.7\n" )
|
|
|
37 inFile.write( "2\tchr1\t3401\t4000\tTE3\t101\t700\t0.0\t950\t97.2\n" )
|
|
|
38 inFile.close()
|
|
|
39 inTable = "dummyPathTable_%s" % self._uniqId
|
|
|
40 db = DbMySql(cfgFileName=self._configFileName)
|
|
|
41 db.createTable(inTable, "path", inFileName)
|
|
|
42 expFile = open( expFileName, "w" )
|
|
|
43 expFile.write( "TE1\tchr1\t1501\t2500\n" )
|
|
|
44 expFile.write( "TE3\tchr1\t3401\t4000\n" )
|
|
|
45 expFile.close()
|
|
|
46 self._getCumulLengthFromTEannot = getCumulLengthFromTEannot()
|
|
|
47 self._getCumulLengthFromTEannot._tableName = inTable
|
|
|
48 self._getCumulLengthFromTEannot._configFileName = self._configFileName
|
|
|
49 self._getCumulLengthFromTEannot.setAdaptatorToTable()
|
|
|
50 obsFileName = self._getCumulLengthFromTEannot.getAllSubjectsAsMapOfQueries()
|
|
|
51 self.assertTrue( self.fileUtils.are2FilesIdentical( obsFileName, expFileName ) )
|
|
|
52 for f in [inFileName,inTable+".map",expFileName]:
|
|
|
53 if os.path.exists(f): os.remove(f)
|
|
|
54 db.dropTable(inTable)
|
|
|
55 db.close()
|
|
|
56
|
|
|
57
|
|
|
58 def test_getCumulLength( self ):
|
|
|
59 mergeFileName = "dummyInFile_%s" % self._uniqId
|
|
|
60 mergeFile = open( mergeFileName, "w" )
|
|
|
61 mergeFile.write( "TE1\tchr1\t1501\t2500\n" )
|
|
|
62 mergeFile.write( "TE3\tchr1\t4000\t3401\n" )
|
|
|
63 mergeFile.close()
|
|
|
64 exp = 1000+600
|
|
|
65 self._getCumulLengthFromTEannot = getCumulLengthFromTEannot()
|
|
|
66 obs = self._getCumulLengthFromTEannot.getCumulLength( mergeFileName )
|
|
|
67 self.assertEqual( obs, exp )
|
|
|
68 os.remove( mergeFileName )
|
|
|
69
|
|
|
70 if __name__ == "__main__":
|
|
|
71 unittest.main() |
