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_lanes_for(self):
162 Check the code that packs the django objects into simple types.
165 lanes = experiments.lanes_for(user)
166 self.failUnlessEqual(len(lanes), 5)
168 response = self.client.get('/experiments/lanes_for/%s/json' % (user,), apidata)
169 lanes_json = json.loads(response.content)
170 self.failUnlessEqual(len(lanes), len(lanes_json))
171 for i in range(len(lanes)):
172 self.failUnlessEqual(lanes[i]['comment'], lanes_json[i]['comment'])
173 self.failUnlessEqual(lanes[i]['lane_number'], lanes_json[i]['lane_number'])
174 self.failUnlessEqual(lanes[i]['flowcell'], lanes_json[i]['flowcell'])
175 self.failUnlessEqual(lanes[i]['run_date'], lanes_json[i]['run_date'])
177 def test_lanes_for_no_lanes(self):
179 Do we get something meaningful back when the user isn't attached to anything?
182 lanes = experiments.lanes_for(user)
183 self.failUnlessEqual(len(lanes), 0)
185 response = self.client.get('/experiments/lanes_for/%s/json' % (user,), apidata)
186 lanes_json = json.loads(response.content)
188 def test_lanes_for_no_user(self):
190 Do we get something meaningful back when its the wrong user
192 user = 'not a real user'
193 self.failUnlessRaises(ObjectDoesNotExist, experiments.lanes_for, user)
195 response = self.client.get('/experiments/lanes_for/%s/json' % (user,), apidata)
196 self.failUnlessEqual(response.status_code, 404)
199 def test_raw_data_dir(self):
200 """Raw data path generator check"""
201 flowcell_id = self.fc1_id
202 raw_dir = os.path.join(settings.RESULT_HOME_DIR, flowcell_id)
204 fc = models.FlowCell.objects.get(flowcell_id=flowcell_id)
205 self.failUnlessEqual(fc.get_raw_data_directory(), raw_dir)
207 fc.flowcell_id = flowcell_id + " (failed)"
208 self.failUnlessEqual(fc.get_raw_data_directory(), raw_dir)
211 def test_data_run_import(self):
212 srf_file_type = models.FileType.objects.get(name='SRF')
213 runxml_file_type = models.FileType.objects.get(name='run_xml')
214 flowcell_id = self.fc1_id
215 flowcell = models.FlowCell.objects.get(flowcell_id=flowcell_id)
216 flowcell.update_data_runs()
217 self.failUnlessEqual(len(flowcell.datarun_set.all()), 1)
219 run = flowcell.datarun_set.all()[0]
220 result_files = run.datafile_set.all()
221 result_dict = dict(((rf.relative_pathname, rf) for rf in result_files))
223 srf4 = result_dict['FC12150/C1-37/woldlab_070829_SERIAL_FC12150_4.srf']
224 self.failUnlessEqual(srf4.file_type, srf_file_type)
225 self.failUnlessEqual(srf4.library_id, '11060')
226 self.failUnlessEqual(srf4.data_run.flowcell.flowcell_id, 'FC12150')
227 self.failUnlessEqual(
228 srf4.data_run.flowcell.lane_set.get(lane_number=4).library_id,
230 self.failUnlessEqual(
232 os.path.join(settings.RESULT_HOME_DIR, srf4.relative_pathname))
234 lane_files = run.lane_files()
235 self.failUnlessEqual(lane_files[4]['srf'], srf4)
237 runxml= result_dict['FC12150/C1-37/run_FC12150_2007-09-27.xml']
238 self.failUnlessEqual(runxml.file_type, runxml_file_type)
239 self.failUnlessEqual(runxml.library_id, None)
242 def test_read_result_file(self):
243 """make sure we can return a result file
245 flowcell_id = self.fc1_id
246 flowcell = models.FlowCell.objects.get(flowcell_id=flowcell_id)
247 flowcell.update_data_runs()
249 #self.client.login(username='supertest', password='BJOKL5kAj6aFZ6A5')
251 result_files = flowcell.datarun_set.all()[0].datafile_set.all()
252 for f in result_files:
253 url = '/experiments/file/%s' % ( f.random_key,)
254 response = self.client.get(url)
255 self.failUnlessEqual(response.status_code, 200)
256 mimetype = f.file_type.mimetype
258 mimetype = 'application/octet-stream'
260 self.failUnlessEqual(mimetype, response['content-type'])
262 class TestFileType(TestCase):
263 def test_file_type_unicode(self):
264 file_type_objects = models.FileType.objects
265 name = 'QSEQ tarfile'
266 file_type_object = file_type_objects.get(name=name)
267 self.failUnlessEqual(u"<FileType: QSEQ tarfile>",
268 unicode(file_type_object))
270 class TestFileType(TestCase):
271 def test_find_file_type(self):
272 file_type_objects = models.FileType.objects
273 cases = [('woldlab_090921_HWUSI-EAS627_0009_42FC3AAXX_l7_r1.tar.bz2',
274 'QSEQ tarfile', 7, 1),
275 ('woldlab_091005_HWUSI-EAS627_0010_42JT2AAXX_1.srf',
277 ('s_1_eland_extended.txt.bz2','ELAND Extended', 1, None),
278 ('s_7_eland_multi.txt.bz2', 'ELAND Multi', 7, None),
279 ('s_3_eland_result.txt.bz2','ELAND Result', 3, None),
280 ('s_1_export.txt.bz2','ELAND Export', 1, None),
281 ('s_1_percent_call.png', 'IVC Percent Call', 1, None),
282 ('s_2_percent_base.png', 'IVC Percent Base', 2, None),
283 ('s_3_percent_all.png', 'IVC Percent All', 3, None),
284 ('s_4_call.png', 'IVC Call', 4, None),
285 ('s_5_all.png', 'IVC All', 5, None),
286 ('Summary.htm', 'Summary.htm', None, None),
287 ('run_42JT2AAXX_2009-10-07.xml', 'run_xml', None, None),
289 for filename, typename, lane, end in cases:
290 ft = models.find_file_type_metadata_from_filename(filename)
291 self.failUnlessEqual(ft['file_type'],
292 file_type_objects.get(name=typename))
293 self.failUnlessEqual(ft.get('lane', None), lane)
294 self.failUnlessEqual(ft.get('end', None), end)
296 def test_assign_file_type_complex_path(self):
297 file_type_objects = models.FileType.objects
298 cases = [('/a/b/c/woldlab_090921_HWUSI-EAS627_0009_42FC3AAXX_l7_r1.tar.bz2',
299 'QSEQ tarfile', 7, 1),
300 ('foo/woldlab_091005_HWUSI-EAS627_0010_42JT2AAXX_1.srf',
302 ('../s_1_eland_extended.txt.bz2','ELAND Extended', 1, None),
303 ('/bleem/s_7_eland_multi.txt.bz2', 'ELAND Multi', 7, None),
304 ('/qwer/s_3_eland_result.txt.bz2','ELAND Result', 3, None),
305 ('/ty///1/s_1_export.txt.bz2','ELAND Export', 1, None),
306 ('/help/s_1_percent_call.png', 'IVC Percent Call', 1, None),
307 ('/bored/s_2_percent_base.png', 'IVC Percent Base', 2, None),
308 ('/example1/s_3_percent_all.png', 'IVC Percent All', 3, None),
309 ('amonkey/s_4_call.png', 'IVC Call', 4, None),
310 ('fishie/s_5_all.png', 'IVC All', 5, None),
311 ('/random/Summary.htm', 'Summary.htm', None, None),
312 ('/notrandom/run_42JT2AAXX_2009-10-07.xml', 'run_xml', None, None),
314 for filename, typename, lane, end in cases:
315 result = models.find_file_type_metadata_from_filename(filename)
316 self.failUnlessEqual(result['file_type'],
317 file_type_objects.get(name=typename))
318 self.failUnlessEqual(result.get('lane',None), lane)
319 self.failUnlessEqual(result.get('end', None), end)
321 class TestEmailNotify(TestCase):
322 fixtures = ['test_flowcells.json']
324 def test_started_email_not_logged_in(self):
325 response = self.client.get('/experiments/started/153/')
326 self.failUnlessEqual(response.status_code, 302)
328 def test_started_email_logged_in_user(self):
329 self.client.login(username='test', password='BJOKL5kAj6aFZ6A5')
330 response = self.client.get('/experiments/started/153/')
331 self.failUnlessEqual(response.status_code, 302)
333 def test_started_email_logged_in_staff(self):
334 self.client.login(username='admintest', password='BJOKL5kAj6aFZ6A5')
335 response = self.client.get('/experiments/started/153/')
336 self.failUnlessEqual(response.status_code, 200)
338 def test_started_email_send(self):
339 self.client.login(username='admintest', password='BJOKL5kAj6aFZ6A5')
340 response = self.client.get('/experiments/started/153/')
341 self.failUnlessEqual(response.status_code, 200)
343 self.failUnless('pk1@example.com' in response.content)
344 self.failUnless('Lane #8 : (11064) Paired ends 104' in response.content)
346 response = self.client.get('/experiments/started/153/', {'send':'1','bcc':'on'})
347 self.failUnlessEqual(response.status_code, 200)
348 self.failUnlessEqual(len(mail.outbox), 4)
349 for m in mail.outbox:
350 self.failUnless(len(m.body) > 0)
352 def test_email_navigation(self):
354 Can we navigate between the flowcell and email forms properly?
356 self.client.login(username='supertest', password='BJOKL5kAj6aFZ6A5')
357 response = self.client.get('/experiments/started/153/')
358 self.failUnlessEqual(response.status_code, 200)
359 self.failUnless(re.search('Flowcell FC12150', response.content))
360 # require that navigation back to the admin page exists
361 self.failUnless(re.search('<a href="/admin/experiments/flowcell/153/">[^<]+</a>', response.content))