f9b7527696362953b2f448a0fb067f631854714e
[htsworkflow.git] / htsworkflow / frontend / experiments / tests.py
1 import re
2 try:
3     import json
4 except ImportError, e:
5     import simplejson as json
6 import sys
7
8 from django.core import mail
9 from django.test import TestCase
10 from htsworkflow.frontend.experiments import models
11 from htsworkflow.frontend.experiments import experiments
12 from htsworkflow.frontend.auth import apidata
13
14 class ExperimentsTestCases(TestCase):
15     fixtures = ['test_flowcells.json']
16
17     def setUp(self):
18         pass
19
20     def test_flowcell_information(self):
21         """
22         Check the code that packs the django objects into simple types.
23         """
24         for fc_id in [u'303TUAAXX', u"42JTNAAXX", "42JU1AAXX"]:
25             fc_dict = experiments.flowcell_information(fc_id)
26             fc_django = models.FlowCell.objects.get(flowcell_id=fc_id)
27             self.failUnlessEqual(fc_dict['flowcell_id'], fc_id)
28             self.failUnlessEqual(fc_django.flowcell_id, fc_id)
29             self.failUnlessEqual(fc_dict['sequencer'], fc_django.sequencer.name)
30             self.failUnlessEqual(fc_dict['read_length'], fc_django.read_length)
31             self.failUnlessEqual(fc_dict['notes'], fc_django.notes)
32             self.failUnlessEqual(fc_dict['cluster_station'], fc_django.cluster_station.name)
33
34             for lane in fc_django.lane_set.all():
35                 lane_dict = fc_dict['lane_set'][lane.lane_number]
36                 self.failUnlessEqual(lane_dict['cluster_estimate'], lane.cluster_estimate)
37                 self.failUnlessEqual(lane_dict['comment'], lane.comment)
38                 self.failUnlessEqual(lane_dict['flowcell'], lane.flowcell.flowcell_id)
39                 self.failUnlessEqual(lane_dict['lane_number'], lane.lane_number)
40                 self.failUnlessEqual(lane_dict['library_name'], lane.library.library_name)
41                 self.failUnlessEqual(lane_dict['library_id'], lane.library.library_id)
42                 self.failUnlessAlmostEqual(lane_dict['pM'], float(lane.pM))
43                 self.failUnlessEqual(lane_dict['library_species'],
44                                      lane.library.library_species.scientific_name)
45                     
46             response = self.client.get('/experiments/config/%s/json' % (fc_id,), apidata)
47             # strptime isoformat string = '%Y-%m-%dT%H:%M:%S'
48             fc_json = json.loads(response.content)
49             self.failUnlessEqual(fc_json['flowcell_id'], fc_id)
50             self.failUnlessEqual(fc_json['sequencer'], fc_django.sequencer.name)
51             self.failUnlessEqual(fc_json['read_length'], fc_django.read_length)
52             self.failUnlessEqual(fc_json['notes'], fc_django.notes)
53             self.failUnlessEqual(fc_json['cluster_station'], fc_django.cluster_station.name)
54
55
56             for lane in fc_django.lane_set.all():
57                 lane_dict = fc_json['lane_set'][unicode(lane.lane_number)]
58                 self.failUnlessEqual(lane_dict['cluster_estimate'], lane.cluster_estimate)
59                 self.failUnlessEqual(lane_dict['comment'], lane.comment)
60                 self.failUnlessEqual(lane_dict['flowcell'], lane.flowcell.flowcell_id)
61                 self.failUnlessEqual(lane_dict['lane_number'], lane.lane_number)
62                 self.failUnlessEqual(lane_dict['library_name'], lane.library.library_name)
63                 self.failUnlessEqual(lane_dict['library_id'], lane.library.library_id)
64                 self.failUnlessAlmostEqual(lane_dict['pM'], float(lane.pM))
65                 self.failUnlessEqual(lane_dict['library_species'],
66                                      lane.library.library_species.scientific_name)
67
68     def test_invalid_flowcell(self):
69         """
70         Make sure we get a 404 if we request an invalid flowcell ID
71         """
72         response = self.client.get('/experiments/config/nottheone/json', apidata)
73         self.failUnlessEqual(response.status_code, 404)
74
75     def test_no_key(self):
76         """
77         Require logging in to retrieve meta data
78         """
79         response = self.client.get(u'/experiments/config/303TUAAXX/json')
80         self.failUnlessEqual(response.status_code, 403)
81
82     def test_library_id(self):
83         """
84         Library IDs should be flexible, so make sure we can retrive a non-numeric ID
85         """
86         response = self.client.get('/experiments/config/303TUAAXX/json', apidata)
87         self.failUnlessEqual(response.status_code, 200)
88         flowcell = json.loads(response.content)
89
90         self.failUnlessEqual(flowcell['lane_set']['3']['library_id'], 'SL039')
91
92         response = self.client.get('/samples/library/SL039/json', apidata)
93         self.failUnlessEqual(response.status_code, 200)
94         library_sl039 = json.loads(response.content)
95
96         self.failUnlessEqual(library_sl039['library_id'], 'SL039')
97
98 class TestEmailNotify(TestCase):
99     fixtures = ['test_flowcells.json']
100
101     def test_started_email_not_logged_in(self):
102         response = self.client.get('/experiments/started/153/')
103         self.failUnlessEqual(response.status_code, 302)
104
105     def test_started_email_logged_in_user(self):
106         self.client.login(username='test', password='BJOKL5kAj6aFZ6A5')
107         response = self.client.get('/experiments/started/153/')
108         self.failUnlessEqual(response.status_code, 302)
109         
110     def test_started_email_logged_in_staff(self):
111         self.client.login(username='admintest', password='BJOKL5kAj6aFZ6A5') 
112         response = self.client.get('/experiments/started/153/')
113         self.failUnlessEqual(response.status_code, 200)
114
115     def test_started_email_send(self):
116         self.client.login(username='admintest', password='BJOKL5kAj6aFZ6A5') 
117         response = self.client.get('/experiments/started/153/')
118         self.failUnlessEqual(response.status_code, 200)
119         
120         self.failUnless('pk1@example.com' in response.content)
121         self.failUnless('Lane #8 : (11064) Paired ends 104' in response.content)
122
123         response = self.client.get('/experiments/started/153/', {'send':'1','bcc':'on'})
124         self.failUnlessEqual(response.status_code, 200)
125         self.failUnlessEqual(len(mail.outbox), 4)
126         for m in mail.outbox:
127             self.failUnless(len(m.body) > 0)
128
129     def test_email_navigation(self):
130         """
131         Can we navigate between the flowcell and email forms properly?
132         """
133         self.client.login(username='supertest', password='BJOKL5kAj6aFZ6A5') 
134         response = self.client.get('/experiments/started/153/')
135         self.failUnlessEqual(response.status_code, 200)
136         self.failUnless(re.search('Flowcell 303TUAAXX', response.content))
137         # require that navigation back to the admin page exists
138         self.failUnless(re.search('<a href="/admin/experiments/flowcell/153/">[^<]+</a>', response.content))
139