OLD | NEW |
1 # Copyright 2014 The Chromium Authors. All rights reserved. | 1 # Copyright 2014 The Chromium Authors. All rights reserved. |
2 # Use of this source code is governed by a BSD-style license that can be | 2 # Use of this source code is governed by a BSD-style license that can be |
3 # found in the LICENSE file. | 3 # found in the LICENSE file. |
4 | 4 |
5 import alerts | 5 import alerts |
6 import json | 6 import json |
7 import random | 7 import random |
8 import string | 8 import string |
9 import unittest | 9 import unittest |
10 import webtest | 10 import webtest |
11 | 11 |
12 from google.appengine.api import memcache | 12 from google.appengine.api import memcache |
13 from google.appengine.ext import ndb | 13 from google.appengine.ext import ndb |
14 from google.appengine.ext import testbed | 14 from google.appengine.ext import testbed |
15 | 15 |
16 | 16 |
17 class AlertsTest(unittest.TestCase): | 17 class AlertsTest(unittest.TestCase): |
18 def setUp(self): | 18 def setUp(self): |
19 self.testbed = testbed.Testbed() | 19 self.testbed = testbed.Testbed() |
20 self.testbed.activate() | 20 self.testbed.activate() |
21 self.testbed.init_user_stub() | |
22 self.testbed.init_memcache_stub() | 21 self.testbed.init_memcache_stub() |
23 self.testbed.init_datastore_v3_stub() | 22 self.testbed.init_datastore_v3_stub() |
24 self.testapp = webtest.TestApp(alerts.app) | 23 self.testapp = webtest.TestApp(alerts.app) |
25 | 24 |
26 def tearDown(self): | 25 def tearDown(self): |
27 self.testbed.deactivate() | 26 self.testbed.deactivate() |
28 | 27 |
29 def check_json_headers(self, res): | 28 def check_json_headers(self, res): |
30 self.assertEqual(res.content_type, 'application/json') | 29 self.assertEqual(res.content_type, 'application/json') |
31 # This is necessary for cross-site tools to retrieve alerts | 30 # This is necessary for cross-site tools to retrieve alerts |
(...skipping 37 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... |
69 self.assertEqual(alerts['alerts'], 'everything is OK') | 68 self.assertEqual(alerts['alerts'], 'everything is OK') |
70 | 69 |
71 def test_alerts_jsons_are_stored_in_history(self): | 70 def test_alerts_jsons_are_stored_in_history(self): |
72 test_alerts1 = {'alerts': ['hello', 'world', '1']} | 71 test_alerts1 = {'alerts': ['hello', 'world', '1']} |
73 test_alerts2 = {'alerts': ['hello', 'world', '2']} | 72 test_alerts2 = {'alerts': ['hello', 'world', '2']} |
74 self.testapp.post('/alerts', {'content': json.dumps(test_alerts1)}) | 73 self.testapp.post('/alerts', {'content': json.dumps(test_alerts1)}) |
75 self.testapp.post('/alerts', {'content': json.dumps(test_alerts2)}) | 74 self.testapp.post('/alerts', {'content': json.dumps(test_alerts2)}) |
76 alerts_query = alerts.AlertsJSON.query().order(alerts.AlertsJSON.date) | 75 alerts_query = alerts.AlertsJSON.query().order(alerts.AlertsJSON.date) |
77 stored_alerts = alerts_query.fetch(limit=3) | 76 stored_alerts = alerts_query.fetch(limit=3) |
78 self.assertEqual(2, len(stored_alerts)) | 77 self.assertEqual(2, len(stored_alerts)) |
79 self.assertEqual(stored_alerts[0].type, 'alerts') | |
80 self.assertEqual(stored_alerts[1].type, 'alerts') | |
81 stored_alerts1 = json.loads(stored_alerts[0].json) | 78 stored_alerts1 = json.loads(stored_alerts[0].json) |
82 stored_alerts2 = json.loads(stored_alerts[1].json) | 79 stored_alerts2 = json.loads(stored_alerts[1].json) |
83 self.assertEqual(test_alerts1['alerts'], stored_alerts1['alerts']) | 80 self.assertEqual(test_alerts1['alerts'], stored_alerts1['alerts']) |
84 self.assertEqual(test_alerts2['alerts'], stored_alerts2['alerts']) | 81 self.assertEqual(test_alerts2['alerts'], stored_alerts2['alerts']) |
85 self.assertTrue('date' in stored_alerts1) | 82 self.assertTrue('date' in stored_alerts1) |
86 self.assertTrue('date' in stored_alerts2) | 83 self.assertTrue('date' in stored_alerts2) |
87 self.assertEqual(type(stored_alerts1['date']), int) | 84 self.assertEqual(type(stored_alerts1['date']), int) |
88 self.assertEqual(type(stored_alerts2['date']), int) | 85 self.assertEqual(type(stored_alerts2['date']), int) |
89 | 86 |
90 def test_repeating_alerts_are_not_stored_to_history(self): | 87 def test_repeating_alerts_are_not_stored_to_history(self): |
91 test_alerts = {'alerts': ['hello', 'world']} | 88 test_alerts = {'alerts': ['hello', 'world']} |
92 self.testapp.post('/alerts', {'content': json.dumps(test_alerts)}) | 89 self.testapp.post('/alerts', {'content': json.dumps(test_alerts)}) |
93 test_alerts['last_builder_info'] = {'some': 'info'} | 90 test_alerts['last_builder_info'] = {'some': 'info'} |
94 self.testapp.post('/alerts', {'content': json.dumps(test_alerts)}) | 91 self.testapp.post('/alerts', {'content': json.dumps(test_alerts)}) |
95 stored_alerts = alerts.AlertsJSON.query().fetch(limit=2) | 92 stored_alerts = alerts.AlertsJSON.query().fetch(limit=2) |
96 self.assertEqual(1, len(stored_alerts)) | 93 self.assertEqual(1, len(stored_alerts)) |
97 | 94 |
98 def test_alerts_jsons_are_retrieved_from_history(self): | |
99 test_alert = {'alerts': ['hello', 'world', '1']} | |
100 alerts.AlertsJSON(json=json.dumps(test_alert), type='alerts').put() | |
101 response = self.testapp.get('/alerts-history') | |
102 self.assertEqual(response.status_int, 200) | |
103 self.assertEqual(response.content_type, 'application/json') | |
104 response_json = json.loads(response.normal_body) | |
105 self.assertEqual(len(response_json['history']), 1) | |
106 self.assertEqual(response_json['history'][0], test_alert) | |
107 | |
108 def test_internal_alerts_in_history_visible_to_internal_users_only(self): | |
109 test_alert = {'alerts': ['hello', 'world', '1']} | |
110 alerts.AlertsJSON(json=json.dumps(test_alert), | |
111 type='internal-alerts').put() | |
112 | |
113 # No signed-in user. | |
114 response = self.testapp.get('/alerts-history') | |
115 self.assertEqual(response.status_int, 200) | |
116 self.assertEqual(response.content_type, 'application/json') | |
117 response_json = json.loads(response.normal_body) | |
118 self.assertEqual(len(response_json['history']), 0) | |
119 | |
120 # Non-internal user. | |
121 self.testbed.setup_env(USER_EMAIL='test@example.com', USER_ID='1', | |
122 USER_IS_ADMIN='1', overwrite=True) | |
123 response = self.testapp.get('/alerts-history') | |
124 self.assertEqual(response.status_int, 200) | |
125 self.assertEqual(response.content_type, 'application/json') | |
126 response_json = json.loads(response.normal_body) | |
127 self.assertEqual(len(response_json['history']), 0) | |
128 | |
129 # Internal user. | |
130 self.testbed.setup_env(USER_EMAIL='test@google.com', USER_ID='2', | |
131 USER_IS_ADMIN='1', overwrite=True) | |
132 response = self.testapp.get('/alerts-history') | |
133 self.assertEqual(response.status_int, 200) | |
134 self.assertEqual(response.content_type, 'application/json') | |
135 response_json = json.loads(response.normal_body) | |
136 self.assertEqual(len(response_json['history']), 1) | |
137 self.assertEqual(response_json['history'][0], test_alert) | |
138 | |
139 def test_returned_alerts_from_history_are_paged(self): | |
140 for i in range(20): | |
141 test_alert = {'alerts': ['hello', 'world', i]} | |
142 alerts.AlertsJSON(json=json.dumps(test_alert), type='alerts').put() | |
143 | |
144 response = self.testapp.get('/alerts-history?limit=15') | |
145 self.assertEqual(response.status_int, 200) | |
146 self.assertEqual(response.content_type, 'application/json') | |
147 response_json = json.loads(response.normal_body) | |
148 self.assertEqual(len(response_json['history']), 15) | |
149 self.assertEqual(response_json['has_more'], True) | |
150 | |
151 url = '/alerts-history?limit=15&cursor=%s' % response_json['cursor'] | |
152 response = self.testapp.get(url) | |
153 self.assertEqual(response.status_int, 200) | |
154 self.assertEqual(response.content_type, 'application/json') | |
155 response_json = json.loads(response.normal_body) | |
156 self.assertEqual(len(response_json['history']), 5) | |
157 self.assertEqual(response_json['has_more'], False) | |
158 | |
159 def test_large_number_of_alerts(self): | 95 def test_large_number_of_alerts(self): |
160 # This generates ~2.5MB of JSON that compresses to ~750K. Real | 96 # This generates ~2.5MB of JSON that compresses to ~750K. Real |
161 # data compresses about 6x better. | 97 # data compresses about 6x better. |
162 random.seed(0xf00f00) | 98 random.seed(0xf00f00) |
163 put_alerts = self.generate_fake_alerts(4000) | 99 put_alerts = self.generate_fake_alerts(4000) |
164 | 100 |
165 params = {'content': json.dumps(put_alerts)} | 101 params = {'content': json.dumps(put_alerts)} |
166 self.testapp.post('/alerts', params) | 102 self.testapp.post('/alerts', params) |
167 | 103 |
168 res = self.testapp.get('/alerts') | 104 res = self.testapp.get('/alerts') |
(...skipping 45 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... |
214 | 150 |
215 def value(): | 151 def value(): |
216 generators = [time, build, revision, test, literal_array, | 152 generators = [time, build, revision, test, literal_array, |
217 literal_map] | 153 literal_map] |
218 return random.choice(generators)() | 154 return random.choice(generators)() |
219 | 155 |
220 alert = {} | 156 alert = {} |
221 for _ in range(random.randint(6, 9)): | 157 for _ in range(random.randint(6, 9)): |
222 alert[label()] = value() | 158 alert[label()] = value() |
223 return alert | 159 return alert |
OLD | NEW |