2 from lxml.html import fromstring
6 import simplejson as json
11 from urlparse import urljoin
13 from django.conf import settings
14 from django.core import mail
15 from django.core.exceptions import ObjectDoesNotExist
16 from django.test import TestCase
17 from htsworkflow.frontend.experiments import models
18 from htsworkflow.frontend.experiments import experiments
19 from htsworkflow.frontend.auth import apidata
20 from htsworkflow.util.ethelp import validate_xhtml
22 from htsworkflow.pipelines.test.simulate_runfolder import TESTDATA_DIR
26 NSMAP = {'libns':'http://jumpgate.caltech.edu/wiki/LibraryOntology#'}
28 class ClusterStationTestCases(TestCase):
29 fixtures = ['test_flowcells.json']
31 def test_default(self):
32 c = models.ClusterStation.default()
33 self.assertEqual(c.id, 2)
38 total = models.ClusterStation.objects.filter(isdefault=True).count()
39 self.assertEqual(total, 0)
41 other_default = models.ClusterStation.default()
42 self.assertEqual(other_default.id, 3)
45 def test_update_default(self):
46 old_default = models.ClusterStation.default()
48 c = models.ClusterStation.objects.get(pk=3)
52 new_default = models.ClusterStation.default()
54 self.assertNotEqual(old_default, new_default)
55 self.assertEqual(new_default, c)
57 total = models.ClusterStation.objects.filter(isdefault=True).count()
58 self.assertEqual(total, 1)
60 def test_update_other(self):
61 old_default = models.ClusterStation.default()
62 total = models.ClusterStation.objects.filter(isdefault=True).count()
63 self.assertEqual(total, 1)
65 c = models.ClusterStation.objects.get(pk=1)
66 c.name = "Primary Key 1"
69 total = models.ClusterStation.objects.filter(isdefault=True).count()
70 self.assertEqual(total, 1)
72 new_default = models.ClusterStation.default()
73 self.assertEqual(old_default, new_default)
76 class SequencerTestCases(TestCase):
77 fixtures = ['test_flowcells.json']
79 def test_default(self):
80 # starting with no default
81 s = models.Sequencer.default()
82 self.assertEqual(s.id, 2)
84 total = models.Sequencer.objects.filter(isdefault=True).count()
85 self.assertEqual(total, 1)
90 total = models.Sequencer.objects.filter(isdefault=True).count()
91 self.assertEqual(total, 0)
93 other_default = models.Sequencer.default()
94 self.assertEqual(other_default.id, 7)
96 def test_update_default(self):
97 old_default = models.Sequencer.default()
99 s = models.Sequencer.objects.get(pk=1)
103 new_default = models.Sequencer.default()
105 self.assertNotEqual(old_default, new_default)
106 self.assertEqual(new_default, s)
108 total = models.Sequencer.objects.filter(isdefault=True).count()
109 self.assertEqual(total, 1)
112 def test_update_other(self):
113 old_default = models.Sequencer.default()
114 total = models.Sequencer.objects.filter(isdefault=True).count()
115 self.assertEqual(total, 1)
117 s = models.Sequencer.objects.get(pk=1)
118 s.name = "Primary Key 1"
121 total = models.Sequencer.objects.filter(isdefault=True).count()
122 self.assertEqual(total, 1)
124 new_default = models.Sequencer.default()
125 self.assertEqual(old_default, new_default)
128 class ExperimentsTestCases(TestCase):
129 fixtures = ['test_flowcells.json',
133 self.tempdir = tempfile.mkdtemp(prefix='htsw-test-experiments-')
134 settings.RESULT_HOME_DIR = self.tempdir
136 self.fc1_id = 'FC12150'
137 self.fc1_root = os.path.join(self.tempdir, self.fc1_id)
138 os.mkdir(self.fc1_root)
139 self.fc1_dir = os.path.join(self.fc1_root, 'C1-37')
140 os.mkdir(self.fc1_dir)
141 runxml = 'run_FC12150_2007-09-27.xml'
142 shutil.copy(os.path.join(TESTDATA_DIR, runxml),
143 os.path.join(self.fc1_dir, runxml))
146 os.path.join(TESTDATA_DIR,
147 'woldlab_070829_USI-EAS44_0017_FC11055_1.srf'),
148 os.path.join(self.fc1_dir,
149 'woldlab_070829_SERIAL_FC12150_%d.srf' %(i,))
152 self.fc2_dir = os.path.join(self.tempdir, '42JTNAAXX')
153 os.mkdir(self.fc2_dir)
154 os.mkdir(os.path.join(self.fc2_dir, 'C1-25'))
155 os.mkdir(os.path.join(self.fc2_dir, 'C1-37'))
156 os.mkdir(os.path.join(self.fc2_dir, 'C1-37', 'Plots'))
159 shutil.rmtree(self.tempdir)
161 def test_flowcell_information(self):
163 Check the code that packs the django objects into simple types.
165 for fc_id in [u'FC12150', u"42JTNAAXX", "42JU1AAXX"]:
166 fc_dict = experiments.flowcell_information(fc_id)
167 fc_django = models.FlowCell.objects.get(flowcell_id=fc_id)
168 self.assertEqual(fc_dict['flowcell_id'], fc_id)
169 self.assertEqual(fc_django.flowcell_id, fc_id)
170 self.assertEqual(fc_dict['sequencer'], fc_django.sequencer.name)
171 self.assertEqual(fc_dict['read_length'], fc_django.read_length)
172 self.assertEqual(fc_dict['notes'], fc_django.notes)
173 self.assertEqual(fc_dict['cluster_station'], fc_django.cluster_station.name)
175 for lane in fc_django.lane_set.all():
176 lane_contents = fc_dict['lane_set'][lane.lane_number]
177 lane_dict = multi_lane_to_dict(lane_contents)[lane.library_id]
178 self.assertEqual(lane_dict['cluster_estimate'], lane.cluster_estimate)
179 self.assertEqual(lane_dict['comment'], lane.comment)
180 self.assertEqual(lane_dict['flowcell'], lane.flowcell.flowcell_id)
181 self.assertEqual(lane_dict['lane_number'], lane.lane_number)
182 self.assertEqual(lane_dict['library_name'], lane.library.library_name)
183 self.assertEqual(lane_dict['library_id'], lane.library.id)
184 self.assertAlmostEqual(float(lane_dict['pM']), float(lane.pM))
185 self.assertEqual(lane_dict['library_species'],
186 lane.library.library_species.scientific_name)
188 response = self.client.get('/experiments/config/%s/json' % (fc_id,), apidata)
189 # strptime isoformat string = '%Y-%m-%dT%H:%M:%S'
190 fc_json = json.loads(response.content)
191 self.assertEqual(fc_json['flowcell_id'], fc_id)
192 self.assertEqual(fc_json['sequencer'], fc_django.sequencer.name)
193 self.assertEqual(fc_json['read_length'], fc_django.read_length)
194 self.assertEqual(fc_json['notes'], fc_django.notes)
195 self.assertEqual(fc_json['cluster_station'], fc_django.cluster_station.name)
198 for lane in fc_django.lane_set.all():
199 lane_contents = fc_json['lane_set'][unicode(lane.lane_number)]
200 lane_dict = multi_lane_to_dict(lane_contents)[lane.library_id]
202 self.assertEqual(lane_dict['cluster_estimate'], lane.cluster_estimate)
203 self.assertEqual(lane_dict['comment'], lane.comment)
204 self.assertEqual(lane_dict['flowcell'], lane.flowcell.flowcell_id)
205 self.assertEqual(lane_dict['lane_number'], lane.lane_number)
206 self.assertEqual(lane_dict['library_name'], lane.library.library_name)
207 self.assertEqual(lane_dict['library_id'], lane.library.id)
208 self.assertAlmostEqual(float(lane_dict['pM']), float(lane.pM))
209 self.assertEqual(lane_dict['library_species'],
210 lane.library.library_species.scientific_name)
212 def test_invalid_flowcell(self):
214 Make sure we get a 404 if we request an invalid flowcell ID
216 response = self.client.get('/experiments/config/nottheone/json', apidata)
217 self.assertEqual(response.status_code, 404)
219 def test_no_key(self):
221 Require logging in to retrieve meta data
223 response = self.client.get(u'/experiments/config/FC12150/json')
224 self.assertEqual(response.status_code, 403)
226 def test_library_id(self):
228 Library IDs should be flexible, so make sure we can retrive a non-numeric ID
230 response = self.client.get('/experiments/config/FC12150/json', apidata)
231 self.assertEqual(response.status_code, 200)
232 flowcell = json.loads(response.content)
234 lane_contents = flowcell['lane_set']['3']
235 lane_library = lane_contents[0]
236 self.assertEqual(lane_library['library_id'], 'SL039')
238 response = self.client.get('/samples/library/SL039/json', apidata)
239 self.assertEqual(response.status_code, 200)
240 library_sl039 = json.loads(response.content)
242 self.assertEqual(library_sl039['library_id'], 'SL039')
244 def test_raw_id_field(self):
248 Library's have IDs, libraries also have primary keys,
249 we eventually had enough libraries that the drop down combo box was too
250 hard to filter through, unfortnately we want a field that uses our library
251 id and not the internal primary key, and raw_id_field uses primary keys.
253 This tests to make sure that the value entered in the raw library id field matches
254 the library id looked up.
256 expected_ids = [u'10981',u'11016',u'SL039',u'11060',
257 u'11061',u'11062',u'11063',u'11064']
258 self.client.login(username='supertest', password='BJOKL5kAj6aFZ6A5')
259 response = self.client.get('/admin/experiments/flowcell/153/')
261 tree = fromstring(response.content)
263 xpath_expression = '//input[@id="id_lane_set-%d-library"]'
264 input_field = tree.xpath(xpath_expression % (i,))[0]
265 library_field = input_field.find('../strong')
266 library_id, library_name = library_field.text.split(':')
267 # strip leading '#' sign from name
268 library_id = library_id[1:]
269 self.assertEqual(library_id, expected_ids[i])
270 self.assertEqual(input_field.attrib['value'], library_id)
272 def test_library_to_flowcell_link(self):
274 Make sure the library page includes links to the flowcell pages.
275 That work with flowcell IDs that have parenthetical comments.
277 self.client.login(username='supertest', password='BJOKL5kAj6aFZ6A5')
278 response = self.client.get('/library/11070/')
279 self.assertEqual(response.status_code, 200)
280 status = validate_xhtml(response.content)
281 if status is not None: self.assertTrue(status)
283 tree = fromstring(response.content)
284 flowcell_spans = tree.xpath('//span[@property="libns:flowcell_id"]',
286 self.assertEqual(flowcell_spans[0].text, '30012AAXX (failed)')
287 failed_fc_span = flowcell_spans[0]
288 failed_fc_a = failed_fc_span.getparent()
289 # make sure some of our RDF made it.
290 self.assertEqual(failed_fc_a.get('typeof'), 'libns:IlluminaFlowcell')
291 self.assertEqual(failed_fc_a.get('href'), '/flowcell/30012AAXX/')
292 fc_response = self.client.get(failed_fc_a.get('href'))
293 self.assertEqual(fc_response.status_code, 200)
294 status = validate_xhtml(response.content)
295 if status is not None: self.assertTrue(status)
297 fc_lane_response = self.client.get('/flowcell/30012AAXX/8/')
298 self.assertEqual(fc_lane_response.status_code, 200)
299 status = validate_xhtml(response.content)
300 if status is not None: self.assertTrue(status)
303 def test_pooled_multiplex_id(self):
304 fc_dict = experiments.flowcell_information('42JU1AAXX')
305 lane_contents = fc_dict['lane_set'][3]
306 self.assertEqual(len(lane_contents), 2)
307 lane_dict = multi_lane_to_dict(lane_contents)
309 self.assertEqual(lane_dict['12044']['index_sequence'],
313 self.assertEqual(lane_dict['11045']['index_sequence'],
318 def test_lanes_for(self):
320 Check the code that packs the django objects into simple types.
323 lanes = experiments.lanes_for(user)
324 self.assertEqual(len(lanes), 5)
326 response = self.client.get('/experiments/lanes_for/%s/json' % (user,), apidata)
327 lanes_json = json.loads(response.content)
328 self.assertEqual(len(lanes), len(lanes_json))
329 for i in range(len(lanes)):
330 self.assertEqual(lanes[i]['comment'], lanes_json[i]['comment'])
331 self.assertEqual(lanes[i]['lane_number'], lanes_json[i]['lane_number'])
332 self.assertEqual(lanes[i]['flowcell'], lanes_json[i]['flowcell'])
333 self.assertEqual(lanes[i]['run_date'], lanes_json[i]['run_date'])
335 def test_lanes_for_no_lanes(self):
337 Do we get something meaningful back when the user isn't attached to anything?
340 lanes = experiments.lanes_for(user)
341 self.assertEqual(len(lanes), 0)
343 response = self.client.get('/experiments/lanes_for/%s/json' % (user,), apidata)
344 lanes_json = json.loads(response.content)
346 def test_lanes_for_no_user(self):
348 Do we get something meaningful back when its the wrong user
350 user = 'not a real user'
351 self.assertRaises(ObjectDoesNotExist, experiments.lanes_for, user)
353 response = self.client.get('/experiments/lanes_for/%s/json' % (user,), apidata)
354 self.assertEqual(response.status_code, 404)
357 def test_raw_data_dir(self):
358 """Raw data path generator check"""
359 flowcell_id = self.fc1_id
360 raw_dir = os.path.join(settings.RESULT_HOME_DIR, flowcell_id)
362 fc = models.FlowCell.objects.get(flowcell_id=flowcell_id)
363 self.assertEqual(fc.get_raw_data_directory(), raw_dir)
365 fc.flowcell_id = flowcell_id + " (failed)"
366 self.assertEqual(fc.get_raw_data_directory(), raw_dir)
369 def test_data_run_import(self):
370 srf_file_type = models.FileType.objects.get(name='SRF')
371 runxml_file_type = models.FileType.objects.get(name='run_xml')
372 flowcell_id = self.fc1_id
373 flowcell = models.FlowCell.objects.get(flowcell_id=flowcell_id)
374 flowcell.update_data_runs()
375 self.assertEqual(len(flowcell.datarun_set.all()), 1)
377 run = flowcell.datarun_set.all()[0]
378 result_files = run.datafile_set.all()
379 result_dict = dict(((rf.relative_pathname, rf) for rf in result_files))
381 srf4 = result_dict['FC12150/C1-37/woldlab_070829_SERIAL_FC12150_4.srf']
382 self.assertEqual(srf4.file_type, srf_file_type)
383 self.assertEqual(srf4.library_id, '11060')
384 self.assertEqual(srf4.data_run.flowcell.flowcell_id, 'FC12150')
386 srf4.data_run.flowcell.lane_set.get(lane_number=4).library_id,
390 os.path.join(settings.RESULT_HOME_DIR, srf4.relative_pathname))
392 lane_files = run.lane_files()
393 self.assertEqual(lane_files[4]['srf'], srf4)
395 runxml= result_dict['FC12150/C1-37/run_FC12150_2007-09-27.xml']
396 self.assertEqual(runxml.file_type, runxml_file_type)
397 self.assertEqual(runxml.library_id, None)
399 import1 = len(models.DataRun.objects.filter(result_dir='FC12150/C1-37'))
400 # what happens if we import twice?
401 flowcell.import_data_run('FC12150/C1-37',
402 'run_FC12150_2007-09-27.xml')
404 len(models.DataRun.objects.filter(result_dir='FC12150/C1-37')),
407 def test_read_result_file(self):
408 """make sure we can return a result file
410 flowcell_id = self.fc1_id
411 flowcell = models.FlowCell.objects.get(flowcell_id=flowcell_id)
412 flowcell.update_data_runs()
414 #self.client.login(username='supertest', password='BJOKL5kAj6aFZ6A5')
416 result_files = flowcell.datarun_set.all()[0].datafile_set.all()
417 for f in result_files:
418 url = '/experiments/file/%s' % ( f.random_key,)
419 response = self.client.get(url)
420 self.assertEqual(response.status_code, 200)
421 mimetype = f.file_type.mimetype
423 mimetype = 'application/octet-stream'
425 self.assertEqual(mimetype, response['content-type'])
427 def test_flowcell_rdf(self):
429 from htsworkflow.util.rdfhelp import get_model, \
431 load_string_into_model, \
438 expected = {'1': ['11034'],
440 '3': ['12044','11045'],
441 '4': ['11047','13044'],
446 url = '/flowcell/42JU1AAXX/'
447 response = self.client.get(url)
448 self.assertEqual(response.status_code, 200)
449 status = validate_xhtml(response.content)
450 if status is not None: self.assertTrue(status)
452 ns = urljoin('http://localhost', url)
453 load_string_into_model(model, 'rdfa', response.content, ns=ns)
454 body = """prefix rdf: <http://www.w3.org/1999/02/22-rdf-syntax-ns#>
455 prefix libns: <http://jumpgate.caltech.edu/wiki/LibraryOntology#>
457 select ?flowcell ?flowcell_id ?lane_id ?library_id
459 ?flowcell a libns:IlluminaFlowcell ;
460 libns:flowcell_id ?flowcell_id ;
461 libns:has_lane ?lane .
462 ?lane libns:lane_number ?lane_id ;
463 libns:library ?library .
464 ?library libns:library_id ?library_id .
466 query = RDF.SPARQLQuery(body)
468 for r in query.execute(model):
470 self.assertEqual(fromTypedNode(r['flowcell_id']), u'42JU1AAXX')
471 lane_id = fromTypedNode(r['lane_id'])
472 library_id = fromTypedNode(r['library_id'])
473 self.assertTrue(library_id in expected[lane_id])
474 self.assertEqual(count, 10)
477 class TestFileType(TestCase):
478 def test_file_type_unicode(self):
479 file_type_objects = models.FileType.objects
480 name = 'QSEQ tarfile'
481 file_type_object = file_type_objects.get(name=name)
482 self.assertEqual(u"<FileType: QSEQ tarfile>",
483 unicode(file_type_object))
485 class TestFileType(TestCase):
486 def test_find_file_type(self):
487 file_type_objects = models.FileType.objects
488 cases = [('woldlab_090921_HWUSI-EAS627_0009_42FC3AAXX_l7_r1.tar.bz2',
489 'QSEQ tarfile', 7, 1),
490 ('woldlab_091005_HWUSI-EAS627_0010_42JT2AAXX_1.srf',
492 ('s_1_eland_extended.txt.bz2','ELAND Extended', 1, None),
493 ('s_7_eland_multi.txt.bz2', 'ELAND Multi', 7, None),
494 ('s_3_eland_result.txt.bz2','ELAND Result', 3, None),
495 ('s_1_export.txt.bz2','ELAND Export', 1, None),
496 ('s_1_percent_call.png', 'IVC Percent Call', 1, None),
497 ('s_2_percent_base.png', 'IVC Percent Base', 2, None),
498 ('s_3_percent_all.png', 'IVC Percent All', 3, None),
499 ('s_4_call.png', 'IVC Call', 4, None),
500 ('s_5_all.png', 'IVC All', 5, None),
501 ('Summary.htm', 'Summary.htm', None, None),
502 ('run_42JT2AAXX_2009-10-07.xml', 'run_xml', None, None),
504 for filename, typename, lane, end in cases:
505 ft = models.find_file_type_metadata_from_filename(filename)
506 self.assertEqual(ft['file_type'],
507 file_type_objects.get(name=typename))
508 self.assertEqual(ft.get('lane', None), lane)
509 self.assertEqual(ft.get('end', None), end)
511 def test_assign_file_type_complex_path(self):
512 file_type_objects = models.FileType.objects
513 cases = [('/a/b/c/woldlab_090921_HWUSI-EAS627_0009_42FC3AAXX_l7_r1.tar.bz2',
514 'QSEQ tarfile', 7, 1),
515 ('foo/woldlab_091005_HWUSI-EAS627_0010_42JT2AAXX_1.srf',
517 ('../s_1_eland_extended.txt.bz2','ELAND Extended', 1, None),
518 ('/bleem/s_7_eland_multi.txt.bz2', 'ELAND Multi', 7, None),
519 ('/qwer/s_3_eland_result.txt.bz2','ELAND Result', 3, None),
520 ('/ty///1/s_1_export.txt.bz2','ELAND Export', 1, None),
521 ('/help/s_1_percent_call.png', 'IVC Percent Call', 1, None),
522 ('/bored/s_2_percent_base.png', 'IVC Percent Base', 2, None),
523 ('/example1/s_3_percent_all.png', 'IVC Percent All', 3, None),
524 ('amonkey/s_4_call.png', 'IVC Call', 4, None),
525 ('fishie/s_5_all.png', 'IVC All', 5, None),
526 ('/random/Summary.htm', 'Summary.htm', None, None),
527 ('/notrandom/run_42JT2AAXX_2009-10-07.xml', 'run_xml', None, None),
529 for filename, typename, lane, end in cases:
530 result = models.find_file_type_metadata_from_filename(filename)
531 self.assertEqual(result['file_type'],
532 file_type_objects.get(name=typename))
533 self.assertEqual(result.get('lane',None), lane)
534 self.assertEqual(result.get('end', None), end)
536 class TestEmailNotify(TestCase):
537 fixtures = ['test_flowcells.json']
539 def test_started_email_not_logged_in(self):
540 response = self.client.get('/experiments/started/153/')
541 self.assertEqual(response.status_code, 302)
543 def test_started_email_logged_in_user(self):
544 self.client.login(username='test', password='BJOKL5kAj6aFZ6A5')
545 response = self.client.get('/experiments/started/153/')
546 self.assertEqual(response.status_code, 302)
548 def test_started_email_logged_in_staff(self):
549 self.client.login(username='admintest', password='BJOKL5kAj6aFZ6A5')
550 response = self.client.get('/experiments/started/153/')
551 self.assertEqual(response.status_code, 200)
553 def test_started_email_send(self):
554 self.client.login(username='admintest', password='BJOKL5kAj6aFZ6A5')
555 response = self.client.get('/experiments/started/153/')
556 self.assertEqual(response.status_code, 200)
558 self.assertTrue('pk1@example.com' in response.content)
559 self.assertTrue('Lane #8 : (11064) Paired ends 104' in response.content)
561 response = self.client.get('/experiments/started/153/', {'send':'1','bcc':'on'})
562 self.assertEqual(response.status_code, 200)
563 self.assertEqual(len(mail.outbox), 4)
564 bcc = set(settings.NOTIFICATION_BCC).intersect(set(settings.MANAGERS))
565 for m in mail.outbox:
566 self.assertTrue(len(m.body) > 0)
567 self.assertEqual(m.bcc, bcc)
569 def test_email_navigation(self):
571 Can we navigate between the flowcell and email forms properly?
573 self.client.login(username='supertest', password='BJOKL5kAj6aFZ6A5')
574 response = self.client.get('/experiments/started/153/')
575 self.assertEqual(response.status_code, 200)
576 self.assertTrue(re.search('Flowcell FC12150', response.content))
577 # require that navigation back to the admin page exists
578 self.assertTrue(re.search('<a href="/admin/experiments/flowcell/153/">[^<]+</a>', response.content))
580 def multi_lane_to_dict(lane):
581 """Convert a list of lane entries into a dictionary indexed by library ID
583 return dict( ((x['library_id'],x) for x in lane) )
585 class TestSequencer(TestCase):
586 fixtures = ['test_flowcells.json',
589 def test_name_generation(self):
590 seq = models.Sequencer()
592 seq.instrument_name = "HWI-SEQ1"
593 seq.model = "Imaginary 5000"
595 self.assertEqual(unicode(seq), "Seq1 (HWI-SEQ1)")
597 def test_lookup(self):
598 fc = models.FlowCell.objects.get(pk=153)
599 self.assertEqual(fc.sequencer.model,
600 "Illumina Genome Analyzer IIx")
601 self.assertEqual(fc.sequencer.instrument_name,
603 # well actually we let the browser tack on the host name
604 url = fc.get_absolute_url()
605 self.assertEqual(url, '/flowcell/FC12150/')
608 response = self.client.get('/flowcell/FC12150/', apidata)
609 tree = fromstring(response.content)
610 seq_by = tree.xpath('//div[@rel="libns:sequenced_by"]',
612 self.assertEqual(len(seq_by), 1)
613 self.assertEqual(seq_by[0].attrib['rel'], 'libns:sequenced_by')
614 seq = seq_by[0].getchildren()
615 self.assertEqual(len(seq), 1)
616 self.assertEqual(seq[0].attrib['about'], '/sequencer/2')
617 self.assertEqual(seq[0].attrib['typeof'], 'libns:Sequencer')
619 name = seq[0].xpath('./span[@property="libns:sequencer_name"]')
620 self.assertEqual(len(name), 1)
621 self.assertEqual(name[0].text, 'Tardigrade')
622 instrument = seq[0].xpath(
623 './span[@property="libns:sequencer_instrument"]')
624 self.assertEqual(len(instrument), 1)
625 self.assertEqual(instrument[0].text, 'ILLUMINA-EC5D15')
626 model = seq[0].xpath(
627 './span[@property="libns:sequencer_model"]')
628 self.assertEqual(len(model), 1)
629 self.assertEqual(model[0].text, 'Illumina Genome Analyzer IIx')
631 def test_flowcell_with_rdf_validation(self):
632 from htsworkflow.util.rdfhelp import add_default_schemas, \
635 load_string_into_model
636 from htsworkflow.util.rdfinfer import Infer
639 add_default_schemas(model)
640 inference = Infer(model)
642 url ='/flowcell/FC12150/'
643 response = self.client.get(url)
644 self.assertEqual(response.status_code, 200)
645 status = validate_xhtml(response.content)
646 if status is not None: self.assertTrue(status)
648 load_string_into_model(model, 'rdfa', response.content)
650 errmsgs = list(inference.run_validation())
651 self.assertEqual(len(errmsgs), 2)
652 for errmsg in errmsgs:
653 self.assertEqual(errmsg, 'Missing type for: http://localhost/')
655 def test_lane_with_rdf_validation(self):
656 from htsworkflow.util.rdfhelp import add_default_schemas, \
659 load_string_into_model
660 from htsworkflow.util.rdfinfer import Infer
663 add_default_schemas(model)
664 inference = Infer(model)
667 response = self.client.get(url)
668 self.assertEqual(response.status_code, 200)
669 status = validate_xhtml(response.content)
670 if status is not None: self.assertTrue(status)
672 load_string_into_model(model, 'rdfa', response.content)
674 errmsgs = list(inference.run_validation())
675 self.assertEqual(len(errmsgs), 2)
676 for errmsg in errmsgs:
677 self.assertEqual(errmsg, 'Missing type for: http://localhost/')