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 sys
   9 import tempfile
  10 from six.moves.urllib.parse import urljoin
  11
  12 from django.conf import settings
  13 from django.core import mail
  14 from django.core.exceptions import ObjectDoesNotExist
  15 from django.test import TestCase
  16 from django.test.utils import setup_test_environment, teardown_test_environment
  17 from django.db import connection
  18 from django.conf import settings
  19 from django.utils.encoding import smart_text
  20
  21 from .models import ClusterStation, cluster_station_default, \
  22     DataRun, Sequencer, FlowCell, FileType
  23 from samples.models import HTSUser
  24 from .experiments import flowcell_information, lanes_for
  25 from .experiments_factory import ClusterStationFactory, FlowCellFactory, LaneFactory
  26 from samples.samples_factory import AffiliationFactory, HTSUserFactory, \
  27     LibraryFactory, LibraryTypeFactory, MultiplexIndexFactory
  28 from htsworkflow.auth import apidata
  29 from htsworkflow.util.ethelp import validate_xhtml
  30
  31 from htsworkflow.pipelines.test.simulate_runfolder import TESTDATA_DIR
  32
  33 LANE_SET = range(1,9)
  34
  35 NSMAP = {'libns':'http://jumpgate.caltech.edu/wiki/LibraryOntology#'}
  36
  37 from django.db import connection
  38
  39
  40 class ExperimentsTestCases(TestCase):
  41     def setUp(self):
  42         # Generate at least one fleshed out example flowcell
  43         self.tempdir = tempfile.mkdtemp(prefix='htsw-test-experiments-')
  44         settings.RESULT_HOME_DIR = self.tempdir
  45
  46         self.password = 'password'
  47         self.user_odd = HTSUserFactory(username='user-odd')
  48         self.user_odd.set_password(self.password)
  49         self.affiliation_odd = AffiliationFactory(name='affiliation-odd', users=[self.user_odd])
  50         self.user_even = HTSUserFactory(username='user-even')
  51         self.user_even.set_password(self.password)
  52         self.affiliation_even = AffiliationFactory(name='affiliation-even', users=[self.user_even])
  53         self.admin = HTSUserFactory.create(username='admin', is_staff=True, is_superuser=True)
  54         self.admin.set_password(self.password)
  55         self.admin.save()
  56
  57         self.fc12150 = FlowCellFactory(flowcell_id='FC12150')
  58         self.fc1_id = 'FC12150'
  59         self.fc1_root = os.path.join(self.tempdir, self.fc1_id)
  60         os.mkdir(self.fc1_root)
  61         self.fc1_dir = os.path.join(self.fc1_root, 'C1-37')
  62         os.mkdir(self.fc1_dir)
  63         runxml = 'run_FC12150_2007-09-27.xml'
  64         shutil.copy(os.path.join(TESTDATA_DIR, runxml),
  65                     os.path.join(self.fc1_dir, runxml))
  66         for i in range(1,9):
  67             affiliation = self.affiliation_odd if i % 2 == 1 else self.affiliation_even
  68             library = LibraryFactory(id="1215" + str(i))
  69             library.affiliations.add(affiliation)
  70             lane = LaneFactory(flowcell=self.fc12150, lane_number=i, library=library)
  71             shutil.copy(
  72                 os.path.join(TESTDATA_DIR,
  73                              'woldlab_070829_USI-EAS44_0017_FC11055_1.srf'),
  74                 os.path.join(self.fc1_dir,
  75                              'woldlab_070829_SERIAL_FC12150_%d.srf' %(i,))
  76                 )
  77         self.fc12150.save()
  78
  79         self.fc42jtn = FlowCellFactory(flowcell_id='42JTNAAXX')
  80         self.fc42jtn_lanes = []
  81         for i in range(1,9):
  82             affiliation = self.affiliation_odd if i % 2 == 1 else self.affiliation_even
  83             library_type = LibraryTypeFactory(can_multiplex=True)
  84             multiplex_index = MultiplexIndexFactory(adapter_type=library_type)
  85             library = LibraryFactory(id="1300" + str(i),
  86                                      library_type=library_type,
  87                                      multiplex_id=multiplex_index.multiplex_id)
  88             library.affiliations.add(affiliation)
  89             lane = LaneFactory(flowcell=self.fc42jtn, lane_number=(i % 2) + 1, library=library)
  90             self.fc42jtn_lanes.append(lane)
  91
  92         self.fc2_dir = os.path.join(self.tempdir, '42JTNAAXX')
  93         os.mkdir(self.fc2_dir)
  94         os.mkdir(os.path.join(self.fc2_dir, 'C1-25'))
  95         os.mkdir(os.path.join(self.fc2_dir, 'C1-37'))
  96         os.mkdir(os.path.join(self.fc2_dir, 'C1-37', 'Plots'))
  97
  98     def tearDown(self):
  99         shutil.rmtree(self.tempdir)
 100
 101     def test_flowcell_information(self):
 102         """
 103         Check the code that packs the django objects into simple types.
 104         """
 105         fc12150 = self.fc12150
 106         fc42jtn = self.fc42jtn
 107         fc42ju1 = FlowCellFactory(flowcell_id='42JU1AAXX')
 108
 109         for fc_id in ['FC12150', '42JTNAAXX', '42JU1AAXX']:
 110             fc_dict = flowcell_information(fc_id)
 111             fc_django = FlowCell.objects.get(flowcell_id=fc_id)
 112             self.assertEqual(fc_dict['flowcell_id'], fc_id)
 113             self.assertEqual(fc_django.flowcell_id, fc_id)
 114             self.assertEqual(fc_dict['sequencer'], fc_django.sequencer.name)
 115             self.assertEqual(fc_dict['read_length'], fc_django.read_length)
 116             self.assertEqual(fc_dict['notes'], fc_django.notes)
 117             self.assertEqual(fc_dict['cluster_station'], fc_django.cluster_station.name)
 118
 119             for lane in fc_django.lane_set.all():
 120                 lane_contents = fc_dict['lane_set'][lane.lane_number]
 121                 lane_dict = multi_lane_to_dict(lane_contents)[lane.library_id]
 122                 self.assertEqual(lane_dict['cluster_estimate'], lane.cluster_estimate)
 123                 self.assertEqual(lane_dict['comment'], lane.comment)
 124                 self.assertEqual(lane_dict['flowcell'], lane.flowcell.flowcell_id)
 125                 self.assertEqual(lane_dict['lane_number'], lane.lane_number)
 126                 self.assertEqual(lane_dict['library_name'], lane.library.library_name)
 127                 self.assertEqual(lane_dict['library_id'], lane.library.id)
 128                 self.assertAlmostEqual(float(lane_dict['pM']), float(lane.pM))
 129                 self.assertEqual(lane_dict['library_species'],
 130                                      lane.library.library_species.scientific_name)
 131
 132             response = self.client.get('/experiments/config/%s/json' % (fc_id,), apidata)
 133             # strptime isoformat string = '%Y-%m-%dT%H:%M:%S'
 134             fc_json = json.loads(smart_text(response.content))['result']
 135             self.assertEqual(fc_json['flowcell_id'], fc_id)
 136             self.assertEqual(fc_json['sequencer'], fc_django.sequencer.name)
 137             self.assertEqual(fc_json['read_length'], fc_django.read_length)
 138             self.assertEqual(fc_json['notes'], fc_django.notes)
 139             self.assertEqual(fc_json['cluster_station'], fc_django.cluster_station.name)
 140
 141
 142             for lane in fc_django.lane_set.all():
 143                 lane_contents = fc_json['lane_set'][str(lane.lane_number)]
 144                 lane_dict = multi_lane_to_dict(lane_contents)[lane.library_id]
 145
 146                 self.assertEqual(lane_dict['cluster_estimate'], lane.cluster_estimate)
 147                 self.assertEqual(lane_dict['comment'], lane.comment)
 148                 self.assertEqual(lane_dict['flowcell'], lane.flowcell.flowcell_id)
 149                 self.assertEqual(lane_dict['lane_number'], lane.lane_number)
 150                 self.assertEqual(lane_dict['library_name'], lane.library.library_name)
 151                 self.assertEqual(lane_dict['library_id'], lane.library.id)
 152                 self.assertAlmostEqual(float(lane_dict['pM']), float(lane.pM))
 153                 self.assertEqual(lane_dict['library_species'],
 154                                      lane.library.library_species.scientific_name)
 155
 156     def test_invalid_flowcell(self):
 157         """
 158         Make sure we get a 404 if we request an invalid flowcell ID
 159         """
 160         response = self.client.get('/experiments/config/nottheone/json', apidata)
 161         self.assertEqual(response.status_code, 404)
 162
 163     def test_no_key(self):
 164         """
 165         Require logging in to retrieve meta data
 166         """
 167         response = self.client.get('/experiments/config/FC12150/json')
 168         self.assertEqual(response.status_code, 403)
 169
 170     def test_library_id(self):
 171         """
 172         Library IDs should be flexible, so make sure we can retrive a non-numeric ID
 173         """
 174         response = self.client.get('/experiments/config/FC12150/json', apidata)
 175         self.assertEqual(response.status_code, 200)
 176         flowcell = json.loads(smart_text(response.content))['result']
 177
 178         # library id is 12150 + lane number (1-8), so 12153
 179         lane_contents = flowcell['lane_set']['3']
 180         lane_library = lane_contents[0]
 181         self.assertEqual(lane_library['library_id'], '12153')
 182
 183         response = self.client.get('/samples/library/12153/json', apidata)
 184         self.assertEqual(response.status_code, 200)
 185         library_12153 = json.loads(smart_text(response.content))['result']
 186
 187         self.assertEqual(library_12153['library_id'], '12153')
 188
 189     def test_raw_id_field(self):
 190         """
 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 box was too
 195         hard to filter through, unfortnately we want a field that uses our library
 196         id and not the internal primary key, and raw_id_field uses primary keys.
 197
 198         This tests to make sure that the value entered in the raw library id field matches
 199         the library id looked up.
 200         """
 201         expected_ids = [ '1215{}'.format(i) for i in range(1,9) ]
 202         self.assertTrue(self.client.login(username=self.admin.username, password=self.password))
 203         response = self.client.get('/admin/experiments/flowcell/{}/'.format(self.fc12150.id))
 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('experiments.views.lanes_for',
 268                     args=[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, reversed(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('experiments.views.lanes_for',
 284                     args=["doesntexist"]))
 285         self.assertEqual(response.status_code, 404)
 286
 287     def test_lanes_for_json(self):
 288         """
 289         Check the code that packs the django objects into simple types.
 290         """
 291         user = self.user_odd.username
 292         lanes = lanes_for(user)
 293         self.assertEqual(len(lanes), 8)
 294
 295         response = self.client.get('/experiments/lanes_for/%s/json' % (user,), apidata)
 296         lanes_json = json.loads(smart_text(response.content))['result']
 297         self.assertEqual(len(lanes), len(lanes_json))
 298         for i in range(len(lanes)):
 299             self.assertEqual(lanes[i]['comment'], lanes_json[i]['comment'])
 300             self.assertEqual(lanes[i]['lane_number'], lanes_json[i]['lane_number'])
 301             self.assertEqual(lanes[i]['flowcell'], lanes_json[i]['flowcell'])
 302             self.assertEqual(lanes[i]['run_date'], lanes_json[i]['run_date'])
 303
 304     def test_lanes_for_no_lanes(self):
 305         """
 306         Do we get something meaningful back when the user isn't attached to anything?
 307         """
 308         user = HTSUserFactory.create(username='supertest')
 309         lanes = lanes_for(user.username)
 310         self.assertEqual(len(lanes), 0)
 311
 312         response = self.client.get('/experiments/lanes_for/%s/json' % (user,), apidata)
 313         self.assertEqual(response.status_code, 404)
 314
 315     def test_lanes_for_no_user(self):
 316         """
 317         Do we get something meaningful back when its the wrong user
 318         """
 319         user = 'not a real user'
 320         self.assertRaises(ObjectDoesNotExist, lanes_for, user)
 321
 322         response = self.client.get('/experiments/lanes_for/%s/json' % (user,), apidata)
 323         self.assertEqual(response.status_code, 404)
 324
 325     def test_raw_data_dir(self):
 326         """Raw data path generator check"""
 327         flowcell_id = self.fc1_id
 328         raw_dir = os.path.join(settings.RESULT_HOME_DIR, flowcell_id)
 329
 330         fc = FlowCell.objects.get(flowcell_id=flowcell_id)
 331         self.assertEqual(fc.get_raw_data_directory(), raw_dir)
 332
 333         fc.flowcell_id = flowcell_id + " (failed)"
 334         self.assertEqual(fc.get_raw_data_directory(), raw_dir)
 335
 336
 337     def test_data_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_data_runs()
 343         self.assertEqual(len(flowcell.datarun_set.all()), 1)
 344
 345         run = flowcell.datarun_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.data_run.flowcell.flowcell_id, 'FC12150')
 353         self.assertEqual(
 354             srf4.data_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(DataRun.objects.filter(result_dir='FC12150/C1-37'))
 368         # what happens if we import twice?
 369         flowcell.import_data_run('FC12150/C1-37',
 370                                  'run_FC12150_2007-09-27.xml')
 371         self.assertEqual(
 372             len(DataRun.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_data_runs()
 381
 382         #self.client.login(username='supertest', password='BJOKL5kAj6aFZ6A5')
 383
 384         result_files = flowcell.datarun_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.super = HTSUserFactory.create(username='supertest', is_staff=True, is_superuser=True)
 454         self.super.set_password(self.password)
 455         self.super.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 test_started_email_not_logged_in(self):
 467         response = self.client.get(self.url)
 468         self.assertEqual(response.status_code, 302)
 469
 470     def test_started_email_logged_in_user(self):
 471         self.assertTrue(self.client.login(username=self.user.username, password=self.password))
 472         response = self.client.get(self.url)
 473         self.assertEqual(response.status_code, 302)
 474
 475     def test_started_email_logged_in_staff(self):
 476         self.assertTrue(self.admin.is_staff)
 477         admin = HTSUser.objects.get(username=self.admin.username)
 478         self.assertTrue(admin.is_staff)
 479         self.assertTrue(admin.check_password(self.password))
 480         self.assertTrue(self.client.login(username=self.admin.username, password=self.password))
 481         response = self.client.get(self.url)
 482         self.assertEqual(response.status_code, 200)
 483
 484     def test_started_email_send(self):
 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         self.assertTrue(self.affiliation.email in smart_text(response.content))
 490         self.assertTrue(self.library.library_name in smart_text(response.content))
 491
 492         response = self.client.get(self.url, {'send':'1','bcc':'on'})
 493         self.assertEqual(response.status_code, 200)
 494         self.assertEqual(len(mail.outbox), 2)
 495         bcc = set(settings.NOTIFICATION_BCC).copy()
 496         bcc.update(set(settings.MANAGERS))
 497         for m in mail.outbox:
 498             self.assertTrue(len(m.body) > 0)
 499             self.assertEqual(set(m.bcc), bcc)
 500
 501     def test_email_navigation(self):
 502         """
 503         Can we navigate between the flowcell and email forms properly?
 504         """
 505         admin_url = '/admin/experiments/flowcell/{}/'.format(self.fc.id)
 506         self.client.login(username=self.admin.username, password=self.password)
 507         response = self.client.get(self.url)
 508         self.assertEqual(response.status_code, 200)
 509         #print("email navigation content:", response.content)
 510         self.assertTrue(re.search(self.fc.flowcell_id, smart_text(response.content)))
 511         # require that navigation back to the admin page exists
 512         self.assertTrue(re.search('<a href="{}">[^<]+</a>'.format(admin_url),
 513                                   smart_text(response.content)))
 514
 515 def multi_lane_to_dict(lane):
 516     """Convert a list of lane entries into a dictionary indexed by library ID
 517     """
 518     return dict( ((x['library_id'],x) for x in lane) )
 519
 520 class TestSequencer(TestCase):
 521     def setUp(self):
 522         self.fc12150 = FlowCellFactory(flowcell_id='FC12150')
 523         self.library = LibraryFactory(id="12150")
 524         self.lane = LaneFactory(flowcell=self.fc12150, lane_number=1, library=self.library)
 525
 526     def test_name_generation(self):
 527         seq = Sequencer()
 528         seq.name = "Seq1"
 529         seq.instrument_name = "HWI-SEQ1"
 530         seq.model = "Imaginary 5000"
 531
 532         self.assertEqual(str(seq), "Seq1 (HWI-SEQ1)")
 533
 534     def test_lookup(self):
 535         fc = self.fc12150
 536         self.assertEqual(fc.sequencer.model, 'HiSeq 1')
 537         self.assertTrue(fc.sequencer.instrument_name.startswith('instrument name')),
 538         # well actually we let the browser tack on the host name
 539         url = fc.get_absolute_url()
 540         self.assertEqual(url, '/flowcell/FC12150/')
 541
 542     def test_rdf(self):
 543         response = self.client.get('/flowcell/FC12150/', apidata)
 544         tree = fromstring(response.content)
 545         seq_by = tree.xpath('//div[@rel="libns:sequenced_by"]',
 546                             namespaces=NSMAP)
 547         self.assertEqual(len(seq_by), 1)
 548         self.assertEqual(seq_by[0].attrib['rel'], 'libns:sequenced_by')
 549         seq = seq_by[0].getchildren()
 550         self.assertEqual(len(seq), 1)
 551         sequencer = '/sequencer/' + str(self.fc12150.sequencer.id)
 552         self.assertEqual(seq[0].attrib['about'], sequencer)
 553         self.assertEqual(seq[0].attrib['typeof'], 'libns:Sequencer')
 554
 555         name = seq[0].xpath('./span[@property="libns:sequencer_name"]')
 556         self.assertEqual(len(name), 1)
 557         self.assertTrue(name[0].text.startswith('sequencer '))
 558         instrument = seq[0].xpath(
 559             './span[@property="libns:sequencer_instrument"]')
 560         self.assertEqual(len(instrument), 1)
 561         self.assertTrue(instrument[0].text.startswith('instrument name'))
 562         model = seq[0].xpath(
 563             './span[@property="libns:sequencer_model"]')
 564         self.assertEqual(len(model), 1)
 565         self.assertEqual(model[0].text, 'HiSeq 1')
 566
 567     def test_flowcell_with_rdf_validation(self):
 568         from htsworkflow.util.rdfhelp import add_default_schemas, \
 569              dump_model, \
 570              get_model, \
 571              load_string_into_model
 572         from htsworkflow.util.rdfinfer import Infer
 573
 574         model = get_model()
 575         add_default_schemas(model)
 576         inference = Infer(model)
 577
 578         url ='/flowcell/FC12150/'
 579         response = self.client.get(url)
 580         self.assertEqual(response.status_code, 200)
 581         status = validate_xhtml(response.content)
 582         if status is not None: self.assertTrue(status)
 583
 584         load_string_into_model(model, 'rdfa', smart_text(response.content))
 585
 586         errmsgs = list(inference.run_validation())
 587         self.assertEqual(len(errmsgs), 0)
 588
 589     def test_lane_with_rdf_validation(self):
 590         from htsworkflow.util.rdfhelp import add_default_schemas, \
 591              dump_model, \
 592              get_model, \
 593              load_string_into_model
 594         from htsworkflow.util.rdfinfer import Infer
 595
 596         model = get_model()
 597         add_default_schemas(model)
 598         inference = Infer(model)
 599
 600         url = '/lane/{}'.format(self.lane.id)
 601         response = self.client.get(url)
 602         rdfbody = smart_text(response.content)
 603         self.assertEqual(response.status_code, 200)
 604         status = validate_xhtml(rdfbody)
 605         if status is not None: self.assertTrue(status)
 606
 607         load_string_into_model(model, 'rdfa', rdfbody)
 608
 609         errmsgs = list(inference.run_validation())
 610         self.assertEqual(len(errmsgs), 0)
 611
 612 def suite():
 613     from unittest import TestSuite, defaultTestLoader
 614     suite = TestSuite()
 615     for testcase in [ExerimentsTestCases,
 616                      TestEmailNotify,
 617                      TestSequencer]:
 618         suite.addTests(defaultTestLoader.loadTestsFromTestCase(testcase))
 619     return suite
 620
 621 if __name__ == "__main__":
 622     from unittest import main
 623     main(defaultTest="suite")