2 from BeautifulSoup import BeautifulSoup
6 import simplejson as json
12 from django.conf import settings
13 from django.core import mail
14 from django.core.exceptions import ObjectDoesNotExist
15 from django.test import TestCase
16 from htsworkflow.frontend.experiments import models
17 from htsworkflow.frontend.experiments import experiments
18 from htsworkflow.frontend.auth import apidata
20 from htsworkflow.pipelines.test.simulate_runfolder import TESTDATA_DIR
24 class ExperimentsTestCases(TestCase):
25 fixtures = ['test_flowcells.json']
28 self.tempdir = tempfile.mkdtemp(prefix='htsw-test-experiments-')
29 settings.RESULT_HOME_DIR = self.tempdir
31 self.fc1_id = 'FC12150'
32 self.fc1_root = os.path.join(self.tempdir, self.fc1_id)
33 os.mkdir(self.fc1_root)
34 self.fc1_dir = os.path.join(self.fc1_root, 'C1-37')
35 os.mkdir(self.fc1_dir)
36 runxml = 'run_FC12150_2007-09-27.xml'
37 shutil.copy(os.path.join(TESTDATA_DIR, runxml),
38 os.path.join(self.fc1_dir, runxml))
41 os.path.join(TESTDATA_DIR,
42 'woldlab_070829_USI-EAS44_0017_FC11055_1.srf'),
43 os.path.join(self.fc1_dir,
44 'woldlab_070829_SERIAL_FC12150_%d.srf' %(i,))
47 self.fc2_dir = os.path.join(self.tempdir, '42JTNAAXX')
48 os.mkdir(self.fc2_dir)
49 os.mkdir(os.path.join(self.fc2_dir, 'C1-25'))
50 os.mkdir(os.path.join(self.fc2_dir, 'C1-37'))
51 os.mkdir(os.path.join(self.fc2_dir, 'C1-37', 'Plots'))
54 shutil.rmtree(self.tempdir)
56 def test_flowcell_information(self):
58 Check the code that packs the django objects into simple types.
60 for fc_id in [u'FC12150', u"42JTNAAXX", "42JU1AAXX"]:
61 fc_dict = experiments.flowcell_information(fc_id)
62 fc_django = models.FlowCell.objects.get(flowcell_id=fc_id)
63 self.failUnlessEqual(fc_dict['flowcell_id'], fc_id)
64 self.failUnlessEqual(fc_django.flowcell_id, fc_id)
65 self.failUnlessEqual(fc_dict['sequencer'], fc_django.sequencer.name)
66 self.failUnlessEqual(fc_dict['read_length'], fc_django.read_length)
67 self.failUnlessEqual(fc_dict['notes'], fc_django.notes)
68 self.failUnlessEqual(fc_dict['cluster_station'], fc_django.cluster_station.name)
70 for lane in fc_django.lane_set.all():
71 lane_dict = fc_dict['lane_set'][lane.lane_number]
72 self.failUnlessEqual(lane_dict['cluster_estimate'], lane.cluster_estimate)
73 self.failUnlessEqual(lane_dict['comment'], lane.comment)
74 self.failUnlessEqual(lane_dict['flowcell'], lane.flowcell.flowcell_id)
75 self.failUnlessEqual(lane_dict['lane_number'], lane.lane_number)
76 self.failUnlessEqual(lane_dict['library_name'], lane.library.library_name)
77 self.failUnlessEqual(lane_dict['library_id'], lane.library.id)
78 self.failUnlessAlmostEqual(float(lane_dict['pM']), float(lane.pM))
79 self.failUnlessEqual(lane_dict['library_species'],
80 lane.library.library_species.scientific_name)
82 response = self.client.get('/experiments/config/%s/json' % (fc_id,), apidata)
83 # strptime isoformat string = '%Y-%m-%dT%H:%M:%S'
84 fc_json = json.loads(response.content)
85 self.failUnlessEqual(fc_json['flowcell_id'], fc_id)
86 self.failUnlessEqual(fc_json['sequencer'], fc_django.sequencer.name)
87 self.failUnlessEqual(fc_json['read_length'], fc_django.read_length)
88 self.failUnlessEqual(fc_json['notes'], fc_django.notes)
89 self.failUnlessEqual(fc_json['cluster_station'], fc_django.cluster_station.name)
92 for lane in fc_django.lane_set.all():
93 lane_dict = fc_json['lane_set'][unicode(lane.lane_number)]
94 self.failUnlessEqual(lane_dict['cluster_estimate'], lane.cluster_estimate)
95 self.failUnlessEqual(lane_dict['comment'], lane.comment)
96 self.failUnlessEqual(lane_dict['flowcell'], lane.flowcell.flowcell_id)
97 self.failUnlessEqual(lane_dict['lane_number'], lane.lane_number)
98 self.failUnlessEqual(lane_dict['library_name'], lane.library.library_name)
99 self.failUnlessEqual(lane_dict['library_id'], lane.library.id)
100 self.failUnlessAlmostEqual(float(lane_dict['pM']), float(lane.pM))
101 self.failUnlessEqual(lane_dict['library_species'],
102 lane.library.library_species.scientific_name)
104 def test_invalid_flowcell(self):
106 Make sure we get a 404 if we request an invalid flowcell ID
108 response = self.client.get('/experiments/config/nottheone/json', apidata)
109 self.failUnlessEqual(response.status_code, 404)
111 def test_no_key(self):
113 Require logging in to retrieve meta data
115 response = self.client.get(u'/experiments/config/FC12150/json')
116 self.failUnlessEqual(response.status_code, 403)
118 def test_library_id(self):
120 Library IDs should be flexible, so make sure we can retrive a non-numeric ID
122 response = self.client.get('/experiments/config/FC12150/json', apidata)
123 self.failUnlessEqual(response.status_code, 200)
124 flowcell = json.loads(response.content)
126 self.failUnlessEqual(flowcell['lane_set']['3']['library_id'], 'SL039')
128 response = self.client.get('/samples/library/SL039/json', apidata)
129 self.failUnlessEqual(response.status_code, 200)
130 library_sl039 = json.loads(response.content)
132 self.failUnlessEqual(library_sl039['library_id'], 'SL039')
134 def test_raw_id_field(self):
138 Library's have IDs, libraries also have primary keys,
139 we eventually had enough libraries that the drop down combo box was too
140 hard to filter through, unfortnately we want a field that uses our library
141 id and not the internal primary key, and raw_id_field uses primary keys.
143 This tests to make sure that the value entered in the raw library id field matches
144 the library id looked up.
146 expected_ids = [u'10981',u'11016',u'SL039',u'11060',
147 u'11061',u'11062',u'11063',u'11064']
148 self.client.login(username='supertest', password='BJOKL5kAj6aFZ6A5')
149 response = self.client.get('/admin/experiments/flowcell/153/')
150 soup = BeautifulSoup(response.content)
152 input_field = soup.find(id='id_lane_set-%d-library' % (i,))
153 library_field = input_field.findNext('strong')
154 library_id, library_name = library_field.string.split(':')
155 # strip leading '#' sign from name
156 library_id = library_id[1:]
157 self.failUnlessEqual(library_id, expected_ids[i])
158 self.failUnlessEqual(input_field['value'], library_id)
160 def test_library_to_flowcell_link(self):
162 Make sure the library page includes links to the flowcell pages.
164 self.client.login(username='supertest', password='BJOKL5kAj6aFZ6A5')
165 response = self.client.get('/library/11070/')
166 soup = BeautifulSoup(response.content)
167 failed_fc_span = soup.find(text='30012AAXX (failed)')
168 failed_fc_a = failed_fc_span.findPrevious('a')
169 # make sure some of our RDF made it.
170 self.failUnlessEqual(failed_fc_a.get('rel'), 'libns:flowcell')
171 self.failUnlessEqual(failed_fc_a.get('href'), '/flowcell/30012AAXX/')
174 def test_lanes_for(self):
176 Check the code that packs the django objects into simple types.
179 lanes = experiments.lanes_for(user)
180 self.failUnlessEqual(len(lanes), 5)
182 response = self.client.get('/experiments/lanes_for/%s/json' % (user,), apidata)
183 lanes_json = json.loads(response.content)
184 self.failUnlessEqual(len(lanes), len(lanes_json))
185 for i in range(len(lanes)):
186 self.failUnlessEqual(lanes[i]['comment'], lanes_json[i]['comment'])
187 self.failUnlessEqual(lanes[i]['lane_number'], lanes_json[i]['lane_number'])
188 self.failUnlessEqual(lanes[i]['flowcell'], lanes_json[i]['flowcell'])
189 self.failUnlessEqual(lanes[i]['run_date'], lanes_json[i]['run_date'])
191 def test_lanes_for_no_lanes(self):
193 Do we get something meaningful back when the user isn't attached to anything?
196 lanes = experiments.lanes_for(user)
197 self.failUnlessEqual(len(lanes), 0)
199 response = self.client.get('/experiments/lanes_for/%s/json' % (user,), apidata)
200 lanes_json = json.loads(response.content)
202 def test_lanes_for_no_user(self):
204 Do we get something meaningful back when its the wrong user
206 user = 'not a real user'
207 self.failUnlessRaises(ObjectDoesNotExist, experiments.lanes_for, user)
209 response = self.client.get('/experiments/lanes_for/%s/json' % (user,), apidata)
210 self.failUnlessEqual(response.status_code, 404)
213 def test_raw_data_dir(self):
214 """Raw data path generator check"""
215 flowcell_id = self.fc1_id
216 raw_dir = os.path.join(settings.RESULT_HOME_DIR, flowcell_id)
218 fc = models.FlowCell.objects.get(flowcell_id=flowcell_id)
219 self.failUnlessEqual(fc.get_raw_data_directory(), raw_dir)
221 fc.flowcell_id = flowcell_id + " (failed)"
222 self.failUnlessEqual(fc.get_raw_data_directory(), raw_dir)
225 def test_data_run_import(self):
226 srf_file_type = models.FileType.objects.get(name='SRF')
227 runxml_file_type = models.FileType.objects.get(name='run_xml')
228 flowcell_id = self.fc1_id
229 flowcell = models.FlowCell.objects.get(flowcell_id=flowcell_id)
230 flowcell.update_data_runs()
231 self.failUnlessEqual(len(flowcell.datarun_set.all()), 1)
233 run = flowcell.datarun_set.all()[0]
234 result_files = run.datafile_set.all()
235 result_dict = dict(((rf.relative_pathname, rf) for rf in result_files))
237 srf4 = result_dict['FC12150/C1-37/woldlab_070829_SERIAL_FC12150_4.srf']
238 self.failUnlessEqual(srf4.file_type, srf_file_type)
239 self.failUnlessEqual(srf4.library_id, '11060')
240 self.failUnlessEqual(srf4.data_run.flowcell.flowcell_id, 'FC12150')
241 self.failUnlessEqual(
242 srf4.data_run.flowcell.lane_set.get(lane_number=4).library_id,
244 self.failUnlessEqual(
246 os.path.join(settings.RESULT_HOME_DIR, srf4.relative_pathname))
248 lane_files = run.lane_files()
249 self.failUnlessEqual(lane_files[4]['srf'], srf4)
251 runxml= result_dict['FC12150/C1-37/run_FC12150_2007-09-27.xml']
252 self.failUnlessEqual(runxml.file_type, runxml_file_type)
253 self.failUnlessEqual(runxml.library_id, None)
256 def test_read_result_file(self):
257 """make sure we can return a result file
259 flowcell_id = self.fc1_id
260 flowcell = models.FlowCell.objects.get(flowcell_id=flowcell_id)
261 flowcell.update_data_runs()
263 #self.client.login(username='supertest', password='BJOKL5kAj6aFZ6A5')
265 result_files = flowcell.datarun_set.all()[0].datafile_set.all()
266 for f in result_files:
267 url = '/experiments/file/%s' % ( f.random_key,)
268 response = self.client.get(url)
269 self.failUnlessEqual(response.status_code, 200)
270 mimetype = f.file_type.mimetype
272 mimetype = 'application/octet-stream'
274 self.failUnlessEqual(mimetype, response['content-type'])
276 class TestFileType(TestCase):
277 def test_file_type_unicode(self):
278 file_type_objects = models.FileType.objects
279 name = 'QSEQ tarfile'
280 file_type_object = file_type_objects.get(name=name)
281 self.failUnlessEqual(u"<FileType: QSEQ tarfile>",
282 unicode(file_type_object))
284 class TestFileType(TestCase):
285 def test_find_file_type(self):
286 file_type_objects = models.FileType.objects
287 cases = [('woldlab_090921_HWUSI-EAS627_0009_42FC3AAXX_l7_r1.tar.bz2',
288 'QSEQ tarfile', 7, 1),
289 ('woldlab_091005_HWUSI-EAS627_0010_42JT2AAXX_1.srf',
291 ('s_1_eland_extended.txt.bz2','ELAND Extended', 1, None),
292 ('s_7_eland_multi.txt.bz2', 'ELAND Multi', 7, None),
293 ('s_3_eland_result.txt.bz2','ELAND Result', 3, None),
294 ('s_1_export.txt.bz2','ELAND Export', 1, None),
295 ('s_1_percent_call.png', 'IVC Percent Call', 1, None),
296 ('s_2_percent_base.png', 'IVC Percent Base', 2, None),
297 ('s_3_percent_all.png', 'IVC Percent All', 3, None),
298 ('s_4_call.png', 'IVC Call', 4, None),
299 ('s_5_all.png', 'IVC All', 5, None),
300 ('Summary.htm', 'Summary.htm', None, None),
301 ('run_42JT2AAXX_2009-10-07.xml', 'run_xml', None, None),
303 for filename, typename, lane, end in cases:
304 ft = models.find_file_type_metadata_from_filename(filename)
305 self.failUnlessEqual(ft['file_type'],
306 file_type_objects.get(name=typename))
307 self.failUnlessEqual(ft.get('lane', None), lane)
308 self.failUnlessEqual(ft.get('end', None), end)
310 def test_assign_file_type_complex_path(self):
311 file_type_objects = models.FileType.objects
312 cases = [('/a/b/c/woldlab_090921_HWUSI-EAS627_0009_42FC3AAXX_l7_r1.tar.bz2',
313 'QSEQ tarfile', 7, 1),
314 ('foo/woldlab_091005_HWUSI-EAS627_0010_42JT2AAXX_1.srf',
316 ('../s_1_eland_extended.txt.bz2','ELAND Extended', 1, None),
317 ('/bleem/s_7_eland_multi.txt.bz2', 'ELAND Multi', 7, None),
318 ('/qwer/s_3_eland_result.txt.bz2','ELAND Result', 3, None),
319 ('/ty///1/s_1_export.txt.bz2','ELAND Export', 1, None),
320 ('/help/s_1_percent_call.png', 'IVC Percent Call', 1, None),
321 ('/bored/s_2_percent_base.png', 'IVC Percent Base', 2, None),
322 ('/example1/s_3_percent_all.png', 'IVC Percent All', 3, None),
323 ('amonkey/s_4_call.png', 'IVC Call', 4, None),
324 ('fishie/s_5_all.png', 'IVC All', 5, None),
325 ('/random/Summary.htm', 'Summary.htm', None, None),
326 ('/notrandom/run_42JT2AAXX_2009-10-07.xml', 'run_xml', None, None),
328 for filename, typename, lane, end in cases:
329 result = models.find_file_type_metadata_from_filename(filename)
330 self.failUnlessEqual(result['file_type'],
331 file_type_objects.get(name=typename))
332 self.failUnlessEqual(result.get('lane',None), lane)
333 self.failUnlessEqual(result.get('end', None), end)
335 class TestEmailNotify(TestCase):
336 fixtures = ['test_flowcells.json']
338 def test_started_email_not_logged_in(self):
339 response = self.client.get('/experiments/started/153/')
340 self.failUnlessEqual(response.status_code, 302)
342 def test_started_email_logged_in_user(self):
343 self.client.login(username='test', password='BJOKL5kAj6aFZ6A5')
344 response = self.client.get('/experiments/started/153/')
345 self.failUnlessEqual(response.status_code, 302)
347 def test_started_email_logged_in_staff(self):
348 self.client.login(username='admintest', password='BJOKL5kAj6aFZ6A5')
349 response = self.client.get('/experiments/started/153/')
350 self.failUnlessEqual(response.status_code, 200)
352 def test_started_email_send(self):
353 self.client.login(username='admintest', password='BJOKL5kAj6aFZ6A5')
354 response = self.client.get('/experiments/started/153/')
355 self.failUnlessEqual(response.status_code, 200)
357 self.failUnless('pk1@example.com' in response.content)
358 self.failUnless('Lane #8 : (11064) Paired ends 104' in response.content)
360 response = self.client.get('/experiments/started/153/', {'send':'1','bcc':'on'})
361 self.failUnlessEqual(response.status_code, 200)
362 self.failUnlessEqual(len(mail.outbox), 4)
363 for m in mail.outbox:
364 self.failUnless(len(m.body) > 0)
366 def test_email_navigation(self):
368 Can we navigate between the flowcell and email forms properly?
370 self.client.login(username='supertest', password='BJOKL5kAj6aFZ6A5')
371 response = self.client.get('/experiments/started/153/')
372 self.failUnlessEqual(response.status_code, 200)
373 self.failUnless(re.search('Flowcell FC12150', response.content))
374 # require that navigation back to the admin page exists
375 self.failUnless(re.search('<a href="/admin/experiments/flowcell/153/">[^<]+</a>', response.content))