18
|
1 import os
|
|
2 import sys
|
|
3 import unittest
|
|
4 from sets import Set
|
|
5 import ConfigParser
|
|
6 from pyRepet.sql.RepetJobMySQL import *
|
|
7 from pyRepet.util.file.FileUtils import *
|
|
8 import commons.pyRepetUnit.hmmer.HmmpfamClusterComponent
|
|
9
|
|
10 CONFIG_FILE = "datas/configTestFunctionalHmmpfamLauncherInparallel.cfg"
|
|
11 CURRENT_DIR = os.getcwd()
|
|
12
|
|
13 class TestFunctionalHmmpfamClusterComponent(unittest.TestCase):
|
|
14
|
|
15 def setUp(self):
|
|
16 os.chdir(CURRENT_DIR)
|
|
17 self._config = ConfigParser.ConfigParser()
|
|
18 self._config.readfp( open(CONFIG_FILE) )
|
|
19 self._hmmer = commons.pyRepetUnit.hmmer.HmmpfamClusterComponent.HmmpfamClusterComponent()
|
|
20
|
|
21 def testHmmpfamHasRunInparallel (self):
|
|
22 user = self._config.get( "db_config","user" )
|
|
23 host = self._config.get( "db_config","host" )
|
|
24 passwd = self._config.get( "db_config","passwd" )
|
|
25 dbName = self._config.get( "db_config","dbname" )
|
|
26 jobDb = RepetJob(user , host , passwd , dbName)
|
|
27 self._hmmer.setJobDb(jobDb)
|
|
28 self._hmmer.loadConfig(self._config)
|
|
29 self._hmmer.run()
|
|
30 fileUtils = FileUtils()
|
|
31 dirFiles = fileUtils.listDir(self._config.get("hmmer_config", "result_dir"))
|
|
32 dirSet = Set(dirFiles)
|
|
33 alignFileSet = Set([self._config.get("hmmer_config", "result_dir") + "/" + "batch_1.fa.hmmpfamOut", self._config.get("hmmer_config", "result_dir") + "/" + "batch_2.fa.hmmpfamOut"])
|
|
34 self.assertTrue(alignFileSet.issubset(dirSet))
|
|
35 self._hmmer.clean()
|
|
36
|
|
37 if __name__ == "__main__":
|
|
38 unittest.main()
|