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     def test_dict_like(self):
  52         """Make sure the result map works like an ordered dictionary
  53         """
  54         results = ResultMap()
  55         results['1000'] = 'dir1000'
  56         results['2000'] = 'dir2000'
  57         results['1500'] = 'dir1500'
  58
  59         self.failUnlessEqual(results.keys(), ['1000', '2000', '1500'])
  60         self.failUnlessEqual(list(results.values()),
  61                              ['dir1000', 'dir2000', 'dir1500'])
  62         self.failUnlessEqual(list(results.items()),
  63                              [('1000', 'dir1000'),
  64                               ('2000', 'dir2000'),
  65                               ('1500', 'dir1500')])
  66
  67         self.failUnlessEqual(results['1000'], 'dir1000')
  68         self.failUnlessEqual(results['1500'], 'dir1500')
  69         self.failUnlessEqual(results['2000'], 'dir2000')
  70
  71         self.assertTrue(u'2000' in results)
  72         self.assertTrue('2000' in results)
  73         self.assertFalse(u'77777' in results)
  74         self.assertFalse('77777' in results)
  75
  76     def test_make_from(self):
  77         results = ResultMap()
  78         results['1000'] =  S1_NAME
  79         results['2000'] =  S2_NAME
  80
  81         results.make_tree_from(self.sourcedir, self.resultdir)
  82
  83         sample1_dir = os.path.join(self.resultdir, S1_NAME)
  84         sample2_dir = os.path.join(self.resultdir, S2_NAME)
  85         self.failUnless(os.path.isdir(sample1_dir))
  86         self.failUnless(os.path.isdir(sample2_dir))
  87
  88         for f in S1_FILES + S2_FILES:
  89             self.failUnless(
  90                 os.path.islink(
  91                     os.path.join(self.resultdir, f)))
  92
  93
  94
  95
  96 def suite():
  97     suite = unittest.makeSuite(TestResultMap, 'test')
  98     return suite
  99
 100 if __name__ == "__main__":
 101     unittest.main(defaultTest='suite')
 102