htsworkflow/submission/test/submission_test_common.py

   1 """Code shared between test cases.
   2 """
   3 import RDF
   4 import logging
   5 import os
   6 import tempfile
   7 import htsworkflow.util.rdfhelp
   8
   9 S1_NAME = '1000-sample'
  10 S2_NAME = '2000-sample'
  11 SCOMBINED_NAME = 'directory'
  12
  13 S1_FILES = [
  14     os.path.join(S1_NAME, 'file1_l8_r1.fastq.bz2'),
  15     os.path.join(S1_NAME, 'file1_l8_r2.fastq.bz2'),
  16 ]
  17
  18 S2_FILES = [
  19     os.path.join(S2_NAME, 'file1.bam'),
  20     os.path.join(S2_NAME, 'file1_l5.fastq.bz2'),
  21 ]
  22
  23 SCOMBINED_FILES = [
  24     os.path.join(SCOMBINED_NAME, 's1_file1.bam'),
  25     os.path.join(SCOMBINED_NAME, 's1_l5.fastq.bz2'),
  26     os.path.join(SCOMBINED_NAME, 's2_file1.bam'),
  27     os.path.join(SCOMBINED_NAME, 's2_l4.read1.fastq.bz2'),
  28     os.path.join(SCOMBINED_NAME, 's2_l4.read2.fastq.bz2'),
  29 ]
  30
  31 TURTLE_PREFIX = htsworkflow.util.rdfhelp.get_turtle_header()
  32
  33 S1_TURTLE = TURTLE_PREFIX + """
  34 <http://localhost/library/1000/>
  35   htswlib:cell_line "Cell1000" ;
  36   htswlib:library_id "1000" ;
  37   htswlib:library_type "Single" ;
  38   htswlib:replicate "1" ;
  39   htswlib:has_lane <http://localhost/lane/1> ;
  40   a htswlib:IlluminaLibrary .
  41
  42 <http://localhost/lane/1>
  43   htswlib:flowcell <http://localhost/flowcel/1234ABXXX> ;
  44   htswlib:lane_number "1"@en;
  45   a htswlib:IlluminaLane .
  46 """
  47
  48 S2_TURTLE = TURTLE_PREFIX + """
  49 <http://localhost/library/2000/>
  50   htswlib:cell_line "Cell2000" ;
  51   htswlib:library_id "2000" ;
  52   htswlib:library_type "Paired End (non-multiplexed)" ;
  53   htswlib:replicate "2" ;
  54   htswlib:has_lane <http://localhost/lane/2> ;
  55   a htswlib:Library .
  56
  57 <http://localhost/lane/2>
  58   htswlib:flowcell <http://localhost/flowcel/1234ABXXX> ;
  59   htswlib:lane_number "2"@en ;
  60   a htswlib:IlluminaLane .
  61 """
  62
  63 class MockAddDetails(object):
  64     def __init__(self, model, turtle=None):
  65         self.model = model
  66         if turtle:
  67             self.add_turtle(turtle)
  68
  69     def add_turtle(self, turtle):
  70         parser = RDF.Parser('turtle')
  71         parser.parse_string_into_model(self.model, turtle, "http://localhost")
  72
  73     def __call__(self, libNode):
  74         q = RDF.Statement(libNode, None, None)
  75         found = False
  76         for s in self.model.find_statements(q):
  77             found = True
  78             break
  79         assert found
  80
  81 def generate_sample_results_tree(obj, prefix):
  82     obj.tempdir = tempfile.mkdtemp(prefix=prefix)
  83     obj.sourcedir = os.path.join(obj.tempdir, 'source')
  84     os.mkdir(obj.sourcedir)
  85     obj.resultdir = os.path.join(obj.tempdir, 'results')
  86     os.mkdir(obj.resultdir)
  87
  88     for d in [os.path.join(obj.sourcedir, S1_NAME),
  89               os.path.join(obj.sourcedir, S2_NAME),
  90               ]:
  91         logging.debug("Creating: %s", d)
  92         os.mkdir(d)
  93
  94     tomake = []
  95     tomake.extend(S1_FILES)
  96     tomake.extend(S2_FILES)
  97     for f in tomake:
  98         target = os.path.join(obj.sourcedir, f)
  99         logging.debug("Creating: %s", target)
 100         stream = open(target, 'w')
 101         stream.write(f)
 102         stream.close()
 103