experiments/test_experiments.py

   1 from __future__ import absolute_import, print_function, unicode_literals
   2
   3 import re
   4 from lxml.html import fromstring
   5 try:
   6     import json
   7 except ImportError as e:
   8     import simplejson as json
   9 import os
  10 import shutil
  11 import sys
  12 import tempfile
  13 from six.moves.urllib.parse import urljoin
  14
  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
  22 from django.utils.encoding import smart_text
  23
  24 from .models import ClusterStation, cluster_station_default, \
  25     DataRun, Sequencer, FlowCell, FileType
  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
  33
  34 from htsworkflow.pipelines.test.simulate_runfolder import TESTDATA_DIR
  35
  36 LANE_SET = range(1,9)
  37
  38 NSMAP = {'libns':'http://jumpgate.caltech.edu/wiki/LibraryOntology#'}
  39
  40 from django.db import connection
  41
  42
  43 class ExperimentsTestCases(TestCase):
  44     def setUp(self):
  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
  48
  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)
  58         self.admin.save()
  59
  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))
  69         for i in range(1,9):
  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)
  74             shutil.copy(
  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,))
  79                 )
  80         self.fc12150.save()
  81
  82         self.fc42jtn = FlowCellFactory(flowcell_id='42JTNAAXX')
  83         self.fc42jtn_lanes = []
  84         for i in range(1,9):
  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)
  94
  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'))
 100
 101     def tearDown(self):
 102         shutil.rmtree(self.tempdir)
 103
 104     def test_flowcell_information(self):
 105         """
 106         Check the code that packs the django objects into simple types.
 107         """
 108         fc12150 = self.fc12150
 109         fc42jtn = self.fc42jtn
 110         fc42ju1 = FlowCellFactory(flowcell_id='42JU1AAXX')
 111
 112         for fc_id in ['FC12150', '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)
 121
 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)
 134
 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(smart_text(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)
 143
 144
 145             for lane in fc_django.lane_set.all():
 146                 lane_contents = fc_json['lane_set'][str(lane.lane_number)]
 147                 lane_dict = multi_lane_to_dict(lane_contents)[lane.library_id]
 148
 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)
 158
 159     def test_invalid_flowcell(self):
 160         """
 161         Make sure we get a 404 if we request an invalid flowcell ID
 162         """
 163         response = self.client.get('/experiments/config/nottheone/json', apidata)
 164         self.assertEqual(response.status_code, 404)
 165
 166     def test_no_key(self):
 167         """
 168         Require logging in to retrieve meta data
 169         """
 170         response = self.client.get('/experiments/config/FC12150/json')
 171         self.assertEqual(response.status_code, 403)
 172
 173     def test_library_id(self):
 174         """
 175         Library IDs should be flexible, so make sure we can retrive a non-numeric ID
 176         """
 177         response = self.client.get('/experiments/config/FC12150/json', apidata)
 178         self.assertEqual(response.status_code, 200)
 179         flowcell = json.loads(smart_text(response.content))['result']
 180
 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')
 185
 186         response = self.client.get('/samples/library/12153/json', apidata)
 187         self.assertEqual(response.status_code, 200)
 188         library_12153 = json.loads(smart_text(response.content))['result']
 189
 190         self.assertEqual(library_12153['library_id'], '12153')
 191
 192     def test_raw_id_field(self):
 193         """
 194         Test ticket:147
 195
 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.
 200
 201         This tests to make sure that the value entered in the raw library id field matches
 202         the library id looked up.
 203         """
 204         expected_ids = [ '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))
 207
 208         tree = fromstring(response.content)
 209         for i in range(0,8):
 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)
 218
 219     def test_library_to_flowcell_link(self):
 220         """
 221         Make sure the library page includes links to the flowcell pages.
 222         That work with flowcell IDs that have parenthetical comments.
 223         """
 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)
 229
 230         tree = fromstring(response.content)
 231         flowcell_spans = tree.xpath('//span[@property="libns:flowcell_id"]',
 232                                     namespaces=NSMAP)
 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)
 243
 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)
 248
 249     def test_pooled_multiplex_id(self):
 250         fc_dict = flowcell_information(self.fc42jtn.flowcell_id)
 251
 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)
 255
 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'])
 260
 261     def test_lanes_for(self):
 262         """
 263         Check the code that packs the django objects into simple types.
 264         """
 265         user = self.user_odd.username
 266         lanes = lanes_for(user)
 267         self.assertEqual(len(lanes), 8)
 268
 269         response = self.client.get('/experiments/lanes_for/%s/json' % (user,), apidata)
 270         lanes_json = json.loads(smart_text(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'])
 277
 278     def test_lanes_for_no_lanes(self):
 279         """
 280         Do we get something meaningful back when the user isn't attached to anything?
 281         """
 282         user = HTSUserFactory.create(username='supertest')
 283         lanes = lanes_for(user.username)
 284         self.assertEqual(len(lanes), 0)
 285
 286         response = self.client.get('/experiments/lanes_for/%s/json' % (user,), apidata)
 287         self.assertEqual(response.status_code, 200)
 288         self.assertEqual(len(json.loads(response.content)['result']), 0)
 289
 290     def test_lanes_for_no_user(self):
 291         """
 292         Do we get something meaningful back when its the wrong user
 293         """
 294         user = 'not a real user'
 295         self.assertRaises(ObjectDoesNotExist, lanes_for, user)
 296
 297         response = self.client.get('/experiments/lanes_for/%s/json' % (user,), apidata)
 298         self.assertEqual(response.status_code, 404)
 299
 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)
 304
 305         fc = FlowCell.objects.get(flowcell_id=flowcell_id)
 306         self.assertEqual(fc.get_raw_data_directory(), raw_dir)
 307
 308         fc.flowcell_id = flowcell_id + " (failed)"
 309         self.assertEqual(fc.get_raw_data_directory(), raw_dir)
 310
 311
 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)
 319
 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))
 323
 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')
 328         self.assertEqual(
 329             srf4.data_run.flowcell.lane_set.get(lane_number=4).library_id,
 330             '12154')
 331         self.assertEqual(
 332             srf4.pathname,
 333             os.path.join(settings.RESULT_HOME_DIR, srf4.relative_pathname))
 334
 335         lane_files = run.lane_files()
 336         self.assertEqual(lane_files[4]['srf'], srf4)
 337
 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)
 341
 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')
 346         self.assertEqual(
 347             len(DataRun.objects.filter(result_dir='FC12150/C1-37')),
 348             import1)
 349
 350     def test_read_result_file(self):
 351         """make sure we can return a result file
 352         """
 353         flowcell_id = self.fc1_id
 354         flowcell = FlowCell.objects.get(flowcell_id=flowcell_id)
 355         flowcell.update_data_runs()
 356
 357         #self.client.login(username='supertest', password='BJOKL5kAj6aFZ6A5')
 358
 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
 365             if mimetype is None:
 366                 mimetype = 'application/octet-stream'
 367
 368             self.assertEqual(mimetype, response['content-type'])
 369
 370     def test_flowcell_rdf(self):
 371         import RDF
 372         from htsworkflow.util.rdfhelp import get_model, \
 373              fromTypedNode, \
 374              load_string_into_model, \
 375              rdfNS, \
 376              libraryOntology, \
 377              dump_model
 378
 379         model = get_model()
 380
 381         expected = {'1': ['12151'],
 382                     '2': ['12152'],
 383                     '3': ['12153'],
 384                     '4': ['12154'],
 385                     '5': ['12155'],
 386                     '6': ['12156'],
 387                     '7': ['12157'],
 388                     '8': ['12158']}
 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)
 394
 395         ns = urljoin('http://localhost', url)
 396         load_string_into_model(model, 'rdfa', smart_text(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#>
 399
 400         select ?flowcell ?flowcell_id ?lane_id ?library_id
 401         where {
 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 .
 408         }"""
 409         query = RDF.SPARQLQuery(body)
 410         count = 0
 411         for r in query.execute(model):
 412             count += 1
 413             self.assertEqual(fromTypedNode(r['flowcell_id']), '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)
 418
 419 class TestEmailNotify(TestCase):
 420     def setUp(self):
 421         self.password = 'foo27'
 422         self.user = HTSUserFactory.create(username='test')
 423         self.user.set_password(self.password)
 424         self.user.save()
 425         self.admin = HTSUserFactory.create(username='admintest', is_staff=True)
 426         self.admin.set_password(self.password)
 427         self.admin.save()
 428         self.super = HTSUserFactory.create(username='supertest', is_staff=True, is_superuser=True)
 429         self.super.set_password(self.password)
 430         self.super.save()
 431
 432         self.library = LibraryFactory.create()
 433         self.affiliation = AffiliationFactory()
 434         self.affiliation.users.add(self.user)
 435         self.library.affiliations.add(self.affiliation)
 436         self.fc = FlowCellFactory.create()
 437         self.lane = LaneFactory(flowcell=self.fc, lane_number=1, library=self.library)
 438
 439         self.url = '/experiments/started/{}/'.format(self.fc.id)
 440
 441     def test_started_email_not_logged_in(self):
 442         response = self.client.get(self.url)
 443         self.assertEqual(response.status_code, 302)
 444
 445     def test_started_email_logged_in_user(self):
 446         self.assertTrue(self.client.login(username=self.user.username, password=self.password))
 447         response = self.client.get(self.url)
 448         self.assertEqual(response.status_code, 302)
 449
 450     def test_started_email_logged_in_staff(self):
 451         self.assertTrue(self.admin.is_staff)
 452         admin = HTSUser.objects.get(username=self.admin.username)
 453         self.assertTrue(admin.is_staff)
 454         self.assertTrue(admin.check_password(self.password))
 455         self.assertTrue(self.client.login(username=self.admin.username, password=self.password))
 456         response = self.client.get(self.url)
 457         self.assertEqual(response.status_code, 200)
 458
 459     def test_started_email_send(self):
 460         self.assertTrue(self.client.login(username=self.admin.username, password=self.password))
 461         response = self.client.get(self.url)
 462         self.assertEqual(response.status_code, 200)
 463
 464         self.assertTrue(self.affiliation.email in smart_text(response.content))
 465         self.assertTrue(self.library.library_name in smart_text(response.content))
 466
 467         response = self.client.get(self.url, {'send':'1','bcc':'on'})
 468         self.assertEqual(response.status_code, 200)
 469         self.assertEqual(len(mail.outbox), 2)
 470         bcc = set(settings.NOTIFICATION_BCC).copy()
 471         bcc.update(set(settings.MANAGERS))
 472         for m in mail.outbox:
 473             self.assertTrue(len(m.body) > 0)
 474             self.assertEqual(set(m.bcc), bcc)
 475
 476     def test_email_navigation(self):
 477         """
 478         Can we navigate between the flowcell and email forms properly?
 479         """
 480         admin_url = '/admin/experiments/flowcell/{}/'.format(self.fc.id)
 481         self.client.login(username=self.admin.username, password=self.password)
 482         response = self.client.get(self.url)
 483         self.assertEqual(response.status_code, 200)
 484         #print("email navigation content:", response.content)
 485         self.assertTrue(re.search(self.fc.flowcell_id, smart_text(response.content)))
 486         # require that navigation back to the admin page exists
 487         self.assertTrue(re.search('<a href="{}">[^<]+</a>'.format(admin_url),
 488                                   smart_text(response.content)))
 489
 490 def multi_lane_to_dict(lane):
 491     """Convert a list of lane entries into a dictionary indexed by library ID
 492     """
 493     return dict( ((x['library_id'],x) for x in lane) )
 494
 495 class TestSequencer(TestCase):
 496     def setUp(self):
 497         self.fc12150 = FlowCellFactory(flowcell_id='FC12150')
 498         self.library = LibraryFactory(id="12150")
 499         self.lane = LaneFactory(flowcell=self.fc12150, lane_number=1, library=self.library)
 500
 501     def test_name_generation(self):
 502         seq = Sequencer()
 503         seq.name = "Seq1"
 504         seq.instrument_name = "HWI-SEQ1"
 505         seq.model = "Imaginary 5000"
 506
 507         self.assertEqual(str(seq), "Seq1 (HWI-SEQ1)")
 508
 509     def test_lookup(self):
 510         fc = self.fc12150
 511         self.assertEqual(fc.sequencer.model, 'HiSeq 1')
 512         self.assertTrue(fc.sequencer.instrument_name.startswith('instrument name')),
 513         # well actually we let the browser tack on the host name
 514         url = fc.get_absolute_url()
 515         self.assertEqual(url, '/flowcell/FC12150/')
 516
 517     def test_rdf(self):
 518         response = self.client.get('/flowcell/FC12150/', apidata)
 519         tree = fromstring(response.content)
 520         seq_by = tree.xpath('//div[@rel="libns:sequenced_by"]',
 521                             namespaces=NSMAP)
 522         self.assertEqual(len(seq_by), 1)
 523         self.assertEqual(seq_by[0].attrib['rel'], 'libns:sequenced_by')
 524         seq = seq_by[0].getchildren()
 525         self.assertEqual(len(seq), 1)
 526         sequencer = '/sequencer/' + str(self.fc12150.sequencer.id)
 527         self.assertEqual(seq[0].attrib['about'], sequencer)
 528         self.assertEqual(seq[0].attrib['typeof'], 'libns:Sequencer')
 529
 530         name = seq[0].xpath('./span[@property="libns:sequencer_name"]')
 531         self.assertEqual(len(name), 1)
 532         self.assertTrue(name[0].text.startswith('sequencer '))
 533         instrument = seq[0].xpath(
 534             './span[@property="libns:sequencer_instrument"]')
 535         self.assertEqual(len(instrument), 1)
 536         self.assertTrue(instrument[0].text.startswith('instrument name'))
 537         model = seq[0].xpath(
 538             './span[@property="libns:sequencer_model"]')
 539         self.assertEqual(len(model), 1)
 540         self.assertEqual(model[0].text, 'HiSeq 1')
 541
 542     def test_flowcell_with_rdf_validation(self):
 543         from htsworkflow.util.rdfhelp import add_default_schemas, \
 544              dump_model, \
 545              get_model, \
 546              load_string_into_model
 547         from htsworkflow.util.rdfinfer import Infer
 548
 549         model = get_model()
 550         add_default_schemas(model)
 551         inference = Infer(model)
 552
 553         url ='/flowcell/FC12150/'
 554         response = self.client.get(url)
 555         self.assertEqual(response.status_code, 200)
 556         status = validate_xhtml(response.content)
 557         if status is not None: self.assertTrue(status)
 558
 559         load_string_into_model(model, 'rdfa', smart_text(response.content))
 560
 561         errmsgs = list(inference.run_validation())
 562         self.assertEqual(len(errmsgs), 0)
 563
 564     def test_lane_with_rdf_validation(self):
 565         from htsworkflow.util.rdfhelp import add_default_schemas, \
 566              dump_model, \
 567              get_model, \
 568              load_string_into_model
 569         from htsworkflow.util.rdfinfer import Infer
 570
 571         model = get_model()
 572         add_default_schemas(model)
 573         inference = Infer(model)
 574
 575         url = '/lane/{}'.format(self.lane.id)
 576         response = self.client.get(url)
 577         rdfbody = smart_text(response.content)
 578         self.assertEqual(response.status_code, 200)
 579         status = validate_xhtml(rdfbody)
 580         if status is not None: self.assertTrue(status)
 581
 582         load_string_into_model(model, 'rdfa', rdfbody)
 583
 584         errmsgs = list(inference.run_validation())
 585         self.assertEqual(len(errmsgs), 0)
 586
 587 def suite():
 588     from unittest import TestSuite, defaultTestLoader
 589     suite = TestSuite()
 590     for testcase in [ExerimentsTestCases,
 591                      TestEmailNotify,
 592                      TestSequencer]:
 593         suite.addTests(defaultTestLoader.loadTestsFromTestCase(testcase))
 594     return suite
 595
 596 if __name__ == "__main__":
 597     from unittest import main
 598     main(defaultTest="suite")