htsworkflow/pipelines/test/test_extract_results.py

   1 #!/usr/bin/env python
   2
   3 from datetime import datetime, date
   4 import logging
   5 import os
   6 import tempfile
   7 import shutil
   8 import sys
   9 from unittest import TestCase
  10
  11 from htsworkflow.pipelines import eland
  12 from htsworkflow.pipelines import ipar
  13 from htsworkflow.pipelines import bustard
  14 from htsworkflow.pipelines import gerald
  15 from htsworkflow.pipelines import runfolder
  16 from htsworkflow.pipelines import ElementTree
  17
  18 from htsworkflow.pipelines.test.simulate_runfolder import *
  19
  20 logging.basicConfig(level=logging.ERROR)
  21
  22 def make_runfolder(obj=None):
  23     """
  24     Make a fake runfolder, attach all the directories to obj if defined
  25     """
  26     # make a fake runfolder directory
  27     temp_dir = tempfile.mkdtemp(prefix='tmp_runfolder_')
  28
  29     runfolder_dir = os.path.join(temp_dir,
  30                                  '090608_HWI-EAS229_0117_4286GAAXX')
  31     os.mkdir(runfolder_dir)
  32
  33     data_dir = os.path.join(runfolder_dir, 'Data')
  34     os.mkdir(data_dir)
  35
  36     intensities_dir = make_rta_intensities_1460(data_dir)
  37
  38     bustard_dir = os.path.join(intensities_dir, 'Bustard1.4.0_16-06-2009_diane')
  39     os.mkdir(bustard_dir)
  40     make_phasing_params(bustard_dir)
  41     make_bustard_config132(bustard_dir)
  42     score_dir = make_scores(bustard_dir)
  43     make_qseqs(bustard_dir)
  44
  45     gerald_dir = os.path.join(bustard_dir,
  46                               'GERALD_16-06-2009_diane')
  47     os.mkdir(gerald_dir)
  48     make_gerald_config_100(gerald_dir)
  49     make_summary_ipar130_htm(gerald_dir)
  50     make_eland_multi(gerald_dir, lane_list=[1,2,3,4,5,6,])
  51     make_scarf(gerald_dir, lane_list=[7,])
  52     make_fastq(gerald_dir, lane_list=[8,])
  53
  54     if obj is not None:
  55         obj.temp_dir = temp_dir
  56         obj.runfolder_dir = runfolder_dir
  57         obj.data_dir = data_dir
  58         obj.image_analysis_dir = intensities_dir
  59         obj.bustard_dir = bustard_dir
  60         obj.gerald_dir = gerald_dir
  61
  62
  63 class RunfolderExtractTests(TestCase):
  64     """
  65     Test the extract result code.
  66     """
  67     def setUp(self):
  68         # attaches all the directories to the object passed in
  69         make_runfolder(self)
  70
  71     def tearDown(self):
  72         shutil.rmtree(self.temp_dir)
  73
  74     def test_extract_results(self):
  75         runs = runfolder.get_runs(self.runfolder_dir)
  76         self.failUnlessEqual(len(runs), 1)
  77         runfolder.extract_results(runs, self.temp_dir, site='asite')
  78         archive = os.listdir(os.path.join(self.temp_dir, '4286GAAXX', 'C1-38'))
  79         self.failUnlessEqual(len(archive), 34)
  80         self.failUnless('asite_090608_HWI-EAS229_0117_4286GAAXX_l6_r1.tar.bz2' in archive)
  81
  82
  83 def suite():
  84     from unittest import TestSuite, defaultTestLoader
  85     suite = TestSuite()
  86     suite.addTests(defaultTestLoader.loadTestsFromTestCase(RunfolderExtractTests))
  87     return suite
  88
  89
  90 if __name__ == "__main__":
  91     from unittest import main
  92     main(defaultTest="suite")
  93