htsworkflow/pipelines/test/test_extract_results.py

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