htsworkflow/submission/test/test_submission.py

   1 import os
   2 from io import StringIO
   3 import shutil
   4 import tempfile
   5 from unittest import TestCase, TestSuite, defaultTestLoader
   6
   7 from htsworkflow.submission import daf, results
   8 from htsworkflow.util.rdfhelp import \
   9      dafTermOntology, \
  10      dump_model, \
  11      fromTypedNode, \
  12      get_turtle_header, \
  13      load_string_into_model, \
  14      rdfNS, \
  15      submissionLog, \
  16      submissionOntology, \
  17      get_model, \
  18      get_serializer
  19 from htsworkflow.submission.submission import list_submissions, Submission
  20 from htsworkflow.submission.results import ResultMap
  21 from .submission_test_common import *
  22
  23 import RDF
  24 #import logging
  25 #logging.basicConfig(level=logging.DEBUG)
  26
  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')
  31
  32     for d in [os.path.join(obj.tempdir, S1_NAME),
  33               os.path.join(obj.tempdir, S2_NAME),
  34               ]:
  35         os.mkdir(os.path.join(obj.tempdir, d))
  36
  37     tomake = []
  38     tomake.extend(S1_FILES)
  39     tomake.extend(S2_FILES)
  40     for f in tomake:
  41         stream = open(os.path.join(obj.tempdir, f), 'w')
  42         stream.write(f)
  43         stream.close()
  44
  45 class TestSubmissionModule(TestCase):
  46     def test_empty_list_submission(self):
  47         model = get_model()
  48         self.assertEqual(len(list(list_submissions(model))), 0)
  49
  50     def test_one_submission(self):
  51         model = get_model()
  52         load_string_into_model(model, "turtle",
  53             """
  54             @prefix subns: <http://jumpgate.caltech.edu/wiki/UcscSubmissionOntology#> .
  55             @prefix test: <http://jumpgate.caltech.edu/wiki/SubmissionsLog/test#> .
  56
  57             <http://jumpgate.caltech.edu/wiki/SubmissionsLog/test#>
  58                subns:has_submission test:lib1 ;
  59                subns:has_submission test:lib2.
  60             """)
  61         submissions = list(list_submissions(model))
  62         self.assertEqual(len(submissions), 1)
  63         self.assertEqual(submissions[0], "test")
  64
  65     def test_two_submission(self):
  66         model = get_model()
  67         load_string_into_model(model, "turtle",
  68             """
  69             @prefix subns: <http://jumpgate.caltech.edu/wiki/UcscSubmissionOntology#> .
  70             @prefix test: <http://jumpgate.caltech.edu/wiki/SubmissionsLog/test#> .
  71
  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 .
  76             """)
  77         submissions = list(list_submissions(model))
  78         self.assertEqual(len(submissions), 2)
  79         truth = set(["test1", "test2"])
  80         testset = set()
  81         for name in submissions:
  82             testset.add(name)
  83         self.assertEqual(testset, truth)
  84
  85 class TestSubmission(TestCase):
  86     def setUp(self):
  87         generate_sample_results_tree(self)
  88         self.model = get_model()
  89
  90     def tearDown(self):
  91         shutil.rmtree(self.tempdir)
  92
  93     def test_create_submission(self):
  94         model = get_model()
  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/')['']))
 102
 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$" ;
 107                a geoSoft:raw ;
 108                geoSoft:fileTypeLabel "fastq" ;
 109                ucscDaf:output_type "read" .
 110 thisView:FastqRead1 ucscDaf:filename_re ".*r1\\.fastq\\.bz2$" ;
 111                a geoSoft:raw ;
 112                geoSoft:fileTypeLabel "fastq" ;
 113                ucscDaf:output_type "read1" .
 114 thisView:FastqRead2 ucscDaf:filename_re ".*r2\\.fastq\\.bz2$" ;
 115                a geoSoft:raw ;
 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" .
 122
 123         """
 124         map = ResultMap()
 125         print(self.tempdir)
 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)
 129
 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)
 136
 137         nodes = list(s.analysis_nodes(map))
 138         self.assertEqual(len(nodes), 2)
 139         expected = set((
 140             'http://jumpgate.caltech.edu/wiki/SubmissionsLog/foo#1000-sample',
 141             'http://jumpgate.caltech.edu/wiki/SubmissionsLog/foo#2000-sample',
 142         ))
 143         got = set((str(nodes[0]), str(nodes[1])))
 144         self.assertEqual(expected, got)
 145
 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$" ;
 150                a geoSoft:raw ;
 151                geoSoft:fileTypeLabel "fastq" ;
 152                ucscDaf:output_type "read" .
 153 thisView:FastqRead1 ucscDaf:filename_re ".*r1\\.fastq\\.bz2$" ;
 154                a geoSoft:raw ;
 155                geoSoft:fileTypeLabel "fastq" ;
 156                ucscDaf:output_type "read1" .
 157 thisView:FastqRead2 ucscDaf:filename_re ".*r2\\.fastq\\.bz2$" ;
 158                a geoSoft:raw ;
 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" .
 165
 166         """
 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)
 172
 173         fastq = s.find_best_match("asdf.fastq.bz2")
 174         self.assertEqual(
 175             str(fastq),
 176             "http://jumpgate.caltech.edu/wiki/SubmissionsLog/test/view/Fastq")
 177
 178         fastq = s.find_best_match("asdf.r2.fastq.bz2")
 179         self.assertEqual(
 180             str(fastq),
 181             "http://jumpgate.caltech.edu/wiki/SubmissionsLog/test/view/FastqRead2")
 182
 183 def suite():
 184     suite = TestSuite()
 185     suite.addTests(
 186         defaultTestLoader.loadTestsFromTestCase(TestSubmissionModule))
 187     suite.addTests(
 188         defaultTestLoader.loadTestsFromTestCase(TestSubmission))
 189     return suite
 190
 191 if __name__ == "__main__":
 192     from unittest import main
 193     main(defaultTest='suite')