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.assertEqual(c.id, 2)
36 total = models.ClusterStation.objects.filter(isdefault=True).count()
37 self.assertEqual(total, 0)
39 other_default = models.ClusterStation.default()
40 self.assertEqual(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.assertNotEqual(old_default, new_default)
53 self.assertEqual(new_default, c)
55 total = models.ClusterStation.objects.filter(isdefault=True).count()
56 self.assertEqual(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.assertEqual(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.assertEqual(total, 1)
70 new_default = models.ClusterStation.default()
71 self.assertEqual(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.assertEqual(s.id, 2)
82 total = models.Sequencer.objects.filter(isdefault=True).count()
83 self.assertEqual(total, 1)
88 total = models.Sequencer.objects.filter(isdefault=True).count()
89 self.assertEqual(total, 0)
91 other_default = models.Sequencer.default()
92 self.assertEqual(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.assertNotEqual(old_default, new_default)
104 self.assertEqual(new_default, s)
106 total = models.Sequencer.objects.filter(isdefault=True).count()
107 self.assertEqual(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.assertEqual(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.assertEqual(total, 1)
122 new_default = models.Sequencer.default()
123 self.assertEqual(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.assertEqual(fc_dict['flowcell_id'], fc_id)
167 self.assertEqual(fc_django.flowcell_id, fc_id)
168 self.assertEqual(fc_dict['sequencer'], fc_django.sequencer.name)
169 self.assertEqual(fc_dict['read_length'], fc_django.read_length)
170 self.assertEqual(fc_dict['notes'], fc_django.notes)
171 self.assertEqual(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.assertEqual(lane_dict['cluster_estimate'], lane.cluster_estimate)
177 self.assertEqual(lane_dict['comment'], lane.comment)
178 self.assertEqual(lane_dict['flowcell'], lane.flowcell.flowcell_id)
179 self.assertEqual(lane_dict['lane_number'], lane.lane_number)
180 self.assertEqual(lane_dict['library_name'], lane.library.library_name)
181 self.assertEqual(lane_dict['library_id'], lane.library.id)
182 self.assertAlmostEqual(float(lane_dict['pM']), float(lane.pM))
183 self.assertEqual(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.assertEqual(fc_json['flowcell_id'], fc_id)
190 self.assertEqual(fc_json['sequencer'], fc_django.sequencer.name)
191 self.assertEqual(fc_json['read_length'], fc_django.read_length)
192 self.assertEqual(fc_json['notes'], fc_django.notes)
193 self.assertEqual(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.assertEqual(lane_dict['cluster_estimate'], lane.cluster_estimate)
201 self.assertEqual(lane_dict['comment'], lane.comment)
202 self.assertEqual(lane_dict['flowcell'], lane.flowcell.flowcell_id)
203 self.assertEqual(lane_dict['lane_number'], lane.lane_number)
204 self.assertEqual(lane_dict['library_name'], lane.library.library_name)
205 self.assertEqual(lane_dict['library_id'], lane.library.id)
206 self.assertAlmostEqual(float(lane_dict['pM']), float(lane.pM))
207 self.assertEqual(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.assertEqual(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.assertEqual(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.assertEqual(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.assertEqual(lane_library['library_id'], 'SL039')
236 response = self.client.get('/samples/library/SL039/json', apidata)
237 self.assertEqual(response.status_code, 200)
238 library_sl039 = json.loads(response.content)
240 self.assertEqual(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.assertEqual(library_id, expected_ids[i])
267 self.assertEqual(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.assertEqual(failed_fc_a.get('rel'), 'libns:flowcell')
284 self.assertEqual(failed_fc_a.get('href'), '/flowcell/30012AAXX/')
285 fc_response = self.client.get(failed_fc_a.get('href'))
286 self.assertEqual(fc_response.status_code, 200)
287 fc_lane_response = self.client.get('/flowcell/30012AAXX/8/')
288 self.assertEqual(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.assertEqual(len(lanes), 5)
313 response = self.client.get('/experiments/lanes_for/%s/json' % (user,), apidata)
314 lanes_json = json.loads(response.content)
315 self.assertEqual(len(lanes), len(lanes_json))
316 for i in range(len(lanes)):
317 self.assertEqual(lanes[i]['comment'], lanes_json[i]['comment'])
318 self.assertEqual(lanes[i]['lane_number'], lanes_json[i]['lane_number'])
319 self.assertEqual(lanes[i]['flowcell'], lanes_json[i]['flowcell'])
320 self.assertEqual(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.assertEqual(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.assertRaises(ObjectDoesNotExist, experiments.lanes_for, user)
340 response = self.client.get('/experiments/lanes_for/%s/json' % (user,), apidata)
341 self.assertEqual(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.assertEqual(fc.get_raw_data_directory(), raw_dir)
352 fc.flowcell_id = flowcell_id + " (failed)"
353 self.assertEqual(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.assertEqual(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.assertEqual(srf4.file_type, srf_file_type)
370 self.assertEqual(srf4.library_id, '11060')
371 self.assertEqual(srf4.data_run.flowcell.flowcell_id, 'FC12150')
373 srf4.data_run.flowcell.lane_set.get(lane_number=4).library_id,
377 os.path.join(settings.RESULT_HOME_DIR, srf4.relative_pathname))
379 lane_files = run.lane_files()
380 self.assertEqual(lane_files[4]['srf'], srf4)
382 runxml= result_dict['FC12150/C1-37/run_FC12150_2007-09-27.xml']
383 self.assertEqual(runxml.file_type, runxml_file_type)
384 self.assertEqual(runxml.library_id, None)
386 import1 = len(models.DataRun.objects.filter(result_dir='FC12150/C1-37'))
387 # what happens if we import twice?
388 flowcell.import_data_run('FC12150/C1-37',
389 'run_FC12150_2007-09-27.xml')
391 len(models.DataRun.objects.filter(result_dir='FC12150/C1-37')),
394 def test_read_result_file(self):
395 """make sure we can return a result file
397 flowcell_id = self.fc1_id
398 flowcell = models.FlowCell.objects.get(flowcell_id=flowcell_id)
399 flowcell.update_data_runs()
401 #self.client.login(username='supertest', password='BJOKL5kAj6aFZ6A5')
403 result_files = flowcell.datarun_set.all()[0].datafile_set.all()
404 for f in result_files:
405 url = '/experiments/file/%s' % ( f.random_key,)
406 response = self.client.get(url)
407 self.assertEqual(response.status_code, 200)
408 mimetype = f.file_type.mimetype
410 mimetype = 'application/octet-stream'
412 self.assertEqual(mimetype, response['content-type'])
414 class TestFileType(TestCase):
415 def test_file_type_unicode(self):
416 file_type_objects = models.FileType.objects
417 name = 'QSEQ tarfile'
418 file_type_object = file_type_objects.get(name=name)
419 self.assertEqual(u"<FileType: QSEQ tarfile>",
420 unicode(file_type_object))
422 class TestFileType(TestCase):
423 def test_find_file_type(self):
424 file_type_objects = models.FileType.objects
425 cases = [('woldlab_090921_HWUSI-EAS627_0009_42FC3AAXX_l7_r1.tar.bz2',
426 'QSEQ tarfile', 7, 1),
427 ('woldlab_091005_HWUSI-EAS627_0010_42JT2AAXX_1.srf',
429 ('s_1_eland_extended.txt.bz2','ELAND Extended', 1, None),
430 ('s_7_eland_multi.txt.bz2', 'ELAND Multi', 7, None),
431 ('s_3_eland_result.txt.bz2','ELAND Result', 3, None),
432 ('s_1_export.txt.bz2','ELAND Export', 1, None),
433 ('s_1_percent_call.png', 'IVC Percent Call', 1, None),
434 ('s_2_percent_base.png', 'IVC Percent Base', 2, None),
435 ('s_3_percent_all.png', 'IVC Percent All', 3, None),
436 ('s_4_call.png', 'IVC Call', 4, None),
437 ('s_5_all.png', 'IVC All', 5, None),
438 ('Summary.htm', 'Summary.htm', None, None),
439 ('run_42JT2AAXX_2009-10-07.xml', 'run_xml', None, None),
441 for filename, typename, lane, end in cases:
442 ft = models.find_file_type_metadata_from_filename(filename)
443 self.assertEqual(ft['file_type'],
444 file_type_objects.get(name=typename))
445 self.assertEqual(ft.get('lane', None), lane)
446 self.assertEqual(ft.get('end', None), end)
448 def test_assign_file_type_complex_path(self):
449 file_type_objects = models.FileType.objects
450 cases = [('/a/b/c/woldlab_090921_HWUSI-EAS627_0009_42FC3AAXX_l7_r1.tar.bz2',
451 'QSEQ tarfile', 7, 1),
452 ('foo/woldlab_091005_HWUSI-EAS627_0010_42JT2AAXX_1.srf',
454 ('../s_1_eland_extended.txt.bz2','ELAND Extended', 1, None),
455 ('/bleem/s_7_eland_multi.txt.bz2', 'ELAND Multi', 7, None),
456 ('/qwer/s_3_eland_result.txt.bz2','ELAND Result', 3, None),
457 ('/ty///1/s_1_export.txt.bz2','ELAND Export', 1, None),
458 ('/help/s_1_percent_call.png', 'IVC Percent Call', 1, None),
459 ('/bored/s_2_percent_base.png', 'IVC Percent Base', 2, None),
460 ('/example1/s_3_percent_all.png', 'IVC Percent All', 3, None),
461 ('amonkey/s_4_call.png', 'IVC Call', 4, None),
462 ('fishie/s_5_all.png', 'IVC All', 5, None),
463 ('/random/Summary.htm', 'Summary.htm', None, None),
464 ('/notrandom/run_42JT2AAXX_2009-10-07.xml', 'run_xml', None, None),
466 for filename, typename, lane, end in cases:
467 result = models.find_file_type_metadata_from_filename(filename)
468 self.assertEqual(result['file_type'],
469 file_type_objects.get(name=typename))
470 self.assertEqual(result.get('lane',None), lane)
471 self.assertEqual(result.get('end', None), end)
473 class TestEmailNotify(TestCase):
474 fixtures = ['test_flowcells.json']
476 def test_started_email_not_logged_in(self):
477 response = self.client.get('/experiments/started/153/')
478 self.assertEqual(response.status_code, 302)
480 def test_started_email_logged_in_user(self):
481 self.client.login(username='test', password='BJOKL5kAj6aFZ6A5')
482 response = self.client.get('/experiments/started/153/')
483 self.assertEqual(response.status_code, 302)
485 def test_started_email_logged_in_staff(self):
486 self.client.login(username='admintest', password='BJOKL5kAj6aFZ6A5')
487 response = self.client.get('/experiments/started/153/')
488 self.assertEqual(response.status_code, 200)
490 def test_started_email_send(self):
491 self.client.login(username='admintest', password='BJOKL5kAj6aFZ6A5')
492 response = self.client.get('/experiments/started/153/')
493 self.assertEqual(response.status_code, 200)
495 self.assertTrue('pk1@example.com' in response.content)
496 self.assertTrue('Lane #8 : (11064) Paired ends 104' in response.content)
498 response = self.client.get('/experiments/started/153/', {'send':'1','bcc':'on'})
499 self.assertEqual(response.status_code, 200)
500 self.assertEqual(len(mail.outbox), 4)
501 for m in mail.outbox:
502 self.assertTrue(len(m.body) > 0)
504 def test_email_navigation(self):
506 Can we navigate between the flowcell and email forms properly?
508 self.client.login(username='supertest', password='BJOKL5kAj6aFZ6A5')
509 response = self.client.get('/experiments/started/153/')
510 self.assertEqual(response.status_code, 200)
511 self.assertTrue(re.search('Flowcell FC12150', response.content))
512 # require that navigation back to the admin page exists
513 self.assertTrue(re.search('<a href="/admin/experiments/flowcell/153/">[^<]+</a>', response.content))
515 def multi_lane_to_dict(lane):
516 """Convert a list of lane entries into a dictionary indexed by library ID
518 return dict( ((x['library_id'],x) for x in lane) )
520 class TestSequencer(TestCase):
521 fixtures = ['test_flowcells.json',
524 def test_name_generation(self):
525 seq = models.Sequencer()
527 seq.instrument_name = "HWI-SEQ1"
528 seq.model = "Imaginary 5000"
530 self.assertEqual(unicode(seq), "Seq1 (HWI-SEQ1)")
532 def test_lookup(self):
533 fc = models.FlowCell.objects.get(pk=153)
534 self.assertEqual(fc.sequencer.model,
535 "Illumina Genome Analyzer IIx")
536 self.assertEqual(fc.sequencer.instrument_name,
540 response = self.client.get('/flowcell/FC12150/', apidata)
541 tree = fromstring(response.content)
542 divs = tree.xpath('//div[@rel="libns:sequenced_by"]',
544 self.assertEqual(len(divs), 1)
545 self.assertEqual(divs[0].attrib['rel'], 'libns:sequenced_by')
546 self.assertEqual(divs[0].attrib['resource'], '/sequencer/2')
548 name = divs[0].xpath('./span[@property="libns:sequencer_name"]')
549 self.assertEqual(len(name), 1)
550 self.assertEqual(name[0].text, 'Tardigrade')
551 instrument = divs[0].xpath(
552 './span[@property="libns:sequencer_instrument"]')
553 self.assertEqual(len(instrument), 1)
554 self.assertEqual(instrument[0].text, 'ILLUMINA-EC5D15')
555 model = divs[0].xpath(
556 './span[@property="libns:sequencer_model"]')
557 self.assertEqual(len(model), 1)
558 self.assertEqual(model[0].text, 'Illumina Genome Analyzer IIx')