htsworkflow/submission/test/test_submission.py

   1
   2 import os
   3 from StringIO 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
  24 import RDF
  25 #import logging
  26 #logging.basicConfig(level=logging.DEBUG)
  27
  28 class TestSubmissionModule(TestCase):
  29     def test_empty_list_submission(self):
  30         model = get_model()
  31         self.assertEqual(len(list(list_submissions(model))), 0)
  32
  33     def test_one_submission(self):
  34         model = get_model()
  35         load_string_into_model(model, "turtle",
  36             """
  37             @prefix subns: <http://jumpgate.caltech.edu/wiki/UcscSubmissionOntology#> .
  38             @prefix test: <http://jumpgate.caltech.edu/wiki/SubmissionsLog/test#> .
  39
  40             <http://jumpgate.caltech.edu/wiki/SubmissionsLog/test#>
  41                subns:has_submission test:lib1 ;
  42                subns:has_submission test:lib2.
  43             """)
  44         submissions = list(list_submissions(model))
  45         self.assertEqual(len(submissions), 1)
  46         self.assertEqual(submissions[0], "test")
  47
  48     def test_two_submission(self):
  49         model = get_model()
  50         load_string_into_model(model, "turtle",
  51             """
  52             @prefix subns: <http://jumpgate.caltech.edu/wiki/UcscSubmissionOntology#> .
  53             @prefix test: <http://jumpgate.caltech.edu/wiki/SubmissionsLog/test#> .
  54
  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 .
  59             """)
  60         submissions = list(list_submissions(model))
  61         self.assertEqual(len(submissions), 2)
  62         truth = set(["test1", "test2"])
  63         testset = set()
  64         for name in submissions:
  65             testset.add(name)
  66         self.assertEqual(testset, truth)
  67
  68 class TestSubmission(TestCase):
  69     def setUp(self):
  70         generate_sample_results_tree(self, 'submission_test')
  71         self.model = get_model()
  72
  73     def tearDown(self):
  74         shutil.rmtree(self.tempdir)
  75
  76     def test_create_submission(self):
  77         model = get_model()
  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/')['']))
  85
  86     def test_scan_submission_dirs(self):
  87         turtle = get_turtle_header() + r"""
  88 @prefix thisView: <http://jumpgate.caltech.edu/wiki/SubmissionsLog/test/view/> .
  89 thisView:Fastq ucscDaf:filename_re ".*[^12]\\.fastq\\.bz2$" ;
  90                a geoSoft:raw ;
  91                geoSoft:fileTypeLabel "fastq" ;
  92                ucscDaf:output_type "read" .
  93 thisView:FastqRead1 ucscDaf:filename_re ".*r1\\.fastq\\.bz2$" ;
  94                a geoSoft:raw ;
  95                geoSoft:fileTypeLabel "fastq" ;
  96                ucscDaf:output_type "read1" .
  97 thisView:FastqRead2 ucscDaf:filename_re ".*r2\\.fastq\\.bz2$" ;
  98                a geoSoft:raw ;
  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" .
 105
 106         """
 107         map = ResultMap()
 108         map['1000'] = os.path.join(self.sourcedir, S1_NAME)
 109         map['2000'] = os.path.join(self.sourcedir, S2_NAME)
 110
 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)
 117
 118         nodes = list(s.analysis_nodes(map))
 119         self.assertEqual(len(nodes), 2)
 120         expected = set((
 121             'http://jumpgate.caltech.edu/wiki/SubmissionsLog/foo#1000-sample',
 122             'http://jumpgate.caltech.edu/wiki/SubmissionsLog/foo#2000-sample',
 123         ))
 124         got = set((str(nodes[0]), str(nodes[1])))
 125         self.assertEqual(expected, got)
 126
 127     def test_find_best_match(self):
 128         turtle = get_turtle_header() + r"""
 129 @prefix thisView: <http://jumpgate.caltech.edu/wiki/SubmissionsLog/test/view/> .
 130 thisView:Fastq ucscDaf:filename_re ".*[^12]\\.fastq\\.bz2$" ;
 131                a geoSoft:raw ;
 132                geoSoft:fileTypeLabel "fastq" ;
 133                ucscDaf:output_type "read" .
 134 thisView:FastqRead1 ucscDaf:filename_re ".*r1\\.fastq\\.bz2$" ;
 135                a geoSoft:raw ;
 136                geoSoft:fileTypeLabel "fastq" ;
 137                ucscDaf:output_type "read1" .
 138 thisView:FastqRead2 ucscDaf:filename_re ".*r2\\.fastq\\.bz2$" ;
 139                a geoSoft:raw ;
 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" .
 146
 147         """
 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)
 153
 154         fastq = s.find_best_match("asdf.fastq.bz2")
 155         self.assertEqual(
 156             str(fastq),
 157             "http://jumpgate.caltech.edu/wiki/SubmissionsLog/test/view/Fastq")
 158
 159         fastq = s.find_best_match("asdf.r2.fastq.bz2")
 160         self.assertEqual(
 161             str(fastq),
 162             "http://jumpgate.caltech.edu/wiki/SubmissionsLog/test/view/FastqRead2")
 163
 164 def suite():
 165     suite = TestSuite()
 166     suite.addTests(
 167         defaultTestLoader.loadTestsFromTestCase(TestSubmissionModule))
 168     suite.addTests(
 169         defaultTestLoader.loadTestsFromTestCase(TestSubmission))
 170     return suite
 171
 172 if __name__ == "__main__":
 173     from unittest import main
 174     main(defaultTest='suite')