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