lib/cros_logging_unittest.py

# -*- coding: utf-8 -*-
# Copyright 2015 The Chromium OS Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.

"""Tests for cros_logging."""

from __future__ import print_function

import os
import sys

from chromite.lib import cros_logging as logging
from chromite.lib import cros_test_lib


class CrosCloudloggingTest(cros_test_lib.MockOutputTestCase):
  """Test google-cloud-logging interacts with logging as expected."""

  # pylint: disable=protected-access

  def setUp(self):
    self.logger = logging.getLogger()
    sh = logging.StreamHandler(sys.stdout)
    self.logger.addHandler(sh)
    # pylint: disable=protected-access
    logging._buildbot_markers_enabled = False
    try:
      import google.cloud.logging  # pylint: disable=import-error,no-name-in-module
    except ImportError:
      self.cloud_logging_import_error = True
      return
    self.cloud_logging_import_error = False
    self.client_mock = self.PatchObject(google.cloud.logging, 'Client')

  def testSetupCloudLogging(self):
    if self.cloud_logging_import_error:
      return
    # Invoke the code that calls logging when env vars are set.
    logging._SetupCloudLogging()

    # Verify that google.cloud.logging.Client() was executed.
    self.client_mock.assert_called_once()

  def testCloudLoggingEnvVariablesAreDefined_notSet(self):
    if self.cloud_logging_import_error:
      return
    with self.OutputCapturer() as output:
      cloud_env_defined = logging._CloudLoggingEnvVariablesAreDefined()
    # If both are not set, there should be no output.
    self.assertEqual(output.GetStdout(), '')
    self.assertEqual(output.GetStderr(), '')
    self.assertFalse(cloud_env_defined)

  def testCloudLoggingEnvVariablesAreDefined_envSet(self):
    if self.cloud_logging_import_error:
      return
    # Set logger to INFO level to check for CHROMITE_CLOUD_LOGGING info message.
    self.logger.setLevel(logging.INFO)
    # Set the env vars
    os.environ['CHROMITE_CLOUD_LOGGING'] = '1'
    os.environ['GOOGLE_APPLICATION_CREDENTIALS'] = '/some/path/to/creds.json'
    with self.OutputCapturer() as output:
      cloud_env_defined = logging._CloudLoggingEnvVariablesAreDefined()
    # Verify that both variables are logged.
    self.assertTrue(cloud_env_defined)
    self.assertIn('CHROMITE_CLOUD_LOGGING', output.GetStdout())

  def testCloudLoggingEnvVariablesAreDefined_noAllEnvSet(self):
    if self.cloud_logging_import_error:
      return
    # Set only one env var.
    os.environ['CHROMITE_CLOUD_LOGGING'] = '1'
    with self.OutputCapturer() as output:
      cloud_env_defined = logging._CloudLoggingEnvVariablesAreDefined()
    # If both are not set, there should be no output.
    self.assertIn('GOOGLE_APPLICATION_CREDENTIALS is not', output.GetStdout())
    self.assertEqual(output.GetStderr(), '')
    self.assertFalse(cloud_env_defined)


class CrosloggingTest(cros_test_lib.OutputTestCase):
  """Test logging works as expected."""

  def setUp(self):
    self.logger = logging.getLogger()
    sh = logging.StreamHandler(sys.stdout)
    self.logger.addHandler(sh)
    # pylint: disable=protected-access
    logging._buildbot_markers_enabled = False

  def AssertLogContainsMsg(self, msg, functor, *args, **kwargs):
    """Asserts that calling functor logs a line that contains msg.

    Args:
      msg: The message to look for.
      functor: A function taking no arguments to test.
      *args, **kwargs: passthrough arguments to AssertLogContainsMsg.
    """
    with self.OutputCapturer():
      functor()
    self.AssertOutputContainsLine(msg, *args, **kwargs)

  def testNotice(self):
    """Test logging.notice works and is between INFO and WARNING."""
    msg = 'notice message'
    self.logger.setLevel(logging.INFO)
    self.AssertLogContainsMsg(msg, lambda: logging.notice(msg))
    self.logger.setLevel(logging.WARNING)
    self.AssertLogContainsMsg(msg, lambda: logging.notice(msg), invert=True)

  def testPrintBuildbotFunctionsNoMarker(self):
    """PrintBuildbot* without markers should not be recognized by buildbot."""
    self.AssertLogContainsMsg('@@@STEP_LINK@',
                              lambda: logging.PrintBuildbotLink('name', 'url'),
                              check_stderr=True, invert=True)
    self.AssertLogContainsMsg('@@@@STEP_TEXT@',
                              lambda: logging.PrintBuildbotStepText('text'),
                              check_stderr=True, invert=True)
    self.AssertLogContainsMsg('@@@STEP_WARNINGS@@@',
                              logging.PrintBuildbotStepWarnings,
                              check_stderr=True, invert=True)
    self.AssertLogContainsMsg('@@@STEP_FAILURE@@@',
                              logging.PrintBuildbotStepFailure,
                              check_stderr=True, invert=True)
    self.AssertLogContainsMsg('@@@BUILD_STEP',
                              lambda: logging.PrintBuildbotStepName('name'),
                              check_stderr=True, invert=True)
    self.AssertLogContainsMsg(
        '@@@SET_BUILD_PROPERTY',
        lambda: logging.PrintKitchenSetBuildProperty('name', {'a': 'value'}),
        check_stderr=True, invert=True)

  def testPrintBuildbotFunctionsWithMarker(self):
    """PrintBuildbot* with markers should be recognized by buildbot."""
    logging.EnableBuildbotMarkers()
    self.AssertLogContainsMsg('@@@STEP_LINK@name@url@@@',
                              lambda: logging.PrintBuildbotLink('name', 'url'),
                              check_stderr=True)
    self.AssertLogContainsMsg('@@@STEP_TEXT@text@@@',
                              lambda: logging.PrintBuildbotStepText('text'),
                              check_stderr=True)
    self.AssertLogContainsMsg('@@@STEP_WARNINGS@@@',
                              logging.PrintBuildbotStepWarnings,
                              check_stderr=True)
    self.AssertLogContainsMsg('@@@STEP_FAILURE@@@',
                              logging.PrintBuildbotStepFailure,
                              check_stderr=True)
    self.AssertLogContainsMsg('@@@BUILD_STEP@name@@@',
                              lambda: logging.PrintBuildbotStepName('name'),
                              check_stderr=True)
    self.AssertLogContainsMsg(
        '@@@SET_BUILD_PROPERTY@name@"value"@@@',
        lambda: logging.PrintKitchenSetBuildProperty('name', 'value'),
        check_stderr=True)