Mercurial > repos > yufei-luo > s_mart
view commons/pyRepetUnit/hmmer/tests/TestFunctionalHmmpfamAndParse2alignLauncherInParallel.py @ 18:94ab73e8a190
Uploaded
author | m-zytnicki |
---|---|
date | Mon, 29 Apr 2013 03:20:15 -0400 |
parents | |
children |
line wrap: on
line source
import os import sys import unittest from sets import Set import ConfigParser from pyRepet.launcher.Launcher import HmmpfamAndParse2alignLauncher from pyRepet.sql.RepetJobMySQL import * from pyRepet.util.file.FileUtils import * CONFIG_FILE = "datas/configTestFunctionalHmmpfamLauncherInparallel.cfg" CURRENT_DIR = os.getcwd() class TestFunctionalHmmpfamLauncherAndParse2alignInparallel(unittest.TestCase): def setUp(self): self._config = ConfigParser.ConfigParser() self._config.readfp( open(CONFIG_FILE) ) os.chdir(CURRENT_DIR) def testHmmpfamAndParse2alignHasRunInparallel (self): user = self._config.get( "db_config","user" ) host = self._config.get( "db_config","host" ) passwd = self._config.get( "db_config","passwd" ) dbName = self._config.get( "db_config","dbname" ) jobDb = RepetJob(user , host , passwd , dbName) params = {"param":"--informat FASTA -E 10", "profilDB": self._config.get("profil_search", "ProfilDatabank"), "outputDir": self._config.get("hmmer_config", "result_dir"), "query": self._config.get("hmmer_config", "query_dir"), "job_table":"jobs", "queue" : self._config.get("sge_config", "queue"), "groupid" : self._config.get("sge_config", "groupid"), "tmpDir": self._config.get("hmmer_config", "tmp_dir"), "scriptToLaunch": self._config.get("parse_config", "scriptToLaunch"), "cDir" : "."} self.hmmpfamAndParse2align = HmmpfamAndParse2alignLauncher(jobDb, params) self.hmmpfamAndParse2align.run() fileUtils = FileUtils() dirFiles = fileUtils.listDir(self._config.get("hmmer_config", "result_dir")) dirSet = Set(dirFiles) 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"]) self.assertTrue( alignFileSet.issubset(dirSet)) self._clean() def _clean(self): dirToClean = self._config.get( "hmmer_config", "tmp_dir" ) os.system("rm " + dirToClean + "/*") dirToClean = self._config.get( "hmmer_config", "result_dir" ) os.system("rm " + dirToClean + "/*") os.system("rm HmmpfamJob*.e*") os.system("rm HmmpfamJob*.o*") if __name__ == "__main__": unittest.main()