Mercurial > repos > yufei-luo > s_mart
comparison commons/pyRepetUnit/hmmer/tests/TestFunctionalHmmpfamAndParse2alignLauncherInParallel.py @ 31:0ab839023fe4
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 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() |