htsworkflow/pipelines/test/test_retrieve_config.py

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