1 from __future__ import absolute_import, print_function
4 from lxml.html import fromstring
8 import simplejson as json
13 from urlparse import urljoin
15 from django.conf import settings
16 from django.core import mail
17 from django.core.exceptions import ObjectDoesNotExist
18 from django.test import TestCase
19 from django.test.utils import setup_test_environment, teardown_test_environment
20 from django.db import connection
21 from django.conf import settings
23 from .models import ClusterStation, cluster_station_default, \
24 DataRun, Sequencer, FlowCell, FileType, \
25 find_file_type_metadata_from_filename
26 from samples.models import HTSUser
27 from .experiments import flowcell_information, lanes_for
28 from .experiments_factory import ClusterStationFactory, FlowCellFactory, LaneFactory
29 from samples.samples_factory import AffiliationFactory, HTSUserFactory, \
30 LibraryFactory, LibraryTypeFactory, MultiplexIndexFactory
31 from htsworkflow.auth import apidata
32 from htsworkflow.util.ethelp import validate_xhtml
34 from htsworkflow.pipelines.test.simulate_runfolder import TESTDATA_DIR
38 NSMAP = {'libns':'http://jumpgate.caltech.edu/wiki/LibraryOntology#'}
40 from django.db import connection
43 class ExperimentsTestCases(TestCase):
45 # Generate at least one fleshed out example flowcell
46 self.tempdir = tempfile.mkdtemp(prefix='htsw-test-experiments-')
47 settings.RESULT_HOME_DIR = self.tempdir
49 self.password = 'password'
50 self.user_odd = HTSUserFactory(username='user-odd')
51 self.user_odd.set_password(self.password)
52 self.affiliation_odd = AffiliationFactory(name='affiliation-odd', users=[self.user_odd])
53 self.user_even = HTSUserFactory(username='user-even')
54 self.user_even.set_password(self.password)
55 self.affiliation_even = AffiliationFactory(name='affiliation-even', users=[self.user_even])
56 self.admin = HTSUserFactory.create(username='admin', is_staff=True, is_superuser=True)
57 self.admin.set_password(self.password)
60 self.fc12150 = FlowCellFactory(flowcell_id='FC12150')
61 self.fc1_id = 'FC12150'
62 self.fc1_root = os.path.join(self.tempdir, self.fc1_id)
63 os.mkdir(self.fc1_root)
64 self.fc1_dir = os.path.join(self.fc1_root, 'C1-37')
65 os.mkdir(self.fc1_dir)
66 runxml = 'run_FC12150_2007-09-27.xml'
67 shutil.copy(os.path.join(TESTDATA_DIR, runxml),
68 os.path.join(self.fc1_dir, runxml))
70 affiliation = self.affiliation_odd if i % 2 == 1 else self.affiliation_even
71 library = LibraryFactory(id="1215" + str(i))
72 library.affiliations.add(affiliation)
73 lane = LaneFactory(flowcell=self.fc12150, lane_number=i, library=library)
75 os.path.join(TESTDATA_DIR,
76 'woldlab_070829_USI-EAS44_0017_FC11055_1.srf'),
77 os.path.join(self.fc1_dir,
78 'woldlab_070829_SERIAL_FC12150_%d.srf' %(i,))
82 self.fc42jtn = FlowCellFactory(flowcell_id='42JTNAAXX')
83 self.fc42jtn_lanes = []
85 affiliation = self.affiliation_odd if i % 2 == 1 else self.affiliation_even
86 library_type = LibraryTypeFactory(can_multiplex=True)
87 multiplex_index = MultiplexIndexFactory(adapter_type=library_type)
88 library = LibraryFactory(id="1300" + str(i),
89 library_type=library_type,
90 multiplex_id=multiplex_index.multiplex_id)
91 library.affiliations.add(affiliation)
92 lane = LaneFactory(flowcell=self.fc42jtn, lane_number=(i % 2) + 1, library=library)
93 self.fc42jtn_lanes.append(lane)
95 self.fc2_dir = os.path.join(self.tempdir, '42JTNAAXX')
96 os.mkdir(self.fc2_dir)
97 os.mkdir(os.path.join(self.fc2_dir, 'C1-25'))
98 os.mkdir(os.path.join(self.fc2_dir, 'C1-37'))
99 os.mkdir(os.path.join(self.fc2_dir, 'C1-37', 'Plots'))
102 shutil.rmtree(self.tempdir)
104 def test_flowcell_information(self):
106 Check the code that packs the django objects into simple types.
108 fc12150 = self.fc12150
109 fc42jtn = self.fc42jtn
110 fc42ju1 = FlowCellFactory(flowcell_id='42JU1AAXX')
112 for fc_id in [u'FC12150', u"42JTNAAXX", "42JU1AAXX"]:
113 fc_dict = flowcell_information(fc_id)
114 fc_django = FlowCell.objects.get(flowcell_id=fc_id)
115 self.assertEqual(fc_dict['flowcell_id'], fc_id)
116 self.assertEqual(fc_django.flowcell_id, fc_id)
117 self.assertEqual(fc_dict['sequencer'], fc_django.sequencer.name)
118 self.assertEqual(fc_dict['read_length'], fc_django.read_length)
119 self.assertEqual(fc_dict['notes'], fc_django.notes)
120 self.assertEqual(fc_dict['cluster_station'], fc_django.cluster_station.name)
122 for lane in fc_django.lane_set.all():
123 lane_contents = fc_dict['lane_set'][lane.lane_number]
124 lane_dict = multi_lane_to_dict(lane_contents)[lane.library_id]
125 self.assertEqual(lane_dict['cluster_estimate'], lane.cluster_estimate)
126 self.assertEqual(lane_dict['comment'], lane.comment)
127 self.assertEqual(lane_dict['flowcell'], lane.flowcell.flowcell_id)
128 self.assertEqual(lane_dict['lane_number'], lane.lane_number)
129 self.assertEqual(lane_dict['library_name'], lane.library.library_name)
130 self.assertEqual(lane_dict['library_id'], lane.library.id)
131 self.assertAlmostEqual(float(lane_dict['pM']), float(lane.pM))
132 self.assertEqual(lane_dict['library_species'],
133 lane.library.library_species.scientific_name)
135 response = self.client.get('/experiments/config/%s/json' % (fc_id,), apidata)
136 # strptime isoformat string = '%Y-%m-%dT%H:%M:%S'
137 fc_json = json.loads(response.content)['result']
138 self.assertEqual(fc_json['flowcell_id'], fc_id)
139 self.assertEqual(fc_json['sequencer'], fc_django.sequencer.name)
140 self.assertEqual(fc_json['read_length'], fc_django.read_length)
141 self.assertEqual(fc_json['notes'], fc_django.notes)
142 self.assertEqual(fc_json['cluster_station'], fc_django.cluster_station.name)
145 for lane in fc_django.lane_set.all():
146 lane_contents = fc_json['lane_set'][unicode(lane.lane_number)]
147 lane_dict = multi_lane_to_dict(lane_contents)[lane.library_id]
149 self.assertEqual(lane_dict['cluster_estimate'], lane.cluster_estimate)
150 self.assertEqual(lane_dict['comment'], lane.comment)
151 self.assertEqual(lane_dict['flowcell'], lane.flowcell.flowcell_id)
152 self.assertEqual(lane_dict['lane_number'], lane.lane_number)
153 self.assertEqual(lane_dict['library_name'], lane.library.library_name)
154 self.assertEqual(lane_dict['library_id'], lane.library.id)
155 self.assertAlmostEqual(float(lane_dict['pM']), float(lane.pM))
156 self.assertEqual(lane_dict['library_species'],
157 lane.library.library_species.scientific_name)
159 def test_invalid_flowcell(self):
161 Make sure we get a 404 if we request an invalid flowcell ID
163 response = self.client.get('/experiments/config/nottheone/json', apidata)
164 self.assertEqual(response.status_code, 404)
166 def test_no_key(self):
168 Require logging in to retrieve meta data
170 response = self.client.get(u'/experiments/config/FC12150/json')
171 self.assertEqual(response.status_code, 403)
173 def test_library_id(self):
175 Library IDs should be flexible, so make sure we can retrive a non-numeric ID
177 response = self.client.get('/experiments/config/FC12150/json', apidata)
178 self.assertEqual(response.status_code, 200)
179 flowcell = json.loads(response.content)['result']
181 # library id is 12150 + lane number (1-8), so 12153
182 lane_contents = flowcell['lane_set']['3']
183 lane_library = lane_contents[0]
184 self.assertEqual(lane_library['library_id'], '12153')
186 response = self.client.get('/samples/library/12153/json', apidata)
187 self.assertEqual(response.status_code, 200)
188 library_12153 = json.loads(response.content)['result']
190 self.assertEqual(library_12153['library_id'], '12153')
192 def test_raw_id_field(self):
196 Library's have IDs, libraries also have primary keys,
197 we eventually had enough libraries that the drop down combo box was too
198 hard to filter through, unfortnately we want a field that uses our library
199 id and not the internal primary key, and raw_id_field uses primary keys.
201 This tests to make sure that the value entered in the raw library id field matches
202 the library id looked up.
204 expected_ids = [ u'1215{}'.format(i) for i in range(1,9) ]
205 self.assertTrue(self.client.login(username=self.admin.username, password=self.password))
206 response = self.client.get('/admin/experiments/flowcell/{}/'.format(self.fc12150.id))
208 tree = fromstring(response.content)
210 xpath_expression = '//input[@id="id_lane_set-%d-library"]'
211 input_field = tree.xpath(xpath_expression % (i,))[0]
212 library_field = input_field.find('../strong')
213 library_id, library_name = library_field.text.split(':')
214 # strip leading '#' sign from name
215 library_id = library_id[1:]
216 self.assertEqual(library_id, expected_ids[i])
217 self.assertEqual(input_field.attrib['value'], library_id)
219 def test_library_to_flowcell_link(self):
221 Make sure the library page includes links to the flowcell pages.
222 That work with flowcell IDs that have parenthetical comments.
224 self.assertTrue(self.client.login(username=self.admin.username, password=self.password))
225 response = self.client.get('/library/12151/')
226 self.assertEqual(response.status_code, 200)
227 status = validate_xhtml(response.content)
228 if status is not None: self.assertTrue(status)
230 tree = fromstring(response.content)
231 flowcell_spans = tree.xpath('//span[@property="libns:flowcell_id"]',
233 self.assertEqual(flowcell_spans[1].text, 'FC12150')
234 failed_fc_span = flowcell_spans[1]
235 failed_fc_a = failed_fc_span.getparent()
236 # make sure some of our RDF made it.
237 self.assertEqual(failed_fc_a.get('typeof'), 'libns:IlluminaFlowcell')
238 self.assertEqual(failed_fc_a.get('href'), '/flowcell/FC12150/')
239 fc_response = self.client.get(failed_fc_a.get('href'))
240 self.assertEqual(fc_response.status_code, 200)
241 status = validate_xhtml(response.content)
242 if status is not None: self.assertTrue(status)
244 fc_lane_response = self.client.get('/flowcell/FC12150/8/')
245 self.assertEqual(fc_lane_response.status_code, 200)
246 status = validate_xhtml(response.content)
247 if status is not None: self.assertTrue(status)
249 def test_pooled_multiplex_id(self):
250 fc_dict = flowcell_information(self.fc42jtn.flowcell_id)
252 lane_contents = fc_dict['lane_set'][2]
253 self.assertEqual(len(lane_contents), len(self.fc42jtn_lanes) / 2)
254 lane_dict = multi_lane_to_dict(lane_contents)
256 self.assertTrue(self.fc42jtn_lanes[0].library.multiplex_id in \
257 lane_dict['13001']['index_sequence'])
258 self.assertTrue(self.fc42jtn_lanes[2].library.multiplex_id in \
259 lane_dict['13003']['index_sequence'])
261 def test_lanes_for(self):
263 Check the code that packs the django objects into simple types.
265 user = self.user_odd.username
266 lanes = lanes_for(user)
267 self.assertEqual(len(lanes), 8)
269 response = self.client.get('/experiments/lanes_for/%s/json' % (user,), apidata)
270 lanes_json = json.loads(response.content)['result']
271 self.assertEqual(len(lanes), len(lanes_json))
272 for i in range(len(lanes)):
273 self.assertEqual(lanes[i]['comment'], lanes_json[i]['comment'])
274 self.assertEqual(lanes[i]['lane_number'], lanes_json[i]['lane_number'])
275 self.assertEqual(lanes[i]['flowcell'], lanes_json[i]['flowcell'])
276 self.assertEqual(lanes[i]['run_date'], lanes_json[i]['run_date'])
278 def test_lanes_for_no_lanes(self):
280 Do we get something meaningful back when the user isn't attached to anything?
282 user = HTSUserFactory.create(username='supertest')
283 lanes = lanes_for(user.username)
284 self.assertEqual(len(lanes), 0)
286 response = self.client.get('/experiments/lanes_for/%s/json' % (user,), apidata)
287 self.assertEqual(response.status_code, 404)
289 def test_lanes_for_no_user(self):
291 Do we get something meaningful back when its the wrong user
293 user = 'not a real user'
294 self.assertRaises(ObjectDoesNotExist, lanes_for, user)
296 response = self.client.get('/experiments/lanes_for/%s/json' % (user,), apidata)
297 self.assertEqual(response.status_code, 404)
300 def test_raw_data_dir(self):
301 """Raw data path generator check"""
302 flowcell_id = self.fc1_id
303 raw_dir = os.path.join(settings.RESULT_HOME_DIR, flowcell_id)
305 fc = FlowCell.objects.get(flowcell_id=flowcell_id)
306 self.assertEqual(fc.get_raw_data_directory(), raw_dir)
308 fc.flowcell_id = flowcell_id + " (failed)"
309 self.assertEqual(fc.get_raw_data_directory(), raw_dir)
312 def test_data_run_import(self):
313 srf_file_type = FileType.objects.get(name='SRF')
314 runxml_file_type = FileType.objects.get(name='run_xml')
315 flowcell_id = self.fc1_id
316 flowcell = FlowCell.objects.get(flowcell_id=flowcell_id)
317 flowcell.update_data_runs()
318 self.assertEqual(len(flowcell.datarun_set.all()), 1)
320 run = flowcell.datarun_set.all()[0]
321 result_files = run.datafile_set.all()
322 result_dict = dict(((rf.relative_pathname, rf) for rf in result_files))
324 srf4 = result_dict['FC12150/C1-37/woldlab_070829_SERIAL_FC12150_4.srf']
325 self.assertEqual(srf4.file_type, srf_file_type)
326 self.assertEqual(srf4.library_id, '12154')
327 self.assertEqual(srf4.data_run.flowcell.flowcell_id, 'FC12150')
329 srf4.data_run.flowcell.lane_set.get(lane_number=4).library_id,
333 os.path.join(settings.RESULT_HOME_DIR, srf4.relative_pathname))
335 lane_files = run.lane_files()
336 self.assertEqual(lane_files[4]['srf'], srf4)
338 runxml= result_dict['FC12150/C1-37/run_FC12150_2007-09-27.xml']
339 self.assertEqual(runxml.file_type, runxml_file_type)
340 self.assertEqual(runxml.library_id, None)
342 import1 = len(DataRun.objects.filter(result_dir='FC12150/C1-37'))
343 # what happens if we import twice?
344 flowcell.import_data_run('FC12150/C1-37',
345 'run_FC12150_2007-09-27.xml')
347 len(DataRun.objects.filter(result_dir='FC12150/C1-37')),
350 def test_read_result_file(self):
351 """make sure we can return a result file
353 flowcell_id = self.fc1_id
354 flowcell = FlowCell.objects.get(flowcell_id=flowcell_id)
355 flowcell.update_data_runs()
357 #self.client.login(username='supertest', password='BJOKL5kAj6aFZ6A5')
359 result_files = flowcell.datarun_set.all()[0].datafile_set.all()
360 for f in result_files:
361 url = '/experiments/file/%s' % ( f.random_key,)
362 response = self.client.get(url)
363 self.assertEqual(response.status_code, 200)
364 mimetype = f.file_type.mimetype
366 mimetype = 'application/octet-stream'
368 self.assertEqual(mimetype, response['content-type'])
370 def test_flowcell_rdf(self):
372 from htsworkflow.util.rdfhelp import get_model, \
374 load_string_into_model, \
381 expected = {'1': ['12151'],
389 url = '/flowcell/{}/'.format(self.fc12150.flowcell_id)
390 response = self.client.get(url)
391 self.assertEqual(response.status_code, 200)
392 status = validate_xhtml(response.content)
393 if status is not None: self.assertTrue(status)
395 ns = urljoin('http://localhost', url)
396 load_string_into_model(model, 'rdfa', response.content, ns=ns)
397 body = """prefix rdf: <http://www.w3.org/1999/02/22-rdf-syntax-ns#>
398 prefix libns: <http://jumpgate.caltech.edu/wiki/LibraryOntology#>
400 select ?flowcell ?flowcell_id ?lane_id ?library_id
402 ?flowcell a libns:IlluminaFlowcell ;
403 libns:flowcell_id ?flowcell_id ;
404 libns:has_lane ?lane .
405 ?lane libns:lane_number ?lane_id ;
406 libns:library ?library .
407 ?library libns:library_id ?library_id .
409 query = RDF.SPARQLQuery(body)
411 for r in query.execute(model):
413 self.assertEqual(fromTypedNode(r['flowcell_id']), u'FC12150')
414 lane_id = fromTypedNode(r['lane_id'])
415 library_id = fromTypedNode(r['library_id'])
416 self.assertTrue(library_id in expected[lane_id])
417 self.assertEqual(count, 8)
420 class TestFileType(TestCase):
421 def test_file_type_unicode(self):
422 file_type_objects = FileType.objects
423 name = 'QSEQ tarfile'
424 file_type_object = file_type_objects.get(name=name)
425 self.assertEqual(u"QSEQ tarfile",
426 unicode(file_type_object))
428 def test_find_file_type(self):
429 file_type_objects = FileType.objects
430 cases = [('woldlab_090921_HWUSI-EAS627_0009_42FC3AAXX_l7_r1.tar.bz2',
431 'QSEQ tarfile', 7, 1),
432 ('woldlab_091005_HWUSI-EAS627_0010_42JT2AAXX_1.srf',
434 ('s_1_eland_extended.txt.bz2','ELAND Extended', 1, None),
435 ('s_7_eland_multi.txt.bz2', 'ELAND Multi', 7, None),
436 ('s_3_eland_result.txt.bz2','ELAND Result', 3, None),
437 ('s_1_export.txt.bz2','ELAND Export', 1, None),
438 ('s_1_percent_call.png', 'IVC Percent Call', 1, None),
439 ('s_2_percent_base.png', 'IVC Percent Base', 2, None),
440 ('s_3_percent_all.png', 'IVC Percent All', 3, None),
441 ('s_4_call.png', 'IVC Call', 4, None),
442 ('s_5_all.png', 'IVC All', 5, None),
443 ('Summary.htm', 'Summary.htm', None, None),
444 ('run_42JT2AAXX_2009-10-07.xml', 'run_xml', None, None),
446 for filename, typename, lane, end in cases:
447 ft = find_file_type_metadata_from_filename(filename)
448 self.assertEqual(ft['file_type'],
449 file_type_objects.get(name=typename))
450 self.assertEqual(ft.get('lane', None), lane)
451 self.assertEqual(ft.get('end', None), end)
453 def test_assign_file_type_complex_path(self):
454 file_type_objects = FileType.objects
455 cases = [('/a/b/c/woldlab_090921_HWUSI-EAS627_0009_42FC3AAXX_l7_r1.tar.bz2',
456 'QSEQ tarfile', 7, 1),
457 ('foo/woldlab_091005_HWUSI-EAS627_0010_42JT2AAXX_1.srf',
459 ('../s_1_eland_extended.txt.bz2','ELAND Extended', 1, None),
460 ('/bleem/s_7_eland_multi.txt.bz2', 'ELAND Multi', 7, None),
461 ('/qwer/s_3_eland_result.txt.bz2','ELAND Result', 3, None),
462 ('/ty///1/s_1_export.txt.bz2','ELAND Export', 1, None),
463 ('/help/s_1_percent_call.png', 'IVC Percent Call', 1, None),
464 ('/bored/s_2_percent_base.png', 'IVC Percent Base', 2, None),
465 ('/example1/s_3_percent_all.png', 'IVC Percent All', 3, None),
466 ('amonkey/s_4_call.png', 'IVC Call', 4, None),
467 ('fishie/s_5_all.png', 'IVC All', 5, None),
468 ('/random/Summary.htm', 'Summary.htm', None, None),
469 ('/notrandom/run_42JT2AAXX_2009-10-07.xml', 'run_xml', None, None),
471 for filename, typename, lane, end in cases:
472 result = find_file_type_metadata_from_filename(filename)
473 self.assertEqual(result['file_type'],
474 file_type_objects.get(name=typename))
475 self.assertEqual(result.get('lane',None), lane)
476 self.assertEqual(result.get('end', None), end)
478 class TestEmailNotify(TestCase):
480 self.password = 'foo27'
481 self.user = HTSUserFactory.create(username='test')
482 self.user.set_password(self.password)
484 self.admin = HTSUserFactory.create(username='admintest', is_staff=True)
485 self.admin.set_password(self.password)
487 self.super = HTSUserFactory.create(username='supertest', is_staff=True, is_superuser=True)
488 self.super.set_password(self.password)
491 self.library = LibraryFactory.create()
492 self.affiliation = AffiliationFactory()
493 self.affiliation.users.add(self.user)
494 self.library.affiliations.add(self.affiliation)
495 self.fc = FlowCellFactory.create()
496 self.lane = LaneFactory(flowcell=self.fc, lane_number=1, library=self.library)
498 self.url = '/experiments/started/{}/'.format(self.fc.id)
500 def test_started_email_not_logged_in(self):
501 response = self.client.get(self.url)
502 self.assertEqual(response.status_code, 302)
504 def test_started_email_logged_in_user(self):
505 self.assertTrue(self.client.login(username=self.user.username, password=self.password))
506 response = self.client.get(self.url)
507 self.assertEqual(response.status_code, 302)
509 def test_started_email_logged_in_staff(self):
510 self.assertTrue(self.admin.is_staff)
511 admin = HTSUser.objects.get(username=self.admin.username)
512 self.assertTrue(admin.is_staff)
513 self.assertTrue(admin.check_password(self.password))
514 self.assertTrue(self.client.login(username=self.admin.username, password=self.password))
515 response = self.client.get(self.url)
516 self.assertEqual(response.status_code, 200)
518 def test_started_email_send(self):
519 self.assertTrue(self.client.login(username=self.admin.username, password=self.password))
520 response = self.client.get(self.url)
521 self.assertEqual(response.status_code, 200)
523 self.assertTrue(self.affiliation.email in response.content)
524 self.assertTrue(self.library.library_name in response.content)
526 response = self.client.get(self.url, {'send':'1','bcc':'on'})
527 self.assertEqual(response.status_code, 200)
528 self.assertEqual(len(mail.outbox), 2)
529 bcc = set(settings.NOTIFICATION_BCC).copy()
530 bcc.update(set(settings.MANAGERS))
531 for m in mail.outbox:
532 self.assertTrue(len(m.body) > 0)
533 self.assertEqual(set(m.bcc), bcc)
535 def test_email_navigation(self):
537 Can we navigate between the flowcell and email forms properly?
539 admin_url = '/admin/experiments/flowcell/{}/'.format(self.fc.id)
540 self.client.login(username=self.admin.username, password=self.password)
541 response = self.client.get(self.url)
542 self.assertEqual(response.status_code, 200)
543 #print("email navigation content:", response.content)
544 self.assertTrue(re.search(self.fc.flowcell_id, response.content))
545 # require that navigation back to the admin page exists
546 self.assertTrue(re.search('<a href="{}">[^<]+</a>'.format(admin_url), response.content))
548 def multi_lane_to_dict(lane):
549 """Convert a list of lane entries into a dictionary indexed by library ID
551 return dict( ((x['library_id'],x) for x in lane) )
553 class TestSequencer(TestCase):
555 self.fc12150 = FlowCellFactory(flowcell_id='FC12150')
556 self.library = LibraryFactory(id="12150")
557 self.lane = LaneFactory(flowcell=self.fc12150, lane_number=1, library=self.library)
559 def test_name_generation(self):
562 seq.instrument_name = "HWI-SEQ1"
563 seq.model = "Imaginary 5000"
565 self.assertEqual(unicode(seq), "Seq1 (HWI-SEQ1)")
567 def test_lookup(self):
569 self.assertEqual(fc.sequencer.model, 'HiSeq 1')
570 self.assertTrue(fc.sequencer.instrument_name.startswith('instrument name')),
571 # well actually we let the browser tack on the host name
572 url = fc.get_absolute_url()
573 self.assertEqual(url, '/flowcell/FC12150/')
576 response = self.client.get('/flowcell/FC12150/', apidata)
577 tree = fromstring(response.content)
578 seq_by = tree.xpath('//div[@rel="libns:sequenced_by"]',
580 self.assertEqual(len(seq_by), 1)
581 self.assertEqual(seq_by[0].attrib['rel'], 'libns:sequenced_by')
582 seq = seq_by[0].getchildren()
583 self.assertEqual(len(seq), 1)
584 sequencer = '/sequencer/' + str(self.fc12150.sequencer.id)
585 self.assertEqual(seq[0].attrib['about'], sequencer)
586 self.assertEqual(seq[0].attrib['typeof'], 'libns:Sequencer')
588 name = seq[0].xpath('./span[@property="libns:sequencer_name"]')
589 self.assertEqual(len(name), 1)
590 self.assertTrue(name[0].text.startswith('sequencer '))
591 instrument = seq[0].xpath(
592 './span[@property="libns:sequencer_instrument"]')
593 self.assertEqual(len(instrument), 1)
594 self.assertTrue(instrument[0].text.startswith('instrument name'))
595 model = seq[0].xpath(
596 './span[@property="libns:sequencer_model"]')
597 self.assertEqual(len(model), 1)
598 self.assertEqual(model[0].text, 'HiSeq 1')
600 def test_flowcell_with_rdf_validation(self):
601 from htsworkflow.util.rdfhelp import add_default_schemas, \
604 load_string_into_model
605 from htsworkflow.util.rdfinfer import Infer
608 add_default_schemas(model)
609 inference = Infer(model)
611 url ='/flowcell/FC12150/'
612 response = self.client.get(url)
613 self.assertEqual(response.status_code, 200)
614 status = validate_xhtml(response.content)
615 if status is not None: self.assertTrue(status)
617 load_string_into_model(model, 'rdfa', response.content)
619 errmsgs = list(inference.run_validation())
620 self.assertEqual(len(errmsgs), 0)
622 def test_lane_with_rdf_validation(self):
623 from htsworkflow.util.rdfhelp import add_default_schemas, \
626 load_string_into_model
627 from htsworkflow.util.rdfinfer import Infer
630 add_default_schemas(model)
631 inference = Infer(model)
633 url = '/lane/{}'.format(self.lane.id)
634 response = self.client.get(url)
635 self.assertEqual(response.status_code, 200)
636 status = validate_xhtml(response.content)
637 if status is not None: self.assertTrue(status)
639 load_string_into_model(model, 'rdfa', response.content)
641 errmsgs = list(inference.run_validation())
642 self.assertEqual(len(errmsgs), 0)
645 from unittest import TestSuite, defaultTestLoader
647 for testcase in [ClusterStationTestCases,
653 suite.addTests(defaultTestLoader.loadTestsFromTestCase(testcase))
656 if __name__ == "__main__":
657 from unittest import main
658 main(defaultTest="suite")