htsworkflow/pipelines/test/test_retrieve_config.py

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