htsworkflow/frontend/experiments/tests.py

   1 import re
   2 from BeautifulSoup import BeautifulSoup
   3 try:
   4     import json
   5 except ImportError, e:
   6     import simplejson as json
   7 import os
   8 import shutil
   9 import sys
  10 import tempfile
  11
  12 from django.conf import settings
  13 from django.core import mail
  14 from django.core.exceptions import ObjectDoesNotExist
  15 from django.test import TestCase
  16 from htsworkflow.frontend.experiments import models
  17 from htsworkflow.frontend.experiments import experiments
  18 from htsworkflow.frontend.auth import apidata
  19
  20 from htsworkflow.pipelines.test.simulate_runfolder import TESTDATA_DIR
  21
  22 LANE_SET = range(1,9)
  23
  24 class ExperimentsTestCases(TestCase):
  25     fixtures = ['test_flowcells.json']
  26
  27     def setUp(self):
  28         self.tempdir = tempfile.mkdtemp(prefix='htsw-test-experiments-')
  29         settings.RESULT_HOME_DIR = self.tempdir
  30
  31         self.fc1_id = 'FC12150'
  32         self.fc1_root = os.path.join(self.tempdir, self.fc1_id)
  33         os.mkdir(self.fc1_root)
  34         self.fc1_dir = os.path.join(self.fc1_root, 'C1-37')
  35         os.mkdir(self.fc1_dir)
  36         runxml = 'run_FC12150_2007-09-27.xml'
  37         shutil.copy(os.path.join(TESTDATA_DIR, runxml),
  38                     os.path.join(self.fc1_dir, runxml))
  39         for i in range(1,9):
  40             shutil.copy(
  41                 os.path.join(TESTDATA_DIR,
  42                              'woldlab_070829_USI-EAS44_0017_FC11055_1.srf'),
  43                 os.path.join(self.fc1_dir,
  44                              'woldlab_070829_SERIAL_FC12150_%d.srf' %(i,))
  45                 )
  46
  47         self.fc2_dir = os.path.join(self.tempdir, '42JTNAAXX')
  48         os.mkdir(self.fc2_dir)
  49         os.mkdir(os.path.join(self.fc2_dir, 'C1-25'))
  50         os.mkdir(os.path.join(self.fc2_dir, 'C1-37'))
  51         os.mkdir(os.path.join(self.fc2_dir, 'C1-37', 'Plots'))
  52
  53     def tearDown(self):
  54         shutil.rmtree(self.tempdir)
  55
  56     def test_flowcell_information(self):
  57         """
  58         Check the code that packs the django objects into simple types.
  59         """
  60         for fc_id in [u'FC12150', u"42JTNAAXX", "42JU1AAXX"]:
  61             fc_dict = experiments.flowcell_information(fc_id)
  62             fc_django = models.FlowCell.objects.get(flowcell_id=fc_id)
  63             self.failUnlessEqual(fc_dict['flowcell_id'], fc_id)
  64             self.failUnlessEqual(fc_django.flowcell_id, fc_id)
  65             self.failUnlessEqual(fc_dict['sequencer'], fc_django.sequencer.name)
  66             self.failUnlessEqual(fc_dict['read_length'], fc_django.read_length)
  67             self.failUnlessEqual(fc_dict['notes'], fc_django.notes)
  68             self.failUnlessEqual(fc_dict['cluster_station'], fc_django.cluster_station.name)
  69
  70             for lane in fc_django.lane_set.all():
  71                 lane_dict = fc_dict['lane_set'][lane.lane_number]
  72                 self.failUnlessEqual(lane_dict['cluster_estimate'], lane.cluster_estimate)
  73                 self.failUnlessEqual(lane_dict['comment'], lane.comment)
  74                 self.failUnlessEqual(lane_dict['flowcell'], lane.flowcell.flowcell_id)
  75                 self.failUnlessEqual(lane_dict['lane_number'], lane.lane_number)
  76                 self.failUnlessEqual(lane_dict['library_name'], lane.library.library_name)
  77                 self.failUnlessEqual(lane_dict['library_id'], lane.library.id)
  78                 self.failUnlessAlmostEqual(float(lane_dict['pM']), float(lane.pM))
  79                 self.failUnlessEqual(lane_dict['library_species'],
  80                                      lane.library.library_species.scientific_name)
  81
  82             response = self.client.get('/experiments/config/%s/json' % (fc_id,), apidata)
  83             # strptime isoformat string = '%Y-%m-%dT%H:%M:%S'
  84             fc_json = json.loads(response.content)
  85             self.failUnlessEqual(fc_json['flowcell_id'], fc_id)
  86             self.failUnlessEqual(fc_json['sequencer'], fc_django.sequencer.name)
  87             self.failUnlessEqual(fc_json['read_length'], fc_django.read_length)
  88             self.failUnlessEqual(fc_json['notes'], fc_django.notes)
  89             self.failUnlessEqual(fc_json['cluster_station'], fc_django.cluster_station.name)
  90
  91
  92             for lane in fc_django.lane_set.all():
  93                 lane_dict = fc_json['lane_set'][unicode(lane.lane_number)]
  94                 self.failUnlessEqual(lane_dict['cluster_estimate'], lane.cluster_estimate)
  95                 self.failUnlessEqual(lane_dict['comment'], lane.comment)
  96                 self.failUnlessEqual(lane_dict['flowcell'], lane.flowcell.flowcell_id)
  97                 self.failUnlessEqual(lane_dict['lane_number'], lane.lane_number)
  98                 self.failUnlessEqual(lane_dict['library_name'], lane.library.library_name)
  99                 self.failUnlessEqual(lane_dict['library_id'], lane.library.id)
 100                 self.failUnlessAlmostEqual(float(lane_dict['pM']), float(lane.pM))
 101                 self.failUnlessEqual(lane_dict['library_species'],
 102                                      lane.library.library_species.scientific_name)
 103
 104     def test_invalid_flowcell(self):
 105         """
 106         Make sure we get a 404 if we request an invalid flowcell ID
 107         """
 108         response = self.client.get('/experiments/config/nottheone/json', apidata)
 109         self.failUnlessEqual(response.status_code, 404)
 110
 111     def test_no_key(self):
 112         """
 113         Require logging in to retrieve meta data
 114         """
 115         response = self.client.get(u'/experiments/config/FC12150/json')
 116         self.failUnlessEqual(response.status_code, 403)
 117
 118     def test_library_id(self):
 119         """
 120         Library IDs should be flexible, so make sure we can retrive a non-numeric ID
 121         """
 122         response = self.client.get('/experiments/config/FC12150/json', apidata)
 123         self.failUnlessEqual(response.status_code, 200)
 124         flowcell = json.loads(response.content)
 125
 126         self.failUnlessEqual(flowcell['lane_set']['3']['library_id'], 'SL039')
 127
 128         response = self.client.get('/samples/library/SL039/json', apidata)
 129         self.failUnlessEqual(response.status_code, 200)
 130         library_sl039 = json.loads(response.content)
 131
 132         self.failUnlessEqual(library_sl039['library_id'], 'SL039')
 133
 134     def test_raw_id_field(self):
 135         """
 136         Test ticket:147
 137
 138         Library's have IDs, libraries also have primary keys,
 139         we eventually had enough libraries that the drop down combo box was too
 140         hard to filter through, unfortnately we want a field that uses our library
 141         id and not the internal primary key, and raw_id_field uses primary keys.
 142
 143         This tests to make sure that the value entered in the raw library id field matches
 144         the library id looked up.
 145         """
 146         expected_ids = [u'10981',u'11016',u'SL039',u'11060',
 147                         u'11061',u'11062',u'11063',u'11064']
 148         self.client.login(username='supertest', password='BJOKL5kAj6aFZ6A5')
 149         response = self.client.get('/admin/experiments/flowcell/153/')
 150         soup = BeautifulSoup(response.content)
 151         for i in range(0,8):
 152             input_field = soup.find(id='id_lane_set-%d-library' % (i,))
 153             library_field = input_field.findNext('strong')
 154             library_id, library_name = library_field.string.split(':')
 155             # strip leading '#' sign from name
 156             library_id = library_id[1:]
 157             self.failUnlessEqual(library_id, expected_ids[i])
 158             self.failUnlessEqual(input_field['value'], library_id)
 159
 160     def test_library_to_flowcell_link(self):
 161         """
 162         Make sure the library page includes links to the flowcell pages.
 163         """
 164         self.client.login(username='supertest', password='BJOKL5kAj6aFZ6A5')
 165         response = self.client.get('/library/11070/')
 166         soup = BeautifulSoup(response.content)
 167         failed_fc_span = soup.find(text='30012AAXX (failed)')
 168         failed_fc_a = failed_fc_span.findPrevious('a')
 169         # make sure some of our RDF made it.
 170         self.failUnlessEqual(failed_fc_a.get('rel'), 'libns:flowcell')
 171         self.failUnlessEqual(failed_fc_a.get('href'), '/flowcell/30012AAXX/')
 172
 173
 174     def test_lanes_for(self):
 175         """
 176         Check the code that packs the django objects into simple types.
 177         """
 178         user = 'test'
 179         lanes = experiments.lanes_for(user)
 180         self.failUnlessEqual(len(lanes), 5)
 181
 182         response = self.client.get('/experiments/lanes_for/%s/json' % (user,), apidata)
 183         lanes_json = json.loads(response.content)
 184         self.failUnlessEqual(len(lanes), len(lanes_json))
 185         for i in range(len(lanes)):
 186             self.failUnlessEqual(lanes[i]['comment'], lanes_json[i]['comment'])
 187             self.failUnlessEqual(lanes[i]['lane_number'], lanes_json[i]['lane_number'])
 188             self.failUnlessEqual(lanes[i]['flowcell'], lanes_json[i]['flowcell'])
 189             self.failUnlessEqual(lanes[i]['run_date'], lanes_json[i]['run_date'])
 190
 191     def test_lanes_for_no_lanes(self):
 192         """
 193         Do we get something meaningful back when the user isn't attached to anything?
 194         """
 195         user = 'supertest'
 196         lanes = experiments.lanes_for(user)
 197         self.failUnlessEqual(len(lanes), 0)
 198
 199         response = self.client.get('/experiments/lanes_for/%s/json' % (user,), apidata)
 200         lanes_json = json.loads(response.content)
 201
 202     def test_lanes_for_no_user(self):
 203         """
 204         Do we get something meaningful back when its the wrong user
 205         """
 206         user = 'not a real user'
 207         self.failUnlessRaises(ObjectDoesNotExist, experiments.lanes_for, user)
 208
 209         response = self.client.get('/experiments/lanes_for/%s/json' % (user,), apidata)
 210         self.failUnlessEqual(response.status_code, 404)
 211
 212
 213     def test_raw_data_dir(self):
 214         """Raw data path generator check"""
 215         flowcell_id = self.fc1_id
 216         raw_dir = os.path.join(settings.RESULT_HOME_DIR, flowcell_id)
 217
 218         fc = models.FlowCell.objects.get(flowcell_id=flowcell_id)
 219         self.failUnlessEqual(fc.get_raw_data_directory(), raw_dir)
 220
 221         fc.flowcell_id = flowcell_id + " (failed)"
 222         self.failUnlessEqual(fc.get_raw_data_directory(), raw_dir)
 223
 224
 225     def test_data_run_import(self):
 226         srf_file_type = models.FileType.objects.get(name='SRF')
 227         runxml_file_type = models.FileType.objects.get(name='run_xml')
 228         flowcell_id = self.fc1_id
 229         flowcell = models.FlowCell.objects.get(flowcell_id=flowcell_id)
 230         flowcell.update_data_runs()
 231         self.failUnlessEqual(len(flowcell.datarun_set.all()), 1)
 232
 233         run = flowcell.datarun_set.all()[0]
 234         result_files = run.datafile_set.all()
 235         result_dict = dict(((rf.relative_pathname, rf) for rf in result_files))
 236
 237         srf4 = result_dict['FC12150/C1-37/woldlab_070829_SERIAL_FC12150_4.srf']
 238         self.failUnlessEqual(srf4.file_type, srf_file_type)
 239         self.failUnlessEqual(srf4.library_id, '11060')
 240         self.failUnlessEqual(srf4.data_run.flowcell.flowcell_id, 'FC12150')
 241         self.failUnlessEqual(
 242             srf4.data_run.flowcell.lane_set.get(lane_number=4).library_id,
 243             '11060')
 244         self.failUnlessEqual(
 245             srf4.pathname,
 246             os.path.join(settings.RESULT_HOME_DIR, srf4.relative_pathname))
 247
 248         lane_files = run.lane_files()
 249         self.failUnlessEqual(lane_files[4]['srf'], srf4)
 250
 251         runxml= result_dict['FC12150/C1-37/run_FC12150_2007-09-27.xml']
 252         self.failUnlessEqual(runxml.file_type, runxml_file_type)
 253         self.failUnlessEqual(runxml.library_id, None)
 254
 255
 256     def test_read_result_file(self):
 257         """make sure we can return a result file
 258         """
 259         flowcell_id = self.fc1_id
 260         flowcell = models.FlowCell.objects.get(flowcell_id=flowcell_id)
 261         flowcell.update_data_runs()
 262
 263         #self.client.login(username='supertest', password='BJOKL5kAj6aFZ6A5')
 264
 265         result_files = flowcell.datarun_set.all()[0].datafile_set.all()
 266         for f in result_files:
 267             url = '/experiments/file/%s' % ( f.random_key,)
 268             response = self.client.get(url)
 269             self.failUnlessEqual(response.status_code, 200)
 270             mimetype = f.file_type.mimetype
 271             if mimetype is None:
 272                 mimetype = 'application/octet-stream'
 273
 274             self.failUnlessEqual(mimetype, response['content-type'])
 275
 276 class TestFileType(TestCase):
 277     def test_file_type_unicode(self):
 278         file_type_objects = models.FileType.objects
 279         name = 'QSEQ tarfile'
 280         file_type_object = file_type_objects.get(name=name)
 281         self.failUnlessEqual(u"<FileType: QSEQ tarfile>",
 282                              unicode(file_type_object))
 283
 284 class TestFileType(TestCase):
 285     def test_find_file_type(self):
 286         file_type_objects = models.FileType.objects
 287         cases = [('woldlab_090921_HWUSI-EAS627_0009_42FC3AAXX_l7_r1.tar.bz2',
 288                   'QSEQ tarfile', 7, 1),
 289                  ('woldlab_091005_HWUSI-EAS627_0010_42JT2AAXX_1.srf',
 290                   'SRF', 1, None),
 291                  ('s_1_eland_extended.txt.bz2','ELAND Extended', 1, None),
 292                  ('s_7_eland_multi.txt.bz2', 'ELAND Multi', 7, None),
 293                  ('s_3_eland_result.txt.bz2','ELAND Result', 3, None),
 294                  ('s_1_export.txt.bz2','ELAND Export', 1, None),
 295                  ('s_1_percent_call.png', 'IVC Percent Call', 1, None),
 296                  ('s_2_percent_base.png', 'IVC Percent Base', 2, None),
 297                  ('s_3_percent_all.png', 'IVC Percent All', 3, None),
 298                  ('s_4_call.png', 'IVC Call', 4, None),
 299                  ('s_5_all.png', 'IVC All', 5, None),
 300                  ('Summary.htm', 'Summary.htm', None, None),
 301                  ('run_42JT2AAXX_2009-10-07.xml', 'run_xml', None, None),
 302          ]
 303         for filename, typename, lane, end in cases:
 304             ft = models.find_file_type_metadata_from_filename(filename)
 305             self.failUnlessEqual(ft['file_type'],
 306                                  file_type_objects.get(name=typename))
 307             self.failUnlessEqual(ft.get('lane', None), lane)
 308             self.failUnlessEqual(ft.get('end', None), end)
 309
 310     def test_assign_file_type_complex_path(self):
 311         file_type_objects = models.FileType.objects
 312         cases = [('/a/b/c/woldlab_090921_HWUSI-EAS627_0009_42FC3AAXX_l7_r1.tar.bz2',
 313                   'QSEQ tarfile', 7, 1),
 314                  ('foo/woldlab_091005_HWUSI-EAS627_0010_42JT2AAXX_1.srf',
 315                   'SRF', 1, None),
 316                  ('../s_1_eland_extended.txt.bz2','ELAND Extended', 1, None),
 317                  ('/bleem/s_7_eland_multi.txt.bz2', 'ELAND Multi', 7, None),
 318                  ('/qwer/s_3_eland_result.txt.bz2','ELAND Result', 3, None),
 319                  ('/ty///1/s_1_export.txt.bz2','ELAND Export', 1, None),
 320                  ('/help/s_1_percent_call.png', 'IVC Percent Call', 1, None),
 321                  ('/bored/s_2_percent_base.png', 'IVC Percent Base', 2, None),
 322                  ('/example1/s_3_percent_all.png', 'IVC Percent All', 3, None),
 323                  ('amonkey/s_4_call.png', 'IVC Call', 4, None),
 324                  ('fishie/s_5_all.png', 'IVC All', 5, None),
 325                  ('/random/Summary.htm', 'Summary.htm', None, None),
 326                  ('/notrandom/run_42JT2AAXX_2009-10-07.xml', 'run_xml', None, None),
 327          ]
 328         for filename, typename, lane, end in cases:
 329             result = models.find_file_type_metadata_from_filename(filename)
 330             self.failUnlessEqual(result['file_type'],
 331                                  file_type_objects.get(name=typename))
 332             self.failUnlessEqual(result.get('lane',None), lane)
 333             self.failUnlessEqual(result.get('end', None), end)
 334
 335 class TestEmailNotify(TestCase):
 336     fixtures = ['test_flowcells.json']
 337
 338     def test_started_email_not_logged_in(self):
 339         response = self.client.get('/experiments/started/153/')
 340         self.failUnlessEqual(response.status_code, 302)
 341
 342     def test_started_email_logged_in_user(self):
 343         self.client.login(username='test', password='BJOKL5kAj6aFZ6A5')
 344         response = self.client.get('/experiments/started/153/')
 345         self.failUnlessEqual(response.status_code, 302)
 346
 347     def test_started_email_logged_in_staff(self):
 348         self.client.login(username='admintest', password='BJOKL5kAj6aFZ6A5')
 349         response = self.client.get('/experiments/started/153/')
 350         self.failUnlessEqual(response.status_code, 200)
 351
 352     def test_started_email_send(self):
 353         self.client.login(username='admintest', password='BJOKL5kAj6aFZ6A5')
 354         response = self.client.get('/experiments/started/153/')
 355         self.failUnlessEqual(response.status_code, 200)
 356
 357         self.failUnless('pk1@example.com' in response.content)
 358         self.failUnless('Lane #8 : (11064) Paired ends 104' in response.content)
 359
 360         response = self.client.get('/experiments/started/153/', {'send':'1','bcc':'on'})
 361         self.failUnlessEqual(response.status_code, 200)
 362         self.failUnlessEqual(len(mail.outbox), 4)
 363         for m in mail.outbox:
 364             self.failUnless(len(m.body) > 0)
 365
 366     def test_email_navigation(self):
 367         """
 368         Can we navigate between the flowcell and email forms properly?
 369         """
 370         self.client.login(username='supertest', password='BJOKL5kAj6aFZ6A5')
 371         response = self.client.get('/experiments/started/153/')
 372         self.failUnlessEqual(response.status_code, 200)
 373         self.failUnless(re.search('Flowcell FC12150', response.content))
 374         # require that navigation back to the admin page exists
 375         self.failUnless(re.search('<a href="/admin/experiments/flowcell/153/">[^<]+</a>', response.content))
 376
 377