diff commons/pyRepetUnit/hmmer/tests/TestFunctionalHmmpfamAndParse2alignLauncherInParallel.py @ 31:0ab839023fe4

Uploaded

--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/commons/pyRepetUnit/hmmer/tests/TestFunctionalHmmpfamAndParse2alignLauncherInParallel.py	Tue Apr 30 14:33:21 2013 -0400
@@ -0,0 +1,56 @@
+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()