2 from StringIO import StringIO
5 from unittest2 import TestCase, TestSuite, defaultTestLoader
7 from htsworkflow.submission import daf, results
8 from htsworkflow.util.rdfhelp import \
13 load_string_into_model, \
19 from htsworkflow.submission.submission import list_submissions, Submission
20 from htsworkflow.submission.results import ResultMap
21 from submission_test_common import *
25 #logging.basicConfig(level=logging.DEBUG)
27 def generate_sample_results_tree(obj):
28 obj.tempdir = tempfile.mkdtemp(prefix="submission_test")
29 obj.sourcedir = os.path.join(obj.tempdir, 'source')
30 obj.resultdir = os.path.join(obj.tempdir, 'results')
32 for d in [os.path.join(obj.tempdir, S1_NAME),
33 os.path.join(obj.tempdir, S2_NAME),
35 os.mkdir(os.path.join(obj.tempdir, d))
38 tomake.extend(S1_FILES)
39 tomake.extend(S2_FILES)
41 stream = open(os.path.join(obj.tempdir, f), 'w')
45 class TestSubmissionModule(TestCase):
46 def test_empty_list_submission(self):
48 self.assertEqual(len(list(list_submissions(model))), 0)
50 def test_one_submission(self):
52 load_string_into_model(model, "turtle",
54 @prefix subns: <http://jumpgate.caltech.edu/wiki/UcscSubmissionOntology#> .
55 @prefix test: <http://jumpgate.caltech.edu/wiki/SubmissionsLog/test#> .
57 <http://jumpgate.caltech.edu/wiki/SubmissionsLog/test#>
58 subns:has_submission test:lib1 ;
59 subns:has_submission test:lib2.
61 submissions = list(list_submissions(model))
62 self.assertEqual(len(submissions), 1)
63 self.assertEqual(submissions[0], "test")
65 def test_two_submission(self):
67 load_string_into_model(model, "turtle",
69 @prefix subns: <http://jumpgate.caltech.edu/wiki/UcscSubmissionOntology#> .
70 @prefix test: <http://jumpgate.caltech.edu/wiki/SubmissionsLog/test#> .
72 <http://jumpgate.caltech.edu/wiki/SubmissionsLog/test1#>
73 subns:has_submission test:lib1 .
74 <http://jumpgate.caltech.edu/wiki/SubmissionsLog/test2#>
75 subns:has_submission test:lib2 .
77 submissions = list(list_submissions(model))
78 self.assertEqual(len(submissions), 2)
79 truth = set(["test1", "test2"])
81 for name in submissions:
83 self.assertEqual(testset, truth)
85 class TestSubmission(TestCase):
87 generate_sample_results_tree(self)
88 self.model = get_model()
91 shutil.rmtree(self.tempdir)
93 def test_create_submission(self):
95 s = Submission('foo', self.model, 'http://localhost')
96 self.assertEqual(str(s.submissionSet),
97 "http://jumpgate.caltech.edu/wiki/SubmissionsLog/foo")
98 self.assertEqual(str(s.submissionSetNS['']),
99 str(RDF.NS(str(s.submissionSet) + '#')['']))
100 self.assertEqual(str(s.libraryNS['']),
101 str(RDF.NS('http://localhost/library/')['']))
103 def test_scan_submission_dirs(self):
104 turtle = get_turtle_header() + """
105 @prefix thisView: <http://jumpgate.caltech.edu/wiki/SubmissionsLog/test/view/> .
106 thisView:Fastq ucscDaf:filename_re ".*[^12]\\.fastq\\.bz2$" ;
108 geoSoft:fileTypeLabel "fastq" ;
109 ucscDaf:output_type "read" .
110 thisView:FastqRead1 ucscDaf:filename_re ".*r1\\.fastq\\.bz2$" ;
112 geoSoft:fileTypeLabel "fastq" ;
113 ucscDaf:output_type "read1" .
114 thisView:FastqRead2 ucscDaf:filename_re ".*r2\\.fastq\\.bz2$" ;
116 geoSoft:fileTypeLabel "fastq" ;
117 ucscDaf:output_type "read2" .
118 thisView:alignments ucscDaf:filename_re ".*\\.bam$" ;
119 a geoSoft:supplemental ;
120 geoSoft:fileTypeLabel "bam" ;
121 ucscDaf:output_type "alignments" .
126 print os.listdir(self.tempdir)
127 map['1000'] = os.path.join(self.tempdir, S1_NAME)
128 map['2000'] = os.path.join(self.tempdir, S2_NAME)
130 s = Submission('foo', self.model, 'http://localhost')
131 mock = MockAddDetails(self.model, turtle)
132 mock.add_turtle(S1_TURTLE)
133 mock.add_turtle(S2_TURTLE)
134 s._add_library_details_to_model = mock
135 s.scan_submission_dirs(map)
137 nodes = list(s.analysis_nodes(map))
138 self.assertEqual(len(nodes), 2)
140 'http://jumpgate.caltech.edu/wiki/SubmissionsLog/foo#1000-sample',
141 'http://jumpgate.caltech.edu/wiki/SubmissionsLog/foo#2000-sample',
143 got = set((str(nodes[0]), str(nodes[1])))
144 self.assertEqual(expected, got)
146 def test_find_best_match(self):
147 turtle = get_turtle_header() + """
148 @prefix thisView: <http://jumpgate.caltech.edu/wiki/SubmissionsLog/test/view/> .
149 thisView:Fastq ucscDaf:filename_re ".*[^12]\\.fastq\\.bz2$" ;
151 geoSoft:fileTypeLabel "fastq" ;
152 ucscDaf:output_type "read" .
153 thisView:FastqRead1 ucscDaf:filename_re ".*r1\\.fastq\\.bz2$" ;
155 geoSoft:fileTypeLabel "fastq" ;
156 ucscDaf:output_type "read1" .
157 thisView:FastqRead2 ucscDaf:filename_re ".*r2\\.fastq\\.bz2$" ;
159 geoSoft:fileTypeLabel "fastq" ;
160 ucscDaf:output_type "read2" .
161 thisView:alignments ucscDaf:filename_re ".*\\.bam$" ;
162 a geoSoft:supplemental ;
163 geoSoft:fileTypeLabel "bam" ;
164 ucscDaf:output_type "alignments" .
167 load_string_into_model(self.model, 'turtle', turtle)
168 s = Submission('foo', self.model, 'http://localhost')
169 q = RDF.Statement(None, dafTermOntology['filename_re'], None)
170 view_map = s._get_filename_view_map()
171 self.assertEqual(len(view_map), 4)
173 fastq = s.find_best_match("asdf.fastq.bz2")
176 "http://jumpgate.caltech.edu/wiki/SubmissionsLog/test/view/Fastq")
178 fastq = s.find_best_match("asdf.r2.fastq.bz2")
181 "http://jumpgate.caltech.edu/wiki/SubmissionsLog/test/view/FastqRead2")
186 defaultTestLoader.loadTestsFromTestCase(TestSubmissionModule))
188 defaultTestLoader.loadTestsFromTestCase(TestSubmission))
191 if __name__ == "__main__":
192 from unittest2 import main
193 main(defaultTest='suite')