htsworkflow/pipelines/test/test_retrive_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 htsworkflow.frontend.auth import apidata
  14 from htsworkflow.pipelines.retrieve_config import \
  15      format_gerald_config, \
  16      getCombinedOptions, \
  17      save_sample_sheet
  18
  19 class RetrieveTestCases(TestCase):
  20     fixtures = ['test_flowcells.json']
  21
  22     def setUp(self):
  23         pass
  24
  25     def test_format_gerald(self):
  26         flowcell_request = self.client.get('/experiments/config/FC12150/json', apidata)
  27         self.failUnlessEqual(flowcell_request.status_code, 200)
  28         flowcell_info = json.loads(flowcell_request.content)
  29
  30         options = getCombinedOptions(['-f','FC12150','-g',os.getcwd()])
  31         genome_map = {u'Homo sapiens': '/tmp/hg18' }
  32
  33         config = format_gerald_config(options, flowcell_info, genome_map)
  34         config_lines = config.split('\n')
  35         lane3 = [ line for line in config_lines if re.search('Lane3', line) ]
  36         self.failUnlessEqual(len(lane3), 1)
  37         self.failUnlessEqual(lane3[0], '# Lane3: SL039 | Paired ends 99 GM12892')
  38         human = [ line for line in config_lines if re.search('hg18', line) ]
  39         self.failUnlessEqual(len(human), 1)
  40         self.failUnlessEqual(human[0], '345678:ELAND_GENOME /tmp/hg18')
  41         # we changed the api to force unknown genomes to be sequencing
  42         sequencing = [ line for line in config_lines if re.search('sequence_pair', line) ]
  43         self.failUnlessEqual(len(sequencing), 2)
  44
  45
  46     def test_format_sample_sheet(self):
  47         fcid = '42JU1AAXX'
  48         url = '/experiments/config/%s/json' % (fcid,)
  49         flowcell_request = self.client.get(url, apidata)
  50         self.failUnlessEqual(flowcell_request.status_code, 200)
  51         flowcell_info = json.loads(flowcell_request.content)
  52
  53         options = getCombinedOptions(['-f',fcid,'-g',os.getcwd(),])
  54
  55         output = StringIO()
  56         save_sample_sheet(output, options, flowcell_info)
  57         output.seek(0)
  58         sheet = list(csv.DictReader(output))
  59         expected = [{'SampleProject': '12044_index1', 'Index': 'ATCACG'},
  60                     {'SampleProject': '12044_index2', 'Index': 'CGATGT'},
  61                     {'SampleProject': '12044_index3', 'Index': 'TTAGGC'},
  62                     {'SampleProject': '11045_index1', 'Index': 'ATCACG'},
  63                     ]
  64         for i in range(4):
  65             self.assertEqual(sheet[i]['SampleProject'],
  66                              expected[i]['SampleProject'])
  67             self.assertEqual(sheet[i]['Index'],
  68                              expected[i]['Index'])
  69             self.assertEqual(sheet[i]['FCID'], fcid)
  70             self.assertEqual(sheet[i]['Lane'], '3')
  71
  72