htsworkflow/submission/test/submission_test_common.py

   1 """Code shared between test cases.
   2 """
   3 import RDF
   4 import os
   5 import tempfile
   6 import htsworkflow.util.rdfhelp
   7
   8 S1_NAME = '1000-sample'
   9 S2_NAME = '2000-sample'
  10
  11 S1_FILES = [
  12     os.path.join(S1_NAME, 'file1_l8_r1.fastq'),
  13     os.path.join(S1_NAME, 'file1_l8_r2.fastq'),
  14 ]
  15
  16 S2_FILES = [
  17     os.path.join(S2_NAME, 'file1.bam'),
  18     os.path.join(S2_NAME, 'file1_l5.fastq'),
  19 ]
  20
  21 TURTLE_PREFIX = htsworkflow.util.rdfhelp.get_turtle_header()
  22
  23 S1_TURTLE = TURTLE_PREFIX + """
  24 <http://localhost/library/1000/>
  25   htswlib:cell_line "Cell1000" ;
  26   htswlib:library_id "1000" ;
  27   htswlib:library_type "Single" ;
  28   htswlib:replicate "1" ;
  29   htswlib:has_lane <http://localhost/lane/1> ;
  30   a htswlib:IlluminaLibrary .
  31
  32 <http://localhost/lane/1>
  33   htswlib:flowcell <http://localhost/flowcel/1234ABXXX> ;
  34   htswlib:lane_number "1"@en;
  35   a htswlib:IlluminaLane .
  36 """
  37
  38 S2_TURTLE = TURTLE_PREFIX + """
  39 <http://localhost/library/2000/>
  40   htswlib:cell_line "Cell2000" ;
  41   htswlib:library_id "2000" ;
  42   htswlib:library_type "Paired End (non-multiplexed)" ;
  43   htswlib:replicate "2" ;
  44   htswlib:has_lane <http://localhost/lane/2> ;
  45   a htswlib:Library .
  46
  47 <http://localhost/lane/2>
  48   htswlib:flowcell <http://localhost/flowcel/1234ABXXX> ;
  49   htswlib:lane_number "2"@en ;
  50   a htswlib:IlluminaLane .
  51 """
  52
  53 class MockAddDetails(object):
  54     def __init__(self, model, turtle=None):
  55         self.model = model
  56         if turtle:
  57             self.add_turtle(turtle)
  58
  59     def add_turtle(self, turtle):
  60         parser = RDF.Parser('turtle')
  61         parser.parse_string_into_model(self.model, turtle, "http://localhost")
  62
  63     def __call__(self, libNode):
  64         q = RDF.Statement(libNode, None, None)
  65         found = False
  66         for s in self.model.find_statements(q):
  67             found = True
  68             break
  69         assert found
  70