2 from lxml.html import fromstring
6 import simplejson as json
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
20 from htsworkflow.pipelines.test.simulate_runfolder import TESTDATA_DIR
24 NSMAP = {'libns':'http://jumpgate.caltech.edu/wiki/LibraryOntology#'}
26 class ClusterStationTestCases(TestCase):
27 fixtures = ['test_flowcells.json']
29 def test_default(self):
30 c = models.ClusterStation.default()
31 self.failUnlessEqual(c.id, 2)
36 total = models.ClusterStation.objects.filter(isdefault=True).count()
37 self.failUnlessEqual(total, 0)
39 other_default = models.ClusterStation.default()
40 self.failUnlessEqual(other_default.id, 3)
43 def test_update_default(self):
44 old_default = models.ClusterStation.default()
46 c = models.ClusterStation.objects.get(pk=3)
50 new_default = models.ClusterStation.default()
52 self.failUnless(old_default != new_default)
53 self.failUnlessEqual(new_default, c)
55 total = models.ClusterStation.objects.filter(isdefault=True).count()
56 self.failUnlessEqual(total, 1)
58 def test_update_other(self):
59 old_default = models.ClusterStation.default()
60 total = models.ClusterStation.objects.filter(isdefault=True).count()
61 self.failUnlessEqual(total, 1)
63 c = models.ClusterStation.objects.get(pk=1)
64 c.name = "Primary Key 1"
67 total = models.ClusterStation.objects.filter(isdefault=True).count()
68 self.failUnlessEqual(total, 1)
70 new_default = models.ClusterStation.default()
71 self.failUnlessEqual(old_default, new_default)
74 class SequencerTestCases(TestCase):
75 fixtures = ['test_flowcells.json']
77 def test_default(self):
78 # starting with no default
79 s = models.Sequencer.default()
80 self.failUnlessEqual(s.id, 2)
82 total = models.Sequencer.objects.filter(isdefault=True).count()
83 self.failUnlessEqual(total, 1)
88 total = models.Sequencer.objects.filter(isdefault=True).count()
89 self.failUnlessEqual(total, 0)
91 other_default = models.Sequencer.default()
92 self.failUnlessEqual(other_default.id, 7)
94 def test_update_default(self):
95 old_default = models.Sequencer.default()
97 s = models.Sequencer.objects.get(pk=1)
101 new_default = models.Sequencer.default()
103 self.failUnless(old_default != new_default)
104 self.failUnlessEqual(new_default, s)
106 total = models.Sequencer.objects.filter(isdefault=True).count()
107 self.failUnlessEqual(total, 1)
110 def test_update_other(self):
111 old_default = models.Sequencer.default()
112 total = models.Sequencer.objects.filter(isdefault=True).count()
113 self.failUnlessEqual(total, 1)
115 s = models.Sequencer.objects.get(pk=1)
116 s.name = "Primary Key 1"
119 total = models.Sequencer.objects.filter(isdefault=True).count()
120 self.failUnlessEqual(total, 1)
122 new_default = models.Sequencer.default()
123 self.failUnlessEqual(old_default, new_default)
126 class ExperimentsTestCases(TestCase):
127 fixtures = ['test_flowcells.json',
131 self.tempdir = tempfile.mkdtemp(prefix='htsw-test-experiments-')
132 settings.RESULT_HOME_DIR = self.tempdir
134 self.fc1_id = 'FC12150'
135 self.fc1_root = os.path.join(self.tempdir, self.fc1_id)
136 os.mkdir(self.fc1_root)
137 self.fc1_dir = os.path.join(self.fc1_root, 'C1-37')
138 os.mkdir(self.fc1_dir)
139 runxml = 'run_FC12150_2007-09-27.xml'
140 shutil.copy(os.path.join(TESTDATA_DIR, runxml),
141 os.path.join(self.fc1_dir, runxml))
144 os.path.join(TESTDATA_DIR,
145 'woldlab_070829_USI-EAS44_0017_FC11055_1.srf'),
146 os.path.join(self.fc1_dir,
147 'woldlab_070829_SERIAL_FC12150_%d.srf' %(i,))
150 self.fc2_dir = os.path.join(self.tempdir, '42JTNAAXX')
151 os.mkdir(self.fc2_dir)
152 os.mkdir(os.path.join(self.fc2_dir, 'C1-25'))
153 os.mkdir(os.path.join(self.fc2_dir, 'C1-37'))
154 os.mkdir(os.path.join(self.fc2_dir, 'C1-37', 'Plots'))
157 shutil.rmtree(self.tempdir)
159 def test_flowcell_information(self):
161 Check the code that packs the django objects into simple types.
163 for fc_id in [u'FC12150', u"42JTNAAXX", "42JU1AAXX"]:
164 fc_dict = experiments.flowcell_information(fc_id)
165 fc_django = models.FlowCell.objects.get(flowcell_id=fc_id)
166 self.failUnlessEqual(fc_dict['flowcell_id'], fc_id)
167 self.failUnlessEqual(fc_django.flowcell_id, fc_id)
168 self.failUnlessEqual(fc_dict['sequencer'], fc_django.sequencer.name)
169 self.failUnlessEqual(fc_dict['read_length'], fc_django.read_length)
170 self.failUnlessEqual(fc_dict['notes'], fc_django.notes)
171 self.failUnlessEqual(fc_dict['cluster_station'], fc_django.cluster_station.name)
173 for lane in fc_django.lane_set.all():
174 lane_contents = fc_dict['lane_set'][lane.lane_number]
175 lane_dict = multi_lane_to_dict(lane_contents)[lane.library_id]
176 self.failUnlessEqual(lane_dict['cluster_estimate'], lane.cluster_estimate)
177 self.failUnlessEqual(lane_dict['comment'], lane.comment)
178 self.failUnlessEqual(lane_dict['flowcell'], lane.flowcell.flowcell_id)
179 self.failUnlessEqual(lane_dict['lane_number'], lane.lane_number)
180 self.failUnlessEqual(lane_dict['library_name'], lane.library.library_name)
181 self.failUnlessEqual(lane_dict['library_id'], lane.library.id)
182 self.failUnlessAlmostEqual(float(lane_dict['pM']), float(lane.pM))
183 self.failUnlessEqual(lane_dict['library_species'],
184 lane.library.library_species.scientific_name)
186 response = self.client.get('/experiments/config/%s/json' % (fc_id,), apidata)
187 # strptime isoformat string = '%Y-%m-%dT%H:%M:%S'
188 fc_json = json.loads(response.content)
189 self.failUnlessEqual(fc_json['flowcell_id'], fc_id)
190 self.failUnlessEqual(fc_json['sequencer'], fc_django.sequencer.name)
191 self.failUnlessEqual(fc_json['read_length'], fc_django.read_length)
192 self.failUnlessEqual(fc_json['notes'], fc_django.notes)
193 self.failUnlessEqual(fc_json['cluster_station'], fc_django.cluster_station.name)
196 for lane in fc_django.lane_set.all():
197 lane_contents = fc_json['lane_set'][unicode(lane.lane_number)]
198 lane_dict = multi_lane_to_dict(lane_contents)[lane.library_id]
200 self.failUnlessEqual(lane_dict['cluster_estimate'], lane.cluster_estimate)
201 self.failUnlessEqual(lane_dict['comment'], lane.comment)
202 self.failUnlessEqual(lane_dict['flowcell'], lane.flowcell.flowcell_id)
203 self.failUnlessEqual(lane_dict['lane_number'], lane.lane_number)
204 self.failUnlessEqual(lane_dict['library_name'], lane.library.library_name)
205 self.failUnlessEqual(lane_dict['library_id'], lane.library.id)
206 self.failUnlessAlmostEqual(float(lane_dict['pM']), float(lane.pM))
207 self.failUnlessEqual(lane_dict['library_species'],
208 lane.library.library_species.scientific_name)
210 def test_invalid_flowcell(self):
212 Make sure we get a 404 if we request an invalid flowcell ID
214 response = self.client.get('/experiments/config/nottheone/json', apidata)
215 self.failUnlessEqual(response.status_code, 404)
217 def test_no_key(self):
219 Require logging in to retrieve meta data
221 response = self.client.get(u'/experiments/config/FC12150/json')
222 self.failUnlessEqual(response.status_code, 403)
224 def test_library_id(self):
226 Library IDs should be flexible, so make sure we can retrive a non-numeric ID
228 response = self.client.get('/experiments/config/FC12150/json', apidata)
229 self.failUnlessEqual(response.status_code, 200)
230 flowcell = json.loads(response.content)
232 lane_contents = flowcell['lane_set']['3']
233 lane_library = lane_contents[0]
234 self.failUnlessEqual(lane_library['library_id'], 'SL039')
236 response = self.client.get('/samples/library/SL039/json', apidata)
237 self.failUnlessEqual(response.status_code, 200)
238 library_sl039 = json.loads(response.content)
240 self.failUnlessEqual(library_sl039['library_id'], 'SL039')
242 def test_raw_id_field(self):
246 Library's have IDs, libraries also have primary keys,
247 we eventually had enough libraries that the drop down combo box was too
248 hard to filter through, unfortnately we want a field that uses our library
249 id and not the internal primary key, and raw_id_field uses primary keys.
251 This tests to make sure that the value entered in the raw library id field matches
252 the library id looked up.
254 expected_ids = [u'10981',u'11016',u'SL039',u'11060',
255 u'11061',u'11062',u'11063',u'11064']
256 self.client.login(username='supertest', password='BJOKL5kAj6aFZ6A5')
257 response = self.client.get('/admin/experiments/flowcell/153/')
258 tree = fromstring(response.content)
260 xpath_expression = '//input[@id="id_lane_set-%d-library"]'
261 input_field = tree.xpath(xpath_expression % (i,))[0]
262 library_field = input_field.find('../strong')
263 library_id, library_name = library_field.text.split(':')
264 # strip leading '#' sign from name
265 library_id = library_id[1:]
266 self.failUnlessEqual(library_id, expected_ids[i])
267 self.failUnlessEqual(input_field.attrib['value'], library_id)
269 def test_library_to_flowcell_link(self):
271 Make sure the library page includes links to the flowcell pages.
272 That work with flowcell IDs that have parenthetical comments.
274 self.client.login(username='supertest', password='BJOKL5kAj6aFZ6A5')
275 response = self.client.get('/library/11070/')
276 tree = fromstring(response.content)
277 flowcell_spans = tree.xpath('//span[@property="libns:flowcell_id"]',
279 self.assertEqual(flowcell_spans[0].text, '30012AAXX (failed)')
280 failed_fc_span = flowcell_spans[0]
281 failed_fc_a = failed_fc_span.getparent()
282 # make sure some of our RDF made it.
283 self.failUnlessEqual(failed_fc_a.get('rel'), 'libns:flowcell')
284 self.failUnlessEqual(failed_fc_a.get('href'), '/flowcell/30012AAXX/')
285 fc_response = self.client.get(failed_fc_a.get('href'))
286 self.failUnlessEqual(fc_response.status_code, 200)
287 fc_lane_response = self.client.get('/flowcell/30012AAXX/8/')
288 self.failUnlessEqual(fc_lane_response.status_code, 200)
290 def test_pooled_multiplex_id(self):
291 fc_dict = experiments.flowcell_information('42JU1AAXX')
292 lane_contents = fc_dict['lane_set'][3]
293 self.assertEqual(len(lane_contents), 2)
294 lane_dict = multi_lane_to_dict(lane_contents)
296 self.assertEqual(lane_dict['12044']['index_sequence'],
300 self.assertEqual(lane_dict['11045']['index_sequence'],
305 def test_lanes_for(self):
307 Check the code that packs the django objects into simple types.
310 lanes = experiments.lanes_for(user)
311 self.failUnlessEqual(len(lanes), 5)
313 response = self.client.get('/experiments/lanes_for/%s/json' % (user,), apidata)
314 lanes_json = json.loads(response.content)
315 self.failUnlessEqual(len(lanes), len(lanes_json))
316 for i in range(len(lanes)):
317 self.failUnlessEqual(lanes[i]['comment'], lanes_json[i]['comment'])
318 self.failUnlessEqual(lanes[i]['lane_number'], lanes_json[i]['lane_number'])
319 self.failUnlessEqual(lanes[i]['flowcell'], lanes_json[i]['flowcell'])
320 self.failUnlessEqual(lanes[i]['run_date'], lanes_json[i]['run_date'])
322 def test_lanes_for_no_lanes(self):
324 Do we get something meaningful back when the user isn't attached to anything?
327 lanes = experiments.lanes_for(user)
328 self.failUnlessEqual(len(lanes), 0)
330 response = self.client.get('/experiments/lanes_for/%s/json' % (user,), apidata)
331 lanes_json = json.loads(response.content)
333 def test_lanes_for_no_user(self):
335 Do we get something meaningful back when its the wrong user
337 user = 'not a real user'
338 self.failUnlessRaises(ObjectDoesNotExist, experiments.lanes_for, user)
340 response = self.client.get('/experiments/lanes_for/%s/json' % (user,), apidata)
341 self.failUnlessEqual(response.status_code, 404)
344 def test_raw_data_dir(self):
345 """Raw data path generator check"""
346 flowcell_id = self.fc1_id
347 raw_dir = os.path.join(settings.RESULT_HOME_DIR, flowcell_id)
349 fc = models.FlowCell.objects.get(flowcell_id=flowcell_id)
350 self.failUnlessEqual(fc.get_raw_data_directory(), raw_dir)
352 fc.flowcell_id = flowcell_id + " (failed)"
353 self.failUnlessEqual(fc.get_raw_data_directory(), raw_dir)
356 def test_data_run_import(self):
357 srf_file_type = models.FileType.objects.get(name='SRF')
358 runxml_file_type = models.FileType.objects.get(name='run_xml')
359 flowcell_id = self.fc1_id
360 flowcell = models.FlowCell.objects.get(flowcell_id=flowcell_id)
361 flowcell.update_data_runs()
362 self.failUnlessEqual(len(flowcell.datarun_set.all()), 1)
364 run = flowcell.datarun_set.all()[0]
365 result_files = run.datafile_set.all()
366 result_dict = dict(((rf.relative_pathname, rf) for rf in result_files))
368 srf4 = result_dict['FC12150/C1-37/woldlab_070829_SERIAL_FC12150_4.srf']
369 self.failUnlessEqual(srf4.file_type, srf_file_type)
370 self.failUnlessEqual(srf4.library_id, '11060')
371 self.failUnlessEqual(srf4.data_run.flowcell.flowcell_id, 'FC12150')
372 self.failUnlessEqual(
373 srf4.data_run.flowcell.lane_set.get(lane_number=4).library_id,
375 self.failUnlessEqual(
377 os.path.join(settings.RESULT_HOME_DIR, srf4.relative_pathname))
379 lane_files = run.lane_files()
380 self.failUnlessEqual(lane_files[4]['srf'], srf4)
382 runxml= result_dict['FC12150/C1-37/run_FC12150_2007-09-27.xml']
383 self.failUnlessEqual(runxml.file_type, runxml_file_type)
384 self.failUnlessEqual(runxml.library_id, None)
387 def test_read_result_file(self):
388 """make sure we can return a result file
390 flowcell_id = self.fc1_id
391 flowcell = models.FlowCell.objects.get(flowcell_id=flowcell_id)
392 flowcell.update_data_runs()
394 #self.client.login(username='supertest', password='BJOKL5kAj6aFZ6A5')
396 result_files = flowcell.datarun_set.all()[0].datafile_set.all()
397 for f in result_files:
398 url = '/experiments/file/%s' % ( f.random_key,)
399 response = self.client.get(url)
400 self.failUnlessEqual(response.status_code, 200)
401 mimetype = f.file_type.mimetype
403 mimetype = 'application/octet-stream'
405 self.failUnlessEqual(mimetype, response['content-type'])
407 class TestFileType(TestCase):
408 def test_file_type_unicode(self):
409 file_type_objects = models.FileType.objects
410 name = 'QSEQ tarfile'
411 file_type_object = file_type_objects.get(name=name)
412 self.failUnlessEqual(u"<FileType: QSEQ tarfile>",
413 unicode(file_type_object))
415 class TestFileType(TestCase):
416 def test_find_file_type(self):
417 file_type_objects = models.FileType.objects
418 cases = [('woldlab_090921_HWUSI-EAS627_0009_42FC3AAXX_l7_r1.tar.bz2',
419 'QSEQ tarfile', 7, 1),
420 ('woldlab_091005_HWUSI-EAS627_0010_42JT2AAXX_1.srf',
422 ('s_1_eland_extended.txt.bz2','ELAND Extended', 1, None),
423 ('s_7_eland_multi.txt.bz2', 'ELAND Multi', 7, None),
424 ('s_3_eland_result.txt.bz2','ELAND Result', 3, None),
425 ('s_1_export.txt.bz2','ELAND Export', 1, None),
426 ('s_1_percent_call.png', 'IVC Percent Call', 1, None),
427 ('s_2_percent_base.png', 'IVC Percent Base', 2, None),
428 ('s_3_percent_all.png', 'IVC Percent All', 3, None),
429 ('s_4_call.png', 'IVC Call', 4, None),
430 ('s_5_all.png', 'IVC All', 5, None),
431 ('Summary.htm', 'Summary.htm', None, None),
432 ('run_42JT2AAXX_2009-10-07.xml', 'run_xml', None, None),
434 for filename, typename, lane, end in cases:
435 ft = models.find_file_type_metadata_from_filename(filename)
436 self.failUnlessEqual(ft['file_type'],
437 file_type_objects.get(name=typename))
438 self.failUnlessEqual(ft.get('lane', None), lane)
439 self.failUnlessEqual(ft.get('end', None), end)
441 def test_assign_file_type_complex_path(self):
442 file_type_objects = models.FileType.objects
443 cases = [('/a/b/c/woldlab_090921_HWUSI-EAS627_0009_42FC3AAXX_l7_r1.tar.bz2',
444 'QSEQ tarfile', 7, 1),
445 ('foo/woldlab_091005_HWUSI-EAS627_0010_42JT2AAXX_1.srf',
447 ('../s_1_eland_extended.txt.bz2','ELAND Extended', 1, None),
448 ('/bleem/s_7_eland_multi.txt.bz2', 'ELAND Multi', 7, None),
449 ('/qwer/s_3_eland_result.txt.bz2','ELAND Result', 3, None),
450 ('/ty///1/s_1_export.txt.bz2','ELAND Export', 1, None),
451 ('/help/s_1_percent_call.png', 'IVC Percent Call', 1, None),
452 ('/bored/s_2_percent_base.png', 'IVC Percent Base', 2, None),
453 ('/example1/s_3_percent_all.png', 'IVC Percent All', 3, None),
454 ('amonkey/s_4_call.png', 'IVC Call', 4, None),
455 ('fishie/s_5_all.png', 'IVC All', 5, None),
456 ('/random/Summary.htm', 'Summary.htm', None, None),
457 ('/notrandom/run_42JT2AAXX_2009-10-07.xml', 'run_xml', None, None),
459 for filename, typename, lane, end in cases:
460 result = models.find_file_type_metadata_from_filename(filename)
461 self.failUnlessEqual(result['file_type'],
462 file_type_objects.get(name=typename))
463 self.failUnlessEqual(result.get('lane',None), lane)
464 self.failUnlessEqual(result.get('end', None), end)
466 class TestEmailNotify(TestCase):
467 fixtures = ['test_flowcells.json']
469 def test_started_email_not_logged_in(self):
470 response = self.client.get('/experiments/started/153/')
471 self.failUnlessEqual(response.status_code, 302)
473 def test_started_email_logged_in_user(self):
474 self.client.login(username='test', password='BJOKL5kAj6aFZ6A5')
475 response = self.client.get('/experiments/started/153/')
476 self.failUnlessEqual(response.status_code, 302)
478 def test_started_email_logged_in_staff(self):
479 self.client.login(username='admintest', password='BJOKL5kAj6aFZ6A5')
480 response = self.client.get('/experiments/started/153/')
481 self.failUnlessEqual(response.status_code, 200)
483 def test_started_email_send(self):
484 self.client.login(username='admintest', password='BJOKL5kAj6aFZ6A5')
485 response = self.client.get('/experiments/started/153/')
486 self.failUnlessEqual(response.status_code, 200)
488 self.failUnless('pk1@example.com' in response.content)
489 self.failUnless('Lane #8 : (11064) Paired ends 104' in response.content)
491 response = self.client.get('/experiments/started/153/', {'send':'1','bcc':'on'})
492 self.failUnlessEqual(response.status_code, 200)
493 self.failUnlessEqual(len(mail.outbox), 4)
494 for m in mail.outbox:
495 self.failUnless(len(m.body) > 0)
497 def test_email_navigation(self):
499 Can we navigate between the flowcell and email forms properly?
501 self.client.login(username='supertest', password='BJOKL5kAj6aFZ6A5')
502 response = self.client.get('/experiments/started/153/')
503 self.failUnlessEqual(response.status_code, 200)
504 self.failUnless(re.search('Flowcell FC12150', response.content))
505 # require that navigation back to the admin page exists
506 self.failUnless(re.search('<a href="/admin/experiments/flowcell/153/">[^<]+</a>', response.content))
508 def multi_lane_to_dict(lane):
509 """Convert a list of lane entries into a dictionary indexed by library ID
511 return dict( ((x['library_id'],x) for x in lane) )
513 class TestSequencer(TestCase):
514 fixtures = ['test_flowcells.json',
517 def test_name_generation(self):
518 seq = models.Sequencer()
520 seq.instrument_name = "HWI-SEQ1"
521 seq.model = "Imaginary 5000"
523 self.failUnlessEqual(unicode(seq), "Seq1 (HWI-SEQ1)")
525 def test_lookup(self):
526 fc = models.FlowCell.objects.get(pk=153)
527 self.failUnlessEqual(fc.sequencer.model,
528 "Illumina Genome Analyzer IIx")
529 self.failUnlessEqual(fc.sequencer.instrument_name,
533 response = self.client.get('/flowcell/FC12150/', apidata)
534 tree = fromstring(response.content)
535 divs = tree.xpath('//div[@rel="libns:sequenced_by"]',
537 self.failUnlessEqual(len(divs), 1)
538 self.failUnlessEqual(divs[0].attrib['rel'], 'libns:sequenced_by')
539 self.failUnlessEqual(divs[0].attrib['resource'], '/sequencer/2')
541 name = divs[0].xpath('./span[@property="libns:sequencer_name"]')
542 self.failUnlessEqual(len(name), 1)
543 self.failUnlessEqual(name[0].text, 'Tardigrade')
544 instrument = divs[0].xpath(
545 './span[@property="libns:sequencer_instrument"]')
546 self.failUnlessEqual(len(instrument), 1)
547 self.failUnlessEqual(instrument[0].text, 'ILLUMINA-EC5D15')
548 model = divs[0].xpath(
549 './span[@property="libns:sequencer_model"]')
550 self.failUnlessEqual(len(model), 1)
551 self.failUnlessEqual(model[0].text, 'Illumina Genome Analyzer IIx')