Mercurial > repos > yufei-luo > s_mart
diff commons/pyRepetUnit/convCoord/test/Test_PathChunkConnector.py @ 31:0ab839023fe4
Uploaded
author | m-zytnicki |
---|---|
date | Tue, 30 Apr 2013 14:33:21 -0400 |
parents | 94ab73e8a190 |
children |
line wrap: on
line diff
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/commons/pyRepetUnit/convCoord/test/Test_PathChunkConnector.py Tue Apr 30 14:33:21 2013 -0400 @@ -0,0 +1,121 @@ +import unittest +import time +import os +from commons.core.sql.TablePathAdaptator import TablePathAdaptator +from commons.pyRepetUnit.convCoord.PathChunkConnector import PathChunkConnector +from commons.core.sql.DbMySql import DbMySql + + +class Test_PathChunkConnector( unittest.TestCase ): + + def setUp(self): + self._uniqId = "%s_%s" % ( time.strftime("%Y%m%d%H%M%S") , os.getpid() ) + self._configFileName = "dummyConfigFile_%s" % ( self._uniqId ) + configF = open(self._configFileName, "w" ) + configF.write( "[repet_env]\n" ) + configF.write( "repet_host: %s\n" % ( os.environ["REPET_HOST"] ) ) + configF.write( "repet_user: %s\n" % ( os.environ["REPET_USER"] ) ) + configF.write( "repet_pw: %s\n" % ( os.environ["REPET_PW"] ) ) + configF.write( "repet_db: %s\n" % ( os.environ["REPET_DB"] ) ) + configF.write( "repet_port: %s\n" % ( os.environ["REPET_PORT"] ) ) + configF.close() + self._db = DbMySql( cfgFileName=self._configFileName ) + self._table = "dummyPathTable_%s" % ( self._uniqId ) + self._tpA = TablePathAdaptator( self._db, self._table ) + self._mapDict = {'chunk1': ('dmel_chr4', 1, 100000),'chunk2': ('dmel_chr4', 90001, 190000),'chunk3': ('dmel_chr4', 180001, 280000), } + self._mapFileName = "map_file.map" + + def tearDown(self): + self._db.close() + + def testTwoQueryOverlapsOnPlusStrand (self): + lines = [ + "1\tdmel_chr4\t95535\t95570\tsbj2\t125423\t125467\t7e-15\t82\t97.78\n", + "2\tdmel_chr4\t95545\t95576\tsbj2\t125457\t133465\t2e-38\t83\t65\n" + ] + + expectedList =[("dmel_chr4", 95535, 95576, "sbj2", 125423, 133465, 2e-38, 83,97.78)] + self._templateTest(lines, expectedList) + + def testTwoQueryOverlapsOnReverseStrand (self): + lines = [ + "1\tdmel_chr4\t95535\t95570\tsbj2\t125467\t125423\t7e-15\t82\t97.78\n", + "2\tdmel_chr4\t95545\t95576\tsbj2\t133465\t125457\t2e-38\t83\t65\n" + ] + + expectedList =[("dmel_chr4", 95535, 95576, "sbj2", 133465, 125423, 2e-38, 83,97.78)] + self._templateTest(lines, expectedList) + + def testTwoQueryOverlapsOnDifferentStrands (self): + lines = [ + "1\tdmel_chr4\t95535\t95570\tsbj2\t125423\t125467\t7e-15\t82\t97.78\n", + "2\tdmel_chr4\t95545\t95576\tsbj2\t133465\t125457\t2e-38\t83\t65\n" + ] + expectedList =[ + ("dmel_chr4", 95535, 95570, "sbj2", 125423, 125467, 7e-15, 82,97.78), + ("dmel_chr4", 95545, 95576, "sbj2", 133465, 125457, 2e-38, 83,65) + ] + self._templateTest(lines, expectedList) + + def _templateTest(self, datas2TestList, expectedList ): + pathFileName = "dummyPathFile_%s" % ( self._uniqId ) + _MockPathFile(pathFileName, datas2TestList) + + + self._db.createTable( self._table, "path", pathFileName ) + + _MockMapFile(self._mapFileName) + chunkConnector = PathChunkConnector(self._mapFileName, self._db, self._table, 0) + chunkConnector.run() + + sql_cmd = 'select * from %s' % (self._table) + self._db.execute(sql_cmd) + res = self._db.fetchall() + + for i in xrange(len(expectedList)): + resultTuple = res[i] + expectedTuple = expectedList[i] + self._assertExpectedTupleEqualsObsTuple(expectedTuple, resultTuple) + + self._db.dropTable( self._table ) + os.remove(pathFileName) + os.remove(self._mapFileName) + + def _assertExpectedTupleEqualsObsTuple(self, expectedTuple, resultTuple): + self.assertEquals(expectedTuple[0], resultTuple[1]) + self.assertEquals(expectedTuple[1], resultTuple[2]) + self.assertEquals(expectedTuple[2], resultTuple[3]) + self.assertEquals(expectedTuple[3], resultTuple[4]) + self.assertEquals(expectedTuple[4], resultTuple[5]) + self.assertEquals(expectedTuple[5], resultTuple[6]) + self.assertEquals(expectedTuple[6], resultTuple[7]) + self.assertEquals(expectedTuple[7], resultTuple[8]) + self.assertEquals(expectedTuple[8], resultTuple[9]) + + +class _MockPathFile: + + def __init__(self, fileName, lines): + path = open(fileName, "w"); + for line in lines: + path.write(line) + path.close + + +class _MockMapFile: + + def __init__ (self, fileName): + map = open(fileName, "w") + line1 = "chunk1" + '\t' + "dmel_chr4" + '\t' + "1" + '\t'+ "100000" + "\n" + line2 = "chunk2" + '\t' + "dmel_chr4" + '\t' + "90001" + '\t'+ "190000" + "\n" + line3 = "chunk3" + '\t' + "dmel_chr4" + '\t' + "180001" + '\t' + "280000" + "\n" + map.write(line1) + map.write(line2) + map.write(line3) + map.close + + +test_suite = unittest.TestSuite() +test_suite.addTest( unittest.makeSuite( Test_PathChunkConnector ) ) +if __name__ == "__main__": + unittest.TextTestRunner(verbosity=2).run( test_suite )