18
|
1 import os
|
|
2 import sys
|
|
3 import unittest
|
|
4 from sets import Set
|
|
5 import ConfigParser
|
|
6 from pyRepet.launcher.Launcher import HmmpfamAndParse2alignLauncher
|
|
7 from pyRepet.sql.RepetJobMySQL import *
|
|
8 from pyRepet.util.file.FileUtils import *
|
|
9
|
|
10 CONFIG_FILE = "datas/configTestFunctionalHmmpfamLauncherInparallel.cfg"
|
|
11 CURRENT_DIR = os.getcwd()
|
|
12
|
|
13 class TestFunctionalHmmpfamLauncherAndParse2alignInparallel(unittest.TestCase):
|
|
14
|
|
15 def setUp(self):
|
|
16 self._config = ConfigParser.ConfigParser()
|
|
17 self._config.readfp( open(CONFIG_FILE) )
|
|
18 os.chdir(CURRENT_DIR)
|
|
19
|
|
20 def testHmmpfamAndParse2alignHasRunInparallel (self):
|
|
21 user = self._config.get( "db_config","user" )
|
|
22 host = self._config.get( "db_config","host" )
|
|
23 passwd = self._config.get( "db_config","passwd" )
|
|
24 dbName = self._config.get( "db_config","dbname" )
|
|
25 jobDb = RepetJob(user , host , passwd , dbName)
|
|
26 params = {"param":"--informat FASTA -E 10",
|
|
27 "profilDB": self._config.get("profil_search", "ProfilDatabank"),
|
|
28 "outputDir": self._config.get("hmmer_config", "result_dir"),
|
|
29 "query": self._config.get("hmmer_config", "query_dir"),
|
|
30 "job_table":"jobs",
|
|
31 "queue" : self._config.get("sge_config", "queue"),
|
|
32 "groupid" : self._config.get("sge_config", "groupid"),
|
|
33 "tmpDir": self._config.get("hmmer_config", "tmp_dir"),
|
|
34 "scriptToLaunch": self._config.get("parse_config", "scriptToLaunch"),
|
|
35 "cDir" : "."}
|
|
36 self.hmmpfamAndParse2align = HmmpfamAndParse2alignLauncher(jobDb, params)
|
|
37 self.hmmpfamAndParse2align.run()
|
|
38 fileUtils = FileUtils()
|
|
39 dirFiles = fileUtils.listDir(self._config.get("hmmer_config", "result_dir"))
|
|
40 dirSet = Set(dirFiles)
|
|
41 alignFileSet = Set([self._config.get("hmmer_config", "result_dir") + "/" + "batch_1.fa.hmmpfamOut.align", self._config.get("hmmer_config", "result_dir") + "/" + "batch_2.fa.hmmpfamOut.align"])
|
|
42 self.assertTrue( alignFileSet.issubset(dirSet))
|
|
43 self._clean()
|
|
44
|
|
45 def _clean(self):
|
|
46
|
|
47 dirToClean = self._config.get( "hmmer_config", "tmp_dir" )
|
|
48 os.system("rm " + dirToClean + "/*")
|
|
49 dirToClean = self._config.get( "hmmer_config", "result_dir" )
|
|
50 os.system("rm " + dirToClean + "/*")
|
|
51 os.system("rm HmmpfamJob*.e*")
|
|
52 os.system("rm HmmpfamJob*.o*")
|
|
53
|
|
54
|
|
55 if __name__ == "__main__":
|
|
56 unittest.main()
|