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         """Test ticket:147
 191
 192         Library's have IDs, libraries also have primary keys,
 193         we eventually had enough libraries that the drop down combo
 194         box was too hard to filter through, unfortnately we want a
 195         field that uses our library id and not the internal
 196         primary key, and raw_id_field uses primary keys.
 197
 198         This tests to make sure that the value entered in the raw
 199         library id field matches the library id looked up.
 200
 201         """
 202         expected_ids = [ '1215{}'.format(i) for i in range(1,9) ]
 203         self.assertTrue(self.client.login(username=self.admin.username, password=self.password))
 204         response = self.client.get('/admin/experiments/flowcell/{}/'.format(self.fc12150.id))
 205
 206         tree = fromstring(response.content)
 207         for i in range(0,8):
 208             xpath_expression = '//input[@id="id_lane_set-%d-library"]'
 209             input_field = tree.xpath(xpath_expression % (i,))[0]
 210             library_field = input_field.find('../strong')
 211             library_id, library_name = library_field.text.split(':')
 212             # strip leading '#' sign from name
 213             library_id = library_id[1:]
 214             self.assertEqual(library_id, expected_ids[i])
 215             self.assertEqual(input_field.attrib['value'], library_id)
 216
 217     def test_library_to_flowcell_link(self):
 218         """
 219         Make sure the library page includes links to the flowcell pages.
 220         That work with flowcell IDs that have parenthetical comments.
 221         """
 222         self.assertTrue(self.client.login(username=self.admin.username, password=self.password))
 223         response = self.client.get('/library/12151/')
 224         self.assertEqual(response.status_code, 200)
 225         status = validate_xhtml(response.content)
 226         if status is not None: self.assertTrue(status)
 227
 228         tree = fromstring(response.content)
 229         flowcell_spans = tree.xpath('//span[@property="libns:flowcell_id"]',
 230                                     namespaces=NSMAP)
 231         self.assertEqual(flowcell_spans[1].text, 'FC12150')
 232         failed_fc_span = flowcell_spans[1]
 233         failed_fc_a = failed_fc_span.getparent()
 234         # make sure some of our RDF made it.
 235         self.assertEqual(failed_fc_a.get('typeof'), 'libns:IlluminaFlowcell')
 236         self.assertEqual(failed_fc_a.get('href'), '/flowcell/FC12150/')
 237         fc_response = self.client.get(failed_fc_a.get('href'))
 238         self.assertEqual(fc_response.status_code, 200)
 239         status = validate_xhtml(response.content)
 240         if status is not None: self.assertTrue(status)
 241
 242         fc_lane_response = self.client.get('/flowcell/FC12150/8/')
 243         self.assertEqual(fc_lane_response.status_code, 200)
 244         status = validate_xhtml(response.content)
 245         if status is not None: self.assertTrue(status)
 246
 247     def test_pooled_multiplex_id(self):
 248         fc_dict = flowcell_information(self.fc42jtn.flowcell_id)
 249
 250         lane_contents = fc_dict['lane_set'][2]
 251         self.assertEqual(len(lane_contents), len(self.fc42jtn_lanes) / 2)
 252         lane_dict = multi_lane_to_dict(lane_contents)
 253
 254         self.assertTrue(self.fc42jtn_lanes[0].library.multiplex_id in \
 255                         lane_dict['13001']['index_sequence'])
 256         self.assertTrue(self.fc42jtn_lanes[2].library.multiplex_id in \
 257                         lane_dict['13003']['index_sequence'])
 258
 259
 260     def test_lanes_for_view_user_odd(self):
 261         """Make sure lanes_for HTML UI works.
 262         """
 263         user = self.user_odd.username
 264         lanes = lanes_for(user)
 265         self.assertEqual(len(lanes), 8)
 266
 267         response = self.client.get(
 268             reverse('experiments.views.lanes_for',
 269                     args=[user]))
 270         self.assertEqual(response.status_code, 200)
 271         tree = fromstring(response.content)
 272         lane_trs = tree.xpath('//div[@id="changelist"]/table/tbody/tr')
 273         self.assertEqual(len(lane_trs), len(lanes))
 274         # lanes is in db order
 275         # lane_trs is in newest to oldest order
 276         for lane_tr, lane_db in zip(lane_trs, reversed(lanes)):
 277             library_id = lane_tr.xpath('td[6]/a')[0].text
 278             self.assertEqual(library_id, lane_db['library'])
 279
 280     def test_lanes_for_view_invalid_user(self):
 281         """Make sure we don't find anything with an invalid user
 282         """
 283         response = self.client.get(
 284             reverse('experiments.views.lanes_for',
 285                     args=["doesntexist"]))
 286         self.assertEqual(response.status_code, 404)
 287
 288     def test_lanes_for_json(self):
 289         """
 290         Check the code that packs the django objects into simple types.
 291         """
 292         user = self.user_odd.username
 293         lanes = lanes_for(user)
 294         self.assertEqual(len(lanes), 8)
 295
 296         response = self.client.get('/experiments/lanes_for/%s/json' % (user,), apidata)
 297         lanes_json = json.loads(smart_text(response.content))['result']
 298         self.assertEqual(len(lanes), len(lanes_json))
 299         for i in range(len(lanes)):
 300             self.assertEqual(lanes[i]['comment'], lanes_json[i]['comment'])
 301             self.assertEqual(lanes[i]['lane_number'], lanes_json[i]['lane_number'])
 302             self.assertEqual(lanes[i]['flowcell'], lanes_json[i]['flowcell'])
 303             self.assertEqual(lanes[i]['run_date'], lanes_json[i]['run_date'])
 304
 305     def test_lanes_for_no_lanes(self):
 306         """
 307         Do we get something meaningful back when the user isn't attached to anything?
 308         """
 309         user = HTSUserFactory.create(username='supertest')
 310         lanes = lanes_for(user.username)
 311         self.assertEqual(len(lanes), 0)
 312
 313         response = self.client.get('/experiments/lanes_for/%s/json' % (user,), apidata)
 314         self.assertEqual(response.status_code, 404)
 315
 316     def test_lanes_for_no_user(self):
 317         """
 318         Do we get something meaningful back when its the wrong user
 319         """
 320         user = 'not a real user'
 321         self.assertRaises(ObjectDoesNotExist, lanes_for, user)
 322
 323         response = self.client.get('/experiments/lanes_for/%s/json' % (user,), apidata)
 324         self.assertEqual(response.status_code, 404)
 325
 326     def test_raw_data_dir(self):
 327         """Raw data path generator check"""
 328         flowcell_id = self.fc1_id
 329         raw_dir = os.path.join(settings.RESULT_HOME_DIR, flowcell_id)
 330
 331         fc = FlowCell.objects.get(flowcell_id=flowcell_id)
 332         self.assertEqual(fc.get_raw_data_directory(), raw_dir)
 333
 334         fc.flowcell_id = flowcell_id + " (failed)"
 335         self.assertEqual(fc.get_raw_data_directory(), raw_dir)
 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")