# -*- coding: utf-8 -*-
# Copyright 2014 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.

"""Test suite for tree_status.py"""

from __future__ import print_function

import mock
import time
import urllib

from chromite.lib import config_lib_unittest
from chromite.lib import constants
from chromite.lib import cros_test_lib
from chromite.lib import timeout_util
from chromite.lib import tree_status


# pylint: disable=protected-access


class TestTreeStatus(cros_test_lib.MockTestCase):
  """Tests TreeStatus method in cros_build_lib."""

  status_url = 'https://chromiumos-status.appspot.com/current?format=json'

  def setUp(self):
    mock_site_config = config_lib_unittest.MockSiteConfig()
    # Add a couple other builders so we have more than one.
    mock_site_config.Add('x86-generic-paladin')
    mock_site_config.Add('arm-generic-paladin')
    self.site_config = mock_site_config

  def _TreeStatusFile(self, message, general_state):
    """Returns a file-like object with the status message writtin in it."""
    json = '{"message": "%s", "general_state": "%s"}' % (message, general_state)
    return mock.MagicMock(json=json, getcode=lambda: 200, read=lambda: json)

  def _SetupMockTreeStatusResponses(self,
                                    final_tree_status='Tree is open.',
                                    final_general_state=constants.TREE_OPEN,
                                    rejected_tree_status='Tree is closed.',
                                    rejected_general_state=
                                    constants.TREE_CLOSED,
                                    rejected_status_count=0,
                                    retries_500=0,
                                    output_final_status=True):
    """Mocks out urllib.urlopen commands to simulate a given tree status.

    Args:
      final_tree_status: The final value of tree status that will be returned
        by urlopen.
      final_general_state: The final value of 'general_state' that will be
        returned by urlopen.
      rejected_tree_status: An intermediate value of tree status that will be
        returned by urlopen and retried upon.
      rejected_general_state: An intermediate value of 'general_state' that
        will be returned by urlopen and retried upon.
      rejected_status_count: The number of times urlopen will return the
        rejected state.
      retries_500: The number of times urlopen will fail with a 500 code.
      output_final_status: If True, the status given by final_tree_status and
        final_general_state will be the last status returned by urlopen. If
        False, final_tree_status will never be returned, and instead an
        unlimited number of times rejected_response will be returned.
    """
    final_response = self._TreeStatusFile(final_tree_status,
                                          final_general_state)
    rejected_response = self._TreeStatusFile(rejected_tree_status,
                                             rejected_general_state)

    error_500_response = mock.MagicMock(getcode=lambda: 500)
    return_value = [error_500_response] * retries_500

    if output_final_status:
      return_value += [rejected_response] * rejected_status_count
      return_value += [final_response]
    else:
      return_value += [rejected_response] * 10

    self.PatchObject(urllib, 'urlopen', autospec=True,
                     side_effect=return_value)

  def testTreeIsOpen(self):
    """Tests that we return True is the tree is open."""
    self._SetupMockTreeStatusResponses(rejected_status_count=5,
                                       retries_500=5)
    self.assertTrue(tree_status.IsTreeOpen(status_url=self.status_url,
                                           period=0))

  def testTreeIsClosed(self):
    """Tests that we return false is the tree is closed."""
    self._SetupMockTreeStatusResponses(output_final_status=False)
    self.assertFalse(tree_status.IsTreeOpen(status_url=self.status_url,
                                            period=0.1))

  def testTreeIsThrottled(self):
    """Tests that we return True if the tree is throttled."""
    self._SetupMockTreeStatusResponses(
        final_tree_status='Tree is throttled (flaky bug on flaky builder)',
        final_general_state=constants.TREE_THROTTLED)
    self.assertTrue(tree_status.IsTreeOpen(status_url=self.status_url,
                                           throttled_ok=True))

  def testTreeIsThrottledNotOk(self):
    """Tests that we respect throttled_ok"""
    self._SetupMockTreeStatusResponses(
        rejected_tree_status='Tree is throttled (flaky bug on flaky builder)',
        rejected_general_state=constants.TREE_THROTTLED,
        output_final_status=False)
    self.assertFalse(tree_status.IsTreeOpen(status_url=self.status_url,
                                            period=0.1))

  def testWaitForStatusOpen(self):
    """Tests that we can wait for a tree open response."""
    self._SetupMockTreeStatusResponses()
    self.assertEqual(tree_status.WaitForTreeStatus(status_url=self.status_url),
                     constants.TREE_OPEN)


  def testWaitForStatusThrottled(self):
    """Tests that we can wait for a tree open response."""
    self._SetupMockTreeStatusResponses(
        final_general_state=constants.TREE_THROTTLED)
    self.assertEqual(tree_status.WaitForTreeStatus(status_url=self.status_url,
                                                   throttled_ok=True),
                     constants.TREE_THROTTLED)

  def testWaitForStatusFailure(self):
    """Tests that we can wait for a tree open response."""
    self._SetupMockTreeStatusResponses(output_final_status=False)
    self.assertRaises(timeout_util.TimeoutError,
                      tree_status.WaitForTreeStatus,
                      status_url=self.status_url,
                      period=0.1)

  def testGetStatusDictParsesMessage(self):
    """Tests that _GetStatusDict parses message correctly."""
    self._SetupMockTreeStatusResponses(
        final_tree_status='Tree is throttled (foo canary: taco investigating)',
        final_general_state=constants.TREE_OPEN)
    data = tree_status._GetStatusDict(self.status_url)
    self.assertEqual(data[tree_status.TREE_STATUS_MESSAGE],
                     'foo canary: taco investigating')

  def testGetStatusDictEmptyMessage(self):
    """Tests that _GetStatusDict stores an empty string for unknown format."""
    self._SetupMockTreeStatusResponses(
        final_tree_status='Tree is throttled. foo canary -> crbug.com/bar',
        final_general_state=constants.TREE_OPEN)
    data = tree_status._GetStatusDict(self.status_url)
    self.assertEqual(data[tree_status.TREE_STATUS_MESSAGE], '')

  def testGetStatusDictRawMessage(self):
    """Tests that _GetStatusDict stores raw message if requested."""
    self._SetupMockTreeStatusResponses(final_tree_status='Tree is open (taco).',
                                       final_general_state=constants.TREE_OPEN)
    data = tree_status._GetStatusDict(self.status_url, raw_message=True)
    self.assertEqual(data[tree_status.TREE_STATUS_MESSAGE],
                     'Tree is open (taco).')

  def testGetExperimentalBuilders(self):
    """Tests that GetExperimentalBuilders parses out EXPERIMENTAL-BUILDERS=."""
    self._SetupMockTreeStatusResponses(
        final_tree_status=(
            'Tree is open (EXPERIMENTAL=amd64-generic-paladin).'),
        final_general_state=constants.TREE_OPEN)
    builders = tree_status.GetExperimentalBuilders(self.status_url)
    self.assertItemsEqual(builders, ['amd64-generic-paladin'])

    self._SetupMockTreeStatusResponses(
        final_tree_status=('Tree is open '
                           '(EXPERIMENTAL=amd64-generic-paladin '
                           'EXPERIMENTAL=arm-generic-paladin).'),
        final_general_state=constants.TREE_OPEN)
    builders = tree_status.GetExperimentalBuilders(self.status_url)
    self.assertItemsEqual(
        builders,
        ['amd64-generic-paladin', 'arm-generic-paladin'])

    # Builders not in the site config are filtered out.
    self._SetupMockTreeStatusResponses(
        final_tree_status=(
            'Tree is open (EXPERIMENTAL=foo-generic-paladin).'),
        final_general_state=constants.TREE_OPEN)
    builders = tree_status.GetExperimentalBuilders(self.status_url)
    self.assertEqual(builders, [])

    # Case insensitive.
    self._SetupMockTreeStatusResponses(
        final_tree_status=(
            'Tree is open (experimental=amd64-generic-paladin).'),
        final_general_state=constants.TREE_OPEN)
    builders = tree_status.GetExperimentalBuilders(self.status_url)
    self.assertItemsEqual(builders, ['amd64-generic-paladin'])

  def testGetExperimentalBuildersTimeout(self):
    """Tests timeout behavior of GetExperimentalBuilders."""
    with mock.patch.object(tree_status, '_GetStatusDict') as m:
      m.side_effect = lambda _: time.sleep(10)
      with self.assertRaises(timeout_util.TimeoutError):
        tree_status.GetExperimentalBuilders(self.status_url, timeout=1)


