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         print output.buf
  58
  59         output.seek(0)
  60         sheet = list(csv.DictReader(output))
  61         expected = [{'SampleProject': '12044_index1',
  62                      'Index': 'ATCACG',
  63                      'Lane': '3',
  64                      },
  65                     {'SampleProject': '12044_index2',
  66                      'Index': 'CGATGT',
  67                      'Lane': '3',
  68                      },
  69                     {'SampleProject': '12044_index3',
  70                      'Index': 'TTAGGC',
  71                      'Lane': '3',
  72                      },
  73                     {'SampleProject': '11045_index1',
  74                      'Index': 'ATCACG',
  75                      'Lane': '3',
  76                      },
  77                     {'SampleProject': '13044_indexN701-N501',
  78                      'Index': 'TAAGGCGA-TAGATCGC',
  79                      'Lane': '4',
  80                      }
  81                     ]
  82         self.failUnlessEqual(len(sheet), len(expected))
  83         for s, e in zip(sheet, expected):
  84             for key in e.keys():
  85                 self.failUnlessEqual(s[key], e[key],
  86                   "%s != %s for key %s" % (s[key],e[key], key))