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 import unittest
   9
  10 from htsworkflow.submission.results import ResultMap
  11
  12 S1_NAME = '1000-sample'
  13 S2_NAME = '2000-sample'
  14
  15 S1_FILES = [
  16     os.path.join(S1_NAME, 'file1_l8_r1.fastq'),
  17     os.path.join(S1_NAME, 'file1_l8_r2.fastq'),
  18 ]
  19
  20 S2_FILES = [
  21     os.path.join(S2_NAME, 'file1.bam'),
  22     os.path.join(S2_NAME, 'file1_l5.fastq'),
  23 ]
  24
  25 def generate_sample_results_tree(obj):
  26     obj.tempdir = tempfile.mkdtemp(prefix="results_test")
  27     obj.sourcedir = os.path.join(obj.tempdir, 'source')
  28     obj.resultdir = os.path.join(obj.tempdir, 'results')
  29
  30     for d in [obj.sourcedir,
  31               os.path.join(obj.sourcedir, S1_NAME),
  32               os.path.join(obj.sourcedir, S2_NAME),
  33               obj.resultdir]:
  34         os.mkdir(os.path.join(obj.tempdir, d))
  35
  36     tomake = []
  37     tomake.extend(S1_FILES)
  38     tomake.extend(S2_FILES)
  39     for f in tomake:
  40         stream = open(os.path.join(obj.sourcedir, f), 'w')
  41         stream.write(f)
  42         stream.close()
  43
  44 class TestResultMap(unittest.TestCase):
  45     def setUp(self):
  46         generate_sample_results_tree(self)
  47
  48     def tearDown(self):
  49         shutil.rmtree(self.tempdir)
  50
  51
  52     def test_dict_like(self):
  53         """Make sure the result map works like an ordered dictionary
  54         """
  55         results = ResultMap()
  56         results.add_result('1000', 'dir1000')
  57         results.add_result('2000', 'dir2000')
  58         results.add_result('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     def test_make_from(self):
  73         results = ResultMap()
  74         results.add_result('1000', S1_NAME)
  75         results.add_result('2000', S2_NAME)
  76
  77         results.make_tree_from(self.sourcedir, self.resultdir)
  78
  79         sample1_dir = os.path.join(self.resultdir, S1_NAME)
  80         sample2_dir = os.path.join(self.resultdir, S2_NAME)
  81         self.failUnless(os.path.isdir(sample1_dir))
  82         self.failUnless(os.path.isdir(sample2_dir))
  83
  84         for f in S1_FILES + S2_FILES:
  85             self.failUnless(
  86                 os.path.islink(
  87                     os.path.join(self.resultdir, f)))
  88
  89
  90
  91
  92 def suite():
  93     suite = unittest.makeSuite(TestResultMap, 'test')
  94     return suite
  95
  96 if __name__ == "__main__":
  97     unittest.main(defaultTest='suite')
  98