htsworkflow/pipelines/test/test_retrieve_config.py

   1 import csv
   2 import os
   3 import re
   4 from six.moves import StringIO
   5
   6 try:
   7     import json
   8 except ImportError as e:
   9     import simplejson as json
  10
  11 from django.test import TestCase
  12 from django.utils.encoding import smart_text
  13
  14 from samples.samples_factory import LibraryFactory, LibraryTypeFactory, \
  15     MultiplexIndexFactory
  16 from experiments.experiments_factory import FlowCellFactory, LaneFactory
  17
  18 from htsworkflow.auth import apidata
  19 from htsworkflow.pipelines.retrieve_config import \
  20      format_gerald_config, \
  21      getCombinedOptions, \
  22      save_sample_sheet, \
  23      format_project_name
  24
  25 class RetrieveTestCases(TestCase):
  26     def setUp(self):
  27         pass
  28
  29     def test_format_gerald(self):
  30         fc = FlowCellFactory.create(flowcell_id='FC12150')
  31         library = LibraryFactory.create()
  32         lane = LaneFactory(flowcell=fc, library=library, lane_number=3)
  33
  34         flowcell_request = self.client.get('/experiments/config/FC12150/json',
  35                                            apidata)
  36         self.failUnlessEqual(flowcell_request.status_code, 200)
  37         flowcell_info = json.loads(smart_text(flowcell_request.content))['result']
  38
  39         options = getCombinedOptions(['-f','FC12150','-g',os.getcwd()])
  40         genome_map = {library.library_species.scientific_name: '/tmp/build' }
  41
  42         config = format_gerald_config(options, flowcell_info, genome_map)
  43         config_lines = config.split('\n')
  44         lane3 = [ line for line in config_lines if re.search('Lane3', line) ]
  45         self.failUnlessEqual(len(lane3), 1)
  46         expected = '# Lane3: {} | {}'.format(library.id, library.library_name)
  47         self.failUnlessEqual(lane3[0], expected)
  48         human = [ line for line in config_lines if re.search('build', line) ]
  49         self.failUnlessEqual(len(human), 1)
  50         self.failUnlessEqual(human[0], '3:ELAND_GENOME /tmp/build')
  51
  52
  53     def test_format_sample_sheet(self):
  54         fcid = '42JU1AAXX'
  55         fc = FlowCellFactory.create(flowcell_id=fcid)
  56         library_type = LibraryTypeFactory(can_multiplex=True)
  57         multiplex_index1 = MultiplexIndexFactory(adapter_type=library_type)
  58         multiplex_index2 = MultiplexIndexFactory(adapter_type=library_type)
  59
  60         library1 = LibraryFactory.create(
  61             library_type=library_type,
  62             multiplex_id=multiplex_index1.multiplex_id)
  63         library2 = LibraryFactory.create(
  64             library_type=library_type,
  65             multiplex_id=multiplex_index2.multiplex_id)
  66
  67         lane1l1 = LaneFactory(flowcell=fc, library=library1, lane_number=1)
  68         lane1l2 = LaneFactory(flowcell=fc, library=library2, lane_number=1)
  69         lane2l1 = LaneFactory(flowcell=fc, library=library1, lane_number=2)
  70         lane2l2 = LaneFactory(flowcell=fc, library=library2, lane_number=2)
  71
  72         url = '/experiments/config/%s/json' % (fcid,)
  73         flowcell_request = self.client.get(url, apidata)
  74         self.failUnlessEqual(flowcell_request.status_code, 200)
  75         flowcell_info = json.loads(smart_text(flowcell_request.content))['result']
  76
  77         options = getCombinedOptions(['-f',fcid,'-g',os.getcwd(),])
  78
  79         output = StringIO()
  80         save_sample_sheet(output, options, flowcell_info)
  81
  82         output.seek(0)
  83         sheet = list(csv.DictReader(output))
  84         name1 = library1.id + '_index' + library1.index_sequences().keys()[0]
  85         name2 = library2.id + '_index' + library2.index_sequences().keys()[0]
  86         expected = [{'SampleProject': name1,
  87                      'Index': library1.index_sequences().values()[0],
  88                      'Lane': '1',
  89                      },
  90                     {'SampleProject': name2,
  91                      'Index': library2.index_sequences().values()[0],
  92                      'Lane': '1',
  93                      },
  94                     {'SampleProject': name1,
  95                      'Index': library1.index_sequences().values()[0],
  96                      'Lane': '2',
  97                      },
  98                     {'SampleProject': name2,
  99                      'Index': library2.index_sequences().values()[0],
 100                      'Lane': '2',
 101                      },
 102                     ]
 103         self.failUnlessEqual(len(sheet), len(expected))
 104         for s, e in zip(sheet, expected):
 105             for key in e.keys():
 106                 self.failUnlessEqual(s[key], e[key],
 107                   "%s != %s for key %s" % (s[key],e[key], key))