htsworkflow/submission/test/test_submission.py

   1
   2 import os
   3 from six.moves import StringIO
   4 import shutil
   5 import tempfile
   6 from unittest import TestCase, TestSuite, defaultTestLoader
   7
   8 from htsworkflow.submission import daf, results
   9 from htsworkflow.util.rdfhelp import \
  10      dafTermOntology, \
  11      dump_model, \
  12      fromTypedNode, \
  13      get_turtle_header, \
  14      load_string_into_model, \
  15      rdfNS, \
  16      submissionLog, \
  17      submissionOntology, \
  18      get_model, \
  19      get_serializer
  20 from htsworkflow.submission.submission import list_submissions, Submission
  21 from htsworkflow.submission.results import ResultMap
  22 from .submission_test_common import  (
  23     generate_sample_results_tree,
  24     S1_NAME,
  25     S2_NAME,
  26     S1_TURTLE,
  27     S2_TURTLE,
  28     MockAddDetails,
  29 )
  30
  31 import RDF
  32 #import logging
  33 #logging.basicConfig(level=logging.DEBUG)
  34
  35 class TestSubmissionModule(TestCase):
  36     def test_empty_list_submission(self):
  37         model = get_model()
  38         self.assertEqual(len(list(list_submissions(model))), 0)
  39
  40     def test_one_submission(self):
  41         model = get_model()
  42         load_string_into_model(model, "turtle",
  43             """
  44             @prefix subns: <http://jumpgate.caltech.edu/wiki/UcscSubmissionOntology#> .
  45             @prefix test: <http://jumpgate.caltech.edu/wiki/SubmissionsLog/test#> .
  46
  47             <http://jumpgate.caltech.edu/wiki/SubmissionsLog/test#>
  48                subns:has_submission test:lib1 ;
  49                subns:has_submission test:lib2.
  50             """)
  51         submissions = list(list_submissions(model))
  52         self.assertEqual(len(submissions), 1)
  53         self.assertEqual(submissions[0], "test")
  54
  55     def test_two_submission(self):
  56         model = get_model()
  57         load_string_into_model(model, "turtle",
  58             """
  59             @prefix subns: <http://jumpgate.caltech.edu/wiki/UcscSubmissionOntology#> .
  60             @prefix test: <http://jumpgate.caltech.edu/wiki/SubmissionsLog/test#> .
  61
  62             <http://jumpgate.caltech.edu/wiki/SubmissionsLog/test1#>
  63                subns:has_submission test:lib1 .
  64             <http://jumpgate.caltech.edu/wiki/SubmissionsLog/test2#>
  65                subns:has_submission test:lib2 .
  66             """)
  67         submissions = list(list_submissions(model))
  68         self.assertEqual(len(submissions), 2)
  69         truth = set(["test1", "test2"])
  70         testset = set()
  71         for name in submissions:
  72             testset.add(name)
  73         self.assertEqual(testset, truth)
  74
  75 class TestSubmission(TestCase):
  76     def setUp(self):
  77         generate_sample_results_tree(self, 'submission_test')
  78         self.model = get_model()
  79
  80     def tearDown(self):
  81         shutil.rmtree(self.tempdir)
  82
  83     def test_create_submission(self):
  84         model = get_model()
  85         s = Submission('foo', self.model, 'http://localhost')
  86         self.assertEqual(str(s.submissionSet),
  87                          "http://jumpgate.caltech.edu/wiki/SubmissionsLog/foo")
  88         self.assertEqual(str(s.submissionSetNS['']),
  89                          str(RDF.NS(str(s.submissionSet) + '#')['']))
  90         self.assertEqual(str(s.libraryNS['']),
  91                          str(RDF.NS('http://localhost/library/')['']))
  92
  93     def test_scan_submission_dirs(self):
  94         turtle = get_turtle_header() + """
  95 @prefix thisView: <http://jumpgate.caltech.edu/wiki/SubmissionsLog/test/view/> .
  96 thisView:Fastq ucscDaf:filename_re ".*[^12]\\.fastq\\.bz2$" ;
  97                a geoSoft:raw ;
  98                geoSoft:fileTypeLabel "fastq" ;
  99                ucscDaf:output_type "read" .
 100 thisView:FastqRead1 ucscDaf:filename_re ".*r1\\.fastq\\.bz2$" ;
 101                a geoSoft:raw ;
 102                geoSoft:fileTypeLabel "fastq" ;
 103                ucscDaf:output_type "read1" .
 104 thisView:FastqRead2 ucscDaf:filename_re ".*r2\\.fastq\\.bz2$" ;
 105                a geoSoft:raw ;
 106                geoSoft:fileTypeLabel "fastq" ;
 107                ucscDaf:output_type "read2" .
 108 thisView:alignments ucscDaf:filename_re ".*\\.bam$" ;
 109                a geoSoft:supplemental ;
 110                geoSoft:fileTypeLabel "bam" ;
 111                ucscDaf:output_type "alignments" .
 112
 113         """
 114         map = ResultMap()
 115         map['1000'] = os.path.join(self.sourcedir, S1_NAME)
 116         map['2000'] = os.path.join(self.sourcedir, S2_NAME)
 117
 118         s = Submission('foo', self.model, 'http://localhost')
 119         mock = MockAddDetails(self.model, turtle)
 120         mock.add_turtle(S1_TURTLE)
 121         mock.add_turtle(S2_TURTLE)
 122         s._add_library_details_to_model =  mock
 123         s.scan_submission_dirs(map)
 124
 125         nodes = list(s.analysis_nodes(map))
 126         self.assertEqual(len(nodes), 2)
 127         expected = set((
 128             'http://jumpgate.caltech.edu/wiki/SubmissionsLog/foo#1000-sample',
 129             'http://jumpgate.caltech.edu/wiki/SubmissionsLog/foo#2000-sample',
 130         ))
 131         got = set((str(nodes[0]), str(nodes[1])))
 132         self.assertEqual(expected, got)
 133
 134     def test_find_best_match(self):
 135         turtle = get_turtle_header() + """
 136 @prefix thisView: <http://jumpgate.caltech.edu/wiki/SubmissionsLog/test/view/> .
 137 thisView:Fastq ucscDaf:filename_re ".*[^12]\\.fastq\\.bz2$" ;
 138                a geoSoft:raw ;
 139                geoSoft:fileTypeLabel "fastq" ;
 140                ucscDaf:output_type "read" .
 141 thisView:FastqRead1 ucscDaf:filename_re ".*r1\\.fastq\\.bz2$" ;
 142                a geoSoft:raw ;
 143                geoSoft:fileTypeLabel "fastq" ;
 144                ucscDaf:output_type "read1" .
 145 thisView:FastqRead2 ucscDaf:filename_re ".*r2\\.fastq\\.bz2$" ;
 146                a geoSoft:raw ;
 147                geoSoft:fileTypeLabel "fastq" ;
 148                ucscDaf:output_type "read2" .
 149 thisView:alignments ucscDaf:filename_re ".*\\.bam$" ;
 150                a geoSoft:supplemental ;
 151                geoSoft:fileTypeLabel "bam" ;
 152                ucscDaf:output_type "alignments" .
 153
 154         """
 155         load_string_into_model(self.model, 'turtle', turtle)
 156         s = Submission('foo', self.model, 'http://localhost')
 157         q = RDF.Statement(None, dafTermOntology['filename_re'], None)
 158         view_map = s._get_filename_view_map()
 159         self.assertEqual(len(view_map), 4)
 160
 161         fastq = s.find_best_match("asdf.fastq.bz2")
 162         self.assertEqual(
 163             str(fastq),
 164             "http://jumpgate.caltech.edu/wiki/SubmissionsLog/test/view/Fastq")
 165
 166         fastq = s.find_best_match("asdf.r2.fastq.bz2")
 167         self.assertEqual(
 168             str(fastq),
 169             "http://jumpgate.caltech.edu/wiki/SubmissionsLog/test/view/FastqRead2")
 170
 171 def suite():
 172     suite = TestSuite()
 173     suite.addTests(
 174         defaultTestLoader.loadTestsFromTestCase(TestSubmissionModule))
 175     suite.addTests(
 176         defaultTestLoader.loadTestsFromTestCase(TestSubmission))
 177     return suite
 178
 179 if __name__ == "__main__":
 180     from unittest import main
 181     main(defaultTest='suite')