htsworkflow/submission/test/test_results.py

   1 #!/usr/bin/env python
   2
   3 import copy
   4 import os
   5 from pprint import pprint
   6 import shutil
   7 import tempfile
   8
   9 from unittest2 import TestCase, defaultTestLoader
  10
  11 from htsworkflow.submission.results import ResultMap
  12
  13 S1_NAME = '1000-sample'
  14 S2_NAME = '2000-sample'
  15
  16 S1_FILES = [
  17     os.path.join(S1_NAME, 'file1_l8_r1.fastq'),
  18     os.path.join(S1_NAME, 'file1_l8_r2.fastq'),
  19 ]
  20
  21 S2_FILES = [
  22     os.path.join(S2_NAME, 'file1.bam'),
  23     os.path.join(S2_NAME, 'file1_l5.fastq'),
  24 ]
  25
  26 def generate_sample_results_tree(obj):
  27     obj.tempdir = tempfile.mkdtemp(prefix="results_test")
  28     obj.sourcedir = os.path.join(obj.tempdir, 'source')
  29     obj.resultdir = os.path.join(obj.tempdir, 'results')
  30
  31     for d in [obj.sourcedir,
  32               os.path.join(obj.sourcedir, S1_NAME),
  33               os.path.join(obj.sourcedir, S2_NAME),
  34               obj.resultdir]:
  35         os.mkdir(os.path.join(obj.tempdir, d))
  36
  37     tomake = []
  38     tomake.extend(S1_FILES)
  39     tomake.extend(S2_FILES)
  40     for f in tomake:
  41         stream = open(os.path.join(obj.sourcedir, f), 'w')
  42         stream.write(f)
  43         stream.close()
  44
  45 class TestResultMap(TestCase):
  46     def setUp(self):
  47         generate_sample_results_tree(self)
  48
  49     def tearDown(self):
  50         shutil.rmtree(self.tempdir)
  51
  52     def test_dict_like(self):
  53         """Make sure the result map works like an ordered dictionary
  54         """
  55         results = ResultMap()
  56         results['1000'] = 'dir1000'
  57         results['2000'] = 'dir2000'
  58         results['1500'] = 'dir1500'
  59
  60         self.failUnlessEqual(results.keys(), ['1000', '2000', '1500'])
  61         self.failUnlessEqual(list(results.values()),
  62                              ['dir1000', 'dir2000', 'dir1500'])
  63         self.failUnlessEqual(list(results.items()),
  64                              [('1000', 'dir1000'),
  65                               ('2000', 'dir2000'),
  66                               ('1500', 'dir1500')])
  67
  68         self.failUnlessEqual(results['1000'], 'dir1000')
  69         self.failUnlessEqual(results['1500'], 'dir1500')
  70         self.failUnlessEqual(results['2000'], 'dir2000')
  71
  72         self.assertTrue(u'2000' in results)
  73         self.assertTrue('2000' in results)
  74         self.assertFalse(u'77777' in results)
  75         self.assertFalse('77777' in results)
  76
  77     def test_make_from_absolute(self):
  78         """Test that make from works if ResultMap has absolute paths
  79         """
  80         results = ResultMap()
  81         sample1_dir = os.path.join(self.resultdir, S1_NAME)
  82         sample2_dir = os.path.join(self.resultdir, S2_NAME)
  83         results['1000'] =  sample1_dir
  84         results['2000'] =  sample2_dir
  85
  86         results.make_tree_from(self.sourcedir, self.resultdir)
  87         self.failUnless(os.path.isdir(sample1_dir))
  88         self.failUnless(os.path.isdir(sample2_dir))
  89
  90         for f in S1_FILES + S2_FILES:
  91             self.failUnless(
  92                 os.path.islink(
  93                     os.path.join(self.resultdir, f)))
  94
  95     def test_make_from_filename(self):
  96         """Test that make from works if ResultMap has no path
  97         """
  98         results = ResultMap()
  99         results['1000'] =  S1_NAME
 100         results['2000'] =  S2_NAME
 101
 102         results.make_tree_from(self.sourcedir, self.resultdir)
 103         sample1_dir = os.path.join(self.resultdir, S1_NAME)
 104         sample2_dir = os.path.join(self.resultdir, S2_NAME)
 105         self.failUnless(os.path.isdir(sample1_dir))
 106         self.failUnless(os.path.isdir(sample2_dir))
 107
 108         for f in S1_FILES + S2_FILES:
 109             self.failUnless(
 110                 os.path.islink(
 111                     os.path.join(self.resultdir, f)))
 112
 113
 114 def suite():
 115     suite = defaultTestLoader.loadTestsFromTestCase(TestResultMap)
 116     return suite
 117
 118 if __name__ == "__main__":
 119     import logging
 120     logging.basicConfig(level=logging.DEBUG)
 121     from unittest2 import main
 122     main(defaultTest='suite')