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