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