diff commons/tools/tests/Test_F_LaunchMatcherInParallel.py @ 18:94ab73e8a190

Uploaded

--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/commons/tools/tests/Test_F_LaunchMatcherInParallel.py	Mon Apr 29 03:20:15 2013 -0400
@@ -0,0 +1,88 @@
+from commons.core.utils.FileUtils import FileUtils
+import shutil
+import unittest
+import os
+from commons.tools.LaunchMatcherInParallel import LaunchMatcherInParallel
+
+class Test_F_LaunchMatcherInParallel(unittest.TestCase):
+
+    CLUSTER_HOST = "compute-2-46.local"
+
+    def setUp(self):
+        self._configFileName = "TE.cfg"
+        self._writeConfig()
+
+    def tearDown(self):
+        FileUtils.removeFilesByPattern("DmelChr4*")
+        os.remove(self._configFileName)
+
+    def test_run_as_class_1_seq(self):
+        inFileName = "DmelChr4.align"
+        fastaFileName = "DmelChr4.fa"
+        inFilePath = "%s/Tools/%s" % (os.environ["REPET_DATA"], inFileName)
+        inFastaPath = "%s/Tools/%s" % (os.environ["REPET_DATA"], fastaFileName)
+        os.symlink(inFilePath, inFileName)
+        os.symlink(inFastaPath, fastaFileName)
+        expPathFileName = "%s/Tools/DmelChr4.align.match.path" % os.environ["REPET_DATA"]
+        expTabFileName = "%s/Tools/DmelChr4.align.match.tab" % os.environ["REPET_DATA"]
+        
+        obsPathFileName = "%s.match.path" % inFileName
+        obsTabFileName = "%s.match.tab" % inFileName
+        
+        iLaunchMatcherInParallel = LaunchMatcherInParallel(align=inFileName, queryFileName=fastaFileName, subjectFileName=fastaFileName, doJoin=True, evalue="1e-20", keepConflict=True, config=self._configFileName)
+        iLaunchMatcherInParallel.run()
+        
+        self.assertTrue(FileUtils.are2FilesIdentical(expPathFileName, obsPathFileName))
+        self.assertTrue(FileUtils.are2FilesIdentical(expTabFileName, obsTabFileName))
+
+    def test_run_as_class_n_seq_less_jobs_than_n(self):
+        inFileName = "DmelChr4_dummy.align"
+        fastaFileName = "DmelChr4.fa"
+        inFilePath = "%s/Tools/%s" % (os.environ["REPET_DATA"], inFileName)
+        inFastaPath = "%s/Tools/%s" % (os.environ["REPET_DATA"], fastaFileName)
+        os.symlink(inFilePath, inFileName)
+        os.symlink(inFastaPath, fastaFileName)
+        expPathFileName = "%s/Tools/DmelChr4_dummy.align.match.path" % os.environ["REPET_DATA"]
+        expTabFileName = "%s/Tools/DmelChr4_dummy.align.match.tab" % os.environ["REPET_DATA"]
+        
+        obsPathFileName = "%s.match.path" % inFileName
+        obsTabFileName = "%s.match.tab" % inFileName
+        
+        iLaunchMatcherInParallel = LaunchMatcherInParallel(align=inFileName, queryFileName=fastaFileName, subjectFileName=fastaFileName, doJoin=True, evalue="1e-20", keepConflict=True, config=self._configFileName, doClean=True)
+        iLaunchMatcherInParallel.run()
+        
+        self.assertTrue(FileUtils.are2FilesIdentical(expPathFileName, obsPathFileName))
+        self.assertTrue(FileUtils.are2FilesIdentical(expTabFileName, obsTabFileName))
+        
+    def test_run_as_class_n_seq_less_jobs_than_n_not_merged_results(self):
+        inFileName = "DmelChr4_dummy.align"
+        fastaFileName = "DmelChr4.fa"
+        inFilePath = "%s/Tools/%s" % (os.environ["REPET_DATA"], inFileName)
+        inFastaPath = "%s/Tools/%s" % (os.environ["REPET_DATA"], fastaFileName)
+        os.symlink(inFilePath, inFileName)
+        os.symlink(inFastaPath, fastaFileName)
+        
+        iLaunchMatcherInParallel = LaunchMatcherInParallel(align=inFileName, queryFileName=fastaFileName, subjectFileName=fastaFileName, doJoin=True, evalue="1e-20", keepConflict=True, config=self._configFileName, maxFileSize=100, mergeResults=False)
+        iLaunchMatcherInParallel.run()
+        
+        baseFileName = os.path.splitext(inFileName)[0]
+        lExpMatchFileNames = ["%s_1.align.match.path" % baseFileName, "%s_2.align.match.path" % baseFileName, "%s_3.align.match.path" % baseFileName]
+        lExpTabFileNames = ["%s_1.align.match.tab" % baseFileName, "%s_2.align.match.tab" % baseFileName, "%s_3.align.match.tab" % baseFileName]
+        for i in range(0,3):
+            self.assertTrue(FileUtils.isRessourceExists("tmpMatcher/%s" % lExpMatchFileNames[i]))
+            self.assertTrue(FileUtils.isRessourceExists("tmpMatcher/%s" % lExpTabFileNames[i]))
+        shutil.rmtree("tmpMatcher")
+
+    def _writeConfig(self):
+        with open(self._configFileName, "w") as fh:
+            fh.write("[jobs]\n")
+            if os.getenv("HOSTNAME") == self.CLUSTER_HOST:
+                fh.write("resources: test\n")
+            else:
+                fh.write("resources:\n")
+            fh.write("tmpDir:\n")
+            fh.write("copy: no\n") 
+            fh.write("clean: yes\n")
+
+if __name__ == "__main__":
+    unittest.main()
\ No newline at end of file