| OLD | NEW |
|---|
| (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 internal_alerts | |
| 7 import json | |
| 8 import random | |
| 9 import string | |
| 10 import unittest | |
| 11 import webtest | |
| 12 | |
| 13 from google.appengine.api import memcache | |
| 14 from google.appengine.ext import testbed | |
| 15 | |
| 16 | |
| 17 class InternalAlertsTest(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(internal_alerts.app) | |
| 25 | |
| 26 def tearDown(self): | |
| 27 self.testbed.deactivate() | |
| 28 | |
| 29 def user_helper(self, email, uid): | |
| 30 self.testbed.setup_env( | |
| 31 USER_EMAIL=email, | |
| 32 USER_ID=uid, | |
| 33 USER_IS_ADMIN='0', | |
| 34 overwrite=True) | |
| 35 | |
| 36 def check_json_headers(self, res): | |
| 37 self.user_helper('tester@google.com', '123') | |
| 38 self.assertEqual(res.content_type, 'application/json') | |
| 39 # This is necessary for cross-site tools to retrieve internal alerts. | |
| 40 self.assertEqual(res.headers['access-control-allow-origin'], '*') | |
| 41 | |
| 42 def test_get_no_data_cached(self): | |
| 43 self.user_helper('tester@google.com', '123') | |
| 44 res = self.testapp.get('/internal-alerts') | |
| 45 self.check_json_headers(res) | |
| 46 self.assertEqual(res.body, '') | |
| 47 | |
| 48 def test_happy_path(self): | |
| 49 self.user_helper('tester@google.com', '123') | |
| 50 # Set it. | |
| 51 params = {'content': '{"alerts": ["hello", "world"]}'} | |
| 52 self.testapp.post('/internal-alerts', params) | |
| 53 | |
| 54 # Get it. | |
| 55 res = self.testapp.get('/internal-alerts') | |
| 56 self.check_json_headers(res) | |
| 57 internal_alerts = json.loads(res.body) | |
| 58 | |
| 59 # The server should have stuck a 'date' on there. | |
| 60 self.assertTrue('date' in internal_alerts) | |
| 61 self.assertEqual(type(internal_alerts['date']), int) | |
| 62 | |
| 63 self.assertEqual(internal_alerts['alerts'], ['hello', 'world']) | |
| 64 | |
| 65 def test_post_invalid_data_not_reflected(self): | |
| 66 self.user_helper('tester@google.com', '123') | |
| 67 params = {'content': '[{"this is not valid JSON'} | |
| 68 self.testapp.post('/internal-alerts', params, status=400) | |
| 69 res = self.testapp.get('/internal-alerts') | |
| 70 self.assertEqual(res.body, '') | |
| 71 | |
| 72 def test_post_invalid_data_does_not_overwrite_valid_data(self): | |
| 73 self.user_helper('tester@google.com', '123') | |
| 74 # Populate the cache with something valid | |
| 75 params = {'content': '{"alerts": "everything is OK"}'} | |
| 76 self.testapp.post('/internal-alerts', params) | |
| 77 self.testapp.post('/internal-alerts', {'content': 'woozlwuzl'}, | |
| 78 status=400) | |
| 79 res = self.testapp.get('/internal-alerts') | |
| 80 self.check_json_headers(res) | |
| 81 internal_alerts = json.loads(res.body) | |
| 82 self.assertEqual(internal_alerts['alerts'], 'everything is OK') | |
| 83 | |
| 84 def test_internal_alerts_stored_in_history_have_correct_type(self): | |
| 85 test_alerts1 = {'alerts': ['hello', 'world', '1']} | |
| 86 test_alerts2 = {'alerts': ['hello', 'world', '2']} | |
| 87 self.testapp.post('/internal-alerts', | |
| 88 {'content': json.dumps(test_alerts1)}) | |
| 89 self.testapp.post('/internal-alerts', | |
| 90 {'content': json.dumps(test_alerts2)}) | |
| 91 alerts_query = alerts.AlertsJSON.query().order(alerts.AlertsJSON.date) | |
| 92 stored_alerts = alerts_query.fetch(limit=3) | |
| 93 self.assertEqual(2, len(stored_alerts)) | |
| 94 self.assertEqual(stored_alerts[0].type, 'internal-alerts') | |
| 95 self.assertEqual(stored_alerts[1].type, 'internal-alerts') | |
| 96 | |
| 97 def test_internal_alerts_same_as_last_alerts_are_added_to_history(self): | |
| 98 test_alerts = {'alerts': ['hello', 'world']} | |
| 99 alerts.AlertsJSON(json=json.dumps(test_alerts), type='alerts').put() | |
| 100 self.testapp.post('/internal-alerts', | |
| 101 {'content': json.dumps(test_alerts)}) | |
| 102 alerts_query = alerts.AlertsJSON.query() | |
| 103 self.assertEqual(2, alerts_query.count(limit=3)) | |
| 104 | |
| 105 def test_large_number_of_internal_alerts(self): | |
| 106 self.user_helper('tester@google.com', '123') | |
| 107 # This generates ~2.5MB of JSON that compresses to ~750K. Real | |
| 108 # data compresses about 6x better. | |
| 109 random.seed(0xf00f00) | |
| 110 put_internal_alerts = self.generate_fake_internal_alerts(4000) | |
| 111 | |
| 112 params = {'content': json.dumps(put_internal_alerts)} | |
| 113 self.testapp.post('/internal-alerts', params) | |
| 114 | |
| 115 res = self.testapp.get('/internal-alerts') | |
| 116 got_internal_alerts = json.loads(res.body) | |
| 117 self.assertEquals(got_internal_alerts['alerts'], | |
| 118 put_internal_alerts['alerts']) | |
| 119 | |
| 120 def test_no_user(self): | |
| 121 # Get it. | |
| 122 res = self.testapp.get('/internal-alerts') | |
| 123 self.check_json_headers(res) | |
| 124 internal_alerts = json.loads(res.body) | |
| 125 | |
| 126 # The server should have stuck a 'date' on there. | |
| 127 self.assertTrue('date' in internal_alerts) | |
| 128 self.assertEqual(type(internal_alerts['date']), int) | |
| 129 | |
| 130 self.assertTrue('redirect-url' in internal_alerts) | |
| 131 self.assertEqual(type(internal_alerts['redirect-url']), unicode) | |
| 132 | |
| 133 def test_invalid_user(self): | |
| 134 self.user_helper('tester@chromium.org', '123') | |
| 135 # Get it. | |
| 136 res = self.testapp.get('/internal-alerts', status=403) | |
| 137 | |
| 138 def generate_fake_internal_alerts(self, n): | |
| 139 self.user_helper('tester@google.com', '123') | |
| 140 return {'alerts': [self.generate_fake_alert() for _ in range(n)]} | |
| 141 | |
| 142 def generate_fake_alert(self): | |
| 143 # fake labels | |
| 144 labels = [['', 'last_', 'latest_', 'failing_', 'passing_'], | |
| 145 ['build', 'builder', 'revision'], | |
| 146 ['', 's', '_url', '_reason', '_name']] | |
| 147 | |
| 148 def label(): | |
| 149 return string.join(map(random.choice, labels), '') | |
| 150 | |
| 151 # fake values | |
| 152 def time(): | |
| 153 return random.randint(1407976107614, 1408076107614) / 101.0 | |
| 154 | |
| 155 def build(): | |
| 156 return random.randint(2737, 2894) | |
| 157 | |
| 158 def revision(): | |
| 159 return random.randint(288849, 289415) | |
| 160 | |
| 161 tests = [['Activity', 'Async', 'Browser', 'Content', 'Input'], | |
| 162 ['Manager', 'Card', 'Sandbox', 'Container'], | |
| 163 ['Test.'], | |
| 164 ['', 'Basic', 'Empty', 'More'], | |
| 165 ['Mouse', 'App', 'Selection', 'Network', 'Grab'], | |
| 166 ['Input', 'Click', 'Failure', 'Capture']] | |
| 167 | |
| 168 def test(): | |
| 169 return string.join(map(random.choice, tests), '') | |
| 170 | |
| 171 def literal_array(): | |
| 172 generator = random.choice([time, build, revision]) | |
| 173 return [generator() for _ in range(random.randint(0, 10))] | |
| 174 | |
| 175 def literal_map(): | |
| 176 generators = [build, revision, test, literal_array] | |
| 177 obj = {} | |
| 178 for _ in range(random.randint(3, 9)): | |
| 179 obj[label()] = random.choice(generators)() | |
| 180 return obj | |
| 181 | |
| 182 def value(): | |
| 183 generators = [time, build, revision, test, literal_array, | |
| 184 literal_map] | |
| 185 return random.choice(generators)() | |
| 186 | |
| 187 alert = {} | |
| 188 for _ in range(random.randint(6, 9)): | |
| 189 alert[label()] = value() | |
| 190 return alert | |
| OLD | NEW |
|---|
Chromium Code Reviews
