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 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 AlertsTest(unittest.TestCase):
17    def setUp(self):
18        self.testbed = testbed.Testbed()
19        self.testbed.activate()
20        self.testbed.init_memcache_stub()
21        self.testapp = webtest.TestApp(alerts.app)
22
23    def tearDown(self):
24        self.testbed.deactivate()
25
26    def check_json_headers(self, res):
27        self.assertEqual(res.content_type, 'application/json')
28        # This is necessary for cross-site tools to retrieve alerts
29        self.assertEqual(res.headers['access-control-allow-origin'], '*')
30
31    def test_get_no_data_cached(self):
32        res = self.testapp.get('/alerts')
33        self.check_json_headers(res)
34        self.assertEqual(res.body, '')
35
36    def test_happy_path(self):
37        # Set it.
38        params = {'content': '{"alerts": ["hello", "world"]}'}
39        self.testapp.post('/alerts', params)
40
41        # Get it.
42        res = self.testapp.get('/alerts')
43        self.check_json_headers(res)
44        alerts = json.loads(res.body)
45
46        # The server should have stuck a 'date' on there.
47        self.assertTrue('date' in alerts)
48        self.assertEqual(type(alerts['date']), int)
49
50        self.assertEqual(alerts['alerts'], ['hello', 'world'])
51
52    def test_post_invalid_data_not_reflected(self):
53        params = {'content': '[{"this is not valid JSON'}
54        self.testapp.post('/alerts', params, status=400)
55        res = self.testapp.get('/alerts')
56        self.assertEqual(res.body, '')
57
58    def test_post_invalid_data_does_not_overwrite_valid_data(self):
59        # Populate the cache with something valid
60        params = {'content': '{"alerts": "everything is OK"}'}
61        self.testapp.post('/alerts', params)
62        self.testapp.post('/alerts', {'content': 'woozlwuzl'}, status=400)
63        res = self.testapp.get('/alerts')
64        self.check_json_headers(res)
65        alerts = json.loads(res.body)
66        self.assertEqual(alerts['alerts'], 'everything is OK')
67
68    def test_large_number_of_alerts(self):
69        # This generates ~2.5MB of JSON that compresses to ~750K. Real
70        # data compresses about 6x better.
71        random.seed(0xf00f00)
72        put_alerts = self.generate_fake_alerts(4000)
73
74        params = {'content': json.dumps(put_alerts)}
75        self.testapp.post('/alerts', params)
76
77        res = self.testapp.get('/alerts')
78        got_alerts = json.loads(res.body)
79        self.assertEquals(got_alerts['alerts'], put_alerts['alerts'])
80
81    def generate_fake_alerts(self, n):
82        return {'alerts': [self.generate_fake_alert() for _ in range(n)]}
83
84    def generate_fake_alert(self):
85        # fake labels
86        labels = [['', 'last_', 'latest_', 'failing_', 'passing_'],
87                  ['build', 'builder', 'revision'],
88                  ['', 's', '_url', '_reason', '_name']]
89
90        def label():
91            return string.join(map(random.choice, labels), '')
92
93        # fake values
94        def time():
95            return random.randint(1407976107614, 1408076107614) / 101.0
96
97        def build():
98            return random.randint(2737, 2894)
99
100        def revision():
101            return random.randint(288849, 289415)
102
103        tests = [['Activity', 'Async', 'Browser', 'Content', 'Input'],
104                 ['Manager', 'Card', 'Sandbox', 'Container'],
105                 ['Test.'],
106                 ['', 'Basic', 'Empty', 'More'],
107                 ['Mouse', 'App', 'Selection', 'Network', 'Grab'],
108                 ['Input', 'Click', 'Failure', 'Capture']]
109
110        def test():
111            return string.join(map(random.choice, tests), '')
112
113        def literal_array():
114            generator = random.choice([time, build, revision])
115            return [generator() for _ in range(random.randint(0, 10))]
116
117        def literal_map():
118            generators = [build, revision, test, literal_array]
119            obj = {}
120            for _ in range(random.randint(3, 9)):
121                obj[label()] = random.choice(generators)()
122            return obj
123
124        def value():
125            generators = [time, build, revision, test, literal_array,
126                          literal_map]
127            return random.choice(generators)()
128
129        alert = {}
130        for _ in range(random.randint(6, 9)):
131            alert[label()] = value()
132        return alert
133