htsworkflow/submission/test/test_results.py

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