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