X-Git-Url: http://woldlab.caltech.edu/gitweb/?p=htsworkflow.git;a=blobdiff_plain;f=htsworkflow%2Ffrontend%2Fexperiments%2Ftests.py;h=ef299e3676a02bbd087917d2bfc8654f3a391668;hp=664547bfb7e5d799fca4e4036eb6221762197af8;hb=bc27b813d0cdd3513323434405ee2ec5126490e5;hpb=0123277f86af843d47170d56d217f45e7d7cb4a6 diff --git a/htsworkflow/frontend/experiments/tests.py b/htsworkflow/frontend/experiments/tests.py index 664547b..ef299e3 100644 --- a/htsworkflow/frontend/experiments/tests.py +++ b/htsworkflow/frontend/experiments/tests.py @@ -19,7 +19,8 @@ from htsworkflow.frontend.auth import apidata from htsworkflow.pipelines.test.simulate_runfolder import TESTDATA_DIR -LANE_SET = range(1,9) +LANE_SET = range(1, 9) + class ExperimentsTestCases(TestCase): fixtures = ['test_flowcells.json'] @@ -36,14 +37,14 @@ class ExperimentsTestCases(TestCase): runxml = 'run_FC12150_2007-09-27.xml' shutil.copy(os.path.join(TESTDATA_DIR, runxml), os.path.join(self.fc1_dir, runxml)) - for i in range(1,9): + for i in range(1, 9): shutil.copy( os.path.join(TESTDATA_DIR, 'woldlab_070829_USI-EAS44_0017_FC11055_1.srf'), os.path.join(self.fc1_dir, - 'woldlab_070829_SERIAL_FC12150_%d.srf' %(i,)) + 'woldlab_070829_SERIAL_FC12150_%d.srf' % (i, )) ) - + self.fc2_dir = os.path.join(self.tempdir, '42JTNAAXX') os.mkdir(self.fc2_dir) os.mkdir(os.path.join(self.fc2_dir, 'C1-25')) @@ -62,50 +63,69 @@ class ExperimentsTestCases(TestCase): fc_django = models.FlowCell.objects.get(flowcell_id=fc_id) self.failUnlessEqual(fc_dict['flowcell_id'], fc_id) self.failUnlessEqual(fc_django.flowcell_id, fc_id) - self.failUnlessEqual(fc_dict['sequencer'], fc_django.sequencer.name) - self.failUnlessEqual(fc_dict['read_length'], fc_django.read_length) + self.failUnlessEqual(fc_dict['sequencer'], + fc_django.sequencer.name) + self.failUnlessEqual(fc_dict['read_length'], + fc_django.read_length) self.failUnlessEqual(fc_dict['notes'], fc_django.notes) - self.failUnlessEqual(fc_dict['cluster_station'], fc_django.cluster_station.name) + self.failUnlessEqual(fc_dict['cluster_station'], + fc_django.cluster_station.name) for lane in fc_django.lane_set.all(): - lane_dict = fc_dict['lane_set'][lane.lane_number] - self.failUnlessEqual(lane_dict['cluster_estimate'], lane.cluster_estimate) + lane_contents = fc_dict['lane_set'][lane.lane_number] + lane_dict = multi_lane_to_dict(lane_contents)[lane.library_id] + self.failUnlessEqual(lane_dict['cluster_estimate'], + lane.cluster_estimate) self.failUnlessEqual(lane_dict['comment'], lane.comment) - self.failUnlessEqual(lane_dict['flowcell'], lane.flowcell.flowcell_id) - self.failUnlessEqual(lane_dict['lane_number'], lane.lane_number) - self.failUnlessEqual(lane_dict['library_name'], lane.library.library_name) + self.failUnlessEqual(lane_dict['flowcell'], + lane.flowcell.flowcell_id) + self.failUnlessEqual(lane_dict['lane_number'], + lane.lane_number) + self.failUnlessEqual(lane_dict['library_name'], + lane.library.library_name) self.failUnlessEqual(lane_dict['library_id'], lane.library.id) - self.failUnlessAlmostEqual(float(lane_dict['pM']), float(lane.pM)) + self.failUnlessAlmostEqual(float(lane_dict['pM']), + float(lane.pM)) self.failUnlessEqual(lane_dict['library_species'], - lane.library.library_species.scientific_name) - - response = self.client.get('/experiments/config/%s/json' % (fc_id,), apidata) + lane.library.library_species.scientific_name) + + flowcell_url = '/experiments/config/%s/json' + response = self.client.get(flowcell_url % (fc_id,), apidata) # strptime isoformat string = '%Y-%m-%dT%H:%M:%S' fc_json = json.loads(response.content) self.failUnlessEqual(fc_json['flowcell_id'], fc_id) - self.failUnlessEqual(fc_json['sequencer'], fc_django.sequencer.name) + self.failUnlessEqual(fc_json['sequencer'], + fc_django.sequencer.name) self.failUnlessEqual(fc_json['read_length'], fc_django.read_length) self.failUnlessEqual(fc_json['notes'], fc_django.notes) - self.failUnlessEqual(fc_json['cluster_station'], fc_django.cluster_station.name) - + self.failUnlessEqual(fc_json['cluster_station'], + fc_django.cluster_station.name) for lane in fc_django.lane_set.all(): - lane_dict = fc_json['lane_set'][unicode(lane.lane_number)] - self.failUnlessEqual(lane_dict['cluster_estimate'], lane.cluster_estimate) + lane_contents = fc_json['lane_set'][unicode(lane.lane_number)] + lane_dict = multi_lane_to_dict(lane_contents)[lane.library_id] + + self.failUnlessEqual(lane_dict['cluster_estimate'], + lane.cluster_estimate) self.failUnlessEqual(lane_dict['comment'], lane.comment) - self.failUnlessEqual(lane_dict['flowcell'], lane.flowcell.flowcell_id) - self.failUnlessEqual(lane_dict['lane_number'], lane.lane_number) - self.failUnlessEqual(lane_dict['library_name'], lane.library.library_name) + self.failUnlessEqual(lane_dict['flowcell'], + lane.flowcell.flowcell_id) + self.failUnlessEqual(lane_dict['lane_number'], + lane.lane_number) + self.failUnlessEqual(lane_dict['library_name'], + lane.library.library_name) self.failUnlessEqual(lane_dict['library_id'], lane.library.id) - self.failUnlessAlmostEqual(float(lane_dict['pM']), float(lane.pM)) + self.failUnlessAlmostEqual(float(lane_dict['pM']), + float(lane.pM)) self.failUnlessEqual(lane_dict['library_species'], - lane.library.library_species.scientific_name) + lane.library.library_species.scientific_name) def test_invalid_flowcell(self): """ Make sure we get a 404 if we request an invalid flowcell ID """ - response = self.client.get('/experiments/config/nottheone/json', apidata) + flowcell_url = '/experiments/config/nottheone/json' + response = self.client.get(flowcell_url, apidata) self.failUnlessEqual(response.status_code, 404) def test_no_key(self): @@ -117,13 +137,16 @@ class ExperimentsTestCases(TestCase): def test_library_id(self): """ - Library IDs should be flexible, so make sure we can retrive a non-numeric ID + Library IDs should be flexible, retrive a non-numeric ID """ - response = self.client.get('/experiments/config/FC12150/json', apidata) + flowcell_url = '/experiments/config/FC12150/json' + response = self.client.get(flowcell_url, apidata) self.failUnlessEqual(response.status_code, 200) flowcell = json.loads(response.content) - self.failUnlessEqual(flowcell['lane_set']['3']['library_id'], 'SL039') + lane_contents = flowcell['lane_set']['3'] + lane_library = lane_contents[0] + self.failUnlessEqual(lane_library['library_id'], 'SL039') response = self.client.get('/samples/library/SL039/json', apidata) self.failUnlessEqual(response.status_code, 200) @@ -137,21 +160,22 @@ class ExperimentsTestCases(TestCase): Library's have IDs, libraries also have primary keys, we eventually had enough libraries that the drop down combo box was too - hard to filter through, unfortnately we want a field that uses our library - id and not the internal primary key, and raw_id_field uses primary keys. + hard to filter through, unfortnately we want a field that uses our + library id and not the internal primary key, and raw_id_field uses + primary keys. - This tests to make sure that the value entered in the raw library id field matches - the library id looked up. + This tests to make sure that the value entered in the raw library id + field matches the library id looked up. """ - expected_ids = [u'10981',u'11016',u'SL039',u'11060', - u'11061',u'11062',u'11063',u'11064'] + expected_ids = [u'10981', u'11016', u'SL039', u'11060', + u'11061', u'11062', u'11063', u'11064'] self.client.login(username='supertest', password='BJOKL5kAj6aFZ6A5') response = self.client.get('/admin/experiments/flowcell/153/') soup = BeautifulSoup(response.content) - for i in range(0,8): + for i in range(0, 8): input_field = soup.find(id='id_lane_set-%d-library' % (i,)) library_field = input_field.findNext('strong') - library_id, library_name = library_field.string.split(':') + library_id, library_name = library_field.text.split(':') # strip leading '#' sign from name library_id = library_id[1:] self.failUnlessEqual(library_id, expected_ids[i]) @@ -174,8 +198,19 @@ class ExperimentsTestCases(TestCase): self.failUnlessEqual(fc_response.status_code, 200) fc_lane_response = self.client.get('/flowcell/30012AAXX/8/') self.failUnlessEqual(fc_lane_response.status_code, 200) - - + + def test_pooled_multiplex_id(self): + fc_dict = experiments.flowcell_information('42JU1AAXX') + lane_contents = fc_dict['lane_set'][3] + self.assertEqual(len(lane_contents), 2) + lane_dict = multi_lane_to_dict(lane_contents) + + self.assertEqual(lane_dict['12044']['index_sequence'], + {u'1': u'ATCACG', + u'2': u'CGATGT', + u'3': u'TTAGGC'}) + self.assertEqual(lane_dict['11045']['index_sequence'], + {u'1': u'ATCACG'}) def test_lanes_for(self): """ @@ -185,24 +220,29 @@ class ExperimentsTestCases(TestCase): lanes = experiments.lanes_for(user) self.failUnlessEqual(len(lanes), 5) - response = self.client.get('/experiments/lanes_for/%s/json' % (user,), apidata) + flowcell_url = '/experiments/lanes_for/%s/json' + response = self.client.get(flowcell_url % (user,), apidata) lanes_json = json.loads(response.content) self.failUnlessEqual(len(lanes), len(lanes_json)) for i in range(len(lanes)): self.failUnlessEqual(lanes[i]['comment'], lanes_json[i]['comment']) - self.failUnlessEqual(lanes[i]['lane_number'], lanes_json[i]['lane_number']) - self.failUnlessEqual(lanes[i]['flowcell'], lanes_json[i]['flowcell']) - self.failUnlessEqual(lanes[i]['run_date'], lanes_json[i]['run_date']) - + self.failUnlessEqual(lanes[i]['lane_number'], + lanes_json[i]['lane_number']) + self.failUnlessEqual(lanes[i]['flowcell'], + lanes_json[i]['flowcell']) + self.failUnlessEqual(lanes[i]['run_date'], + lanes_json[i]['run_date']) + def test_lanes_for_no_lanes(self): """ - Do we get something meaningful back when the user isn't attached to anything? + What happens to user who haven't submitted anything """ user = 'supertest' lanes = experiments.lanes_for(user) self.failUnlessEqual(len(lanes), 0) - response = self.client.get('/experiments/lanes_for/%s/json' % (user,), apidata) + url = '/experiments/lanes_for/%s/json' + response = self.client.get(url % (user,), apidata) lanes_json = json.loads(response.content) def test_lanes_for_no_user(self): @@ -212,22 +252,21 @@ class ExperimentsTestCases(TestCase): user = 'not a real user' self.failUnlessRaises(ObjectDoesNotExist, experiments.lanes_for, user) - response = self.client.get('/experiments/lanes_for/%s/json' % (user,), apidata) + url = '/experiments/lanes_for/%s/json' + response = self.client.get(url % (user,), apidata) self.failUnlessEqual(response.status_code, 404) - def test_raw_data_dir(self): """Raw data path generator check""" flowcell_id = self.fc1_id raw_dir = os.path.join(settings.RESULT_HOME_DIR, flowcell_id) - + fc = models.FlowCell.objects.get(flowcell_id=flowcell_id) self.failUnlessEqual(fc.get_raw_data_directory(), raw_dir) fc.flowcell_id = flowcell_id + " (failed)" self.failUnlessEqual(fc.get_raw_data_directory(), raw_dir) - def test_data_run_import(self): srf_file_type = models.FileType.objects.get(name='SRF') runxml_file_type = models.FileType.objects.get(name='run_xml') @@ -254,10 +293,9 @@ class ExperimentsTestCases(TestCase): lane_files = run.lane_files() self.failUnlessEqual(lane_files[4]['srf'], srf4) - runxml= result_dict['FC12150/C1-37/run_FC12150_2007-09-27.xml'] + runxml = result_dict['FC12150/C1-37/run_FC12150_2007-09-27.xml'] self.failUnlessEqual(runxml.file_type, runxml_file_type) self.failUnlessEqual(runxml.library_id, None) - def test_read_result_file(self): """make sure we can return a result file @@ -265,20 +303,21 @@ class ExperimentsTestCases(TestCase): flowcell_id = self.fc1_id flowcell = models.FlowCell.objects.get(flowcell_id=flowcell_id) flowcell.update_data_runs() - - #self.client.login(username='supertest', password='BJOKL5kAj6aFZ6A5') + + #self.client.login(username='supertest', password='BJOKL5kAj6aFZ6A5') result_files = flowcell.datarun_set.all()[0].datafile_set.all() for f in result_files: - url = '/experiments/file/%s' % ( f.random_key,) + url = '/experiments/file/%s' % (f.random_key, ) response = self.client.get(url) self.failUnlessEqual(response.status_code, 200) mimetype = f.file_type.mimetype if mimetype is None: mimetype = 'application/octet-stream' - + self.failUnlessEqual(mimetype, response['content-type']) - + + class TestFileType(TestCase): def test_file_type_unicode(self): file_type_objects = models.FileType.objects @@ -286,7 +325,8 @@ class TestFileType(TestCase): file_type_object = file_type_objects.get(name=name) self.failUnlessEqual(u"", unicode(file_type_object)) - + + class TestFileType(TestCase): def test_find_file_type(self): file_type_objects = models.FileType.objects @@ -294,9 +334,9 @@ class TestFileType(TestCase): 'QSEQ tarfile', 7, 1), ('woldlab_091005_HWUSI-EAS627_0010_42JT2AAXX_1.srf', 'SRF', 1, None), - ('s_1_eland_extended.txt.bz2','ELAND Extended', 1, None), + ('s_1_eland_extended.txt.bz2', 'ELAND Extended', 1, None), ('s_7_eland_multi.txt.bz2', 'ELAND Multi', 7, None), - ('s_3_eland_result.txt.bz2','ELAND Result', 3, None), + ('s_3_eland_result.txt.bz2', 'ELAND Result', 3, None), ('s_1_export.txt.bz2','ELAND Export', 1, None), ('s_1_percent_call.png', 'IVC Percent Call', 1, None), ('s_2_percent_base.png', 'IVC Percent Base', 2, None), @@ -319,10 +359,10 @@ class TestFileType(TestCase): 'QSEQ tarfile', 7, 1), ('foo/woldlab_091005_HWUSI-EAS627_0010_42JT2AAXX_1.srf', 'SRF', 1, None), - ('../s_1_eland_extended.txt.bz2','ELAND Extended', 1, None), + ('../s_1_eland_extended.txt.bz2', 'ELAND Extended', 1, None), ('/bleem/s_7_eland_multi.txt.bz2', 'ELAND Multi', 7, None), - ('/qwer/s_3_eland_result.txt.bz2','ELAND Result', 3, None), - ('/ty///1/s_1_export.txt.bz2','ELAND Export', 1, None), + ('/qwer/s_3_eland_result.txt.bz2', 'ELAND Result', 3, None), + ('/ty///1/s_1_export.txt.bz2', 'ELAND Export', 1, None), ('/help/s_1_percent_call.png', 'IVC Percent Call', 1, None), ('/bored/s_2_percent_base.png', 'IVC Percent Base', 2, None), ('/example1/s_3_percent_all.png', 'IVC Percent All', 3, None), @@ -335,9 +375,10 @@ class TestFileType(TestCase): result = models.find_file_type_metadata_from_filename(filename) self.failUnlessEqual(result['file_type'], file_type_objects.get(name=typename)) - self.failUnlessEqual(result.get('lane',None), lane) + self.failUnlessEqual(result.get('lane', None), lane) self.failUnlessEqual(result.get('end', None), end) - + + class TestEmailNotify(TestCase): fixtures = ['test_flowcells.json'] @@ -349,21 +390,22 @@ class TestEmailNotify(TestCase): self.client.login(username='test', password='BJOKL5kAj6aFZ6A5') response = self.client.get('/experiments/started/153/') self.failUnlessEqual(response.status_code, 302) - + def test_started_email_logged_in_staff(self): - self.client.login(username='admintest', password='BJOKL5kAj6aFZ6A5') + self.client.login(username='admintest', password='BJOKL5kAj6aFZ6A5') response = self.client.get('/experiments/started/153/') self.failUnlessEqual(response.status_code, 200) def test_started_email_send(self): - self.client.login(username='admintest', password='BJOKL5kAj6aFZ6A5') + self.client.login(username='admintest', password='BJOKL5kAj6aFZ6A5') response = self.client.get('/experiments/started/153/') self.failUnlessEqual(response.status_code, 200) - + self.failUnless('pk1@example.com' in response.content) self.failUnless('Lane #8 : (11064) Paired ends 104' in response.content) - response = self.client.get('/experiments/started/153/', {'send':'1','bcc':'on'}) + response = self.client.get('/experiments/started/153/', + {'send': '1','bcc': 'on'}) self.failUnlessEqual(response.status_code, 200) self.failUnlessEqual(len(mail.outbox), 4) for m in mail.outbox: @@ -373,11 +415,16 @@ class TestEmailNotify(TestCase): """ Can we navigate between the flowcell and email forms properly? """ - self.client.login(username='supertest', password='BJOKL5kAj6aFZ6A5') + self.client.login(username='supertest', password='BJOKL5kAj6aFZ6A5') response = self.client.get('/experiments/started/153/') self.failUnlessEqual(response.status_code, 200) self.failUnless(re.search('Flowcell FC12150', response.content)) # require that navigation back to the admin page exists - self.failUnless(re.search('[^<]+', response.content)) - + flowcell_a_re = '[^<]+' + self.failUnless(re.search(flowcell_a_re, response.content)) + +def multi_lane_to_dict(lane): + """Convert a list of lane entries into a dictionary indexed by library ID + """ + return dict(((x['library_id'], x) for x in lane))