Monitoring proxy for time series data

appengine/chrome_infra_mon_proxy/test/main_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 logging
+import os
+import webtest
+
+from google.appengine.api import taskqueue, users
+from testing_utils import testing
+
+import main
+from components import auth
+
+
+class IsGroupMemberTest(testing.AppengineTestCase):
+  def test_is_group_member(self):
+    # This is a smoke test for coverage. The function is otherwise trivial.
+    self.mock(os, 'environ', {'SERVER_SOFTWARE': 'Development server'})
+    main.is_group_member('foo')
+    self.mock(os, 'environ', {'SERVER_SOFTWARE': 'GAE production server'})
+    main.is_group_member('foo')
+
+
+class LoadBalancerTest(testing.AppengineTestCase):
+  def test_choose_module(self):
+    # Smoke test for coverage.
+    self.mock(main.LoadBalancer, '_healthy_modules', lambda _: [])
+    with self.assertRaises(main.NoBackendException):
+      lb = main.LoadBalancer()
+      lb.choose_module()
+
+
+class MonacqHandlerTest(testing.AppengineTestCase):
+  @property
+  def app_module(self):
+    return main.app
+
+  def setUp(self):
+    super(MonacqHandlerTest, self).setUp()
+    # Disable auth module checks.
+    self.mock(users, 'get_current_user',
+              lambda: users.User('test@user.com', 'auth_domain'))
+    self.mock(main.MonacqHandler, 'xsrf_token_enforce_on', [])
+    self.mock(auth, 'is_group_member', lambda _: True) # pragma: no branch
+
+  def tearDown(self):
+    super(MonacqHandlerTest, self).tearDown()
+
+  def test_get(self):
+    # GET request is not allowed.
+    with self.assertRaises(webtest.AppError) as cm:
+      self.test_app.get('/monacq')
+    logging.info('exception = %s', cm.exception)
+    self.assertIn('403', str(cm.exception))
+
+  def test_post(self):
+    self.mock(taskqueue, 'add', lambda *_args, **_kw: None)
+    # Dev appserver.
+    self.mock(os, 'environ', {'SERVER_SOFTWARE': 'Development server'})
+    self.test_app.post('/monacq', 'deadbeafdata')
+    # Production server.
+    self.mock(os, 'environ', {'SERVER_SOFTWARE': 'GAE production server'})
+    self.test_app.post('/monacq', 'deadbeafdata')
+
+
+class MainHandlerTest(testing.AppengineTestCase):
+  @property
+  def app_module(self):
+    return main.app
+
+  def test_get(self):
+    response = self.test_app.get('/')
+    logging.info('response = %s', response)
+    self.assertEquals(200, response.status_int)