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