Mercurial > repos > yufei-luo > s_mart

comparison commons/pyRepetUnit/convCoord/test/Test_PathChunkConnector.py @ 31:0ab839023fe4

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
Uploaded
author m-zytnicki
date Tue, 30 Apr 2013 14:33:21 -0400
parents 94ab73e8a190
children
comparison
equal deleted inserted replaced
30:5677346472b5 31:0ab839023fe4
1 import unittest
2 import time
3 import os
4 from commons.core.sql.TablePathAdaptator import TablePathAdaptator
5 from commons.pyRepetUnit.convCoord.PathChunkConnector import PathChunkConnector
6 from commons.core.sql.DbMySql import DbMySql
7
8
9 class Test_PathChunkConnector( unittest.TestCase ):
10
11 def setUp(self):
12 self._uniqId = "%s_%s" % ( time.strftime("%Y%m%d%H%M%S") , os.getpid() )
13 self._configFileName = "dummyConfigFile_%s" % ( self._uniqId )
14 configF = open(self._configFileName, "w" )
15 configF.write( "[repet_env]\n" )
16 configF.write( "repet_host: %s\n" % ( os.environ["REPET_HOST"] ) )
17 configF.write( "repet_user: %s\n" % ( os.environ["REPET_USER"] ) )
18 configF.write( "repet_pw: %s\n" % ( os.environ["REPET_PW"] ) )
19 configF.write( "repet_db: %s\n" % ( os.environ["REPET_DB"] ) )
20 configF.write( "repet_port: %s\n" % ( os.environ["REPET_PORT"] ) )
21 configF.close()
22 self._db = DbMySql( cfgFileName=self._configFileName )
23 self._table = "dummyPathTable_%s" % ( self._uniqId )
24 self._tpA = TablePathAdaptator( self._db, self._table )
25 self._mapDict = {'chunk1': ('dmel_chr4', 1, 100000),'chunk2': ('dmel_chr4', 90001, 190000),'chunk3': ('dmel_chr4', 180001, 280000), }
26 self._mapFileName = "map_file.map"
27
28 def tearDown(self):
29 self._db.close()
30
31 def testTwoQueryOverlapsOnPlusStrand (self):
32 lines = [
33 "1\tdmel_chr4\t95535\t95570\tsbj2\t125423\t125467\t7e-15\t82\t97.78\n",
34 "2\tdmel_chr4\t95545\t95576\tsbj2\t125457\t133465\t2e-38\t83\t65\n"
35 ]
36
37 expectedList =[("dmel_chr4", 95535, 95576, "sbj2", 125423, 133465, 2e-38, 83,97.78)]
38 self._templateTest(lines, expectedList)
39
40 def testTwoQueryOverlapsOnReverseStrand (self):
41 lines = [
42 "1\tdmel_chr4\t95535\t95570\tsbj2\t125467\t125423\t7e-15\t82\t97.78\n",
43 "2\tdmel_chr4\t95545\t95576\tsbj2\t133465\t125457\t2e-38\t83\t65\n"
44 ]
45
46 expectedList =[("dmel_chr4", 95535, 95576, "sbj2", 133465, 125423, 2e-38, 83,97.78)]
47 self._templateTest(lines, expectedList)
48
49 def testTwoQueryOverlapsOnDifferentStrands (self):
50 lines = [
51 "1\tdmel_chr4\t95535\t95570\tsbj2\t125423\t125467\t7e-15\t82\t97.78\n",
52 "2\tdmel_chr4\t95545\t95576\tsbj2\t133465\t125457\t2e-38\t83\t65\n"
53 ]
54 expectedList =[
55 ("dmel_chr4", 95535, 95570, "sbj2", 125423, 125467, 7e-15, 82,97.78),
56 ("dmel_chr4", 95545, 95576, "sbj2", 133465, 125457, 2e-38, 83,65)
57 ]
58 self._templateTest(lines, expectedList)
59
60 def _templateTest(self, datas2TestList, expectedList ):
61 pathFileName = "dummyPathFile_%s" % ( self._uniqId )
62 _MockPathFile(pathFileName, datas2TestList)
63
64
65 self._db.createTable( self._table, "path", pathFileName )
66
67 _MockMapFile(self._mapFileName)
68 chunkConnector = PathChunkConnector(self._mapFileName, self._db, self._table, 0)
69 chunkConnector.run()
70
71 sql_cmd = 'select * from %s' % (self._table)
72 self._db.execute(sql_cmd)
73 res = self._db.fetchall()
74
75 for i in xrange(len(expectedList)):
76 resultTuple = res[i]
77 expectedTuple = expectedList[i]
78 self._assertExpectedTupleEqualsObsTuple(expectedTuple, resultTuple)
79
80 self._db.dropTable( self._table )
81 os.remove(pathFileName)
82 os.remove(self._mapFileName)
83
84 def _assertExpectedTupleEqualsObsTuple(self, expectedTuple, resultTuple):
85 self.assertEquals(expectedTuple[0], resultTuple[1])
86 self.assertEquals(expectedTuple[1], resultTuple[2])
87 self.assertEquals(expectedTuple[2], resultTuple[3])
88 self.assertEquals(expectedTuple[3], resultTuple[4])
89 self.assertEquals(expectedTuple[4], resultTuple[5])
90 self.assertEquals(expectedTuple[5], resultTuple[6])
91 self.assertEquals(expectedTuple[6], resultTuple[7])
92 self.assertEquals(expectedTuple[7], resultTuple[8])
93 self.assertEquals(expectedTuple[8], resultTuple[9])
94
95
96 class _MockPathFile:
97
98 def __init__(self, fileName, lines):
99 path = open(fileName, "w");
100 for line in lines:
101 path.write(line)
102 path.close
103
104
105 class _MockMapFile:
106
107 def __init__ (self, fileName):
108 map = open(fileName, "w")
109 line1 = "chunk1" + '\t' + "dmel_chr4" + '\t' + "1" + '\t'+ "100000" + "\n"
110 line2 = "chunk2" + '\t' + "dmel_chr4" + '\t' + "90001" + '\t'+ "190000" + "\n"
111 line3 = "chunk3" + '\t' + "dmel_chr4" + '\t' + "180001" + '\t' + "280000" + "\n"
112 map.write(line1)
113 map.write(line2)
114 map.write(line3)
115 map.close
116
117
118 test_suite = unittest.TestSuite()
119 test_suite.addTest( unittest.makeSuite( Test_PathChunkConnector ) )
120 if __name__ == "__main__":
121 unittest.TextTestRunner(verbosity=2).run( test_suite )