Monitoring proxy for time series data

appengine/chrome_infra_mon_proxy/test/admin_handler_test.py

+# Copyright 2015 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 collections
+import logging
+import webtest
+
+from testing_utils import testing
+from google.appengine.api import users
+
+import admin_handler
+import common
+import main
+
+class AdminTest(testing.AppengineTestCase):
+  @property
+  def app_module(self):
+    return main.app
+
+  def test_admin_page(self):
+    # Unauthorized GET request: redirect to login page.
+    response = self.test_app.get('/admin/')
+    logging.info('response = %s', response)
+    self.assertEquals(302, response.status_int)
+    # Unauthorized POST request: 403.
+    with self.assertRaises(webtest.AppError) as cm:
+      self.test_app.post('/admin/')
+    logging.info('exception = %s', cm.exception)
+    self.assertIn('403', str(cm.exception))
+
+    self.mock(users, 'is_current_user_admin', lambda: True)
+
+    # Authorized GET request.
+    response = self.test_app.get('/admin/')
+    logging.info('response = %s', response)
+    self.assertEquals(200, response.status_int)
+    # Authorized POST request: 403 (POST not allowed on /admin/).
+    with self.assertRaises(webtest.AppError) as cm:
+      self.test_app.post('/admin/')
+    logging.info('exception = %s', cm.exception)
+    self.assertIn('403', str(cm.exception))
+
+  def test_set_credentials(self):
+    class MonAcqDataMock(object):
+      def __init__(self, data):
+        self.data = data
+
+      def get_by_id(self, _id):
+        return self.data
+
+      def get_or_insert(self, _id):
+        return self.data
+
+    class DataMock(object):
+      def __init__(self, credentials=None, url='http://',
+                   scopes=None, headers=None):
+        self.credentials = credentials or {}
+        self.url = url
+        self.scopes = scopes or ['a', 'b']
+        self.headers = headers or {}
+        self.updated = False
+
+      def to_dict(self):
+        return {
+            'credentials': self.credentials,
+            'url': self.url,
+            'scopes': self.scopes,
+            'headers': self.headers,
+        }
+
+      def put(self):
+        self.updated = True
+        logging.debug('Saving NDB data: %s', self.to_dict())
+
+    # Unauthorized GET request: redirect to login page.
+    response = self.test_app.get('/admin/set-credentials')
+    logging.info('response = %s', response)
+    self.assertEquals(302, response.status_int)
+
+    # Unauthorized POST request: 403.
+    with self.assertRaises(webtest.AppError) as cm:
+      self.test_app.post('/admin/set-credentials')
+    logging.info('exception = %s', cm.exception)
+    self.assertIn('403', str(cm.exception))
+
+    # Authorized GET request, no data in NDB.
+    self.mock(users, 'is_current_user_admin', lambda: True)
+    self.mock(common, 'MonAcqData', MonAcqDataMock(None))
+    response = self.test_app.get('/admin/set-credentials')
+    self.assertEquals(200, response.status_int)
+
+    # Authorized GET request, data exists in NDB.
+    self.mock(common, 'MonAcqData', MonAcqDataMock(DataMock()))
+    response = self.test_app.get('/admin/set-credentials')
+    self.assertEquals(200, response.status_int)
+
+    # POST request with bad data (for coverage of serialize()).
+    self.mock(common, 'MonAcqData', MonAcqDataMock(DataMock(headers=set('a'))))
+    response = self.test_app.post('/admin/set-credentials')
+    self.assertEquals(200, response.status_int)
+
+    # Valid POST request.
+    data = DataMock()
+    self.mock(common, 'MonAcqData', MonAcqDataMock(data))
+    params = collections.OrderedDict([
+        ('url', 'https://new.url'),
+        ('credentials', '{"client_id": "john@doe"}'),
+        ('scopes', 'foo bar'),
+    ])
+    response = self.test_app.post('/admin/set-credentials', params)
+    self.assertEquals(200, response.status_int)
+    self.assertTrue(data.updated)
+
+    # Invalid POST request.
+    data = DataMock()
+    self.mock(common, 'MonAcqData', MonAcqDataMock(data))
+    params = collections.OrderedDict([
+        ('credentials', '{"client_id": '), # Bad JSON.
+    ])
+    response = self.test_app.post('/admin/set-credentials', params)
+    self.assertEquals(200, response.status_int)
+    self.assertFalse(data.updated)
+
+  def test_deserialize(self):
+    # Mostly covered by other tests, except for the unsupported type case.
+    class UnsupportedType(object):
+      pass
+
+    with self.assertRaises(ValueError):
+      admin_handler.deserialize('blah', UnsupportedType)