3 from six.moves import StringIO
6 from unittest 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" .
108 map['1000'] = os.path.join(self.sourcedir, S1_NAME)
109 map['2000'] = os.path.join(self.sourcedir, S2_NAME)
111 s = Submission('foo', self.model, 'http://localhost')
112 mock = MockAddDetails(self.model, turtle)
113 mock.add_turtle(S1_TURTLE)
114 mock.add_turtle(S2_TURTLE)
115 s._add_library_details_to_model = mock
116 s.scan_submission_dirs(map)
118 nodes = list(s.analysis_nodes(map))
119 self.assertEqual(len(nodes), 2)
121 'http://jumpgate.caltech.edu/wiki/SubmissionsLog/foo#1000-sample',
122 'http://jumpgate.caltech.edu/wiki/SubmissionsLog/foo#2000-sample',
124 got = set((str(nodes[0]), str(nodes[1])))
125 self.assertEqual(expected, got)
127 def test_find_best_match(self):
128 turtle = get_turtle_header() + """
129 @prefix thisView: <http://jumpgate.caltech.edu/wiki/SubmissionsLog/test/view/> .
130 thisView:Fastq ucscDaf:filename_re ".*[^12]\\.fastq\\.bz2$" ;
132 geoSoft:fileTypeLabel "fastq" ;
133 ucscDaf:output_type "read" .
134 thisView:FastqRead1 ucscDaf:filename_re ".*r1\\.fastq\\.bz2$" ;
136 geoSoft:fileTypeLabel "fastq" ;
137 ucscDaf:output_type "read1" .
138 thisView:FastqRead2 ucscDaf:filename_re ".*r2\\.fastq\\.bz2$" ;
140 geoSoft:fileTypeLabel "fastq" ;
141 ucscDaf:output_type "read2" .
142 thisView:alignments ucscDaf:filename_re ".*\\.bam$" ;
143 a geoSoft:supplemental ;
144 geoSoft:fileTypeLabel "bam" ;
145 ucscDaf:output_type "alignments" .
148 load_string_into_model(self.model, 'turtle', turtle)
149 s = Submission('foo', self.model, 'http://localhost')
150 q = RDF.Statement(None, dafTermOntology['filename_re'], None)
151 view_map = s._get_filename_view_map()
152 self.assertEqual(len(view_map), 4)
154 fastq = s.find_best_match("asdf.fastq.bz2")
157 "http://jumpgate.caltech.edu/wiki/SubmissionsLog/test/view/Fastq")
159 fastq = s.find_best_match("asdf.r2.fastq.bz2")
162 "http://jumpgate.caltech.edu/wiki/SubmissionsLog/test/view/FastqRead2")
167 defaultTestLoader.loadTestsFromTestCase(TestSubmissionModule))
169 defaultTestLoader.loadTestsFromTestCase(TestSubmission))
172 if __name__ == "__main__":
173 from unittest import main
174 main(defaultTest='suite')