experiments/test_experiments.py

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