5 from pprint import pprint
10 from htsworkflow.submission.results import ResultMap
12 S1_NAME = '1000-sample'
13 S2_NAME = '2000-sample'
16 os.path.join(S1_NAME, 'file1_l8_r1.fastq'),
17 os.path.join(S1_NAME, 'file1_l8_r2.fastq'),
21 os.path.join(S2_NAME, 'file1.bam'),
22 os.path.join(S2_NAME, 'file1_l5.fastq'),
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')
30 for d in [obj.sourcedir,
31 os.path.join(obj.sourcedir, S1_NAME),
32 os.path.join(obj.sourcedir, S2_NAME),
34 os.mkdir(os.path.join(obj.tempdir, d))
37 tomake.extend(S1_FILES)
38 tomake.extend(S2_FILES)
40 stream = open(os.path.join(obj.sourcedir, f), 'w')
44 class TestResultMap(unittest.TestCase):
46 generate_sample_results_tree(self)
49 shutil.rmtree(self.tempdir)
52 def test_dict_like(self):
53 """Make sure the result map works like an ordered dictionary
56 results.add_result('1000', 'dir1000')
57 results.add_result('2000', 'dir2000')
58 results.add_result('1500', 'dir1500')
60 self.failUnlessEqual(results.keys(), ['1000', '2000', '1500'])
61 self.failUnlessEqual(list(results.values()),
62 ['dir1000', 'dir2000', 'dir1500'])
63 self.failUnlessEqual(list(results.items()),
68 self.failUnlessEqual(results['1000'], 'dir1000')
69 self.failUnlessEqual(results['1500'], 'dir1500')
70 self.failUnlessEqual(results['2000'], 'dir2000')
72 def test_make_from(self):
74 results.add_result('1000', S1_NAME)
75 results.add_result('2000', S2_NAME)
77 results.make_tree_from(self.sourcedir, self.resultdir)
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))
84 for f in S1_FILES + S2_FILES:
87 os.path.join(self.resultdir, f)))
93 suite = unittest.makeSuite(TestResultMap, 'test')
96 if __name__ == "__main__":
97 unittest.main(defaultTest='suite')