Index: infra_libs/event_mon/test/config_test.py |
diff --git a/infra_libs/event_mon/test/config_test.py b/infra_libs/event_mon/test/config_test.py |
deleted file mode 100644 |
index 59b87be37e0df4f64c4a7f3309a4b24aa43058cc..0000000000000000000000000000000000000000 |
--- a/infra_libs/event_mon/test/config_test.py |
+++ /dev/null |
@@ -1,144 +0,0 @@ |
-# 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 argparse |
-import os |
-import traceback |
-import unittest |
- |
-import mock |
- |
-import infra_libs |
-from infra_libs import event_mon |
-from infra_libs.event_mon import config, router |
-from infra_libs.event_mon.protos import chrome_infra_log_pb2 |
- |
- |
-class ConfigTest(unittest.TestCase): |
- def tearDown(self): |
- self._close() |
- |
- def _set_up_args(self, args=None): # pragma: no cover |
- parser = argparse.ArgumentParser() |
- event_mon.add_argparse_options(parser) |
- args = parser.parse_args((args or [])) |
- # Safety net. We really don't want to send something to a real endpoint. |
- self.assertTrue((args.event_mon_run_type not in ('test', 'prod')) |
- or args.dry_run) |
- event_mon.process_argparse_options(args) |
- r = config._router |
- self.assertIsInstance(r, router._Router) |
- # Check that process_argparse_options is idempotent |
- event_mon.process_argparse_options(args) |
- self.assertIs(config._router, r) |
- |
- def _close(self): |
- self.assertTrue(event_mon.close()) |
- self.assertFalse(config._cache) |
- # Test that calling it twice does not raise an exception. |
- self.assertTrue(event_mon.close()) |
- |
- def test_no_args_smoke(self): # pragma: no cover |
- self._set_up_args() |
- |
- def test_args_and_default_event(self): # pragma: no cover |
- # The protobuf structure is actually an API not an implementation detail |
- # so it's sane to test for changes. |
- hostname = 'a' |
- service_name = 'b' |
- appengine_name = 'c' |
- |
- args = ['--event-mon-run-type', 'dry', |
- '--event-mon-hostname', hostname, |
- '--event-mon-service-name', service_name, |
- '--event-mon-appengine-name', appengine_name] |
- self._set_up_args(args=args) |
- event = event_mon.get_default_event() |
- self.assertEquals(event.event_source.host_name, hostname) |
- self.assertEquals(event.event_source.service_name, service_name) |
- self.assertEquals(event.event_source.appengine_name, appengine_name) |
- |
- def test_run_type_file_good(self): |
- try: |
- with infra_libs.temporary_directory(prefix='config_test-') as tempdir: |
- filename = os.path.join(tempdir, 'output.db') |
- self._set_up_args(args=['--event-mon-run-type', 'file', |
- '--event-mon-output-file', filename]) |
- self.assertEqual(config._router.output_file, filename) |
- except Exception: # pragma: no cover |
- # help for debugging |
- traceback.print_exc() |
- raise |
- |
- # Direct setup_monitoring testing below this line. |
- def test_default_event(self): |
- # The protobuf structure is actually an API not an implementation detail |
- # so it's sane to test for changes. |
- event_mon.setup_monitoring() |
- event = event_mon.get_default_event() |
- self.assertTrue(event.event_source.HasField('host_name')) |
- self.assertFalse(event.event_source.HasField('service_name')) |
- self.assertFalse(event.event_source.HasField('appengine_name')) |
- |
- @mock.patch('os.environ') |
- def test_logging_on_appengine(self, environ): |
- environ.get.return_value = 'Google App Engine/1.2.3' |
- event_mon.setup_monitoring() |
- self.assertIsInstance(config._router, router._LoggingStreamRouter) |
- |
- environ.get.return_value = 'Development/1.2' |
- event_mon.setup_monitoring() |
- self.assertIsInstance(config._router, router._LoggingStreamRouter) |
- |
- def test_default_event_with_values(self): |
- # The protobuf structure is actually an API not an implementation detail |
- # so it's sane to test for changes. |
- hostname = 'a' |
- service_name = 'b' |
- appengine_name = 'c' |
- |
- event_mon.setup_monitoring( |
- hostname=hostname, |
- service_name=service_name, |
- appengine_name=appengine_name |
- ) |
- event = event_mon.get_default_event() |
- self.assertEquals(event.event_source.host_name, hostname) |
- self.assertEquals(event.event_source.service_name, service_name) |
- self.assertEquals(event.event_source.appengine_name, appengine_name) |
- |
- def test_set_default_event(self): |
- event_mon.setup_monitoring() |
- orig_event = event_mon.get_default_event() |
- |
- # Set the new default event to something different from orig_event |
- # to make sure it has changed. |
- event = chrome_infra_log_pb2.ChromeInfraEvent() |
- new_hostname = orig_event.event_source.host_name + '.foo' |
- event.event_source.host_name = new_hostname |
- event_mon.set_default_event(event) |
- |
- new_event = event_mon.get_default_event() |
- self.assertEquals(new_event.event_source.host_name, new_hostname) |
- |
- def test_set_default_event_bad_type(self): |
- event_mon.setup_monitoring() |
- |
- # bad type |
- with self.assertRaises(TypeError): |
- event_mon.set_default_event({'hostname': 'foo'}) |
- |
- def test_get_default_event(self): |
- event_mon.setup_monitoring() |
- orig_event = config._cache['default_event'] |
- self.assertEqual(orig_event, event_mon.get_default_event()) |
- self.assertIsNot(orig_event, event_mon.get_default_event()) |
- |
- def test_run_type_test(self): |
- event_mon.setup_monitoring(run_type='test', service_account_creds='creds') |
- self.assertEquals(config._cache['service_account_creds'], 'creds') |
- |
- def test_run_type_invalid(self): |
- event_mon.setup_monitoring(run_type='invalid.') |
- self.assertFalse(isinstance(config._router, router._HttpRouter)) |