Implemented retrieval of alerts history

Tools/GardeningServer/alerts_test.py

 # Copyright 2014 The Chromium Authors. All rights reserved.
 # Use of this source code is governed by a BSD-style license that can be
 # found in the LICENSE file.
 
 import alerts
 import json
 import random
 import string
 import unittest
 import webtest
 
 from google.appengine.api import memcache
 from google.appengine.ext import ndb
 from google.appengine.ext import testbed
 
 
 class AlertsTest(unittest.TestCase):
     def setUp(self):
         self.testbed = testbed.Testbed()
         self.testbed.activate()
+        self.testbed.init_user_stub()
         self.testbed.init_memcache_stub()
         self.testbed.init_datastore_v3_stub()
         self.testapp = webtest.TestApp(alerts.app)
 
     def tearDown(self):
         self.testbed.deactivate()
 
     def check_json_headers(self, res):
         self.assertEqual(res.content_type, 'application/json')
         # This is necessary for cross-site tools to retrieve alerts
@@ skipping 37 matching lines @@
         self.assertEqual(alerts['alerts'], 'everything is OK')
 
     def test_alerts_jsons_are_stored_in_history(self):
         test_alerts1 = {'alerts': ['hello', 'world', '1']}
         test_alerts2 = {'alerts': ['hello', 'world', '2']}
         self.testapp.post('/alerts', {'content': json.dumps(test_alerts1)})
         self.testapp.post('/alerts', {'content': json.dumps(test_alerts2)})
         alerts_query = alerts.AlertsJSON.query().order(alerts.AlertsJSON.date)
         stored_alerts = alerts_query.fetch(limit=3)
         self.assertEqual(2, len(stored_alerts))
+        self.assertEqual(stored_alerts[0].key, 'alerts')
+        self.assertEqual(stored_alerts[1].key, 'alerts')
         stored_alerts1 = json.loads(stored_alerts[0].json)
         stored_alerts2 = json.loads(stored_alerts[1].json)
         self.assertEqual(test_alerts1['alerts'], stored_alerts1['alerts'])
         self.assertEqual(test_alerts2['alerts'], stored_alerts2['alerts'])
         self.assertTrue('date' in stored_alerts1)
         self.assertTrue('date' in stored_alerts2)
         self.assertEqual(type(stored_alerts1['date']), int)
         self.assertEqual(type(stored_alerts2['date']), int)
 
     def test_repeating_alerts_are_not_stored_to_history(self):
         test_alerts = {'alerts': ['hello', 'world']}
         self.testapp.post('/alerts', {'content': json.dumps(test_alerts)})
         test_alerts['last_builder_info'] = {'some': 'info'}
         self.testapp.post('/alerts', {'content': json.dumps(test_alerts)})
         stored_alerts = alerts.AlertsJSON.query().fetch(limit=2)
         self.assertEqual(1, len(stored_alerts))
 
+    def test_alerts_jsons_are_retrieved_from_history(self):
+        test_alert = {'alerts': ['hello', 'world', '1']}
+        alerts.AlertsJSON(json=json.dumps(test_alert), key='alerts').put()
+        response = self.testapp.get('/alerts-history')
+        self.assertEqual(response.status_int, 200)
+        self.assertEqual(response.content_type, 'application/json')
+        response_json = json.loads(response.normal_body)
+        self.assertEqual(len(response_json['history']), 1)
+        self.assertEqual(response_json['history'][0], test_alert)
+
+    def test_internal_alerts_in_history_visible_to_internal_users_only(self):
+        test_alert = {'alerts': ['hello', 'world', '1']}
+        alerts.AlertsJSON(json=json.dumps(test_alert),
+                          key='internal-alerts').put()
+
+        # No signed-in user.
+        response = self.testapp.get('/alerts-history')
+        self.assertEqual(response.status_int, 200)
+        self.assertEqual(response.content_type, 'application/json')
+        response_json = json.loads(response.normal_body)
+        self.assertEqual(len(response_json['history']), 0)
+
+        # Non-internal user.
+        self.testbed.setup_env(USER_EMAIL='test@example.com', USER_ID='1',
+                               USER_IS_ADMIN='1', overwrite=True)
+        response = self.testapp.get('/alerts-history')
+        self.assertEqual(response.status_int, 200)
+        self.assertEqual(response.content_type, 'application/json')
+        response_json = json.loads(response.normal_body)
+        self.assertEqual(len(response_json['history']), 0)
+
+        # Internal user.
+        self.testbed.setup_env(USER_EMAIL='test@google.com', USER_ID='2',
+                               USER_IS_ADMIN='1', overwrite=True)
+        response = self.testapp.get('/alerts-history')
+        self.assertEqual(response.status_int, 200)
+        self.assertEqual(response.content_type, 'application/json')
+        response_json = json.loads(response.normal_body)
+        self.assertEqual(len(response_json['history']), 1)
+        self.assertEqual(response_json['history'][0], test_alert)
+
+    def test_returned_alerts_from_history_are_paged(self):
+        for i in range(20):
+            test_alert = {'alerts': ['hello', 'world', i]}
+            alerts.AlertsJSON(json=json.dumps(test_alert), key='alerts').put()
+
+        response = self.testapp.get('/alerts-history?limit=15')
+        self.assertEqual(response.status_int, 200)
+        self.assertEqual(response.content_type, 'application/json')
+        response_json = json.loads(response.normal_body)
+        self.assertEqual(len(response_json['history']), 15)
+        self.assertEqual(response_json['has_more'], True)
+
+        url = '/alerts-history?limit=15&cursor=%s' % response_json['cursor']
+        response = self.testapp.get(url)
+        self.assertEqual(response.status_int, 200)
+        self.assertEqual(response.content_type, 'application/json')
+        response_json = json.loads(response.normal_body)
+        self.assertEqual(len(response_json['history']), 5)
+        self.assertEqual(response_json['has_more'], False)
+
     def test_large_number_of_alerts(self):
         # This generates ~2.5MB of JSON that compresses to ~750K. Real
         # data compresses about 6x better.
         random.seed(0xf00f00)
         put_alerts = self.generate_fake_alerts(4000)
 
         params = {'content': json.dumps(put_alerts)}
         self.testapp.post('/alerts', params)
 
         res = self.testapp.get('/alerts')
@@ skipping 45 matching lines @@
 
         def value():
             generators = [time, build, revision, test, literal_array,
                           literal_map]
             return random.choice(generators)()
 
         alert = {}
         for _ in range(random.randint(6, 9)):
             alert[label()] = value()
         return alert