view commons/pyRepetUnit/hmmer/tests/TestFunctionalHmmpfamClusterComponent.py @ 31:0ab839023fe4

Uploaded

import os
import sys
import unittest
from sets import Set
import ConfigParser 
from pyRepet.sql.RepetJobMySQL import *
from pyRepet.util.file.FileUtils import *
import commons.pyRepetUnit.hmmer.HmmpfamClusterComponent

CONFIG_FILE = "datas/configTestFunctionalHmmpfamLauncherInparallel.cfg"
CURRENT_DIR = os.getcwd()

class TestFunctionalHmmpfamClusterComponent(unittest.TestCase):
    
    def setUp(self):
        os.chdir(CURRENT_DIR)
        self._config = ConfigParser.ConfigParser()
        self._config.readfp( open(CONFIG_FILE) )
        self._hmmer = commons.pyRepetUnit.hmmer.HmmpfamClusterComponent.HmmpfamClusterComponent()

    def testHmmpfamHasRunInparallel (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)
        self._hmmer.setJobDb(jobDb)
        self._hmmer.loadConfig(self._config)
        self._hmmer.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", self._config.get("hmmer_config", "result_dir") + "/" + "batch_2.fa.hmmpfamOut"])
        self.assertTrue(alignFileSet.issubset(dirSet))
        self._hmmer.clean()
        
if __name__ == "__main__":
        unittest.main()