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

Uploaded

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()