htsworkflow/frontend/experiments/tests.py

   1 import re
   2 from lxml.html import fromstring
   3 try:
   4     import json
   5 except ImportError, e:
   6     import simplejson as json
   7 import os
   8 import shutil
   9 import sys
  10 import tempfile
  11 from urlparse import urljoin
  12
  13 from django.conf import settings
  14 from django.core import mail
  15 from django.core.exceptions import ObjectDoesNotExist
  16 from django.test import TestCase
  17 from htsworkflow.frontend.experiments import models
  18 from htsworkflow.frontend.experiments import experiments
  19 from htsworkflow.frontend.auth import apidata
  20 from htsworkflow.util.ethelp import validate_xhtml
  21
  22 from htsworkflow.pipelines.test.simulate_runfolder import TESTDATA_DIR
  23
  24 LANE_SET = range(1,9)
  25
  26 NSMAP = {'libns':'http://jumpgate.caltech.edu/wiki/LibraryOntology#'}
  27
  28 class ClusterStationTestCases(TestCase):
  29     fixtures = ['test_flowcells.json']
  30
  31     def test_default(self):
  32         c = models.ClusterStation.default()
  33         self.assertEqual(c.id, 2)
  34
  35         c.isdefault = False
  36         c.save()
  37
  38         total = models.ClusterStation.objects.filter(isdefault=True).count()
  39         self.assertEqual(total, 0)
  40
  41         other_default = models.ClusterStation.default()
  42         self.assertEqual(other_default.id, 3)
  43
  44
  45     def test_update_default(self):
  46         old_default = models.ClusterStation.default()
  47
  48         c = models.ClusterStation.objects.get(pk=3)
  49         c.isdefault = True
  50         c.save()
  51
  52         new_default = models.ClusterStation.default()
  53
  54         self.assertNotEqual(old_default, new_default)
  55         self.assertEqual(new_default, c)
  56
  57         total = models.ClusterStation.objects.filter(isdefault=True).count()
  58         self.assertEqual(total, 1)
  59
  60     def test_update_other(self):
  61         old_default = models.ClusterStation.default()
  62         total = models.ClusterStation.objects.filter(isdefault=True).count()
  63         self.assertEqual(total, 1)
  64
  65         c = models.ClusterStation.objects.get(pk=1)
  66         c.name = "Primary Key 1"
  67         c.save()
  68
  69         total = models.ClusterStation.objects.filter(isdefault=True).count()
  70         self.assertEqual(total, 1)
  71
  72         new_default = models.ClusterStation.default()
  73         self.assertEqual(old_default, new_default)
  74
  75
  76 class SequencerTestCases(TestCase):
  77     fixtures = ['test_flowcells.json']
  78
  79     def test_default(self):
  80         # starting with no default
  81         s = models.Sequencer.default()
  82         self.assertEqual(s.id, 2)
  83
  84         total = models.Sequencer.objects.filter(isdefault=True).count()
  85         self.assertEqual(total, 1)
  86
  87         s.isdefault = False
  88         s.save()
  89
  90         total = models.Sequencer.objects.filter(isdefault=True).count()
  91         self.assertEqual(total, 0)
  92
  93         other_default = models.Sequencer.default()
  94         self.assertEqual(other_default.id, 7)
  95
  96     def test_update_default(self):
  97         old_default = models.Sequencer.default()
  98
  99         s = models.Sequencer.objects.get(pk=1)
 100         s.isdefault = True
 101         s.save()
 102
 103         new_default = models.Sequencer.default()
 104
 105         self.assertNotEqual(old_default, new_default)
 106         self.assertEqual(new_default, s)
 107
 108         total = models.Sequencer.objects.filter(isdefault=True).count()
 109         self.assertEqual(total, 1)
 110
 111
 112     def test_update_other(self):
 113         old_default = models.Sequencer.default()
 114         total = models.Sequencer.objects.filter(isdefault=True).count()
 115         self.assertEqual(total, 1)
 116
 117         s = models.Sequencer.objects.get(pk=1)
 118         s.name = "Primary Key 1"
 119         s.save()
 120
 121         total = models.Sequencer.objects.filter(isdefault=True).count()
 122         self.assertEqual(total, 1)
 123
 124         new_default = models.Sequencer.default()
 125         self.assertEqual(old_default, new_default)
 126
 127
 128 class ExperimentsTestCases(TestCase):
 129     fixtures = ['test_flowcells.json',
 130                 ]
 131
 132     def setUp(self):
 133         self.tempdir = tempfile.mkdtemp(prefix='htsw-test-experiments-')
 134         settings.RESULT_HOME_DIR = self.tempdir
 135
 136         self.fc1_id = 'FC12150'
 137         self.fc1_root = os.path.join(self.tempdir, self.fc1_id)
 138         os.mkdir(self.fc1_root)
 139         self.fc1_dir = os.path.join(self.fc1_root, 'C1-37')
 140         os.mkdir(self.fc1_dir)
 141         runxml = 'run_FC12150_2007-09-27.xml'
 142         shutil.copy(os.path.join(TESTDATA_DIR, runxml),
 143                     os.path.join(self.fc1_dir, runxml))
 144         for i in range(1,9):
 145             shutil.copy(
 146                 os.path.join(TESTDATA_DIR,
 147                              'woldlab_070829_USI-EAS44_0017_FC11055_1.srf'),
 148                 os.path.join(self.fc1_dir,
 149                              'woldlab_070829_SERIAL_FC12150_%d.srf' %(i,))
 150                 )
 151
 152         self.fc2_dir = os.path.join(self.tempdir, '42JTNAAXX')
 153         os.mkdir(self.fc2_dir)
 154         os.mkdir(os.path.join(self.fc2_dir, 'C1-25'))
 155         os.mkdir(os.path.join(self.fc2_dir, 'C1-37'))
 156         os.mkdir(os.path.join(self.fc2_dir, 'C1-37', 'Plots'))
 157
 158     def tearDown(self):
 159         shutil.rmtree(self.tempdir)
 160
 161     def test_flowcell_information(self):
 162         """
 163         Check the code that packs the django objects into simple types.
 164         """
 165         for fc_id in [u'FC12150', u"42JTNAAXX", "42JU1AAXX"]:
 166             fc_dict = experiments.flowcell_information(fc_id)
 167             fc_django = models.FlowCell.objects.get(flowcell_id=fc_id)
 168             self.assertEqual(fc_dict['flowcell_id'], fc_id)
 169             self.assertEqual(fc_django.flowcell_id, fc_id)
 170             self.assertEqual(fc_dict['sequencer'], fc_django.sequencer.name)
 171             self.assertEqual(fc_dict['read_length'], fc_django.read_length)
 172             self.assertEqual(fc_dict['notes'], fc_django.notes)
 173             self.assertEqual(fc_dict['cluster_station'], fc_django.cluster_station.name)
 174
 175             for lane in fc_django.lane_set.all():
 176                 lane_contents = fc_dict['lane_set'][lane.lane_number]
 177                 lane_dict = multi_lane_to_dict(lane_contents)[lane.library_id]
 178                 self.assertEqual(lane_dict['cluster_estimate'], lane.cluster_estimate)
 179                 self.assertEqual(lane_dict['comment'], lane.comment)
 180                 self.assertEqual(lane_dict['flowcell'], lane.flowcell.flowcell_id)
 181                 self.assertEqual(lane_dict['lane_number'], lane.lane_number)
 182                 self.assertEqual(lane_dict['library_name'], lane.library.library_name)
 183                 self.assertEqual(lane_dict['library_id'], lane.library.id)
 184                 self.assertAlmostEqual(float(lane_dict['pM']), float(lane.pM))
 185                 self.assertEqual(lane_dict['library_species'],
 186                                      lane.library.library_species.scientific_name)
 187
 188             response = self.client.get('/experiments/config/%s/json' % (fc_id,), apidata)
 189             # strptime isoformat string = '%Y-%m-%dT%H:%M:%S'
 190             fc_json = json.loads(response.content)
 191             self.assertEqual(fc_json['flowcell_id'], fc_id)
 192             self.assertEqual(fc_json['sequencer'], fc_django.sequencer.name)
 193             self.assertEqual(fc_json['read_length'], fc_django.read_length)
 194             self.assertEqual(fc_json['notes'], fc_django.notes)
 195             self.assertEqual(fc_json['cluster_station'], fc_django.cluster_station.name)
 196
 197
 198             for lane in fc_django.lane_set.all():
 199                 lane_contents = fc_json['lane_set'][unicode(lane.lane_number)]
 200                 lane_dict = multi_lane_to_dict(lane_contents)[lane.library_id]
 201
 202                 self.assertEqual(lane_dict['cluster_estimate'], lane.cluster_estimate)
 203                 self.assertEqual(lane_dict['comment'], lane.comment)
 204                 self.assertEqual(lane_dict['flowcell'], lane.flowcell.flowcell_id)
 205                 self.assertEqual(lane_dict['lane_number'], lane.lane_number)
 206                 self.assertEqual(lane_dict['library_name'], lane.library.library_name)
 207                 self.assertEqual(lane_dict['library_id'], lane.library.id)
 208                 self.assertAlmostEqual(float(lane_dict['pM']), float(lane.pM))
 209                 self.assertEqual(lane_dict['library_species'],
 210                                      lane.library.library_species.scientific_name)
 211
 212     def test_invalid_flowcell(self):
 213         """
 214         Make sure we get a 404 if we request an invalid flowcell ID
 215         """
 216         response = self.client.get('/experiments/config/nottheone/json', apidata)
 217         self.assertEqual(response.status_code, 404)
 218
 219     def test_no_key(self):
 220         """
 221         Require logging in to retrieve meta data
 222         """
 223         response = self.client.get(u'/experiments/config/FC12150/json')
 224         self.assertEqual(response.status_code, 403)
 225
 226     def test_library_id(self):
 227         """
 228         Library IDs should be flexible, so make sure we can retrive a non-numeric ID
 229         """
 230         response = self.client.get('/experiments/config/FC12150/json', apidata)
 231         self.assertEqual(response.status_code, 200)
 232         flowcell = json.loads(response.content)
 233
 234         lane_contents = flowcell['lane_set']['3']
 235         lane_library = lane_contents[0]
 236         self.assertEqual(lane_library['library_id'], 'SL039')
 237
 238         response = self.client.get('/samples/library/SL039/json', apidata)
 239         self.assertEqual(response.status_code, 200)
 240         library_sl039 = json.loads(response.content)
 241
 242         self.assertEqual(library_sl039['library_id'], 'SL039')
 243
 244     def test_raw_id_field(self):
 245         """
 246         Test ticket:147
 247
 248         Library's have IDs, libraries also have primary keys,
 249         we eventually had enough libraries that the drop down combo box was too
 250         hard to filter through, unfortnately we want a field that uses our library
 251         id and not the internal primary key, and raw_id_field uses primary keys.
 252
 253         This tests to make sure that the value entered in the raw library id field matches
 254         the library id looked up.
 255         """
 256         expected_ids = [u'10981',u'11016',u'SL039',u'11060',
 257                         u'11061',u'11062',u'11063',u'11064']
 258         self.client.login(username='supertest', password='BJOKL5kAj6aFZ6A5')
 259         response = self.client.get('/admin/experiments/flowcell/153/')
 260
 261         tree = fromstring(response.content)
 262         for i in range(0,8):
 263             xpath_expression = '//input[@id="id_lane_set-%d-library"]'
 264             input_field = tree.xpath(xpath_expression % (i,))[0]
 265             library_field = input_field.find('../strong')
 266             library_id, library_name = library_field.text.split(':')
 267             # strip leading '#' sign from name
 268             library_id = library_id[1:]
 269             self.assertEqual(library_id, expected_ids[i])
 270             self.assertEqual(input_field.attrib['value'], library_id)
 271
 272     def test_library_to_flowcell_link(self):
 273         """
 274         Make sure the library page includes links to the flowcell pages.
 275         That work with flowcell IDs that have parenthetical comments.
 276         """
 277         self.client.login(username='supertest', password='BJOKL5kAj6aFZ6A5')
 278         response = self.client.get('/library/11070/')
 279         self.assertEqual(response.status_code, 200)
 280         status = validate_xhtml(response.content)
 281         if status is not None: self.assertTrue(status)
 282
 283         tree = fromstring(response.content)
 284         flowcell_spans = tree.xpath('//span[@property="libns:flowcell_id"]',
 285                                     namespaces=NSMAP)
 286         self.assertEqual(flowcell_spans[0].text, '30012AAXX (failed)')
 287         failed_fc_span = flowcell_spans[0]
 288         failed_fc_a = failed_fc_span.getparent()
 289         # make sure some of our RDF made it.
 290         self.assertEqual(failed_fc_a.get('typeof'), 'libns:IlluminaFlowcell')
 291         self.assertEqual(failed_fc_a.get('href'), '/flowcell/30012AAXX/')
 292         fc_response = self.client.get(failed_fc_a.get('href'))
 293         self.assertEqual(fc_response.status_code, 200)
 294         status = validate_xhtml(response.content)
 295         if status is not None: self.assertTrue(status)
 296
 297         fc_lane_response = self.client.get('/flowcell/30012AAXX/8/')
 298         self.assertEqual(fc_lane_response.status_code, 200)
 299         status = validate_xhtml(response.content)
 300         if status is not None: self.assertTrue(status)
 301
 302
 303     def test_pooled_multiplex_id(self):
 304         fc_dict = experiments.flowcell_information('42JU1AAXX')
 305         lane_contents = fc_dict['lane_set'][3]
 306         self.assertEqual(len(lane_contents), 2)
 307         lane_dict = multi_lane_to_dict(lane_contents)
 308
 309         self.assertEqual(lane_dict['12044']['index_sequence'],
 310                          {u'1': u'ATCACG',
 311                           u'2': u'CGATGT',
 312                           u'3': u'TTAGGC'})
 313         self.assertEqual(lane_dict['11045']['index_sequence'],
 314                          {u'1': u'ATCACG'})
 315
 316
 317
 318     def test_lanes_for(self):
 319         """
 320         Check the code that packs the django objects into simple types.
 321         """
 322         user = 'test'
 323         lanes = experiments.lanes_for(user)
 324         self.assertEqual(len(lanes), 5)
 325
 326         response = self.client.get('/experiments/lanes_for/%s/json' % (user,), apidata)
 327         lanes_json = json.loads(response.content)
 328         self.assertEqual(len(lanes), len(lanes_json))
 329         for i in range(len(lanes)):
 330             self.assertEqual(lanes[i]['comment'], lanes_json[i]['comment'])
 331             self.assertEqual(lanes[i]['lane_number'], lanes_json[i]['lane_number'])
 332             self.assertEqual(lanes[i]['flowcell'], lanes_json[i]['flowcell'])
 333             self.assertEqual(lanes[i]['run_date'], lanes_json[i]['run_date'])
 334
 335     def test_lanes_for_no_lanes(self):
 336         """
 337         Do we get something meaningful back when the user isn't attached to anything?
 338         """
 339         user = 'supertest'
 340         lanes = experiments.lanes_for(user)
 341         self.assertEqual(len(lanes), 0)
 342
 343         response = self.client.get('/experiments/lanes_for/%s/json' % (user,), apidata)
 344         lanes_json = json.loads(response.content)
 345
 346     def test_lanes_for_no_user(self):
 347         """
 348         Do we get something meaningful back when its the wrong user
 349         """
 350         user = 'not a real user'
 351         self.assertRaises(ObjectDoesNotExist, experiments.lanes_for, user)
 352
 353         response = self.client.get('/experiments/lanes_for/%s/json' % (user,), apidata)
 354         self.assertEqual(response.status_code, 404)
 355
 356
 357     def test_raw_data_dir(self):
 358         """Raw data path generator check"""
 359         flowcell_id = self.fc1_id
 360         raw_dir = os.path.join(settings.RESULT_HOME_DIR, flowcell_id)
 361
 362         fc = models.FlowCell.objects.get(flowcell_id=flowcell_id)
 363         self.assertEqual(fc.get_raw_data_directory(), raw_dir)
 364
 365         fc.flowcell_id = flowcell_id + " (failed)"
 366         self.assertEqual(fc.get_raw_data_directory(), raw_dir)
 367
 368
 369     def test_data_run_import(self):
 370         srf_file_type = models.FileType.objects.get(name='SRF')
 371         runxml_file_type = models.FileType.objects.get(name='run_xml')
 372         flowcell_id = self.fc1_id
 373         flowcell = models.FlowCell.objects.get(flowcell_id=flowcell_id)
 374         flowcell.update_data_runs()
 375         self.assertEqual(len(flowcell.datarun_set.all()), 1)
 376
 377         run = flowcell.datarun_set.all()[0]
 378         result_files = run.datafile_set.all()
 379         result_dict = dict(((rf.relative_pathname, rf) for rf in result_files))
 380
 381         srf4 = result_dict['FC12150/C1-37/woldlab_070829_SERIAL_FC12150_4.srf']
 382         self.assertEqual(srf4.file_type, srf_file_type)
 383         self.assertEqual(srf4.library_id, '11060')
 384         self.assertEqual(srf4.data_run.flowcell.flowcell_id, 'FC12150')
 385         self.assertEqual(
 386             srf4.data_run.flowcell.lane_set.get(lane_number=4).library_id,
 387             '11060')
 388         self.assertEqual(
 389             srf4.pathname,
 390             os.path.join(settings.RESULT_HOME_DIR, srf4.relative_pathname))
 391
 392         lane_files = run.lane_files()
 393         self.assertEqual(lane_files[4]['srf'], srf4)
 394
 395         runxml= result_dict['FC12150/C1-37/run_FC12150_2007-09-27.xml']
 396         self.assertEqual(runxml.file_type, runxml_file_type)
 397         self.assertEqual(runxml.library_id, None)
 398
 399         import1 = len(models.DataRun.objects.filter(result_dir='FC12150/C1-37'))
 400         # what happens if we import twice?
 401         flowcell.import_data_run('FC12150/C1-37',
 402                                  'run_FC12150_2007-09-27.xml')
 403         self.assertEqual(
 404             len(models.DataRun.objects.filter(result_dir='FC12150/C1-37')),
 405             import1)
 406
 407     def test_read_result_file(self):
 408         """make sure we can return a result file
 409         """
 410         flowcell_id = self.fc1_id
 411         flowcell = models.FlowCell.objects.get(flowcell_id=flowcell_id)
 412         flowcell.update_data_runs()
 413
 414         #self.client.login(username='supertest', password='BJOKL5kAj6aFZ6A5')
 415
 416         result_files = flowcell.datarun_set.all()[0].datafile_set.all()
 417         for f in result_files:
 418             url = '/experiments/file/%s' % ( f.random_key,)
 419             response = self.client.get(url)
 420             self.assertEqual(response.status_code, 200)
 421             mimetype = f.file_type.mimetype
 422             if mimetype is None:
 423                 mimetype = 'application/octet-stream'
 424
 425             self.assertEqual(mimetype, response['content-type'])
 426
 427     def test_flowcell_rdf(self):
 428         import RDF
 429         from htsworkflow.util.rdfhelp import get_model, \
 430              fromTypedNode, \
 431              load_string_into_model, \
 432              rdfNS, \
 433              libraryOntology, \
 434              dump_model
 435
 436         model = get_model()
 437
 438         expected = {'1': ['11034'],
 439                     '2': ['11036'],
 440                     '3': ['12044','11045'],
 441                     '4': ['11047','13044'],
 442                     '5': ['11055'],
 443                     '6': ['11067'],
 444                     '7': ['11069'],
 445                     '8': ['11070']}
 446         url = '/flowcell/42JU1AAXX/'
 447         response = self.client.get(url)
 448         self.assertEqual(response.status_code, 200)
 449         status = validate_xhtml(response.content)
 450         if status is not None: self.assertTrue(status)
 451
 452         ns = urljoin('http://localhost', url)
 453         load_string_into_model(model, 'rdfa', response.content, ns=ns)
 454         body = """prefix rdf: <http://www.w3.org/1999/02/22-rdf-syntax-ns#>
 455         prefix libns: <http://jumpgate.caltech.edu/wiki/LibraryOntology#>
 456
 457         select ?flowcell ?flowcell_id ?lane_id ?library_id
 458         where {
 459           ?flowcell a libns:IlluminaFlowcell ;
 460                     libns:flowcell_id ?flowcell_id ;
 461                     libns:has_lane ?lane .
 462           ?lane libns:lane_number ?lane_id ;
 463                 libns:library ?library .
 464           ?library libns:library_id ?library_id .
 465         }"""
 466         query = RDF.SPARQLQuery(body)
 467         count = 0
 468         for r in query.execute(model):
 469             count += 1
 470             self.assertEqual(fromTypedNode(r['flowcell_id']), u'42JU1AAXX')
 471             lane_id = fromTypedNode(r['lane_id'])
 472             library_id = fromTypedNode(r['library_id'])
 473             self.assertTrue(library_id in expected[lane_id])
 474         self.assertEqual(count, 10)
 475
 476
 477 class TestFileType(TestCase):
 478     def test_file_type_unicode(self):
 479         file_type_objects = models.FileType.objects
 480         name = 'QSEQ tarfile'
 481         file_type_object = file_type_objects.get(name=name)
 482         self.assertEqual(u"<FileType: QSEQ tarfile>",
 483                              unicode(file_type_object))
 484
 485 class TestFileType(TestCase):
 486     def test_find_file_type(self):
 487         file_type_objects = models.FileType.objects
 488         cases = [('woldlab_090921_HWUSI-EAS627_0009_42FC3AAXX_l7_r1.tar.bz2',
 489                   'QSEQ tarfile', 7, 1),
 490                  ('woldlab_091005_HWUSI-EAS627_0010_42JT2AAXX_1.srf',
 491                   'SRF', 1, None),
 492                  ('s_1_eland_extended.txt.bz2','ELAND Extended', 1, None),
 493                  ('s_7_eland_multi.txt.bz2', 'ELAND Multi', 7, None),
 494                  ('s_3_eland_result.txt.bz2','ELAND Result', 3, None),
 495                  ('s_1_export.txt.bz2','ELAND Export', 1, None),
 496                  ('s_1_percent_call.png', 'IVC Percent Call', 1, None),
 497                  ('s_2_percent_base.png', 'IVC Percent Base', 2, None),
 498                  ('s_3_percent_all.png', 'IVC Percent All', 3, None),
 499                  ('s_4_call.png', 'IVC Call', 4, None),
 500                  ('s_5_all.png', 'IVC All', 5, None),
 501                  ('Summary.htm', 'Summary.htm', None, None),
 502                  ('run_42JT2AAXX_2009-10-07.xml', 'run_xml', None, None),
 503          ]
 504         for filename, typename, lane, end in cases:
 505             ft = models.find_file_type_metadata_from_filename(filename)
 506             self.assertEqual(ft['file_type'],
 507                                  file_type_objects.get(name=typename))
 508             self.assertEqual(ft.get('lane', None), lane)
 509             self.assertEqual(ft.get('end', None), end)
 510
 511     def test_assign_file_type_complex_path(self):
 512         file_type_objects = models.FileType.objects
 513         cases = [('/a/b/c/woldlab_090921_HWUSI-EAS627_0009_42FC3AAXX_l7_r1.tar.bz2',
 514                   'QSEQ tarfile', 7, 1),
 515                  ('foo/woldlab_091005_HWUSI-EAS627_0010_42JT2AAXX_1.srf',
 516                   'SRF', 1, None),
 517                  ('../s_1_eland_extended.txt.bz2','ELAND Extended', 1, None),
 518                  ('/bleem/s_7_eland_multi.txt.bz2', 'ELAND Multi', 7, None),
 519                  ('/qwer/s_3_eland_result.txt.bz2','ELAND Result', 3, None),
 520                  ('/ty///1/s_1_export.txt.bz2','ELAND Export', 1, None),
 521                  ('/help/s_1_percent_call.png', 'IVC Percent Call', 1, None),
 522                  ('/bored/s_2_percent_base.png', 'IVC Percent Base', 2, None),
 523                  ('/example1/s_3_percent_all.png', 'IVC Percent All', 3, None),
 524                  ('amonkey/s_4_call.png', 'IVC Call', 4, None),
 525                  ('fishie/s_5_all.png', 'IVC All', 5, None),
 526                  ('/random/Summary.htm', 'Summary.htm', None, None),
 527                  ('/notrandom/run_42JT2AAXX_2009-10-07.xml', 'run_xml', None, None),
 528          ]
 529         for filename, typename, lane, end in cases:
 530             result = models.find_file_type_metadata_from_filename(filename)
 531             self.assertEqual(result['file_type'],
 532                                  file_type_objects.get(name=typename))
 533             self.assertEqual(result.get('lane',None), lane)
 534             self.assertEqual(result.get('end', None), end)
 535
 536 class TestEmailNotify(TestCase):
 537     fixtures = ['test_flowcells.json']
 538
 539     def test_started_email_not_logged_in(self):
 540         response = self.client.get('/experiments/started/153/')
 541         self.assertEqual(response.status_code, 302)
 542
 543     def test_started_email_logged_in_user(self):
 544         self.client.login(username='test', password='BJOKL5kAj6aFZ6A5')
 545         response = self.client.get('/experiments/started/153/')
 546         self.assertEqual(response.status_code, 302)
 547
 548     def test_started_email_logged_in_staff(self):
 549         self.client.login(username='admintest', password='BJOKL5kAj6aFZ6A5')
 550         response = self.client.get('/experiments/started/153/')
 551         self.assertEqual(response.status_code, 200)
 552
 553     def test_started_email_send(self):
 554         self.client.login(username='admintest', password='BJOKL5kAj6aFZ6A5')
 555         response = self.client.get('/experiments/started/153/')
 556         self.assertEqual(response.status_code, 200)
 557
 558         self.assertTrue('pk1@example.com' in response.content)
 559         self.assertTrue('Lane #8 : (11064) Paired ends 104' in response.content)
 560
 561         response = self.client.get('/experiments/started/153/', {'send':'1','bcc':'on'})
 562         self.assertEqual(response.status_code, 200)
 563         self.assertEqual(len(mail.outbox), 4)
 564         for m in mail.outbox:
 565             self.assertTrue(len(m.body) > 0)
 566
 567     def test_email_navigation(self):
 568         """
 569         Can we navigate between the flowcell and email forms properly?
 570         """
 571         self.client.login(username='supertest', password='BJOKL5kAj6aFZ6A5')
 572         response = self.client.get('/experiments/started/153/')
 573         self.assertEqual(response.status_code, 200)
 574         self.assertTrue(re.search('Flowcell FC12150', response.content))
 575         # require that navigation back to the admin page exists
 576         self.assertTrue(re.search('<a href="/admin/experiments/flowcell/153/">[^<]+</a>', response.content))
 577
 578 def multi_lane_to_dict(lane):
 579     """Convert a list of lane entries into a dictionary indexed by library ID
 580     """
 581     return dict( ((x['library_id'],x) for x in lane) )
 582
 583 class TestSequencer(TestCase):
 584     fixtures = ['test_flowcells.json',
 585                 ]
 586
 587     def test_name_generation(self):
 588         seq = models.Sequencer()
 589         seq.name = "Seq1"
 590         seq.instrument_name = "HWI-SEQ1"
 591         seq.model = "Imaginary 5000"
 592
 593         self.assertEqual(unicode(seq), "Seq1 (HWI-SEQ1)")
 594
 595     def test_lookup(self):
 596         fc = models.FlowCell.objects.get(pk=153)
 597         self.assertEqual(fc.sequencer.model,
 598                              "Illumina Genome Analyzer IIx")
 599         self.assertEqual(fc.sequencer.instrument_name,
 600                              "ILLUMINA-EC5D15")
 601         # well actually we let the browser tack on the host name
 602         url = fc.get_absolute_url()
 603         self.assertEqual(url, '/flowcell/FC12150/')
 604
 605     def test_rdf(self):
 606         response = self.client.get('/flowcell/FC12150/', apidata)
 607         tree = fromstring(response.content)
 608         seq_by = tree.xpath('//div[@rel="libns:sequenced_by"]',
 609                             namespaces=NSMAP)
 610         self.assertEqual(len(seq_by), 1)
 611         self.assertEqual(seq_by[0].attrib['rel'], 'libns:sequenced_by')
 612         seq = seq_by[0].getchildren()
 613         self.assertEqual(len(seq), 1)
 614         self.assertEqual(seq[0].attrib['about'], '/sequencer/2')
 615         self.assertEqual(seq[0].attrib['typeof'], 'libns:Sequencer')
 616
 617         name = seq[0].xpath('./span[@property="libns:sequencer_name"]')
 618         self.assertEqual(len(name), 1)
 619         self.assertEqual(name[0].text, 'Tardigrade')
 620         instrument = seq[0].xpath(
 621             './span[@property="libns:sequencer_instrument"]')
 622         self.assertEqual(len(instrument), 1)
 623         self.assertEqual(instrument[0].text, 'ILLUMINA-EC5D15')
 624         model = seq[0].xpath(
 625             './span[@property="libns:sequencer_model"]')
 626         self.assertEqual(len(model), 1)
 627         self.assertEqual(model[0].text, 'Illumina Genome Analyzer IIx')
 628
 629     def test_flowcell_with_rdf_validation(self):
 630         from htsworkflow.util.rdfhelp import add_default_schemas, \
 631              dump_model, \
 632              get_model, \
 633              load_string_into_model
 634         from htsworkflow.util.rdfinfer import Infer
 635
 636         model = get_model()
 637         add_default_schemas(model)
 638         inference = Infer(model)
 639
 640         url ='/flowcell/FC12150/'
 641         response = self.client.get(url)
 642         self.assertEqual(response.status_code, 200)
 643         status = validate_xhtml(response.content)
 644         if status is not None: self.assertTrue(status)
 645
 646         load_string_into_model(model, 'rdfa', response.content)
 647
 648         errmsgs = list(inference.run_validation())
 649         self.assertEqual(len(errmsgs), 2)
 650         for errmsg in errmsgs:
 651             self.assertEqual(errmsg, 'Missing type for: http://localhost/')
 652
 653     def test_lane_with_rdf_validation(self):
 654         from htsworkflow.util.rdfhelp import add_default_schemas, \
 655              dump_model, \
 656              get_model, \
 657              load_string_into_model
 658         from htsworkflow.util.rdfinfer import Infer
 659
 660         model = get_model()
 661         add_default_schemas(model)
 662         inference = Infer(model)
 663
 664         url = '/lane/1193'
 665         response = self.client.get(url)
 666         self.assertEqual(response.status_code, 200)
 667         status = validate_xhtml(response.content)
 668         if status is not None: self.assertTrue(status)
 669
 670         load_string_into_model(model, 'rdfa', response.content)
 671
 672         errmsgs = list(inference.run_validation())
 673         self.assertEqual(len(errmsgs), 2)
 674         for errmsg in errmsgs:
 675             self.assertEqual(errmsg, 'Missing type for: http://localhost/')