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 import json
   6 import os
   7 import shutil
   8 import tempfile
   9 from six.moves.urllib.parse import urljoin
  10
  11 from django.conf import settings
  12 from django.core import mail
  13 from django.core.exceptions import ObjectDoesNotExist
  14 from django.test import TestCase
  15 from django.utils.encoding import smart_text
  16
  17 from .models import DataRun, Sequencer, FlowCell, FileType
  18 from samples.models import HTSUser
  19 from .experiments import flowcell_information, lanes_for
  20 from .experiments_factory import FlowCellFactory, LaneFactory
  21 from samples.samples_factory import AffiliationFactory, HTSUserFactory, \
  22     LibraryFactory, LibraryTypeFactory, MultiplexIndexFactory
  23 from htsworkflow.auth import apidata
  24 from htsworkflow.util.ethelp import validate_xhtml
  25
  26 from htsworkflow.pipelines.test.simulate_runfolder import TESTDATA_DIR
  27
  28 LANE_SET = range(1, 9)
  29
  30 NSMAP = {'libns': 'http://jumpgate.caltech.edu/wiki/LibraryOntology#'}
  31
  32
  33 class ExperimentsTestCases(TestCase):
  34     def setUp(self):
  35         # Generate at least one fleshed out example flowcell
  36         self.tempdir = tempfile.mkdtemp(prefix='htsw-test-experiments-')
  37         settings.RESULT_HOME_DIR = self.tempdir
  38
  39         self.password = 'password'
  40         self.user_odd = HTSUserFactory(username='user-odd')
  41         self.user_odd.set_password(self.password)
  42         self.affiliation_odd = AffiliationFactory(name='affiliation-odd', users=[self.user_odd])
  43         self.user_even = HTSUserFactory(username='user-even')
  44         self.user_even.set_password(self.password)
  45         self.affiliation_even = AffiliationFactory(name='affiliation-even', users=[self.user_even])
  46         self.admin = HTSUserFactory.create(username='admin', is_staff=True, is_superuser=True)
  47         self.admin.set_password(self.password)
  48         self.admin.save()
  49
  50         self.fc12150 = FlowCellFactory(flowcell_id='FC12150')
  51         self.fc1_id = 'FC12150'
  52         self.fc1_root = os.path.join(self.tempdir, self.fc1_id)
  53         os.mkdir(self.fc1_root)
  54         self.fc1_dir = os.path.join(self.fc1_root, 'C1-37')
  55         os.mkdir(self.fc1_dir)
  56         runxml = 'run_FC12150_2007-09-27.xml'
  57         shutil.copy(os.path.join(TESTDATA_DIR, runxml),
  58                     os.path.join(self.fc1_dir, runxml))
  59         for i in range(1, 9):
  60             affiliation = self.affiliation_odd if i % 2 == 1 else self.affiliation_even
  61             library = LibraryFactory(id="1215" + str(i))
  62             library.affiliations.add(affiliation)
  63             lane = LaneFactory(flowcell=self.fc12150, lane_number=i, library=library)
  64             shutil.copy(
  65                 os.path.join(TESTDATA_DIR,
  66                              'woldlab_070829_USI-EAS44_0017_FC11055_1.srf'),
  67                 os.path.join(self.fc1_dir,
  68                              'woldlab_070829_SERIAL_FC12150_%d.srf' %(i,))
  69                 )
  70         self.fc12150.save()
  71
  72         self.fc42jtn = FlowCellFactory(flowcell_id='42JTNAAXX')
  73         self.fc42jtn_lanes = []
  74         for i in range(1, 9):
  75             affiliation = self.affiliation_odd if i % 2 == 1 else self.affiliation_even
  76             library_type = LibraryTypeFactory(can_multiplex=True)
  77             multiplex_index = MultiplexIndexFactory(adapter_type=library_type)
  78             library = LibraryFactory(id="1300" + str(i),
  79                                      library_type=library_type,
  80                                      multiplex_id=multiplex_index.multiplex_id)
  81             library.affiliations.add(affiliation)
  82             lane = LaneFactory(flowcell=self.fc42jtn, lane_number=(i % 2) + 1, library=library)
  83             self.fc42jtn_lanes.append(lane)
  84
  85         self.fc2_dir = os.path.join(self.tempdir, '42JTNAAXX')
  86         os.mkdir(self.fc2_dir)
  87         os.mkdir(os.path.join(self.fc2_dir, 'C1-25'))
  88         os.mkdir(os.path.join(self.fc2_dir, 'C1-37'))
  89         os.mkdir(os.path.join(self.fc2_dir, 'C1-37', 'Plots'))
  90
  91     def tearDown(self):
  92         shutil.rmtree(self.tempdir)
  93
  94     def test_flowcell_information(self):
  95         """
  96         Check the code that packs the django objects into simple types.
  97         """
  98         fc12150 = self.fc12150
  99         fc42jtn = self.fc42jtn
 100         fc42ju1 = FlowCellFactory(flowcell_id='42JU1AAXX')
 101
 102         for fc_id in ['FC12150', '42JTNAAXX', '42JU1AAXX']:
 103             fc_dict = flowcell_information(fc_id)
 104             fc_django = FlowCell.objects.get(flowcell_id=fc_id)
 105             self.assertEqual(fc_dict['flowcell_id'], fc_id)
 106             self.assertEqual(fc_django.flowcell_id, fc_id)
 107             self.assertEqual(fc_dict['sequencer'], fc_django.sequencer.name)
 108             self.assertEqual(fc_dict['read_length'], fc_django.read_length)
 109             self.assertEqual(fc_dict['notes'], fc_django.notes)
 110             self.assertEqual(fc_dict['cluster_station'], fc_django.cluster_station.name)
 111
 112             for lane in fc_django.lane_set.all():
 113                 lane_contents = fc_dict['lane_set'][lane.lane_number]
 114                 lane_dict = multi_lane_to_dict(lane_contents)[lane.library_id]
 115                 self.assertEqual(lane_dict['cluster_estimate'], lane.cluster_estimate)
 116                 self.assertEqual(lane_dict['comment'], lane.comment)
 117                 self.assertEqual(lane_dict['flowcell'], lane.flowcell.flowcell_id)
 118                 self.assertEqual(lane_dict['lane_number'], lane.lane_number)
 119                 self.assertEqual(lane_dict['library_name'], lane.library.library_name)
 120                 self.assertEqual(lane_dict['library_id'], lane.library.id)
 121                 self.assertAlmostEqual(float(lane_dict['pM']), float(lane.pM))
 122                 self.assertEqual(lane_dict['library_species'],
 123                                      lane.library.library_species.scientific_name)
 124
 125             response = self.client.get('/experiments/config/%s/json' % (fc_id,), apidata)
 126             # strptime isoformat string = '%Y-%m-%dT%H:%M:%S'
 127             fc_json = json.loads(smart_text(response.content))['result']
 128             self.assertEqual(fc_json['flowcell_id'], fc_id)
 129             self.assertEqual(fc_json['sequencer'], fc_django.sequencer.name)
 130             self.assertEqual(fc_json['read_length'], fc_django.read_length)
 131             self.assertEqual(fc_json['notes'], fc_django.notes)
 132             self.assertEqual(fc_json['cluster_station'], fc_django.cluster_station.name)
 133
 134
 135             for lane in fc_django.lane_set.all():
 136                 lane_contents = fc_json['lane_set'][str(lane.lane_number)]
 137                 lane_dict = multi_lane_to_dict(lane_contents)[lane.library_id]
 138
 139                 self.assertEqual(lane_dict['cluster_estimate'], lane.cluster_estimate)
 140                 self.assertEqual(lane_dict['comment'], lane.comment)
 141                 self.assertEqual(lane_dict['flowcell'], lane.flowcell.flowcell_id)
 142                 self.assertEqual(lane_dict['lane_number'], lane.lane_number)
 143                 self.assertEqual(lane_dict['library_name'], lane.library.library_name)
 144                 self.assertEqual(lane_dict['library_id'], lane.library.id)
 145                 self.assertAlmostEqual(float(lane_dict['pM']), float(lane.pM))
 146                 self.assertEqual(lane_dict['library_species'],
 147                                      lane.library.library_species.scientific_name)
 148
 149     def test_invalid_flowcell(self):
 150         """
 151         Make sure we get a 404 if we request an invalid flowcell ID
 152         """
 153         response = self.client.get('/experiments/config/nottheone/json', apidata)
 154         self.assertEqual(response.status_code, 404)
 155
 156     def test_no_key(self):
 157         """
 158         Require logging in to retrieve meta data
 159         """
 160         response = self.client.get('/experiments/config/FC12150/json')
 161         self.assertEqual(response.status_code, 403)
 162
 163     def test_library_id(self):
 164         """
 165         Library IDs should be flexible, so make sure we can retrive a non-numeric ID
 166         """
 167         response = self.client.get('/experiments/config/FC12150/json', apidata)
 168         self.assertEqual(response.status_code, 200)
 169         flowcell = json.loads(smart_text(response.content))['result']
 170
 171         # library id is 12150 + lane number (1-8), so 12153
 172         lane_contents = flowcell['lane_set']['3']
 173         lane_library = lane_contents[0]
 174         self.assertEqual(lane_library['library_id'], '12153')
 175
 176         response = self.client.get('/samples/library/12153/json', apidata)
 177         self.assertEqual(response.status_code, 200)
 178         library_12153 = json.loads(smart_text(response.content))['result']
 179
 180         self.assertEqual(library_12153['library_id'], '12153')
 181
 182     def test_raw_id_field(self):
 183         """Test ticket:147
 184
 185         Library's have IDs, libraries also have primary keys,
 186         we eventually had enough libraries that the drop down combo
 187         box was too hard to filter through, unfortnately we want a
 188         field that uses our library id and not the internal
 189         primary key, and raw_id_field uses primary keys.
 190
 191         This tests to make sure that the value entered in the raw
 192         library id field matches the library id looked up.
 193
 194         """
 195         expected_ids = [ '1215{}'.format(i) for i in range(1,9) ]
 196         self.assertTrue(self.client.login(username=self.admin.username, password=self.password))
 197         response = self.client.get('/admin/experiments/flowcell/{}/'.format(self.fc12150.id))
 198
 199         tree = fromstring(response.content)
 200         for i in range(0,8):
 201             xpath_expression = '//input[@id="id_lane_set-%d-library"]'
 202             input_field = tree.xpath(xpath_expression % (i,))[0]
 203             library_field = input_field.find('../strong')
 204             library_id, library_name = library_field.text.split(':')
 205             # strip leading '#' sign from name
 206             library_id = library_id[1:]
 207             self.assertEqual(library_id, expected_ids[i])
 208             self.assertEqual(input_field.attrib['value'], library_id)
 209
 210     def test_library_to_flowcell_link(self):
 211         """
 212         Make sure the library page includes links to the flowcell pages.
 213         That work with flowcell IDs that have parenthetical comments.
 214         """
 215         self.assertTrue(self.client.login(username=self.admin.username, password=self.password))
 216         response = self.client.get('/library/12151/')
 217         self.assertEqual(response.status_code, 200)
 218         status = validate_xhtml(response.content)
 219         if status is not None: self.assertTrue(status)
 220
 221         tree = fromstring(response.content)
 222         flowcell_spans = tree.xpath('//span[@property="libns:flowcell_id"]',
 223                                     namespaces=NSMAP)
 224         self.assertEqual(flowcell_spans[1].text, 'FC12150')
 225         failed_fc_span = flowcell_spans[1]
 226         failed_fc_a = failed_fc_span.getparent()
 227         # make sure some of our RDF made it.
 228         self.assertEqual(failed_fc_a.get('typeof'), 'libns:IlluminaFlowcell')
 229         self.assertEqual(failed_fc_a.get('href'), '/flowcell/FC12150/')
 230         fc_response = self.client.get(failed_fc_a.get('href'))
 231         self.assertEqual(fc_response.status_code, 200)
 232         status = validate_xhtml(response.content)
 233         if status is not None: self.assertTrue(status)
 234
 235         fc_lane_response = self.client.get('/flowcell/FC12150/8/')
 236         self.assertEqual(fc_lane_response.status_code, 200)
 237         status = validate_xhtml(response.content)
 238         if status is not None: self.assertTrue(status)
 239
 240     def test_pooled_multiplex_id(self):
 241         fc_dict = flowcell_information(self.fc42jtn.flowcell_id)
 242
 243         lane_contents = fc_dict['lane_set'][2]
 244         self.assertEqual(len(lane_contents), len(self.fc42jtn_lanes) / 2)
 245         lane_dict = multi_lane_to_dict(lane_contents)
 246
 247         self.assertTrue(self.fc42jtn_lanes[0].library.multiplex_id in \
 248                         lane_dict['13001']['index_sequence'])
 249         self.assertTrue(self.fc42jtn_lanes[2].library.multiplex_id in \
 250                         lane_dict['13003']['index_sequence'])
 251
 252
 253     def test_lanes_for_view_user_odd(self):
 254         """Make sure lanes_for HTML UI works.
 255         """
 256         user = self.user_odd.username
 257         lanes = lanes_for(user)
 258         self.assertEqual(len(lanes), 8)
 259
 260         response = self.client.get(
 261             reverse('experiments.views.lanes_for',
 262                     args=[user]))
 263         self.assertEqual(response.status_code, 200)
 264         tree = fromstring(response.content)
 265         lane_trs = tree.xpath('//div[@id="changelist"]/table/tbody/tr')
 266         self.assertEqual(len(lane_trs), len(lanes))
 267         # lanes is in db order
 268         # lane_trs is in newest to oldest order
 269         for lane_tr, lane_db in zip(lane_trs, reversed(lanes)):
 270             library_id = lane_tr.xpath('td[6]/a')[0].text
 271             self.assertEqual(library_id, lane_db['library'])
 272
 273     def test_lanes_for_view_invalid_user(self):
 274         """Make sure we don't find anything with an invalid user
 275         """
 276         response = self.client.get(
 277             reverse('experiments.views.lanes_for',
 278                     args=["doesntexist"]))
 279         self.assertEqual(response.status_code, 404)
 280
 281     def test_lanes_for_json(self):
 282         """
 283         Check the code that packs the django objects into simple types.
 284         """
 285         user = self.user_odd.username
 286         lanes = lanes_for(user)
 287         self.assertEqual(len(lanes), 8)
 288
 289         response = self.client.get('/experiments/lanes_for/%s/json' % (user,), apidata)
 290         lanes_json = json.loads(smart_text(response.content))['result']
 291         self.assertEqual(len(lanes), len(lanes_json))
 292         for i in range(len(lanes)):
 293             self.assertEqual(lanes[i]['comment'], lanes_json[i]['comment'])
 294             self.assertEqual(lanes[i]['lane_number'], lanes_json[i]['lane_number'])
 295             self.assertEqual(lanes[i]['flowcell'], lanes_json[i]['flowcell'])
 296             self.assertEqual(lanes[i]['run_date'], lanes_json[i]['run_date'])
 297
 298     def test_lanes_for_no_lanes(self):
 299         """
 300         Do we get something meaningful back when the user isn't attached to anything?
 301         """
 302         user = HTSUserFactory.create(username='supertest')
 303         lanes = lanes_for(user.username)
 304         self.assertEqual(len(lanes), 0)
 305
 306         response = self.client.get('/experiments/lanes_for/%s/json' % (user,), apidata)
 307         self.assertEqual(response.status_code, 404)
 308
 309     def test_lanes_for_no_user(self):
 310         """
 311         Do we get something meaningful back when its the wrong user
 312         """
 313         user = 'not a real user'
 314         self.assertRaises(ObjectDoesNotExist, lanes_for, user)
 315
 316         response = self.client.get('/experiments/lanes_for/%s/json' % (user,), apidata)
 317         self.assertEqual(response.status_code, 404)
 318
 319     def test_raw_data_dir(self):
 320         """Raw data path generator check"""
 321         flowcell_id = self.fc1_id
 322         raw_dir = os.path.join(settings.RESULT_HOME_DIR, flowcell_id)
 323
 324         fc = FlowCell.objects.get(flowcell_id=flowcell_id)
 325         self.assertEqual(fc.get_raw_data_directory(), raw_dir)
 326
 327         fc.flowcell_id = flowcell_id + " (failed)"
 328         self.assertEqual(fc.get_raw_data_directory(), raw_dir)
 329
 330     def test_data_run_import(self):
 331         srf_file_type = FileType.objects.get(name='SRF')
 332         runxml_file_type = FileType.objects.get(name='run_xml')
 333         flowcell_id = self.fc1_id
 334         flowcell = FlowCell.objects.get(flowcell_id=flowcell_id)
 335         flowcell.update_data_runs()
 336         self.assertEqual(len(flowcell.datarun_set.all()), 1)
 337
 338         run = flowcell.datarun_set.all()[0]
 339         result_files = run.datafile_set.all()
 340         result_dict = dict(((rf.relative_pathname, rf) for rf in result_files))
 341
 342         srf4 = result_dict['FC12150/C1-37/woldlab_070829_SERIAL_FC12150_4.srf']
 343         self.assertEqual(srf4.file_type, srf_file_type)
 344         self.assertEqual(srf4.library_id, '12154')
 345         self.assertEqual(srf4.data_run.flowcell.flowcell_id, 'FC12150')
 346         self.assertEqual(
 347             srf4.data_run.flowcell.lane_set.get(lane_number=4).library_id,
 348             '12154')
 349         self.assertEqual(
 350             srf4.pathname,
 351             os.path.join(settings.RESULT_HOME_DIR, srf4.relative_pathname))
 352
 353         lane_files = run.lane_files()
 354         self.assertEqual(lane_files[4]['srf'], srf4)
 355
 356         runxml= result_dict['FC12150/C1-37/run_FC12150_2007-09-27.xml']
 357         self.assertEqual(runxml.file_type, runxml_file_type)
 358         self.assertEqual(runxml.library_id, None)
 359
 360         import1 = len(DataRun.objects.filter(result_dir='FC12150/C1-37'))
 361         # what happens if we import twice?
 362         flowcell.import_data_run('FC12150/C1-37',
 363                                  'run_FC12150_2007-09-27.xml')
 364         self.assertEqual(
 365             len(DataRun.objects.filter(result_dir='FC12150/C1-37')),
 366             import1)
 367
 368     def test_read_result_file(self):
 369         """make sure we can return a result file
 370         """
 371         flowcell_id = self.fc1_id
 372         flowcell = FlowCell.objects.get(flowcell_id=flowcell_id)
 373         flowcell.update_data_runs()
 374
 375         #self.client.login(username='supertest', password='BJOKL5kAj6aFZ6A5')
 376
 377         result_files = flowcell.datarun_set.all()[0].datafile_set.all()
 378         for f in result_files:
 379             url = '/experiments/file/%s' % ( f.random_key,)
 380             response = self.client.get(url)
 381             self.assertEqual(response.status_code, 200)
 382             mimetype = f.file_type.mimetype
 383             if mimetype is None:
 384                 mimetype = 'application/octet-stream'
 385
 386             self.assertEqual(mimetype, response['content-type'])
 387
 388     def test_flowcell_rdf(self):
 389         import RDF
 390         from htsworkflow.util.rdfhelp import get_model, \
 391              fromTypedNode, \
 392              load_string_into_model, \
 393              rdfNS, \
 394              libraryOntology, \
 395              dump_model
 396
 397         model = get_model()
 398
 399         expected = {'1': ['12151'],
 400                     '2': ['12152'],
 401                     '3': ['12153'],
 402                     '4': ['12154'],
 403                     '5': ['12155'],
 404                     '6': ['12156'],
 405                     '7': ['12157'],
 406                     '8': ['12158']}
 407         url = '/flowcell/{}/'.format(self.fc12150.flowcell_id)
 408         response = self.client.get(url)
 409         self.assertEqual(response.status_code, 200)
 410         status = validate_xhtml(response.content)
 411         if status is not None: self.assertTrue(status)
 412
 413         ns = urljoin('http://localhost', url)
 414         load_string_into_model(model, 'rdfa', smart_text(response.content), ns=ns)
 415         body = """prefix rdf: <http://www.w3.org/1999/02/22-rdf-syntax-ns#>
 416         prefix libns: <http://jumpgate.caltech.edu/wiki/LibraryOntology#>
 417
 418         select ?flowcell ?flowcell_id ?lane_id ?library_id
 419         where {
 420           ?flowcell a libns:IlluminaFlowcell ;
 421                     libns:flowcell_id ?flowcell_id ;
 422                     libns:has_lane ?lane .
 423           ?lane libns:lane_number ?lane_id ;
 424                 libns:library ?library .
 425           ?library libns:library_id ?library_id .
 426         }"""
 427         query = RDF.SPARQLQuery(body)
 428         count = 0
 429         for r in query.execute(model):
 430             count += 1
 431             self.assertEqual(fromTypedNode(r['flowcell_id']), 'FC12150')
 432             lane_id = fromTypedNode(r['lane_id'])
 433             library_id = fromTypedNode(r['library_id'])
 434             self.assertTrue(library_id in expected[lane_id])
 435         self.assertEqual(count, 8)
 436
 437 class TestEmailNotify(TestCase):
 438     def setUp(self):
 439         self.password = 'foo27'
 440         self.user = HTSUserFactory.create(username='test')
 441         self.user.set_password(self.password)
 442         self.user.save()
 443         self.admin = HTSUserFactory.create(username='admintest', is_staff=True)
 444         self.admin.set_password(self.password)
 445         self.admin.save()
 446         self.super = HTSUserFactory.create(username='supertest', is_staff=True, is_superuser=True)
 447         self.super.set_password(self.password)
 448         self.super.save()
 449
 450         self.library = LibraryFactory.create()
 451         self.affiliation = AffiliationFactory()
 452         self.affiliation.users.add(self.user)
 453         self.library.affiliations.add(self.affiliation)
 454         self.fc = FlowCellFactory.create()
 455         self.lane = LaneFactory(flowcell=self.fc, lane_number=1, library=self.library)
 456
 457         self.url = '/experiments/started/{}/'.format(self.fc.id)
 458
 459     def test_started_email_not_logged_in(self):
 460         response = self.client.get(self.url)
 461         self.assertEqual(response.status_code, 302)
 462
 463     def test_started_email_logged_in_user(self):
 464         self.assertTrue(self.client.login(username=self.user.username, password=self.password))
 465         response = self.client.get(self.url)
 466         self.assertEqual(response.status_code, 302)
 467
 468     def test_started_email_logged_in_staff(self):
 469         self.assertTrue(self.admin.is_staff)
 470         admin = HTSUser.objects.get(username=self.admin.username)
 471         self.assertTrue(admin.is_staff)
 472         self.assertTrue(admin.check_password(self.password))
 473         self.assertTrue(self.client.login(username=self.admin.username, password=self.password))
 474         response = self.client.get(self.url)
 475         self.assertEqual(response.status_code, 200)
 476
 477     def test_started_email_send(self):
 478         self.assertTrue(self.client.login(username=self.admin.username, password=self.password))
 479         response = self.client.get(self.url)
 480         self.assertEqual(response.status_code, 200)
 481
 482         self.assertTrue(self.affiliation.email in smart_text(response.content))
 483         self.assertTrue(self.library.library_name in smart_text(response.content))
 484
 485         response = self.client.get(self.url, {'send':'1','bcc':'on'})
 486         self.assertEqual(response.status_code, 200)
 487         self.assertEqual(len(mail.outbox), 2)
 488         bcc = set(settings.NOTIFICATION_BCC).copy()
 489         bcc.update(set(settings.MANAGERS))
 490         for m in mail.outbox:
 491             self.assertTrue(len(m.body) > 0)
 492             self.assertEqual(set(m.bcc), bcc)
 493
 494     def test_email_navigation(self):
 495         """
 496         Can we navigate between the flowcell and email forms properly?
 497         """
 498         admin_url = '/admin/experiments/flowcell/{}/'.format(self.fc.id)
 499         self.client.login(username=self.admin.username, password=self.password)
 500         response = self.client.get(self.url)
 501         self.assertEqual(response.status_code, 200)
 502         #print("email navigation content:", response.content)
 503         self.assertTrue(re.search(self.fc.flowcell_id, smart_text(response.content)))
 504         # require that navigation back to the admin page exists
 505         self.assertTrue(re.search('<a href="{}">[^<]+</a>'.format(admin_url),
 506                                   smart_text(response.content)))
 507
 508 def multi_lane_to_dict(lane):
 509     """Convert a list of lane entries into a dictionary indexed by library ID
 510     """
 511     return dict( ((x['library_id'],x) for x in lane) )
 512
 513 class TestSequencer(TestCase):
 514     def setUp(self):
 515         self.fc12150 = FlowCellFactory(flowcell_id='FC12150')
 516         self.library = LibraryFactory(id="12150")
 517         self.lane = LaneFactory(flowcell=self.fc12150, lane_number=1, library=self.library)
 518
 519     def test_name_generation(self):
 520         seq = Sequencer()
 521         seq.name = "Seq1"
 522         seq.instrument_name = "HWI-SEQ1"
 523         seq.model = "Imaginary 5000"
 524
 525         self.assertEqual(str(seq), "Seq1 (HWI-SEQ1)")
 526
 527     def test_lookup(self):
 528         fc = self.fc12150
 529         self.assertEqual(fc.sequencer.model, 'HiSeq 1')
 530         self.assertTrue(fc.sequencer.instrument_name.startswith('instrument name')),
 531         # well actually we let the browser tack on the host name
 532         url = fc.get_absolute_url()
 533         self.assertEqual(url, '/flowcell/FC12150/')
 534
 535     def test_rdf(self):
 536         response = self.client.get('/flowcell/FC12150/', apidata)
 537         tree = fromstring(response.content)
 538         seq_by = tree.xpath('//div[@rel="libns:sequenced_by"]',
 539                             namespaces=NSMAP)
 540         self.assertEqual(len(seq_by), 1)
 541         self.assertEqual(seq_by[0].attrib['rel'], 'libns:sequenced_by')
 542         seq = seq_by[0].getchildren()
 543         self.assertEqual(len(seq), 1)
 544         sequencer = '/sequencer/' + str(self.fc12150.sequencer.id)
 545         self.assertEqual(seq[0].attrib['about'], sequencer)
 546         self.assertEqual(seq[0].attrib['typeof'], 'libns:Sequencer')
 547
 548         name = seq[0].xpath('./span[@property="libns:sequencer_name"]')
 549         self.assertEqual(len(name), 1)
 550         self.assertTrue(name[0].text.startswith('sequencer '))
 551         instrument = seq[0].xpath(
 552             './span[@property="libns:sequencer_instrument"]')
 553         self.assertEqual(len(instrument), 1)
 554         self.assertTrue(instrument[0].text.startswith('instrument name'))
 555         model = seq[0].xpath(
 556             './span[@property="libns:sequencer_model"]')
 557         self.assertEqual(len(model), 1)
 558         self.assertEqual(model[0].text, 'HiSeq 1')
 559
 560     def test_flowcell_with_rdf_validation(self):
 561         from htsworkflow.util.rdfhelp import add_default_schemas, \
 562              dump_model, \
 563              get_model, \
 564              load_string_into_model
 565         from htsworkflow.util.rdfinfer import Infer
 566
 567         model = get_model()
 568         add_default_schemas(model)
 569         inference = Infer(model)
 570
 571         url ='/flowcell/FC12150/'
 572         response = self.client.get(url)
 573         self.assertEqual(response.status_code, 200)
 574         status = validate_xhtml(response.content)
 575         if status is not None: self.assertTrue(status)
 576
 577         load_string_into_model(model, 'rdfa', smart_text(response.content))
 578
 579         errmsgs = list(inference.run_validation())
 580         self.assertEqual(len(errmsgs), 0)
 581
 582     def test_lane_with_rdf_validation(self):
 583         from htsworkflow.util.rdfhelp import add_default_schemas, \
 584              dump_model, \
 585              get_model, \
 586              load_string_into_model
 587         from htsworkflow.util.rdfinfer import Infer
 588
 589         model = get_model()
 590         add_default_schemas(model)
 591         inference = Infer(model)
 592
 593         url = '/lane/{}'.format(self.lane.id)
 594         response = self.client.get(url)
 595         rdfbody = smart_text(response.content)
 596         self.assertEqual(response.status_code, 200)
 597         status = validate_xhtml(rdfbody)
 598         if status is not None: self.assertTrue(status)
 599
 600         load_string_into_model(model, 'rdfa', rdfbody)
 601
 602         errmsgs = list(inference.run_validation())
 603         self.assertEqual(len(errmsgs), 0)
 604
 605 def suite():
 606     from unittest import TestSuite, defaultTestLoader
 607     suite = TestSuite()
 608     for testcase in [ExerimentsTestCases,
 609                      TestEmailNotify,
 610                      TestSequencer]:
 611         suite.addTests(defaultTestLoader.loadTestsFromTestCase(testcase))
 612     return suite
 613
 614 if __name__ == "__main__":
 615     from unittest import main
 616     main(defaultTest="suite")