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 = ['initial_data.json',
  21                 'test_flowcells.json']
  22
  23     def setUp(self):
  24         pass
  25
  26     def test_format_gerald(self):
  27         flowcell_request = self.client.get('/experiments/config/FC12150/json', apidata)
  28         self.failUnlessEqual(flowcell_request.status_code, 200)
  29         flowcell_info = json.loads(flowcell_request.content)
  30
  31         options = getCombinedOptions(['-f','FC12150','-g',os.getcwd()])
  32         genome_map = {u'Homo sapiens': '/tmp/hg18' }
  33
  34         config = format_gerald_config(options, flowcell_info, genome_map)
  35         config_lines = config.split('\n')
  36         lane3 = [ line for line in config_lines if re.search('Lane3', line) ]
  37         self.failUnlessEqual(len(lane3), 1)
  38         self.failUnlessEqual(lane3[0], '# Lane3: SL039 | Paired ends 99 GM12892')
  39         human = [ line for line in config_lines if re.search('hg18', line) ]
  40         self.failUnlessEqual(len(human), 1)
  41         self.failUnlessEqual(human[0], '345678:ELAND_GENOME /tmp/hg18')
  42         # we changed the api to force unknown genomes to be sequencing
  43         sequencing = [ line for line in config_lines if re.search('sequence_pair', line) ]
  44         self.failUnlessEqual(len(sequencing), 2)
  45
  46
  47     def test_format_sample_sheet(self):
  48         fcid = '42JU1AAXX'
  49         url = '/experiments/config/%s/json' % (fcid,)
  50         flowcell_request = self.client.get(url, apidata)
  51         self.failUnlessEqual(flowcell_request.status_code, 200)
  52         flowcell_info = json.loads(flowcell_request.content)
  53
  54         options = getCombinedOptions(['-f',fcid,'-g',os.getcwd(),])
  55
  56         output = StringIO()
  57         save_sample_sheet(output, options, flowcell_info)
  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))