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