# Copyright 2018 The Chromium Authors
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
import json
import logging
import sys
import unittest
from unittest import mock
from unittest.mock import call

from pathlib import Path

# Add tools/perf to sys.path.
FILE_PATH = Path(__file__).resolve()
sys.path.append(str(FILE_PATH.parents[1]))

from core import results_dashboard


class ResultsDashboardTest(unittest.TestCase):
  def setUp(self):
    self.dummy_token_generator = lambda: 'Arthur-Merlin'
    self.perf_data = {'foo': 1, 'bar': 2}
    self.dashboard_url = 'https://chromeperf.appspot.com'
    self.dummy_token = {'token': 'dummy1234', 'file': 'path/to/file'}

  def testRetryForSendResultRetryException(self):

    def raise_retry_exception(url, histogramset_json, token_generator_callback):
      del url, histogramset_json  # unused
      del token_generator_callback  # unused
      raise results_dashboard.SendResultsRetryException('Should retry')

    with mock.patch('core.results_dashboard.time.sleep') as sleep_mock:
      with mock.patch(
        'core.results_dashboard._SendHistogramJson',
        side_effect=raise_retry_exception,
      ) as m:
        upload_result = results_dashboard.SendResults(
          self.perf_data,
          'dummy_benchmark',
          self.dashboard_url,
          send_as_histograms=True,
          token_generator_callback=self.dummy_token_generator,
          num_retries=5,
        )
        self.assertFalse(upload_result)
        self.assertEqual(m.call_count, 5)
        self.assertEqual(
          sleep_mock.mock_calls,
          [call(15), call(30), call(60), call(120), call(240)],
        )

  def testNoRetryForSendResultFatalException(self):

    def raise_retry_exception(url, histogramset_json, token_generator_callback):
      del url, histogramset_json  # unused
      del token_generator_callback  # unused
      raise results_dashboard.SendResultsFatalException('Do not retry')

    with mock.patch('core.results_dashboard.time.sleep') as sleep_mock:
      with mock.patch(
        'core.results_dashboard._SendHistogramJson',
        side_effect=raise_retry_exception,
      ) as m:
        upload_result = results_dashboard.SendResults(
          self.perf_data,
          'dummy_benchmark',
          self.dashboard_url,
          send_as_histograms=True,
          token_generator_callback=self.dummy_token_generator,
          num_retries=5,
        )
        self.assertFalse(upload_result)
        self.assertEqual(m.call_count, 1)
        self.assertFalse(sleep_mock.mock_calls)

  def testNoRetryForSuccessfulSendResult(self):
    with mock.patch('core.results_dashboard.time.sleep') as sleep_mock:
      with mock.patch('core.results_dashboard._SendHistogramJson') as m:
        upload_result = results_dashboard.SendResults(
          self.perf_data,
          'dummy_benchmark',
          self.dashboard_url,
          send_as_histograms=True,
          token_generator_callback=self.dummy_token_generator,
          num_retries=5,
        )
        self.assertTrue(upload_result)
        self.assertEqual(m.call_count, 1)
        self.assertFalse(sleep_mock.mock_calls)

  def testNoRetryAfterSucessfulSendResult(self):
    counter = [0]

    def raise_retry_exception_first_two_times(
      url, histogramset_json, token_generator_callback
    ):
      del url, histogramset_json  # unused
      del token_generator_callback  # unused
      counter[0] += 1
      if counter[0] <= 2:
        raise results_dashboard.SendResultsRetryException('Please retry')

    with mock.patch('core.results_dashboard.time.sleep') as sleep_mock:
      with mock.patch(
        'core.results_dashboard._SendHistogramJson',
        side_effect=raise_retry_exception_first_two_times,
      ) as m:
        upload_result = results_dashboard.SendResults(
          self.perf_data,
          'dummy_benchmark',
          self.dashboard_url,
          send_as_histograms=True,
          token_generator_callback=self.dummy_token_generator,
          num_retries=5,
        )
        self.assertTrue(upload_result)
        self.assertEqual(m.call_count, 3)
        self.assertEqual(sleep_mock.mock_calls, [call(15), call(30)])

  def testSendHistogramJsonRetryException(self):

    def request(*args, **kwargs):
      del args, kwargs
      content = json.dumps(self.dummy_token)

      class Response(object):
        status = 403
        reason = 'dummy auth error'

      return Response(), content

    with mock.patch('core.results_dashboard.time.sleep') as sleep_mock:
      with mock.patch.object(
        results_dashboard.httplib2.Http, 'request', new=request
      ):
        upload_result = results_dashboard.SendResults(
          self.perf_data,
          'dummy_benchmark',
          self.dashboard_url,
          send_as_histograms=True,
          token_generator_callback=self.dummy_token_generator,
          num_retries=5,
        )
        self.assertFalse(upload_result)
        self.assertEqual(
          sleep_mock.mock_calls,
          [call(15), call(30), call(60), call(120), call(240)],
        )

  def testSendHistogramJsonTypeWarning(self):

    def request(*args, **kwargs):
      del args, kwargs
      content = 'non json response'

      class Response(object):
        status = 200

      return Response(), content

    with mock.patch('core.results_dashboard.time.sleep') as sleep_mock:
      with mock.patch.object(
        results_dashboard.httplib2.Http, 'request', new=request
      ):
        upload_result = results_dashboard.SendResults(
          self.perf_data,
          'dummy_benchmark',
          self.dashboard_url,
          send_as_histograms=True,
          token_generator_callback=self.dummy_token_generator,
          num_retries=5,
        )

        # upload should still be successful, despite failing to fetch token.
        self.assertTrue(upload_result)
        self.assertFalse(sleep_mock.mock_calls)

  def testSendHistogramJsonValueWarning(self):

    def request(*args, **kwargs):
      del args, kwargs
      content = json.dumps(None)

      class Response(object):
        status = 200

      return Response(), content

    with mock.patch('core.results_dashboard.time.sleep') as sleep_mock:
      with mock.patch.object(
        results_dashboard.httplib2.Http, 'request', new=request
      ):
        upload_result = results_dashboard.SendResults(
          self.perf_data,
          'dummy_benchmark',
          self.dashboard_url,
          send_as_histograms=True,
          token_generator_callback=self.dummy_token_generator,
          num_retries=5,
        )

        # upload should still be successful, despite failing to fetch token.
        self.assertTrue(upload_result)
        self.assertFalse(sleep_mock.mock_calls)

  def testSuccessfulSendHistogramJson(self):

    def request(*args, **kwargs):
      del args, kwargs
      content = json.dumps(self.dummy_token)

      class Response(object):
        status = 200

      return Response(), content

    with mock.patch('core.results_dashboard.time.sleep') as sleep_mock:
      with mock.patch.object(
        results_dashboard.httplib2.Http, 'request', new=request
      ):
        with mock.patch.object(logging, 'info') as log_mock:
          upload_result = results_dashboard.SendResults(
            self.perf_data,
            'dummy_benchmark',
            self.dashboard_url,
            send_as_histograms=True,
            token_generator_callback=self.dummy_token_generator,
            num_retries=5,
          )

          self.assertTrue(upload_result)
          self.assertFalse(sleep_mock.mock_calls)

          # check token id is logged to info.
          found_token = False
          for log_call in log_mock.call_args_list:
            log = log_call[0][0] % tuple(log_call[0][1:])
            if 'dummy1234' in log:
              found_token = True
              break
          self.assertTrue(
            found_token, msg='Upload completion token not found in logs.'
          )

  def testMakeBuildStatusUrlEscapesPathParts(self):
    # pylint: disable=protected-access
    url = results_dashboard._MakeBuildStatusUrl(
      'chrome project', 'try bucket', 'builder name', '123 456'
    )
    self.assertEqual(
      'https://ci.chromium.org/ui/p/chrome%20project/builders/'
      'try%20bucket/builder%20name/123%20456',
      url,
    )

  @mock.patch('urllib.request.urlopen')
  def testSendResultsJson(self, urlopen_mock):
    # pylint: disable=protected-access
    results_dashboard._SendResultsJson(
      self.dashboard_url, '{"foo": "bar baz"}', self.dummy_token_generator
    )

    urlopen_mock.assert_called_once_with(mock.ANY, timeout=60 * 5)
    req = urlopen_mock.call_args[0][0]
    self.assertEqual(
      self.dashboard_url + results_dashboard.SEND_RESULTS_PATH, req.full_url
    )
    self.assertEqual('Bearer Arthur-Merlin', req.headers['Authorization'])
    self.assertEqual(b'data=%7B%22foo%22%3A+%22bar+baz%22%7D', req.data)


if __name__ == '__main__':
  unittest.main()