2 from lxml.html import fromstring
6 import simplejson as json
7 from datetime import timedelta
12 from urlparse import urljoin
14 from django.conf import settings
15 from django.core import mail
16 from django.core.exceptions import ObjectDoesNotExist
17 from django.test import TestCase
18 from django.test.utils import setup_test_environment, teardown_test_environment
19 from django.db import connection
20 from django.conf import settings
21 from htsworkflow.frontend.experiments import models
22 from htsworkflow.frontend.experiments import experiments
23 from htsworkflow.frontend.auth import apidata
24 from htsworkflow.util.ethelp import validate_xhtml
26 from htsworkflow.pipelines.test.simulate_runfolder import TESTDATA_DIR
30 NSMAP = {'libns':'http://jumpgate.caltech.edu/wiki/LibraryOntology#'}
32 from django.db import connection
34 class ClusterStationTestCases(TestCase):
35 fixtures = ['initial_data.json',
36 'test_flowcells.json']
38 def test_default(self):
39 c = models.ClusterStation.default()
40 self.assertEqual(c.id, 2)
45 total = models.ClusterStation.objects.filter(isdefault=True).count()
46 self.assertEqual(total, 0)
48 other_default = models.ClusterStation.default()
49 self.assertEqual(other_default.id, 3)
52 def test_update_default(self):
53 old_default = models.ClusterStation.default()
55 c = models.ClusterStation.objects.get(pk=3)
59 new_default = models.ClusterStation.default()
61 self.assertNotEqual(old_default, new_default)
62 self.assertEqual(new_default, c)
64 total = models.ClusterStation.objects.filter(isdefault=True).count()
65 self.assertEqual(total, 1)
67 def test_update_other(self):
68 old_default = models.ClusterStation.default()
69 total = models.ClusterStation.objects.filter(isdefault=True).count()
70 self.assertEqual(total, 1)
72 c = models.ClusterStation.objects.get(pk=1)
73 c.name = "Primary Key 1"
76 total = models.ClusterStation.objects.filter(isdefault=True).count()
77 self.assertEqual(total, 1)
79 new_default = models.ClusterStation.default()
80 self.assertEqual(old_default, new_default)
83 class SequencerTestCases(TestCase):
84 fixtures = ['initial_data.json',
85 'test_flowcells.json']
87 def test_default(self):
88 # starting with no default
89 s = models.Sequencer.default()
90 self.assertEqual(s.id, 2)
92 total = models.Sequencer.objects.filter(isdefault=True).count()
93 self.assertEqual(total, 1)
98 total = models.Sequencer.objects.filter(isdefault=True).count()
99 self.assertEqual(total, 0)
101 other_default = models.Sequencer.default()
102 self.assertEqual(other_default.id, 7)
104 def test_update_default(self):
105 old_default = models.Sequencer.default()
107 s = models.Sequencer.objects.get(pk=1)
111 new_default = models.Sequencer.default()
113 self.assertNotEqual(old_default, new_default)
114 self.assertEqual(new_default, s)
116 total = models.Sequencer.objects.filter(isdefault=True).count()
117 self.assertEqual(total, 1)
120 def test_update_other(self):
121 old_default = models.Sequencer.default()
122 total = models.Sequencer.objects.filter(isdefault=True).count()
123 self.assertEqual(total, 1)
125 s = models.Sequencer.objects.get(pk=1)
126 s.name = "Primary Key 1"
129 total = models.Sequencer.objects.filter(isdefault=True).count()
130 self.assertEqual(total, 1)
132 new_default = models.Sequencer.default()
133 self.assertEqual(old_default, new_default)
136 class ExperimentsTestCases(TestCase):
137 fixtures = ['initial_data.json',
138 'test_flowcells.json',
142 self.tempdir = tempfile.mkdtemp(prefix='htsw-test-experiments-')
143 settings.RESULT_HOME_DIR = self.tempdir
145 self.fc1_id = 'FC12150'
146 self.fc1_root = os.path.join(self.tempdir, self.fc1_id)
147 os.mkdir(self.fc1_root)
148 self.fc1_dir = os.path.join(self.fc1_root, 'C1-37')
149 os.mkdir(self.fc1_dir)
150 runxml = 'run_FC12150_2007-09-27.xml'
151 shutil.copy(os.path.join(TESTDATA_DIR, runxml),
152 os.path.join(self.fc1_dir, runxml))
155 os.path.join(TESTDATA_DIR,
156 'woldlab_070829_USI-EAS44_0017_FC11055_1.srf'),
157 os.path.join(self.fc1_dir,
158 'woldlab_070829_SERIAL_FC12150_%d.srf' %(i,))
161 self.fc2_dir = os.path.join(self.tempdir, '42JTNAAXX')
162 os.mkdir(self.fc2_dir)
163 os.mkdir(os.path.join(self.fc2_dir, 'C1-25'))
164 os.mkdir(os.path.join(self.fc2_dir, 'C1-37'))
165 os.mkdir(os.path.join(self.fc2_dir, 'C1-37', 'Plots'))
168 shutil.rmtree(self.tempdir)
170 def test_flowcell_information(self):
172 Check the code that packs the django objects into simple types.
174 for fc_id in [u'FC12150', u"42JTNAAXX", "42JU1AAXX"]:
175 fc_dict = experiments.flowcell_information(fc_id)
176 fc_django = models.FlowCell.objects.get(flowcell_id=fc_id)
177 self.assertEqual(fc_dict['flowcell_id'], fc_id)
178 self.assertEqual(fc_django.flowcell_id, fc_id)
179 self.assertEqual(fc_dict['sequencer'], fc_django.sequencer.name)
180 self.assertEqual(fc_dict['read_length'], fc_django.read_length)
181 self.assertEqual(fc_dict['notes'], fc_django.notes)
182 self.assertEqual(fc_dict['cluster_station'], fc_django.cluster_station.name)
184 for lane in fc_django.lane_set.all():
185 lane_contents = fc_dict['lane_set'][lane.lane_number]
186 lane_dict = multi_lane_to_dict(lane_contents)[lane.library_id]
187 self.assertEqual(lane_dict['cluster_estimate'], lane.cluster_estimate)
188 self.assertEqual(lane_dict['comment'], lane.comment)
189 self.assertEqual(lane_dict['flowcell'], lane.flowcell.flowcell_id)
190 self.assertEqual(lane_dict['lane_number'], lane.lane_number)
191 self.assertEqual(lane_dict['library_name'], lane.library.library_name)
192 self.assertEqual(lane_dict['library_id'], lane.library.id)
193 self.assertAlmostEqual(float(lane_dict['pM']), float(lane.pM))
194 self.assertEqual(lane_dict['library_species'],
195 lane.library.library_species.scientific_name)
197 response = self.client.get('/experiments/config/%s/json' % (fc_id,), apidata)
198 # strptime isoformat string = '%Y-%m-%dT%H:%M:%S'
199 fc_json = json.loads(response.content)
200 self.assertEqual(fc_json['flowcell_id'], fc_id)
201 self.assertEqual(fc_json['sequencer'], fc_django.sequencer.name)
202 self.assertEqual(fc_json['read_length'], fc_django.read_length)
203 self.assertEqual(fc_json['notes'], fc_django.notes)
204 self.assertEqual(fc_json['cluster_station'], fc_django.cluster_station.name)
207 for lane in fc_django.lane_set.all():
208 lane_contents = fc_json['lane_set'][unicode(lane.lane_number)]
209 lane_dict = multi_lane_to_dict(lane_contents)[lane.library_id]
211 self.assertEqual(lane_dict['cluster_estimate'], lane.cluster_estimate)
212 self.assertEqual(lane_dict['comment'], lane.comment)
213 self.assertEqual(lane_dict['flowcell'], lane.flowcell.flowcell_id)
214 self.assertEqual(lane_dict['lane_number'], lane.lane_number)
215 self.assertEqual(lane_dict['library_name'], lane.library.library_name)
216 self.assertEqual(lane_dict['library_id'], lane.library.id)
217 self.assertAlmostEqual(float(lane_dict['pM']), float(lane.pM))
218 self.assertEqual(lane_dict['library_species'],
219 lane.library.library_species.scientific_name)
221 def test_invalid_flowcell(self):
223 Make sure we get a 404 if we request an invalid flowcell ID
225 response = self.client.get('/experiments/config/nottheone/json', apidata)
226 self.assertEqual(response.status_code, 404)
228 def test_no_key(self):
230 Require logging in to retrieve meta data
232 response = self.client.get(u'/experiments/config/FC12150/json')
233 self.assertEqual(response.status_code, 403)
235 def test_library_id(self):
237 Library IDs should be flexible, so make sure we can retrive a non-numeric ID
239 response = self.client.get('/experiments/config/FC12150/json', apidata)
240 self.assertEqual(response.status_code, 200)
241 flowcell = json.loads(response.content)
243 lane_contents = flowcell['lane_set']['3']
244 lane_library = lane_contents[0]
245 self.assertEqual(lane_library['library_id'], 'SL039')
247 response = self.client.get('/samples/library/SL039/json', apidata)
248 self.assertEqual(response.status_code, 200)
249 library_sl039 = json.loads(response.content)
251 self.assertEqual(library_sl039['library_id'], 'SL039')
253 def test_raw_id_field(self):
257 Library's have IDs, libraries also have primary keys,
258 we eventually had enough libraries that the drop down combo box was too
259 hard to filter through, unfortnately we want a field that uses our library
260 id and not the internal primary key, and raw_id_field uses primary keys.
262 This tests to make sure that the value entered in the raw library id field matches
263 the library id looked up.
265 expected_ids = [u'10981',u'11016',u'SL039',u'11060',
266 u'11061',u'11062',u'11063',u'11064']
267 self.client.login(username='supertest', password='BJOKL5kAj6aFZ6A5')
268 response = self.client.get('/admin/experiments/flowcell/153/')
270 tree = fromstring(response.content)
272 xpath_expression = '//input[@id="id_lane_set-%d-library"]'
273 input_field = tree.xpath(xpath_expression % (i,))[0]
274 library_field = input_field.find('../strong')
275 library_id, library_name = library_field.text.split(':')
276 # strip leading '#' sign from name
277 library_id = library_id[1:]
278 self.assertEqual(library_id, expected_ids[i])
279 self.assertEqual(input_field.attrib['value'], library_id)
281 def test_library_to_flowcell_link(self):
283 Make sure the library page includes links to the flowcell pages.
284 That work with flowcell IDs that have parenthetical comments.
286 self.client.login(username='supertest', password='BJOKL5kAj6aFZ6A5')
287 response = self.client.get('/library/11070/')
288 self.assertEqual(response.status_code, 200)
289 status = validate_xhtml(response.content)
290 if status is not None: self.assertTrue(status)
292 tree = fromstring(response.content)
293 flowcell_spans = tree.xpath('//span[@property="libns:flowcell_id"]',
295 self.assertEqual(flowcell_spans[1].text, '30012AAXX (failed)')
296 failed_fc_span = flowcell_spans[1]
297 failed_fc_a = failed_fc_span.getparent()
298 # make sure some of our RDF made it.
299 self.assertEqual(failed_fc_a.get('typeof'), 'libns:IlluminaFlowcell')
300 self.assertEqual(failed_fc_a.get('href'), '/flowcell/30012AAXX/')
301 fc_response = self.client.get(failed_fc_a.get('href'))
302 self.assertEqual(fc_response.status_code, 200)
303 status = validate_xhtml(response.content)
304 if status is not None: self.assertTrue(status)
306 fc_lane_response = self.client.get('/flowcell/30012AAXX/8/')
307 self.assertEqual(fc_lane_response.status_code, 200)
308 status = validate_xhtml(response.content)
309 if status is not None: self.assertTrue(status)
312 def test_pooled_multiplex_id(self):
313 fc_dict = experiments.flowcell_information('42JU1AAXX')
314 lane_contents = fc_dict['lane_set'][3]
315 self.assertEqual(len(lane_contents), 2)
316 lane_dict = multi_lane_to_dict(lane_contents)
318 self.assertEqual(lane_dict['12044']['index_sequence'],
322 self.assertEqual(lane_dict['11045']['index_sequence'],
327 def test_lanes_for(self):
329 Check the code that packs the django objects into simple types.
332 lanes = experiments.lanes_for(user)
333 self.assertEqual(len(lanes), 5)
335 response = self.client.get('/experiments/lanes_for/%s/json' % (user,), apidata)
336 lanes_json = json.loads(response.content)
337 self.assertEqual(len(lanes), len(lanes_json))
338 for i in range(len(lanes)):
339 self.assertEqual(lanes[i]['comment'], lanes_json[i]['comment'])
340 self.assertEqual(lanes[i]['lane_number'], lanes_json[i]['lane_number'])
341 self.assertEqual(lanes[i]['flowcell'], lanes_json[i]['flowcell'])
342 self.assertEqual(lanes[i]['run_date'], lanes_json[i]['run_date'])
344 def test_lanes_for_no_lanes(self):
346 Do we get something meaningful back when the user isn't attached to anything?
349 lanes = experiments.lanes_for(user)
350 self.assertEqual(len(lanes), 0)
352 response = self.client.get('/experiments/lanes_for/%s/json' % (user,), apidata)
353 lanes_json = json.loads(response.content)
355 def test_lanes_for_no_user(self):
357 Do we get something meaningful back when its the wrong user
359 user = 'not a real user'
360 self.assertRaises(ObjectDoesNotExist, experiments.lanes_for, user)
362 response = self.client.get('/experiments/lanes_for/%s/json' % (user,), apidata)
363 self.assertEqual(response.status_code, 404)
366 def test_raw_data_dir(self):
367 """Raw data path generator check"""
368 flowcell_id = self.fc1_id
369 raw_dir = os.path.join(settings.RESULT_HOME_DIR, flowcell_id)
371 fc = models.FlowCell.objects.get(flowcell_id=flowcell_id)
372 self.assertEqual(fc.get_raw_data_directory(), raw_dir)
374 fc.flowcell_id = flowcell_id + " (failed)"
375 self.assertEqual(fc.get_raw_data_directory(), raw_dir)
378 def test_data_run_import(self):
379 srf_file_type = models.FileType.objects.get(name='SRF')
380 runxml_file_type = models.FileType.objects.get(name='run_xml')
381 flowcell_id = self.fc1_id
382 flowcell = models.FlowCell.objects.get(flowcell_id=flowcell_id)
383 flowcell.update_data_runs()
384 self.assertEqual(len(flowcell.datarun_set.all()), 1)
386 run = flowcell.datarun_set.all()[0]
387 result_files = run.datafile_set.all()
388 result_dict = dict(((rf.relative_pathname, rf) for rf in result_files))
390 srf4 = result_dict['FC12150/C1-37/woldlab_070829_SERIAL_FC12150_4.srf']
391 self.assertEqual(srf4.file_type, srf_file_type)
392 self.assertEqual(srf4.library_id, '11060')
393 self.assertEqual(srf4.data_run.flowcell.flowcell_id, 'FC12150')
395 srf4.data_run.flowcell.lane_set.get(lane_number=4).library_id,
399 os.path.join(settings.RESULT_HOME_DIR, srf4.relative_pathname))
401 lane_files = run.lane_files()
402 self.assertEqual(lane_files[4]['srf'], srf4)
404 runxml= result_dict['FC12150/C1-37/run_FC12150_2007-09-27.xml']
405 self.assertEqual(runxml.file_type, runxml_file_type)
406 self.assertEqual(runxml.library_id, None)
408 import1 = len(models.DataRun.objects.filter(result_dir='FC12150/C1-37'))
409 # what happens if we import twice?
410 flowcell.import_data_run('FC12150/C1-37',
411 'run_FC12150_2007-09-27.xml')
413 len(models.DataRun.objects.filter(result_dir='FC12150/C1-37')),
416 def test_read_result_file(self):
417 """make sure we can return a result file
419 flowcell_id = self.fc1_id
420 flowcell = models.FlowCell.objects.get(flowcell_id=flowcell_id)
421 flowcell.update_data_runs()
423 #self.client.login(username='supertest', password='BJOKL5kAj6aFZ6A5')
425 result_files = flowcell.datarun_set.all()[0].datafile_set.all()
426 for f in result_files:
427 url = '/experiments/file/%s' % ( f.random_key,)
428 response = self.client.get(url)
429 self.assertEqual(response.status_code, 200)
430 mimetype = f.file_type.mimetype
432 mimetype = 'application/octet-stream'
434 self.assertEqual(mimetype, response['content-type'])
436 def test_flowcell_rdf(self):
438 from htsworkflow.util.rdfhelp import get_model, \
440 load_string_into_model, \
447 expected = {'1': ['11034'],
449 '3': ['12044','11045'],
450 '4': ['11047','13044'],
455 url = '/flowcell/42JU1AAXX/'
456 response = self.client.get(url)
457 self.assertEqual(response.status_code, 200)
458 status = validate_xhtml(response.content)
459 if status is not None: self.assertTrue(status)
461 ns = urljoin('http://localhost', url)
462 load_string_into_model(model, 'rdfa', response.content, ns=ns)
463 body = """prefix rdf: <http://www.w3.org/1999/02/22-rdf-syntax-ns#>
464 prefix libns: <http://jumpgate.caltech.edu/wiki/LibraryOntology#>
466 select ?flowcell ?flowcell_id ?lane_id ?library_id
468 ?flowcell a libns:IlluminaFlowcell ;
469 libns:flowcell_id ?flowcell_id ;
470 libns:has_lane ?lane .
471 ?lane libns:lane_number ?lane_id ;
472 libns:library ?library .
473 ?library libns:library_id ?library_id .
475 query = RDF.SPARQLQuery(body)
477 for r in query.execute(model):
479 self.assertEqual(fromTypedNode(r['flowcell_id']), u'42JU1AAXX')
480 lane_id = fromTypedNode(r['lane_id'])
481 library_id = fromTypedNode(r['library_id'])
482 self.assertTrue(library_id in expected[lane_id])
483 self.assertEqual(count, 10)
485 def test_flowcell_estimates(self):
486 classic_flowcell = models.FlowCell.objects.get(pk=153)
487 classic_mid = experiments.estimateFlowcellDuration(classic_flowcell)
489 self.assertEqual(classic_mid, timedelta(4, 44000))
491 rapid_flowcell = models.FlowCell.objects.get(pk=300)
492 rapid_mid = experiments.estimateFlowcellDuration(rapid_flowcell)
493 self.assertEqual(rapid_mid, timedelta(seconds=60800))
495 def test_round_to_days(self):
497 [timedelta(2, 12345), (timedelta(days=2), timedelta(days=3))],
498 [timedelta(0, 345), (timedelta(days=0), timedelta(days=1))],
501 for estimate, expected in data:
502 rounded = experiments.roundToDays(estimate)
503 self.assertEqual(rounded, expected)
505 class TestFileType(TestCase):
506 fixtures = ['initial_data.json',
507 'test_flowcells.json',
510 def test_file_type_unicode(self):
511 file_type_objects = models.FileType.objects
512 name = 'QSEQ tarfile'
513 file_type_object = file_type_objects.get(name=name)
514 self.assertEqual(u"QSEQ tarfile",
515 unicode(file_type_object))
517 def test_find_file_type(self):
518 file_type_objects = models.FileType.objects
519 cases = [('woldlab_090921_HWUSI-EAS627_0009_42FC3AAXX_l7_r1.tar.bz2',
520 'QSEQ tarfile', 7, 1),
521 ('woldlab_091005_HWUSI-EAS627_0010_42JT2AAXX_1.srf',
523 ('s_1_eland_extended.txt.bz2','ELAND Extended', 1, None),
524 ('s_7_eland_multi.txt.bz2', 'ELAND Multi', 7, None),
525 ('s_3_eland_result.txt.bz2','ELAND Result', 3, None),
526 ('s_1_export.txt.bz2','ELAND Export', 1, None),
527 ('s_1_percent_call.png', 'IVC Percent Call', 1, None),
528 ('s_2_percent_base.png', 'IVC Percent Base', 2, None),
529 ('s_3_percent_all.png', 'IVC Percent All', 3, None),
530 ('s_4_call.png', 'IVC Call', 4, None),
531 ('s_5_all.png', 'IVC All', 5, None),
532 ('Summary.htm', 'Summary.htm', None, None),
533 ('run_42JT2AAXX_2009-10-07.xml', 'run_xml', None, None),
535 for filename, typename, lane, end in cases:
536 ft = models.find_file_type_metadata_from_filename(filename)
537 self.assertEqual(ft['file_type'],
538 file_type_objects.get(name=typename))
539 self.assertEqual(ft.get('lane', None), lane)
540 self.assertEqual(ft.get('end', None), end)
542 def test_assign_file_type_complex_path(self):
543 file_type_objects = models.FileType.objects
544 cases = [('/a/b/c/woldlab_090921_HWUSI-EAS627_0009_42FC3AAXX_l7_r1.tar.bz2',
545 'QSEQ tarfile', 7, 1),
546 ('foo/woldlab_091005_HWUSI-EAS627_0010_42JT2AAXX_1.srf',
548 ('../s_1_eland_extended.txt.bz2','ELAND Extended', 1, None),
549 ('/bleem/s_7_eland_multi.txt.bz2', 'ELAND Multi', 7, None),
550 ('/qwer/s_3_eland_result.txt.bz2','ELAND Result', 3, None),
551 ('/ty///1/s_1_export.txt.bz2','ELAND Export', 1, None),
552 ('/help/s_1_percent_call.png', 'IVC Percent Call', 1, None),
553 ('/bored/s_2_percent_base.png', 'IVC Percent Base', 2, None),
554 ('/example1/s_3_percent_all.png', 'IVC Percent All', 3, None),
555 ('amonkey/s_4_call.png', 'IVC Call', 4, None),
556 ('fishie/s_5_all.png', 'IVC All', 5, None),
557 ('/random/Summary.htm', 'Summary.htm', None, None),
558 ('/notrandom/run_42JT2AAXX_2009-10-07.xml', 'run_xml', None, None),
560 for filename, typename, lane, end in cases:
561 result = models.find_file_type_metadata_from_filename(filename)
562 self.assertEqual(result['file_type'],
563 file_type_objects.get(name=typename))
564 self.assertEqual(result.get('lane',None), lane)
565 self.assertEqual(result.get('end', None), end)
567 class TestEmailNotify(TestCase):
568 fixtures = ['initial_data.json',
569 'test_flowcells.json']
571 def test_started_email_not_logged_in(self):
572 response = self.client.get('/experiments/started/153/')
573 self.assertEqual(response.status_code, 302)
575 def test_started_email_logged_in_user(self):
576 self.client.login(username='test', password='BJOKL5kAj6aFZ6A5')
577 response = self.client.get('/experiments/started/153/')
578 self.assertEqual(response.status_code, 302)
580 def test_started_email_logged_in_staff(self):
581 self.client.login(username='admintest', password='BJOKL5kAj6aFZ6A5')
582 response = self.client.get('/experiments/started/153/')
583 self.assertEqual(response.status_code, 200)
585 def test_started_email_send(self):
586 self.client.login(username='admintest', password='BJOKL5kAj6aFZ6A5')
587 response = self.client.get('/experiments/started/153/')
588 self.assertEqual(response.status_code, 200)
590 self.assertTrue('pk1@example.com' in response.content)
591 self.assertTrue('Lane #8 : (11064) Paired ends 104' in response.content)
593 response = self.client.get('/experiments/started/153/', {'send':'1','bcc':'on'})
594 self.assertEqual(response.status_code, 200)
595 self.assertEqual(len(mail.outbox), 4)
596 bcc = set(settings.NOTIFICATION_BCC).copy()
597 bcc.update(set(settings.MANAGERS))
598 for m in mail.outbox:
599 self.assertTrue(len(m.body) > 0)
600 self.assertEqual(set(m.bcc), bcc)
602 def test_email_navigation(self):
604 Can we navigate between the flowcell and email forms properly?
606 self.client.login(username='supertest', password='BJOKL5kAj6aFZ6A5')
607 response = self.client.get('/experiments/started/153/')
608 self.assertEqual(response.status_code, 200)
609 self.assertTrue(re.search('Flowcell FC12150', response.content))
610 # require that navigation back to the admin page exists
611 self.assertTrue(re.search('<a href="/admin/experiments/flowcell/153/">[^<]+</a>', response.content))
613 def multi_lane_to_dict(lane):
614 """Convert a list of lane entries into a dictionary indexed by library ID
616 return dict( ((x['library_id'],x) for x in lane) )
618 class TestSequencer(TestCase):
619 fixtures = ['initial_data.json',
620 'test_flowcells.json',
623 def test_name_generation(self):
624 seq = models.Sequencer()
626 seq.instrument_name = "HWI-SEQ1"
627 seq.model = "Imaginary 5000"
629 self.assertEqual(unicode(seq), "Seq1 (HWI-SEQ1)")
631 def test_lookup(self):
632 fc = models.FlowCell.objects.get(pk=153)
633 self.assertEqual(fc.sequencer.model,
634 "Illumina Genome Analyzer IIx")
635 self.assertEqual(fc.sequencer.instrument_name,
637 # well actually we let the browser tack on the host name
638 url = fc.get_absolute_url()
639 self.assertEqual(url, '/flowcell/FC12150/')
642 response = self.client.get('/flowcell/FC12150/', apidata)
643 tree = fromstring(response.content)
644 seq_by = tree.xpath('//div[@rel="libns:sequenced_by"]',
646 self.assertEqual(len(seq_by), 1)
647 self.assertEqual(seq_by[0].attrib['rel'], 'libns:sequenced_by')
648 seq = seq_by[0].getchildren()
649 self.assertEqual(len(seq), 1)
650 self.assertEqual(seq[0].attrib['about'], '/sequencer/2')
651 self.assertEqual(seq[0].attrib['typeof'], 'libns:Sequencer')
653 name = seq[0].xpath('./span[@property="libns:sequencer_name"]')
654 self.assertEqual(len(name), 1)
655 self.assertEqual(name[0].text, 'Tardigrade')
656 instrument = seq[0].xpath(
657 './span[@property="libns:sequencer_instrument"]')
658 self.assertEqual(len(instrument), 1)
659 self.assertEqual(instrument[0].text, 'ILLUMINA-EC5D15')
660 model = seq[0].xpath(
661 './span[@property="libns:sequencer_model"]')
662 self.assertEqual(len(model), 1)
663 self.assertEqual(model[0].text, 'Illumina Genome Analyzer IIx')
665 def test_flowcell_with_rdf_validation(self):
666 from htsworkflow.util.rdfhelp import add_default_schemas, \
669 load_string_into_model
670 from htsworkflow.util.rdfinfer import Infer
673 add_default_schemas(model)
674 inference = Infer(model)
676 url ='/flowcell/FC12150/'
677 response = self.client.get(url)
678 self.assertEqual(response.status_code, 200)
679 status = validate_xhtml(response.content)
680 if status is not None: self.assertTrue(status)
682 load_string_into_model(model, 'rdfa', response.content)
684 errmsgs = list(inference.run_validation())
685 self.assertEqual(len(errmsgs), 0)
687 def test_lane_with_rdf_validation(self):
688 from htsworkflow.util.rdfhelp import add_default_schemas, \
691 load_string_into_model
692 from htsworkflow.util.rdfinfer import Infer
695 add_default_schemas(model)
696 inference = Infer(model)
699 response = self.client.get(url)
700 self.assertEqual(response.status_code, 200)
701 status = validate_xhtml(response.content)
702 if status is not None: self.assertTrue(status)
704 load_string_into_model(model, 'rdfa', response.content)
706 errmsgs = list(inference.run_validation())
707 self.assertEqual(len(errmsgs), 0)
710 from unittest import TestSuite, defaultTestLoader
712 for testcase in [ClusterStationTestCases,
718 suite.addTests(defaultTestLoader.loadTestsFromTestCase(testcase))
721 if __name__ == "__main__":
722 from unittest import main
723 main(defaultTest="suite")