| 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 |