htsworkflow/submission/test/submission_test_common.py

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