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