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