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         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(1,9) ]
 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')
 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: self.assertTrue(status)
 226
 227         tree = fromstring(response.content)
 228         flowcell_spans = tree.xpath('//span[@property="libns:flowcell_id"]',
 229                                     namespaces=NSMAP)
 230         self.assertEqual(flowcell_spans[1].text, 'FC12150')
 231         failed_fc_span = flowcell_spans[1]
 232         failed_fc_a = failed_fc_span.getparent()
 233         # make sure some of our RDF made it.
 234         self.assertEqual(failed_fc_a.get('typeof'), 'libns:IlluminaFlowcell')
 235         self.assertEqual(failed_fc_a.get('href'), '/flowcell/FC12150/')
 236         fc_response = self.client.get(failed_fc_a.get('href'))
 237         self.assertEqual(fc_response.status_code, 200)
 238         status = validate_xhtml(response.content)
 239         if status is not None: self.assertTrue(status)
 240
 241         fc_lane_response = self.client.get('/flowcell/FC12150/8/')
 242         self.assertEqual(fc_lane_response.status_code, 200)
 243         status = validate_xhtml(response.content)
 244         if status is not None: self.assertTrue(status)
 245
 246     def test_pooled_multiplex_id(self):
 247         fc_dict = flowcell_information(self.fc42jtn.flowcell_id)
 248
 249         lane_contents = fc_dict['lane_set'][2]
 250         self.assertEqual(len(lane_contents), len(self.fc42jtn_lanes) / 2)
 251         lane_dict = multi_lane_to_dict(lane_contents)
 252
 253         self.assertTrue(self.fc42jtn_lanes[0].library.multiplex_id in \
 254                         lane_dict['13001']['index_sequence'])
 255         self.assertTrue(self.fc42jtn_lanes[2].library.multiplex_id in \
 256                         lane_dict['13003']['index_sequence'])
 257
 258
 259     def test_lanes_for_view_user_odd(self):
 260         """Make sure lanes_for HTML UI works.
 261         """
 262         user = self.user_odd.username
 263         lanes = lanes_for(user)
 264         self.assertEqual(len(lanes), 8)
 265
 266         response = self.client.get(
 267             reverse('lanes_for', kwargs={'username': user}))
 268         self.assertEqual(response.status_code, 200)
 269         tree = fromstring(response.content)
 270         lane_trs = tree.xpath('//div[@id="changelist"]/table/tbody/tr')
 271         self.assertEqual(len(lane_trs), len(lanes))
 272         # lanes is in db order
 273         # lane_trs is in newest to oldest order
 274         for lane_tr, lane_db in zip(lane_trs, lanes):
 275             library_id = lane_tr.xpath('td[6]/a')[0].text
 276             self.assertEqual(library_id, lane_db['library'])
 277
 278     def test_lanes_for_view_invalid_user(self):
 279         """Make sure we don't find anything with an invalid user
 280         """
 281         response = self.client.get(
 282             reverse('lanes_for', kwargs={'username': 'doesntexist'}))
 283         self.assertEqual(response.status_code, 404)
 284
 285     def test_lanes_for_json(self):
 286         """
 287         Check the code that packs the django objects into simple types.
 288         """
 289         user = self.user_odd.username
 290         lanes = lanes_for(user)
 291         self.assertEqual(len(lanes), 8)
 292
 293         response = self.client.get('/experiments/lanes_for/%s/json' % (user,), apidata)
 294         lanes_json = json.loads(smart_text(response.content))['result']
 295         self.assertEqual(len(lanes), len(lanes_json))
 296         for i in range(len(lanes)):
 297             self.assertEqual(lanes[i]['comment'], lanes_json[i]['comment'])
 298             self.assertEqual(lanes[i]['lane_number'], lanes_json[i]['lane_number'])
 299             self.assertEqual(lanes[i]['flowcell'], lanes_json[i]['flowcell'])
 300             self.assertEqual(lanes[i]['run_date'], lanes_json[i]['run_date'])
 301
 302     def test_lanes_for_no_lanes(self):
 303         """
 304         Do we get something meaningful back when the user isn't attached to anything?
 305         """
 306         user = HTSUserFactory.create(username='supertest')
 307         lanes = lanes_for(user.username)
 308         self.assertEqual(len(lanes), 0)
 309
 310         response = self.client.get('/experiments/lanes_for/%s/json' % (user,), apidata)
 311         self.assertEqual(response.status_code, 404)
 312
 313     def test_lanes_for_no_user(self):
 314         """
 315         Do we get something meaningful back when its the wrong user
 316         """
 317         user = 'not a real user'
 318         self.assertRaises(ObjectDoesNotExist, lanes_for, user)
 319
 320         response = self.client.get('/experiments/lanes_for/%s/json' % (user,), apidata)
 321         self.assertEqual(response.status_code, 404)
 322
 323     def test_raw_data_dir(self):
 324         """Raw data path generator check"""
 325         flowcell_id = self.fc1_id
 326         raw_dir = os.path.join(settings.RESULT_HOME_DIR, flowcell_id)
 327
 328         fc = FlowCell.objects.get(flowcell_id=flowcell_id)
 329         self.assertEqual(fc.get_raw_data_directory(), raw_dir)
 330
 331         fc.flowcell_id = flowcell_id + " (failed)"
 332         self.assertEqual(fc.get_raw_data_directory(), raw_dir)
 333
 334     def test_sequencing_run_import(self):
 335         srf_file_type = FileType.objects.get(name='SRF')
 336         runxml_file_type = FileType.objects.get(name='run_xml')
 337         flowcell_id = self.fc1_id
 338         flowcell = FlowCell.objects.get(flowcell_id=flowcell_id)
 339         flowcell.update_sequencing_runs()
 340         self.assertEqual(len(flowcell.sequencingrun_set.all()), 1)
 341
 342         run = flowcell.sequencingrun_set.all()[0]
 343         result_files = run.datafile_set.all()
 344         result_dict = dict(((rf.relative_pathname, rf) for rf in result_files))
 345
 346         srf4 = result_dict['FC12150/C1-37/woldlab_070829_SERIAL_FC12150_4.srf']
 347         self.assertEqual(srf4.file_type, srf_file_type)
 348         self.assertEqual(srf4.library_id, '12154')
 349         self.assertEqual(srf4.sequencing_run.flowcell.flowcell_id, 'FC12150')
 350         self.assertEqual(
 351             srf4.sequencing_run.flowcell.lane_set.get(lane_number=4).library_id,
 352             '12154')
 353         self.assertEqual(
 354             srf4.pathname,
 355             os.path.join(settings.RESULT_HOME_DIR, srf4.relative_pathname))
 356
 357         lane_files = run.lane_files()
 358         self.assertEqual(lane_files[4]['srf'], srf4)
 359
 360         runxml= result_dict['FC12150/C1-37/run_FC12150_2007-09-27.xml']
 361         self.assertEqual(runxml.file_type, runxml_file_type)
 362         self.assertEqual(runxml.library_id, None)
 363
 364         import1 = len(SequencingRun.objects.filter(result_dir='FC12150/C1-37'))
 365         # what happens if we import twice?
 366         flowcell.import_sequencing_run('FC12150/C1-37',
 367                                        'run_FC12150_2007-09-27.xml')
 368         self.assertEqual(
 369             len(SequencingRun.objects.filter(result_dir='FC12150/C1-37')),
 370             import1)
 371
 372     def test_read_result_file(self):
 373         """make sure we can return a result file
 374         """
 375         flowcell_id = self.fc1_id
 376         flowcell = FlowCell.objects.get(flowcell_id=flowcell_id)
 377         flowcell.update_sequencing_runs()
 378
 379         #self.client.login(username='supertest', password='BJOKL5kAj6aFZ6A5')
 380
 381         result_files = flowcell.sequencingrun_set.all()[0].datafile_set.all()
 382         for f in result_files:
 383             url = '/experiments/file/%s' % ( f.random_key,)
 384             response = self.client.get(url)
 385             self.assertEqual(response.status_code, 200)
 386             mimetype = f.file_type.mimetype
 387             if mimetype is None:
 388                 mimetype = 'application/octet-stream'
 389
 390             self.assertEqual(mimetype, response['content-type'])
 391
 392     def test_flowcell_rdf(self):
 393         import RDF
 394         from htsworkflow.util.rdfhelp import get_model, \
 395              fromTypedNode, \
 396              load_string_into_model, \
 397              rdfNS, \
 398              libraryOntology, \
 399              dump_model
 400
 401         model = get_model()
 402
 403         expected = {'1': ['12151'],
 404                     '2': ['12152'],
 405                     '3': ['12153'],
 406                     '4': ['12154'],
 407                     '5': ['12155'],
 408                     '6': ['12156'],
 409                     '7': ['12157'],
 410                     '8': ['12158']}
 411         url = '/flowcell/{}/'.format(self.fc12150.flowcell_id)
 412         response = self.client.get(url)
 413         self.assertEqual(response.status_code, 200)
 414         status = validate_xhtml(response.content)
 415         if status is not None: self.assertTrue(status)
 416
 417         ns = urljoin('http://localhost', url)
 418         load_string_into_model(model, 'rdfa', smart_text(response.content), ns=ns)
 419         body = """prefix rdf: <http://www.w3.org/1999/02/22-rdf-syntax-ns#>
 420         prefix libns: <http://jumpgate.caltech.edu/wiki/LibraryOntology#>
 421
 422         select ?flowcell ?flowcell_id ?lane_id ?library_id
 423         where {
 424           ?flowcell a libns:IlluminaFlowcell ;
 425                     libns:flowcell_id ?flowcell_id ;
 426                     libns:has_lane ?lane .
 427           ?lane libns:lane_number ?lane_id ;
 428                 libns:library ?library .
 429           ?library libns:library_id ?library_id .
 430         }"""
 431         query = RDF.SPARQLQuery(body)
 432         count = 0
 433         for r in query.execute(model):
 434             count += 1
 435             self.assertEqual(fromTypedNode(r['flowcell_id']), 'FC12150')
 436             lane_id = fromTypedNode(r['lane_id'])
 437             library_id = fromTypedNode(r['library_id'])
 438             self.assertTrue(library_id in expected[lane_id])
 439         self.assertEqual(count, 8)
 440
 441 class TestEmailNotify(TestCase):
 442     def setUp(self):
 443         self.password = 'foo27'
 444         self.user = HTSUserFactory.create(username='test')
 445         self.user.set_password(self.password)
 446         self.user.save()
 447         self.admin = HTSUserFactory.create(username='admintest', is_staff=True)
 448         self.admin.set_password(self.password)
 449         self.admin.save()
 450         self.superuser = HTSUserFactory.create(username='supertest', is_staff=True, is_superuser=True)
 451         self.superuser.set_password(self.password)
 452         self.superuser.save()
 453
 454         self.library = LibraryFactory.create()
 455         self.affiliation = AffiliationFactory()
 456         self.affiliation.users.add(self.user)
 457         self.library.affiliations.add(self.affiliation)
 458         self.fc = FlowCellFactory.create()
 459         self.lane = LaneFactory(flowcell=self.fc, lane_number=1, library=self.library)
 460
 461         self.url = '/experiments/started/{}/'.format(self.fc.id)
 462
 463     def tearDown(self):
 464         # with django 1.10 running against postgresql I had to delete these
 465         # test objects or else I get a constraint error
 466         self.affiliation.delete()
 467         self.user.delete()
 468         self.admin.delete()
 469         self.superuser.delete()
 470
 471     def test_started_email_not_logged_in(self):
 472         response = self.client.get(self.url)
 473         self.assertEqual(response.status_code, 302)
 474
 475     def test_started_email_logged_in_user(self):
 476         self.assertTrue(self.client.login(username=self.user.username, password=self.password))
 477         response = self.client.get(self.url)
 478         self.assertEqual(response.status_code, 302)
 479
 480     def test_started_email_logged_in_staff(self):
 481         self.assertTrue(self.admin.is_staff)
 482         admin = HTSUser.objects.get(username=self.admin.username)
 483         self.assertTrue(admin.is_staff)
 484         self.assertTrue(admin.check_password(self.password))
 485         self.assertTrue(self.client.login(username=self.admin.username, password=self.password))
 486         response = self.client.get(self.url)
 487         self.assertEqual(response.status_code, 200)
 488
 489     def test_started_email_send(self):
 490         self.assertTrue(self.client.login(username=self.admin.username, password=self.password))
 491         response = self.client.get(self.url)
 492         self.assertEqual(response.status_code, 200)
 493
 494         self.assertTrue(self.affiliation.email in smart_text(response.content))
 495         self.assertTrue(self.library.library_name in smart_text(response.content))
 496
 497         response = self.client.get(self.url, {'send':'1','bcc':'on'})
 498         self.assertEqual(response.status_code, 200)
 499         self.assertEqual(len(mail.outbox), 2)
 500         bcc = set(settings.NOTIFICATION_BCC).copy()
 501         bcc.update(set(settings.MANAGERS))
 502         for m in mail.outbox:
 503             self.assertTrue(len(m.body) > 0)
 504             self.assertEqual(set(m.bcc), bcc)
 505
 506     def test_email_navigation(self):
 507         """
 508         Can we navigate between the flowcell and email forms properly?
 509         """
 510         admin_url = '/admin/experiments/flowcell/{}/change/'.format(self.fc.id)
 511         self.client.login(username=self.admin.username, password=self.password)
 512         response = self.client.get(self.url)
 513         self.assertEqual(response.status_code, 200)
 514         self.assertTrue(re.search(self.fc.flowcell_id, smart_text(response.content)))
 515         # require that navigation back to the admin page exists
 516         admin_a_tag = '<a href="{}">[^<]+</a>'.format(admin_url)
 517         self.assertTrue(re.search(admin_a_tag, smart_text(response.content)))
 518
 519 def multi_lane_to_dict(lane):
 520     """Convert a list of lane entries into a dictionary indexed by library ID
 521     """
 522     return dict( ((x['library_id'],x) for x in lane) )
 523
 524 class TestSequencer(TestCase):
 525     def setUp(self):
 526         self.fc12150 = FlowCellFactory(flowcell_id='FC12150')
 527         self.library = LibraryFactory(id="12150")
 528         self.lane = LaneFactory(flowcell=self.fc12150, lane_number=1, library=self.library)
 529
 530     def test_name_generation(self):
 531         seq = Sequencer()
 532         seq.name = "Seq1"
 533         seq.instrument_name = "HWI-SEQ1"
 534         seq.model = "Imaginary 5000"
 535
 536         self.assertEqual(str(seq), "Seq1 (HWI-SEQ1)")
 537
 538     def test_lookup(self):
 539         fc = self.fc12150
 540         self.assertEqual(fc.sequencer.model, 'HiSeq 1')
 541         self.assertTrue(fc.sequencer.instrument_name.startswith('instrument name')),
 542         # well actually we let the browser tack on the host name
 543         url = fc.get_absolute_url()
 544         self.assertEqual(url, '/flowcell/FC12150/')
 545
 546     def test_rdf(self):
 547         response = self.client.get('/flowcell/FC12150/', apidata)
 548         tree = fromstring(response.content)
 549         seq_by = tree.xpath('//div[@rel="libns:sequenced_by"]',
 550                             namespaces=NSMAP)
 551         self.assertEqual(len(seq_by), 1)
 552         self.assertEqual(seq_by[0].attrib['rel'], 'libns:sequenced_by')
 553         seq = seq_by[0].getchildren()
 554         self.assertEqual(len(seq), 1)
 555         sequencer = '/sequencer/' + str(self.fc12150.sequencer.id)
 556         self.assertEqual(seq[0].attrib['about'], sequencer)
 557         self.assertEqual(seq[0].attrib['typeof'], 'libns:Sequencer')
 558
 559         name = seq[0].xpath('./span[@property="libns:sequencer_name"]')
 560         self.assertEqual(len(name), 1)
 561         self.assertTrue(name[0].text.startswith('sequencer '))
 562         instrument = seq[0].xpath(
 563             './span[@property="libns:sequencer_instrument"]')
 564         self.assertEqual(len(instrument), 1)
 565         self.assertTrue(instrument[0].text.startswith('instrument name'))
 566         model = seq[0].xpath(
 567             './span[@property="libns:sequencer_model"]')
 568         self.assertEqual(len(model), 1)
 569         self.assertEqual(model[0].text, 'HiSeq 1')
 570
 571     def test_flowcell_with_rdf_validation(self):
 572         from htsworkflow.util.rdfhelp import add_default_schemas, \
 573              dump_model, \
 574              get_model, \
 575              load_string_into_model
 576         from htsworkflow.util.rdfinfer import Infer
 577
 578         model = get_model()
 579         add_default_schemas(model)
 580         inference = Infer(model)
 581
 582         url ='/flowcell/FC12150/'
 583         response = self.client.get(url)
 584         self.assertEqual(response.status_code, 200)
 585         status = validate_xhtml(response.content)
 586         if status is not None: self.assertTrue(status)
 587
 588         load_string_into_model(model, 'rdfa', smart_text(response.content))
 589
 590         errmsgs = list(inference.run_validation())
 591         self.assertEqual(len(errmsgs), 0)
 592
 593     def test_lane_with_rdf_validation(self):
 594         from htsworkflow.util.rdfhelp import add_default_schemas, \
 595              dump_model, \
 596              get_model, \
 597              load_string_into_model
 598         from htsworkflow.util.rdfinfer import Infer
 599
 600         model = get_model()
 601         add_default_schemas(model)
 602         inference = Infer(model)
 603
 604         url = '/lane/{}'.format(self.lane.id)
 605         response = self.client.get(url)
 606         rdfbody = smart_text(response.content)
 607         self.assertEqual(response.status_code, 200)
 608         status = validate_xhtml(rdfbody)
 609         if status is not None: self.assertTrue(status)
 610
 611         load_string_into_model(model, 'rdfa', rdfbody)
 612
 613         errmsgs = list(inference.run_validation())
 614         self.assertEqual(len(errmsgs), 0)
 615
 616 def suite():
 617     from unittest import TestSuite, defaultTestLoader
 618     suite = TestSuite()
 619     for testcase in [ExerimentsTestCases,
 620                      TestEmailNotify,
 621                      TestSequencer]:
 622         suite.addTests(defaultTestLoader.loadTestsFromTestCase(testcase))
 623     return suite
 624
 625 if __name__ == "__main__":
 626     from unittest import main
 627     main(defaultTest="suite")