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