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, 404)
 288
 289     def test_lanes_for_no_user(self):
 290         """
 291         Do we get something meaningful back when its the wrong user
 292         """
 293         user = 'not a real user'
 294         self.assertRaises(ObjectDoesNotExist, lanes_for, user)
 295
 296         response = self.client.get('/experiments/lanes_for/%s/json' % (user,), apidata)
 297         self.assertEqual(response.status_code, 404)
 298
 299     def test_raw_data_dir(self):
 300         """Raw data path generator check"""
 301         flowcell_id = self.fc1_id
 302         raw_dir = os.path.join(settings.RESULT_HOME_DIR, flowcell_id)
 303
 304         fc = FlowCell.objects.get(flowcell_id=flowcell_id)
 305         self.assertEqual(fc.get_raw_data_directory(), raw_dir)
 306
 307         fc.flowcell_id = flowcell_id + " (failed)"
 308         self.assertEqual(fc.get_raw_data_directory(), raw_dir)
 309
 310
 311     def test_data_run_import(self):
 312         srf_file_type = FileType.objects.get(name='SRF')
 313         runxml_file_type = FileType.objects.get(name='run_xml')
 314         flowcell_id = self.fc1_id
 315         flowcell = FlowCell.objects.get(flowcell_id=flowcell_id)
 316         flowcell.update_data_runs()
 317         self.assertEqual(len(flowcell.datarun_set.all()), 1)
 318
 319         run = flowcell.datarun_set.all()[0]
 320         result_files = run.datafile_set.all()
 321         result_dict = dict(((rf.relative_pathname, rf) for rf in result_files))
 322
 323         srf4 = result_dict['FC12150/C1-37/woldlab_070829_SERIAL_FC12150_4.srf']
 324         self.assertEqual(srf4.file_type, srf_file_type)
 325         self.assertEqual(srf4.library_id, '12154')
 326         self.assertEqual(srf4.data_run.flowcell.flowcell_id, 'FC12150')
 327         self.assertEqual(
 328             srf4.data_run.flowcell.lane_set.get(lane_number=4).library_id,
 329             '12154')
 330         self.assertEqual(
 331             srf4.pathname,
 332             os.path.join(settings.RESULT_HOME_DIR, srf4.relative_pathname))
 333
 334         lane_files = run.lane_files()
 335         self.assertEqual(lane_files[4]['srf'], srf4)
 336
 337         runxml= result_dict['FC12150/C1-37/run_FC12150_2007-09-27.xml']
 338         self.assertEqual(runxml.file_type, runxml_file_type)
 339         self.assertEqual(runxml.library_id, None)
 340
 341         import1 = len(DataRun.objects.filter(result_dir='FC12150/C1-37'))
 342         # what happens if we import twice?
 343         flowcell.import_data_run('FC12150/C1-37',
 344                                  'run_FC12150_2007-09-27.xml')
 345         self.assertEqual(
 346             len(DataRun.objects.filter(result_dir='FC12150/C1-37')),
 347             import1)
 348
 349     def test_read_result_file(self):
 350         """make sure we can return a result file
 351         """
 352         flowcell_id = self.fc1_id
 353         flowcell = FlowCell.objects.get(flowcell_id=flowcell_id)
 354         flowcell.update_data_runs()
 355
 356         #self.client.login(username='supertest', password='BJOKL5kAj6aFZ6A5')
 357
 358         result_files = flowcell.datarun_set.all()[0].datafile_set.all()
 359         for f in result_files:
 360             url = '/experiments/file/%s' % ( f.random_key,)
 361             response = self.client.get(url)
 362             self.assertEqual(response.status_code, 200)
 363             mimetype = f.file_type.mimetype
 364             if mimetype is None:
 365                 mimetype = 'application/octet-stream'
 366
 367             self.assertEqual(mimetype, response['content-type'])
 368
 369     def test_flowcell_rdf(self):
 370         import RDF
 371         from htsworkflow.util.rdfhelp import get_model, \
 372              fromTypedNode, \
 373              load_string_into_model, \
 374              rdfNS, \
 375              libraryOntology, \
 376              dump_model
 377
 378         model = get_model()
 379
 380         expected = {'1': ['12151'],
 381                     '2': ['12152'],
 382                     '3': ['12153'],
 383                     '4': ['12154'],
 384                     '5': ['12155'],
 385                     '6': ['12156'],
 386                     '7': ['12157'],
 387                     '8': ['12158']}
 388         url = '/flowcell/{}/'.format(self.fc12150.flowcell_id)
 389         response = self.client.get(url)
 390         self.assertEqual(response.status_code, 200)
 391         status = validate_xhtml(response.content)
 392         if status is not None: self.assertTrue(status)
 393
 394         ns = urljoin('http://localhost', url)
 395         load_string_into_model(model, 'rdfa', smart_text(response.content), ns=ns)
 396         body = """prefix rdf: <http://www.w3.org/1999/02/22-rdf-syntax-ns#>
 397         prefix libns: <http://jumpgate.caltech.edu/wiki/LibraryOntology#>
 398
 399         select ?flowcell ?flowcell_id ?lane_id ?library_id
 400         where {
 401           ?flowcell a libns:IlluminaFlowcell ;
 402                     libns:flowcell_id ?flowcell_id ;
 403                     libns:has_lane ?lane .
 404           ?lane libns:lane_number ?lane_id ;
 405                 libns:library ?library .
 406           ?library libns:library_id ?library_id .
 407         }"""
 408         query = RDF.SPARQLQuery(body)
 409         count = 0
 410         for r in query.execute(model):
 411             count += 1
 412             self.assertEqual(fromTypedNode(r['flowcell_id']), 'FC12150')
 413             lane_id = fromTypedNode(r['lane_id'])
 414             library_id = fromTypedNode(r['library_id'])
 415             self.assertTrue(library_id in expected[lane_id])
 416         self.assertEqual(count, 8)
 417
 418 class TestEmailNotify(TestCase):
 419     def setUp(self):
 420         self.password = 'foo27'
 421         self.user = HTSUserFactory.create(username='test')
 422         self.user.set_password(self.password)
 423         self.user.save()
 424         self.admin = HTSUserFactory.create(username='admintest', is_staff=True)
 425         self.admin.set_password(self.password)
 426         self.admin.save()
 427         self.super = HTSUserFactory.create(username='supertest', is_staff=True, is_superuser=True)
 428         self.super.set_password(self.password)
 429         self.super.save()
 430
 431         self.library = LibraryFactory.create()
 432         self.affiliation = AffiliationFactory()
 433         self.affiliation.users.add(self.user)
 434         self.library.affiliations.add(self.affiliation)
 435         self.fc = FlowCellFactory.create()
 436         self.lane = LaneFactory(flowcell=self.fc, lane_number=1, library=self.library)
 437
 438         self.url = '/experiments/started/{}/'.format(self.fc.id)
 439
 440     def test_started_email_not_logged_in(self):
 441         response = self.client.get(self.url)
 442         self.assertEqual(response.status_code, 302)
 443
 444     def test_started_email_logged_in_user(self):
 445         self.assertTrue(self.client.login(username=self.user.username, password=self.password))
 446         response = self.client.get(self.url)
 447         self.assertEqual(response.status_code, 302)
 448
 449     def test_started_email_logged_in_staff(self):
 450         self.assertTrue(self.admin.is_staff)
 451         admin = HTSUser.objects.get(username=self.admin.username)
 452         self.assertTrue(admin.is_staff)
 453         self.assertTrue(admin.check_password(self.password))
 454         self.assertTrue(self.client.login(username=self.admin.username, password=self.password))
 455         response = self.client.get(self.url)
 456         self.assertEqual(response.status_code, 200)
 457
 458     def test_started_email_send(self):
 459         self.assertTrue(self.client.login(username=self.admin.username, password=self.password))
 460         response = self.client.get(self.url)
 461         self.assertEqual(response.status_code, 200)
 462
 463         self.assertTrue(self.affiliation.email in smart_text(response.content))
 464         self.assertTrue(self.library.library_name in smart_text(response.content))
 465
 466         response = self.client.get(self.url, {'send':'1','bcc':'on'})
 467         self.assertEqual(response.status_code, 200)
 468         self.assertEqual(len(mail.outbox), 2)
 469         bcc = set(settings.NOTIFICATION_BCC).copy()
 470         bcc.update(set(settings.MANAGERS))
 471         for m in mail.outbox:
 472             self.assertTrue(len(m.body) > 0)
 473             self.assertEqual(set(m.bcc), bcc)
 474
 475     def test_email_navigation(self):
 476         """
 477         Can we navigate between the flowcell and email forms properly?
 478         """
 479         admin_url = '/admin/experiments/flowcell/{}/'.format(self.fc.id)
 480         self.client.login(username=self.admin.username, password=self.password)
 481         response = self.client.get(self.url)
 482         self.assertEqual(response.status_code, 200)
 483         #print("email navigation content:", response.content)
 484         self.assertTrue(re.search(self.fc.flowcell_id, smart_text(response.content)))
 485         # require that navigation back to the admin page exists
 486         self.assertTrue(re.search('<a href="{}">[^<]+</a>'.format(admin_url),
 487                                   smart_text(response.content)))
 488
 489 def multi_lane_to_dict(lane):
 490     """Convert a list of lane entries into a dictionary indexed by library ID
 491     """
 492     return dict( ((x['library_id'],x) for x in lane) )
 493
 494 class TestSequencer(TestCase):
 495     def setUp(self):
 496         self.fc12150 = FlowCellFactory(flowcell_id='FC12150')
 497         self.library = LibraryFactory(id="12150")
 498         self.lane = LaneFactory(flowcell=self.fc12150, lane_number=1, library=self.library)
 499
 500     def test_name_generation(self):
 501         seq = Sequencer()
 502         seq.name = "Seq1"
 503         seq.instrument_name = "HWI-SEQ1"
 504         seq.model = "Imaginary 5000"
 505
 506         self.assertEqual(str(seq), "Seq1 (HWI-SEQ1)")
 507
 508     def test_lookup(self):
 509         fc = self.fc12150
 510         self.assertEqual(fc.sequencer.model, 'HiSeq 1')
 511         self.assertTrue(fc.sequencer.instrument_name.startswith('instrument name')),
 512         # well actually we let the browser tack on the host name
 513         url = fc.get_absolute_url()
 514         self.assertEqual(url, '/flowcell/FC12150/')
 515
 516     def test_rdf(self):
 517         response = self.client.get('/flowcell/FC12150/', apidata)
 518         tree = fromstring(response.content)
 519         seq_by = tree.xpath('//div[@rel="libns:sequenced_by"]',
 520                             namespaces=NSMAP)
 521         self.assertEqual(len(seq_by), 1)
 522         self.assertEqual(seq_by[0].attrib['rel'], 'libns:sequenced_by')
 523         seq = seq_by[0].getchildren()
 524         self.assertEqual(len(seq), 1)
 525         sequencer = '/sequencer/' + str(self.fc12150.sequencer.id)
 526         self.assertEqual(seq[0].attrib['about'], sequencer)
 527         self.assertEqual(seq[0].attrib['typeof'], 'libns:Sequencer')
 528
 529         name = seq[0].xpath('./span[@property="libns:sequencer_name"]')
 530         self.assertEqual(len(name), 1)
 531         self.assertTrue(name[0].text.startswith('sequencer '))
 532         instrument = seq[0].xpath(
 533             './span[@property="libns:sequencer_instrument"]')
 534         self.assertEqual(len(instrument), 1)
 535         self.assertTrue(instrument[0].text.startswith('instrument name'))
 536         model = seq[0].xpath(
 537             './span[@property="libns:sequencer_model"]')
 538         self.assertEqual(len(model), 1)
 539         self.assertEqual(model[0].text, 'HiSeq 1')
 540
 541     def test_flowcell_with_rdf_validation(self):
 542         from htsworkflow.util.rdfhelp import add_default_schemas, \
 543              dump_model, \
 544              get_model, \
 545              load_string_into_model
 546         from htsworkflow.util.rdfinfer import Infer
 547
 548         model = get_model()
 549         add_default_schemas(model)
 550         inference = Infer(model)
 551
 552         url ='/flowcell/FC12150/'
 553         response = self.client.get(url)
 554         self.assertEqual(response.status_code, 200)
 555         status = validate_xhtml(response.content)
 556         if status is not None: self.assertTrue(status)
 557
 558         load_string_into_model(model, 'rdfa', smart_text(response.content))
 559
 560         errmsgs = list(inference.run_validation())
 561         self.assertEqual(len(errmsgs), 0)
 562
 563     def test_lane_with_rdf_validation(self):
 564         from htsworkflow.util.rdfhelp import add_default_schemas, \
 565              dump_model, \
 566              get_model, \
 567              load_string_into_model
 568         from htsworkflow.util.rdfinfer import Infer
 569
 570         model = get_model()
 571         add_default_schemas(model)
 572         inference = Infer(model)
 573
 574         url = '/lane/{}'.format(self.lane.id)
 575         response = self.client.get(url)
 576         rdfbody = smart_text(response.content)
 577         self.assertEqual(response.status_code, 200)
 578         status = validate_xhtml(rdfbody)
 579         if status is not None: self.assertTrue(status)
 580
 581         load_string_into_model(model, 'rdfa', rdfbody)
 582
 583         errmsgs = list(inference.run_validation())
 584         self.assertEqual(len(errmsgs), 0)
 585
 586 def suite():
 587     from unittest import TestSuite, defaultTestLoader
 588     suite = TestSuite()
 589     for testcase in [ExerimentsTestCases,
 590                      TestEmailNotify,
 591                      TestSequencer]:
 592         suite.addTests(defaultTestLoader.loadTestsFromTestCase(testcase))
 593     return suite
 594
 595 if __name__ == "__main__":
 596     from unittest import main
 597     main(defaultTest="suite")