class TestGettingGardenerEmails(cros_test_lib.MockTestCase):
  """Tests functions related to retrieving the gardener's email address."""

  def _SetEmails(self, emails):
    gardener_json = ('{"updated_unix_timestamp":1547254144,'
                     '"emails":[%s]}' % emails)
    response = mock.MagicMock(json=gardener_json, getcode=lambda: 200,
                              read=lambda: gardener_json)
    self.PatchObject(urllib, 'urlopen', autospec=True,
                     side_effect=[response])

  def testParsingGardenerEmails(self):
    self._SetEmails('"gardener@google.com"')
    self.assertEqual(tree_status.GetGardenerEmailAddresses(),
                     ['gardener@google.com'])

    # Test multiple gardeners.
    self._SetEmails('"gardener@google.com", "gardener2@chromium.org"')
    self.assertEqual(tree_status.GetGardenerEmailAddresses(),
                     ['gardener@google.com', 'gardener2@chromium.org'])


class TestUrlConstruction(cros_test_lib.TestCase):
  """Tests functions that create URLs."""

  def testConstructLegolandBuildURL(self):
    """Tests generating Legoland build URLs."""
    output = tree_status.ConstructLegolandBuildURL('bbid')
    expected = ('https://cros-goldeneye.corp.google.com/chromeos/'
                'healthmonitoring/buildDetails?buildbucketId=bbid')

    self.assertEqual(output, expected)