Chromium Code Reviews
chromiumcodereview-hr@appspot.gserviceaccount.com (chromiumcodereview-hr) | Please choose your nickname with Settings | Help | Chromium Project | Gerrit Changes | Sign out
(410)

Issues Search
    My Issues | Starred     Open | Closed | All

Side by Side Diff: Tools/GardeningServer/alerts_test.py

Issue 728023004: Remove GardeningServer. (Closed) Base URL: svn://svn.chromium.org/blink/trunk
Patch Set: Created 6 years, 1 month ago
Use n/p to move between diff chunks; N/P to move between comments. Draft comments are only viewable by you.
Jump to:
View unified diff | Download patch | Annotate | Revision Log
« no previous file with comments | « Tools/GardeningServer/alerts.py ('k') | Tools/GardeningServer/app.yaml » ('j') | no next file with comments »
Toggle Intra-line Diffs ('i') | Expand Comments ('e') | Collapse Comments ('c') | Show Comments Hide Comments ('s')
OLDNEW
(Empty)
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
3 # found in the LICENSE file.
4
5 import alerts
6 import json
7 import random
8 import string
9 import unittest
10 import webtest
11
12 from google.appengine.api import memcache
13 from google.appengine.ext import ndb
14 from google.appengine.ext import testbed
15
16
17 class AlertsTest(unittest.TestCase):
18 def setUp(self):
19 self.testbed = testbed.Testbed()
20 self.testbed.activate()
21 self.testbed.init_user_stub()
22 self.testbed.init_memcache_stub()
23 self.testbed.init_datastore_v3_stub()
24 self.testapp = webtest.TestApp(alerts.app)
25
26 def tearDown(self):
27 self.testbed.deactivate()
28
29 def check_json_headers(self, res):
30 self.assertEqual(res.content_type, 'application/json')
31 # This is necessary for cross-site tools to retrieve alerts
32 self.assertEqual(res.headers['access-control-allow-origin'], '*')
33
34 def test_get_no_data_cached(self):
35 res = self.testapp.get('/alerts')
36 self.check_json_headers(res)
37 self.assertEqual(res.body, '')
38
39 def test_happy_path(self):
40 # Set it.
41 params = {'content': '{"alerts": ["hello", "world"]}'}
42 self.testapp.post('/alerts', params)
43
44 # Get it.
45 res = self.testapp.get('/alerts')
46 self.check_json_headers(res)
47 alerts = json.loads(res.body)
48
49 # The server should have stuck a 'date' on there.
50 self.assertTrue('date' in alerts)
51 self.assertEqual(type(alerts['date']), int)
52
53 self.assertEqual(alerts['alerts'], ['hello', 'world'])
54
55 def test_post_invalid_data_not_reflected(self):
56 params = {'content': '[{"this is not valid JSON'}
57 self.testapp.post('/alerts', params, status=400)
58 res = self.testapp.get('/alerts')
59 self.assertEqual(res.body, '')
60
61 def test_post_invalid_data_does_not_overwrite_valid_data(self):
62 # Populate the cache with something valid
63 params = {'content': '{"alerts": "everything is OK"}'}
64 self.testapp.post('/alerts', params)
65 self.testapp.post('/alerts', {'content': 'woozlwuzl'}, status=400)
66 res = self.testapp.get('/alerts')
67 self.check_json_headers(res)
68 alerts = json.loads(res.body)
69 self.assertEqual(alerts['alerts'], 'everything is OK')
70
71 def test_alerts_jsons_are_stored_in_history(self):
72 test_alerts1 = {'alerts': ['hello', 'world', '1']}
73 test_alerts2 = {'alerts': ['hello', 'world', '2']}
74 self.testapp.post('/alerts', {'content': json.dumps(test_alerts1)})
75 self.testapp.post('/alerts', {'content': json.dumps(test_alerts2)})
76 alerts_query = alerts.AlertsJSON.query().order(alerts.AlertsJSON.date)
77 stored_alerts = alerts_query.fetch(limit=3)
78 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)
82 stored_alerts2 = json.loads(stored_alerts[1].json)
83 self.assertEqual(test_alerts1['alerts'], stored_alerts1['alerts'])
84 self.assertEqual(test_alerts2['alerts'], stored_alerts2['alerts'])
85 self.assertTrue('date' in stored_alerts1)
86 self.assertTrue('date' in stored_alerts2)
87 self.assertEqual(type(stored_alerts1['date']), int)
88 self.assertEqual(type(stored_alerts2['date']), int)
89
90 def test_repeating_alerts_are_not_stored_to_history(self):
91 test_alerts = {'alerts': ['hello', 'world']}
92 self.testapp.post('/alerts', {'content': json.dumps(test_alerts)})
93 test_alerts['last_builder_info'] = {'some': 'info'}
94 self.testapp.post('/alerts', {'content': json.dumps(test_alerts)})
95 stored_alerts = alerts.AlertsJSON.query().fetch(limit=2)
96 self.assertEqual(1, len(stored_alerts))
97
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 parsed_json = json.loads(response.normal_body)
105 self.assertEqual(len(parsed_json['history']), 1)
106
107 entry_id = parsed_json['history'][0]
108 response = self.testapp.get('/alerts-history/%s' % entry_id)
109 self.assertEqual(response.status_int, 200)
110 self.assertEqual(response.content_type, 'application/json')
111 parsed_json = json.loads(response.normal_body)
112 self.assertEqual(parsed_json['alerts'], test_alert['alerts'])
113
114 def test_provides_login_url(self):
115 response = self.testapp.get('/alerts-history')
116 self.assertIn('login-url', response)
117
118 def test_invalid_keys_return_400(self):
119 response = self.testapp.get('/alerts-history/kjhg$%T',
120 expect_errors=True)
121 self.assertEqual(response.status_int, 400)
122 self.assertEqual(response.content_type, 'application/json')
123
124 def test_non_existing_keys_return_404(self):
125 response = self.testapp.get('/alerts-history/5348024557502464',
126 expect_errors=True)
127 self.assertEqual(response.status_int, 404)
128 self.assertEqual(response.content_type, 'application/json')
129
130 def test_internal_alerts_can_only_retrieved_by_internal_users(self):
131 test_alert = {'alerts': ['hello', 'world', '1']}
132 internal_alert = alerts.AlertsJSON(json=json.dumps(test_alert),
133 type='internal-alerts')
134 internal_alert_key = internal_alert.put().integer_id()
135
136 # No signed-in user.
137 response = self.testapp.get('/alerts-history/%s' % internal_alert_key,
138 expect_errors=True)
139 self.assertEqual(response.status_int, 404)
140 self.assertEqual(response.content_type, 'application/json')
141 parsed_json = json.loads(response.normal_body)
142 self.assertNotIn('alerts', parsed_json)
143
144 # Non-internal user.
145 self.testbed.setup_env(USER_EMAIL='test@example.com', USER_ID='1',
146 USER_IS_ADMIN='1', overwrite=True)
147 response = self.testapp.get('/alerts-history/%s' % internal_alert_key,
148 expect_errors=True)
149 self.assertEqual(response.status_int, 404)
150 self.assertEqual(response.content_type, 'application/json')
151 parsed_json = json.loads(response.normal_body)
152 self.assertNotIn('alerts', parsed_json)
153
154 def test_lists_internal_alerts_to_internal_users_only(self):
155 test_alert = {'alerts': ['hello', 'world', '1']}
156 alerts.AlertsJSON(json=json.dumps(test_alert),
157 type='internal-alerts').put()
158
159 # No signed-in user.
160 response = self.testapp.get('/alerts-history')
161 self.assertEqual(response.status_int, 200)
162 self.assertEqual(response.content_type, 'application/json')
163 response_json = json.loads(response.normal_body)
164 self.assertEqual(len(response_json['history']), 0)
165
166 # Non-internal user.
167 self.testbed.setup_env(USER_EMAIL='test@example.com', USER_ID='1',
168 USER_IS_ADMIN='1', overwrite=True)
169 response = self.testapp.get('/alerts-history')
170 self.assertEqual(response.status_int, 200)
171 self.assertEqual(response.content_type, 'application/json')
172 response_json = json.loads(response.normal_body)
173 self.assertEqual(len(response_json['history']), 0)
174
175 # Internal user.
176 self.testbed.setup_env(USER_EMAIL='test@google.com', USER_ID='2',
177 USER_IS_ADMIN='1', overwrite=True)
178 response = self.testapp.get('/alerts-history')
179 self.assertEqual(response.status_int, 200)
180 self.assertEqual(response.content_type, 'application/json')
181 response_json = json.loads(response.normal_body)
182 self.assertEqual(len(response_json['history']), 1)
183
184 def test_returned_alerts_from_history_are_paged(self):
185 for i in range(20):
186 test_alert = {'alerts': ['hello', 'world', i]}
187 alerts.AlertsJSON(json=json.dumps(test_alert), type='alerts').put()
188
189 response = self.testapp.get('/alerts-history?limit=15')
190 self.assertEqual(response.status_int, 200)
191 self.assertEqual(response.content_type, 'application/json')
192 response_json = json.loads(response.normal_body)
193 self.assertEqual(len(response_json['history']), 15)
194 self.assertEqual(response_json['has_more'], True)
195
196 url = '/alerts-history?limit=15&cursor=%s' % response_json['cursor']
197 response = self.testapp.get(url)
198 self.assertEqual(response.status_int, 200)
199 self.assertEqual(response.content_type, 'application/json')
200 response_json = json.loads(response.normal_body)
201 self.assertEqual(len(response_json['history']), 5)
202 self.assertEqual(response_json['has_more'], False)
203
204 def test_large_number_of_alerts(self):
205 # This generates ~2.5MB of JSON that compresses to ~750K. Real
206 # data compresses about 6x better.
207 random.seed(0xf00f00)
208 put_alerts = self.generate_fake_alerts(4000)
209
210 params = {'content': json.dumps(put_alerts)}
211 self.testapp.post('/alerts', params)
212
213 res = self.testapp.get('/alerts')
214 got_alerts = json.loads(res.body)
215 self.assertEquals(got_alerts['alerts'], put_alerts['alerts'])
216
217 def generate_fake_alerts(self, n):
218 return {'alerts': [self.generate_fake_alert() for _ in range(n)]}
219
220 def generate_fake_alert(self):
221 # fake labels
222 labels = [['', 'last_', 'latest_', 'failing_', 'passing_'],
223 ['build', 'builder', 'revision'],
224 ['', 's', '_url', '_reason', '_name']]
225
226 def label():
227 return string.join(map(random.choice, labels), '')
228
229 # fake values
230 def time():
231 return random.randint(1407976107614, 1408076107614) / 101.0
232
233 def build():
234 return random.randint(2737, 2894)
235
236 def revision():
237 return random.randint(288849, 289415)
238
239 tests = [['Activity', 'Async', 'Browser', 'Content', 'Input'],
240 ['Manager', 'Card', 'Sandbox', 'Container'],
241 ['Test.'],
242 ['', 'Basic', 'Empty', 'More'],
243 ['Mouse', 'App', 'Selection', 'Network', 'Grab'],
244 ['Input', 'Click', 'Failure', 'Capture']]
245
246 def test():
247 return string.join(map(random.choice, tests), '')
248
249 def literal_array():
250 generator = random.choice([time, build, revision])
251 return [generator() for _ in range(random.randint(0, 10))]
252
253 def literal_map():
254 generators = [build, revision, test, literal_array]
255 obj = {}
256 for _ in range(random.randint(3, 9)):
257 obj[label()] = random.choice(generators)()
258 return obj
259
260 def value():
261 generators = [time, build, revision, test, literal_array,
262 literal_map]
263 return random.choice(generators)()
264
265 alert = {}
266 for _ in range(random.randint(6, 9)):
267 alert[label()] = value()
268 return alert
OLDNEW
« no previous file with comments | « Tools/GardeningServer/alerts.py ('k') | Tools/GardeningServer/app.yaml » ('j') | no next file with comments »

Powered by Google App Engine
This is Rietveld 408576698