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         That work with flowcell IDs that have parenthetical comments.
 164         """
 165         self.client.login(username='supertest', password='BJOKL5kAj6aFZ6A5')
 166         response = self.client.get('/library/11070/')
 167         soup = BeautifulSoup(response.content)
 168         failed_fc_span = soup.find(text='30012AAXX (failed)')
 169         failed_fc_a = failed_fc_span.findPrevious('a')
 170         # make sure some of our RDF made it.
 171         self.failUnlessEqual(failed_fc_a.get('rel'), 'libns:flowcell')
 172         self.failUnlessEqual(failed_fc_a.get('href'), '/flowcell/30012AAXX/')
 173         fc_response = self.client.get(failed_fc_a.get('href'))
 174         self.failUnlessEqual(fc_response.status_code, 200)
 175         fc_lane_response = self.client.get('/flowcell/30012AAXX/8/')
 176         self.failUnlessEqual(fc_lane_response.status_code, 200)
 177
 178
 179
 180     def test_lanes_for(self):
 181         """
 182         Check the code that packs the django objects into simple types.
 183         """
 184         user = 'test'
 185         lanes = experiments.lanes_for(user)
 186         self.failUnlessEqual(len(lanes), 5)
 187
 188         response = self.client.get('/experiments/lanes_for/%s/json' % (user,), apidata)
 189         lanes_json = json.loads(response.content)
 190         self.failUnlessEqual(len(lanes), len(lanes_json))
 191         for i in range(len(lanes)):
 192             self.failUnlessEqual(lanes[i]['comment'], lanes_json[i]['comment'])
 193             self.failUnlessEqual(lanes[i]['lane_number'], lanes_json[i]['lane_number'])
 194             self.failUnlessEqual(lanes[i]['flowcell'], lanes_json[i]['flowcell'])
 195             self.failUnlessEqual(lanes[i]['run_date'], lanes_json[i]['run_date'])
 196
 197     def test_lanes_for_no_lanes(self):
 198         """
 199         Do we get something meaningful back when the user isn't attached to anything?
 200         """
 201         user = 'supertest'
 202         lanes = experiments.lanes_for(user)
 203         self.failUnlessEqual(len(lanes), 0)
 204
 205         response = self.client.get('/experiments/lanes_for/%s/json' % (user,), apidata)
 206         lanes_json = json.loads(response.content)
 207
 208     def test_lanes_for_no_user(self):
 209         """
 210         Do we get something meaningful back when its the wrong user
 211         """
 212         user = 'not a real user'
 213         self.failUnlessRaises(ObjectDoesNotExist, experiments.lanes_for, user)
 214
 215         response = self.client.get('/experiments/lanes_for/%s/json' % (user,), apidata)
 216         self.failUnlessEqual(response.status_code, 404)
 217
 218
 219     def test_raw_data_dir(self):
 220         """Raw data path generator check"""
 221         flowcell_id = self.fc1_id
 222         raw_dir = os.path.join(settings.RESULT_HOME_DIR, flowcell_id)
 223
 224         fc = models.FlowCell.objects.get(flowcell_id=flowcell_id)
 225         self.failUnlessEqual(fc.get_raw_data_directory(), raw_dir)
 226
 227         fc.flowcell_id = flowcell_id + " (failed)"
 228         self.failUnlessEqual(fc.get_raw_data_directory(), raw_dir)
 229
 230
 231     def test_data_run_import(self):
 232         srf_file_type = models.FileType.objects.get(name='SRF')
 233         runxml_file_type = models.FileType.objects.get(name='run_xml')
 234         flowcell_id = self.fc1_id
 235         flowcell = models.FlowCell.objects.get(flowcell_id=flowcell_id)
 236         flowcell.update_data_runs()
 237         self.failUnlessEqual(len(flowcell.datarun_set.all()), 1)
 238
 239         run = flowcell.datarun_set.all()[0]
 240         result_files = run.datafile_set.all()
 241         result_dict = dict(((rf.relative_pathname, rf) for rf in result_files))
 242
 243         srf4 = result_dict['FC12150/C1-37/woldlab_070829_SERIAL_FC12150_4.srf']
 244         self.failUnlessEqual(srf4.file_type, srf_file_type)
 245         self.failUnlessEqual(srf4.library_id, '11060')
 246         self.failUnlessEqual(srf4.data_run.flowcell.flowcell_id, 'FC12150')
 247         self.failUnlessEqual(
 248             srf4.data_run.flowcell.lane_set.get(lane_number=4).library_id,
 249             '11060')
 250         self.failUnlessEqual(
 251             srf4.pathname,
 252             os.path.join(settings.RESULT_HOME_DIR, srf4.relative_pathname))
 253
 254         lane_files = run.lane_files()
 255         self.failUnlessEqual(lane_files[4]['srf'], srf4)
 256
 257         runxml= result_dict['FC12150/C1-37/run_FC12150_2007-09-27.xml']
 258         self.failUnlessEqual(runxml.file_type, runxml_file_type)
 259         self.failUnlessEqual(runxml.library_id, None)
 260
 261
 262     def test_read_result_file(self):
 263         """make sure we can return a result file
 264         """
 265         flowcell_id = self.fc1_id
 266         flowcell = models.FlowCell.objects.get(flowcell_id=flowcell_id)
 267         flowcell.update_data_runs()
 268
 269         #self.client.login(username='supertest', password='BJOKL5kAj6aFZ6A5')
 270
 271         result_files = flowcell.datarun_set.all()[0].datafile_set.all()
 272         for f in result_files:
 273             url = '/experiments/file/%s' % ( f.random_key,)
 274             response = self.client.get(url)
 275             self.failUnlessEqual(response.status_code, 200)
 276             mimetype = f.file_type.mimetype
 277             if mimetype is None:
 278                 mimetype = 'application/octet-stream'
 279
 280             self.failUnlessEqual(mimetype, response['content-type'])
 281
 282 class TestFileType(TestCase):
 283     def test_file_type_unicode(self):
 284         file_type_objects = models.FileType.objects
 285         name = 'QSEQ tarfile'
 286         file_type_object = file_type_objects.get(name=name)
 287         self.failUnlessEqual(u"<FileType: QSEQ tarfile>",
 288                              unicode(file_type_object))
 289
 290 class TestFileType(TestCase):
 291     def test_find_file_type(self):
 292         file_type_objects = models.FileType.objects
 293         cases = [('woldlab_090921_HWUSI-EAS627_0009_42FC3AAXX_l7_r1.tar.bz2',
 294                   'QSEQ tarfile', 7, 1),
 295                  ('woldlab_091005_HWUSI-EAS627_0010_42JT2AAXX_1.srf',
 296                   'SRF', 1, None),
 297                  ('s_1_eland_extended.txt.bz2','ELAND Extended', 1, None),
 298                  ('s_7_eland_multi.txt.bz2', 'ELAND Multi', 7, None),
 299                  ('s_3_eland_result.txt.bz2','ELAND Result', 3, None),
 300                  ('s_1_export.txt.bz2','ELAND Export', 1, None),
 301                  ('s_1_percent_call.png', 'IVC Percent Call', 1, None),
 302                  ('s_2_percent_base.png', 'IVC Percent Base', 2, None),
 303                  ('s_3_percent_all.png', 'IVC Percent All', 3, None),
 304                  ('s_4_call.png', 'IVC Call', 4, None),
 305                  ('s_5_all.png', 'IVC All', 5, None),
 306                  ('Summary.htm', 'Summary.htm', None, None),
 307                  ('run_42JT2AAXX_2009-10-07.xml', 'run_xml', None, None),
 308          ]
 309         for filename, typename, lane, end in cases:
 310             ft = models.find_file_type_metadata_from_filename(filename)
 311             self.failUnlessEqual(ft['file_type'],
 312                                  file_type_objects.get(name=typename))
 313             self.failUnlessEqual(ft.get('lane', None), lane)
 314             self.failUnlessEqual(ft.get('end', None), end)
 315
 316     def test_assign_file_type_complex_path(self):
 317         file_type_objects = models.FileType.objects
 318         cases = [('/a/b/c/woldlab_090921_HWUSI-EAS627_0009_42FC3AAXX_l7_r1.tar.bz2',
 319                   'QSEQ tarfile', 7, 1),
 320                  ('foo/woldlab_091005_HWUSI-EAS627_0010_42JT2AAXX_1.srf',
 321                   'SRF', 1, None),
 322                  ('../s_1_eland_extended.txt.bz2','ELAND Extended', 1, None),
 323                  ('/bleem/s_7_eland_multi.txt.bz2', 'ELAND Multi', 7, None),
 324                  ('/qwer/s_3_eland_result.txt.bz2','ELAND Result', 3, None),
 325                  ('/ty///1/s_1_export.txt.bz2','ELAND Export', 1, None),
 326                  ('/help/s_1_percent_call.png', 'IVC Percent Call', 1, None),
 327                  ('/bored/s_2_percent_base.png', 'IVC Percent Base', 2, None),
 328                  ('/example1/s_3_percent_all.png', 'IVC Percent All', 3, None),
 329                  ('amonkey/s_4_call.png', 'IVC Call', 4, None),
 330                  ('fishie/s_5_all.png', 'IVC All', 5, None),
 331                  ('/random/Summary.htm', 'Summary.htm', None, None),
 332                  ('/notrandom/run_42JT2AAXX_2009-10-07.xml', 'run_xml', None, None),
 333          ]
 334         for filename, typename, lane, end in cases:
 335             result = models.find_file_type_metadata_from_filename(filename)
 336             self.failUnlessEqual(result['file_type'],
 337                                  file_type_objects.get(name=typename))
 338             self.failUnlessEqual(result.get('lane',None), lane)
 339             self.failUnlessEqual(result.get('end', None), end)
 340
 341 class TestEmailNotify(TestCase):
 342     fixtures = ['test_flowcells.json']
 343
 344     def test_started_email_not_logged_in(self):
 345         response = self.client.get('/experiments/started/153/')
 346         self.failUnlessEqual(response.status_code, 302)
 347
 348     def test_started_email_logged_in_user(self):
 349         self.client.login(username='test', password='BJOKL5kAj6aFZ6A5')
 350         response = self.client.get('/experiments/started/153/')
 351         self.failUnlessEqual(response.status_code, 302)
 352
 353     def test_started_email_logged_in_staff(self):
 354         self.client.login(username='admintest', password='BJOKL5kAj6aFZ6A5')
 355         response = self.client.get('/experiments/started/153/')
 356         self.failUnlessEqual(response.status_code, 200)
 357
 358     def test_started_email_send(self):
 359         self.client.login(username='admintest', password='BJOKL5kAj6aFZ6A5')
 360         response = self.client.get('/experiments/started/153/')
 361         self.failUnlessEqual(response.status_code, 200)
 362
 363         self.failUnless('pk1@example.com' in response.content)
 364         self.failUnless('Lane #8 : (11064) Paired ends 104' in response.content)
 365
 366         response = self.client.get('/experiments/started/153/', {'send':'1','bcc':'on'})
 367         self.failUnlessEqual(response.status_code, 200)
 368         self.failUnlessEqual(len(mail.outbox), 4)
 369         for m in mail.outbox:
 370             self.failUnless(len(m.body) > 0)
 371
 372     def test_email_navigation(self):
 373         """
 374         Can we navigate between the flowcell and email forms properly?
 375         """
 376         self.client.login(username='supertest', password='BJOKL5kAj6aFZ6A5')
 377         response = self.client.get('/experiments/started/153/')
 378         self.failUnlessEqual(response.status_code, 200)
 379         self.failUnless(re.search('Flowcell FC12150', response.content))
 380         # require that navigation back to the admin page exists
 381         self.failUnless(re.search('<a href="/admin/experiments/flowcell/153/">[^<]+</a>', response.content))
 382
 383