3 from StringIO import StringIO
6 from unittest2 import TestCase, TestSuite, defaultTestLoader
8 from htsworkflow.submission import daf, results
9 from htsworkflow.util.rdfhelp import \
14 load_string_into_model, \
20 from htsworkflow.submission.submission import list_submissions, Submission
21 from htsworkflow.submission.results import ResultMap
22 from submission_test_common import *
26 #logging.basicConfig(level=logging.DEBUG)
28 class TestSubmissionModule(TestCase):
29 def test_empty_list_submission(self):
31 self.assertEqual(len(list(list_submissions(model))), 0)
33 def test_one_submission(self):
35 load_string_into_model(model, "turtle",
37 @prefix subns: <http://jumpgate.caltech.edu/wiki/UcscSubmissionOntology#> .
38 @prefix test: <http://jumpgate.caltech.edu/wiki/SubmissionsLog/test#> .
40 <http://jumpgate.caltech.edu/wiki/SubmissionsLog/test#>
41 subns:has_submission test:lib1 ;
42 subns:has_submission test:lib2.
44 submissions = list(list_submissions(model))
45 self.assertEqual(len(submissions), 1)
46 self.assertEqual(submissions[0], "test")
48 def test_two_submission(self):
50 load_string_into_model(model, "turtle",
52 @prefix subns: <http://jumpgate.caltech.edu/wiki/UcscSubmissionOntology#> .
53 @prefix test: <http://jumpgate.caltech.edu/wiki/SubmissionsLog/test#> .
55 <http://jumpgate.caltech.edu/wiki/SubmissionsLog/test1#>
56 subns:has_submission test:lib1 .
57 <http://jumpgate.caltech.edu/wiki/SubmissionsLog/test2#>
58 subns:has_submission test:lib2 .
60 submissions = list(list_submissions(model))
61 self.assertEqual(len(submissions), 2)
62 truth = set(["test1", "test2"])
64 for name in submissions:
66 self.assertEqual(testset, truth)
68 class TestSubmission(TestCase):
70 generate_sample_results_tree(self, 'submission_test')
71 self.model = get_model()
74 shutil.rmtree(self.tempdir)
76 def test_create_submission(self):
78 s = Submission('foo', self.model, 'http://localhost')
79 self.assertEqual(str(s.submissionSet),
80 "http://jumpgate.caltech.edu/wiki/SubmissionsLog/foo")
81 self.assertEqual(str(s.submissionSetNS['']),
82 str(RDF.NS(str(s.submissionSet) + '#')['']))
83 self.assertEqual(str(s.libraryNS['']),
84 str(RDF.NS('http://localhost/library/')['']))
86 def test_scan_submission_dirs(self):
87 turtle = get_turtle_header() + """
88 @prefix thisView: <http://jumpgate.caltech.edu/wiki/SubmissionsLog/test/view/> .
89 thisView:Fastq ucscDaf:filename_re ".*[^12]\\.fastq\\.bz2$" ;
91 geoSoft:fileTypeLabel "fastq" ;
92 ucscDaf:output_type "read" .
93 thisView:FastqRead1 ucscDaf:filename_re ".*r1\\.fastq\\.bz2$" ;
95 geoSoft:fileTypeLabel "fastq" ;
96 ucscDaf:output_type "read1" .
97 thisView:FastqRead2 ucscDaf:filename_re ".*r2\\.fastq\\.bz2$" ;
99 geoSoft:fileTypeLabel "fastq" ;
100 ucscDaf:output_type "read2" .
101 thisView:alignments ucscDaf:filename_re ".*\\.bam$" ;
102 a geoSoft:supplemental ;
103 geoSoft:fileTypeLabel "bam" ;
104 ucscDaf:output_type "alignments" .
109 print os.listdir(self.tempdir)
110 map['1000'] = os.path.join(self.sourcedir, S1_NAME)
111 map['2000'] = os.path.join(self.sourcedir, S2_NAME)
113 s = Submission('foo', self.model, 'http://localhost')
114 mock = MockAddDetails(self.model, turtle)
115 mock.add_turtle(S1_TURTLE)
116 mock.add_turtle(S2_TURTLE)
117 s._add_library_details_to_model = mock
118 s.scan_submission_dirs(map)
120 nodes = list(s.analysis_nodes(map))
121 self.assertEqual(len(nodes), 2)
123 'http://jumpgate.caltech.edu/wiki/SubmissionsLog/foo#1000-sample',
124 'http://jumpgate.caltech.edu/wiki/SubmissionsLog/foo#2000-sample',
126 got = set((str(nodes[0]), str(nodes[1])))
127 self.assertEqual(expected, got)
129 def test_find_best_match(self):
130 turtle = get_turtle_header() + """
131 @prefix thisView: <http://jumpgate.caltech.edu/wiki/SubmissionsLog/test/view/> .
132 thisView:Fastq ucscDaf:filename_re ".*[^12]\\.fastq\\.bz2$" ;
134 geoSoft:fileTypeLabel "fastq" ;
135 ucscDaf:output_type "read" .
136 thisView:FastqRead1 ucscDaf:filename_re ".*r1\\.fastq\\.bz2$" ;
138 geoSoft:fileTypeLabel "fastq" ;
139 ucscDaf:output_type "read1" .
140 thisView:FastqRead2 ucscDaf:filename_re ".*r2\\.fastq\\.bz2$" ;
142 geoSoft:fileTypeLabel "fastq" ;
143 ucscDaf:output_type "read2" .
144 thisView:alignments ucscDaf:filename_re ".*\\.bam$" ;
145 a geoSoft:supplemental ;
146 geoSoft:fileTypeLabel "bam" ;
147 ucscDaf:output_type "alignments" .
150 load_string_into_model(self.model, 'turtle', turtle)
151 s = Submission('foo', self.model, 'http://localhost')
152 q = RDF.Statement(None, dafTermOntology['filename_re'], None)
153 view_map = s._get_filename_view_map()
154 self.assertEqual(len(view_map), 4)
156 fastq = s.find_best_match("asdf.fastq.bz2")
159 "http://jumpgate.caltech.edu/wiki/SubmissionsLog/test/view/Fastq")
161 fastq = s.find_best_match("asdf.r2.fastq.bz2")
164 "http://jumpgate.caltech.edu/wiki/SubmissionsLog/test/view/FastqRead2")
169 defaultTestLoader.loadTestsFromTestCase(TestSubmissionModule))
171 defaultTestLoader.loadTestsFromTestCase(TestSubmission))
174 if __name__ == "__main__":
175 from unittest2 import main
176 main(defaultTest='suite')