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
  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         """make sure we can retrive a non-numeric library ID
 166         """
 167         response = self.client.get('/experiments/config/FC12150/json', apidata)
 168         self.assertEqual(response.status_code, 200)
 169         flowcell = json.loads(smart_text(response.content))['result']
 170
 171         # library id is 12150 + lane number (1-8), so 12153
 172         lane_contents = flowcell['lane_set']['3']
 173         lane_library = lane_contents[0]
 174         self.assertEqual(lane_library['library_id'], '12153')
 175
 176         response = self.client.get('/samples/library/12153/json', apidata)
 177         self.assertEqual(response.status_code, 200)
 178         library_12153 = json.loads(smart_text(response.content))['result']
 179
 180         self.assertEqual(library_12153['library_id'], '12153')
 181
 182     def test_raw_id_field(self):
 183         """Test ticket:147
 184
 185         Library's have IDs, libraries also have primary keys,
 186         we eventually had enough libraries that the drop down combo
 187         box was too hard to filter through, unfortnately we want a
 188         field that uses our library id and not the internal
 189         primary key, and raw_id_field uses primary keys.
 190
 191         This tests to make sure that the value entered in the raw
 192         library id field matches the library id looked up.
 193
 194         """
 195         expected_ids = [ '1215{}'.format(i) for i in range(1,9) ]
 196         self.assertTrue(self.client.login(username=self.admin.username, password=self.password))
 197         response = self.client.get(
 198             '/admin/experiments/flowcell/{}/change/'.format(self.fc12150.id),
 199             follow=True)
 200         self.assertEquals(response.status_code, 200)
 201
 202         tree = fromstring(response.content)
 203         for i in range(0,8):
 204             xpath_expression = '//input[@id="id_lane_set-%d-library"]'
 205             input_field = tree.xpath(xpath_expression % (i,))[0]
 206             library_field = input_field.find('../strong')
 207             library_id, library_name = library_field.text.split(':')
 208             # strip leading '#' sign from name
 209             library_id = library_id[1:]
 210             self.assertEqual(library_id, expected_ids[i])
 211             self.assertEqual(input_field.attrib['value'], library_id)
 212
 213     def test_library_to_flowcell_link(self):
 214         """
 215         Make sure the library page includes links to the flowcell pages.
 216         That work with flowcell IDs that have parenthetical comments.
 217         """
 218         self.assertTrue(self.client.login(username=self.admin.username, password=self.password))
 219         response = self.client.get('/library/12151/')
 220         self.assertEqual(response.status_code, 200)
 221         status = validate_xhtml(response.content)
 222         if status is not None: self.assertTrue(status)
 223
 224         tree = fromstring(response.content)
 225         flowcell_spans = tree.xpath('//span[@property="libns:flowcell_id"]',
 226                                     namespaces=NSMAP)
 227         self.assertEqual(flowcell_spans[1].text, 'FC12150')
 228         failed_fc_span = flowcell_spans[1]
 229         failed_fc_a = failed_fc_span.getparent()
 230         # make sure some of our RDF made it.
 231         self.assertEqual(failed_fc_a.get('typeof'), 'libns:IlluminaFlowcell')
 232         self.assertEqual(failed_fc_a.get('href'), '/flowcell/FC12150/')
 233         fc_response = self.client.get(failed_fc_a.get('href'))
 234         self.assertEqual(fc_response.status_code, 200)
 235         status = validate_xhtml(response.content)
 236         if status is not None: self.assertTrue(status)
 237
 238         fc_lane_response = self.client.get('/flowcell/FC12150/8/')
 239         self.assertEqual(fc_lane_response.status_code, 200)
 240         status = validate_xhtml(response.content)
 241         if status is not None: self.assertTrue(status)
 242
 243     def test_pooled_multiplex_id(self):
 244         fc_dict = flowcell_information(self.fc42jtn.flowcell_id)
 245
 246         lane_contents = fc_dict['lane_set'][2]
 247         self.assertEqual(len(lane_contents), len(self.fc42jtn_lanes) / 2)
 248         lane_dict = multi_lane_to_dict(lane_contents)
 249
 250         self.assertTrue(self.fc42jtn_lanes[0].library.multiplex_id in \
 251                         lane_dict['13001']['index_sequence'])
 252         self.assertTrue(self.fc42jtn_lanes[2].library.multiplex_id in \
 253                         lane_dict['13003']['index_sequence'])
 254
 255
 256     def test_lanes_for_view_user_odd(self):
 257         """Make sure lanes_for HTML UI works.
 258         """
 259         user = self.user_odd.username
 260         lanes = lanes_for(user)
 261         self.assertEqual(len(lanes), 8)
 262
 263         response = self.client.get(
 264             reverse('lanes_for', kwargs={'username': user}))
 265         self.assertEqual(response.status_code, 200)
 266         tree = fromstring(response.content)
 267         lane_trs = tree.xpath('//div[@id="changelist"]/table/tbody/tr')
 268         self.assertEqual(len(lane_trs), len(lanes))
 269         # lanes is in db order
 270         # lane_trs is in newest to oldest order
 271         for lane_tr, lane_db in zip(lane_trs, reversed(lanes)):
 272             library_id = lane_tr.xpath('td[6]/a')[0].text
 273             self.assertEqual(library_id, lane_db['library'])
 274
 275     def test_lanes_for_view_invalid_user(self):
 276         """Make sure we don't find anything with an invalid user
 277         """
 278         response = self.client.get(
 279             reverse('lanes_for', kwargs={'username': '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_sequencing_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_sequencing_runs()
 337         self.assertEqual(len(flowcell.sequencingrun_set.all()), 1)
 338
 339         run = flowcell.sequencingrun_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.sequencing_run.flowcell.flowcell_id, 'FC12150')
 347         self.assertEqual(
 348             srf4.sequencing_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(SequencingRun.objects.filter(result_dir='FC12150/C1-37'))
 362         # what happens if we import twice?
 363         flowcell.import_sequencing_run('FC12150/C1-37',
 364                                        'run_FC12150_2007-09-27.xml')
 365         self.assertEqual(
 366             len(SequencingRun.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_sequencing_runs()
 375
 376         #self.client.login(username='supertest', password='BJOKL5kAj6aFZ6A5')
 377
 378         result_files = flowcell.sequencingrun_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/{}/change/'.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         self.assertTrue(re.search(self.fc.flowcell_id, smart_text(response.content)))
 504         # require that navigation back to the admin page exists
 505         admin_a_tag = '<a href="{}">[^<]+</a>'.format(admin_url)
 506         self.assertTrue(re.search(admin_a_tag, smart_text(response.content)))
 507
 508 def multi_lane_to_dict(lane):
 509     """Convert a list of lane entries into a dictionary indexed by library ID
 510     """
 511     return dict( ((x['library_id'],x) for x in lane) )
 512
 513 class TestSequencer(TestCase):
 514     def setUp(self):
 515         self.fc12150 = FlowCellFactory(flowcell_id='FC12150')
 516         self.library = LibraryFactory(id="12150")
 517         self.lane = LaneFactory(flowcell=self.fc12150, lane_number=1, library=self.library)
 518
 519     def test_name_generation(self):
 520         seq = Sequencer()
 521         seq.name = "Seq1"
 522         seq.instrument_name = "HWI-SEQ1"
 523         seq.model = "Imaginary 5000"
 524
 525         self.assertEqual(str(seq), "Seq1 (HWI-SEQ1)")
 526
 527     def test_lookup(self):
 528         fc = self.fc12150
 529         self.assertEqual(fc.sequencer.model, 'HiSeq 1')
 530         self.assertTrue(fc.sequencer.instrument_name.startswith('instrument name')),
 531         # well actually we let the browser tack on the host name
 532         url = fc.get_absolute_url()
 533         self.assertEqual(url, '/flowcell/FC12150/')
 534
 535     def test_rdf(self):
 536         response = self.client.get('/flowcell/FC12150/', apidata)
 537         tree = fromstring(response.content)
 538         seq_by = tree.xpath('//div[@rel="libns:sequenced_by"]',
 539                             namespaces=NSMAP)
 540         self.assertEqual(len(seq_by), 1)
 541         self.assertEqual(seq_by[0].attrib['rel'], 'libns:sequenced_by')
 542         seq = seq_by[0].getchildren()
 543         self.assertEqual(len(seq), 1)
 544         sequencer = '/sequencer/' + str(self.fc12150.sequencer.id)
 545         self.assertEqual(seq[0].attrib['about'], sequencer)
 546         self.assertEqual(seq[0].attrib['typeof'], 'libns:Sequencer')
 547
 548         name = seq[0].xpath('./span[@property="libns:sequencer_name"]')
 549         self.assertEqual(len(name), 1)
 550         self.assertTrue(name[0].text.startswith('sequencer '))
 551         instrument = seq[0].xpath(
 552             './span[@property="libns:sequencer_instrument"]')
 553         self.assertEqual(len(instrument), 1)
 554         self.assertTrue(instrument[0].text.startswith('instrument name'))
 555         model = seq[0].xpath(
 556             './span[@property="libns:sequencer_model"]')
 557         self.assertEqual(len(model), 1)
 558         self.assertEqual(model[0].text, 'HiSeq 1')
 559
 560     def test_flowcell_with_rdf_validation(self):
 561         from htsworkflow.util.rdfhelp import add_default_schemas, \
 562              dump_model, \
 563              get_model, \
 564              load_string_into_model
 565         from htsworkflow.util.rdfinfer import Infer
 566
 567         model = get_model()
 568         add_default_schemas(model)
 569         inference = Infer(model)
 570
 571         url ='/flowcell/FC12150/'
 572         response = self.client.get(url)
 573         self.assertEqual(response.status_code, 200)
 574         status = validate_xhtml(response.content)
 575         if status is not None: self.assertTrue(status)
 576
 577         load_string_into_model(model, 'rdfa', smart_text(response.content))
 578
 579         errmsgs = list(inference.run_validation())
 580         self.assertEqual(len(errmsgs), 0)
 581
 582     def test_lane_with_rdf_validation(self):
 583         from htsworkflow.util.rdfhelp import add_default_schemas, \
 584              dump_model, \
 585              get_model, \
 586              load_string_into_model
 587         from htsworkflow.util.rdfinfer import Infer
 588
 589         model = get_model()
 590         add_default_schemas(model)
 591         inference = Infer(model)
 592
 593         url = '/lane/{}'.format(self.lane.id)
 594         response = self.client.get(url)
 595         rdfbody = smart_text(response.content)
 596         self.assertEqual(response.status_code, 200)
 597         status = validate_xhtml(rdfbody)
 598         if status is not None: self.assertTrue(status)
 599
 600         load_string_into_model(model, 'rdfa', rdfbody)
 601
 602         errmsgs = list(inference.run_validation())
 603         self.assertEqual(len(errmsgs), 0)
 604
 605 def suite():
 606     from unittest import TestSuite, defaultTestLoader
 607     suite = TestSuite()
 608     for testcase in [ExerimentsTestCases,
 609                      TestEmailNotify,
 610                      TestSequencer]:
 611         suite.addTests(defaultTestLoader.loadTestsFromTestCase(testcase))
 612     return suite
 613
 614 if __name__ == "__main__":
 615     from unittest import main
 616     main(defaultTest="suite")