# 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 unittest

import mock
import webapp2
import webtest

from dashboard import alerts
from dashboard import testing_common
from dashboard import utils
from dashboard.models import anomaly
from dashboard.models import bug_data
from dashboard.models import graph_data
from dashboard.models import sheriff
from dashboard.models import stoppage_alert


class AlertsTest(testing_common.TestCase):

  # TODO(qyearsley): Simplify this unit test.

  def setUp(self):
    super(AlertsTest, self).setUp()
    app = webapp2.WSGIApplication([('/alerts', alerts.AlertsHandler)])
    self.testapp = webtest.TestApp(app)

  def _AddAlertsToDataStore(self):
    """Adds sample data, including triaged and non-triaged alerts."""
    key_map = {}

    sheriff_key = sheriff.Sheriff(
        id='Chromium Perf Sheriff', email='sullivan@google.com').put()
    testing_common.AddTests(['ChromiumGPU'], ['linux-release'], {
        'scrolling-benchmark': {
            'first_paint': {},
            'mean_frame_time': {},
        }
    })
    first_paint = utils.TestKey(
        'ChromiumGPU/linux-release/scrolling-benchmark/first_paint')
    mean_frame_time = utils.TestKey(
        'ChromiumGPU/linux-release/scrolling-benchmark/mean_frame_time')

    # By default, all Test entities have an improvement_direction of UNKNOWN,
    # meaning that neither direction is considered an improvement.
    # Here we set the improvement direction so that some anomalies are
    # considered improvements.
    for test_key in [first_paint, mean_frame_time]:
      test = test_key.get()
      test.improvement_direction = anomaly.DOWN
      test.put()

    # Add some (12) non-triaged alerts.
    for end_rev in range(10000, 10120, 10):
      test_key = first_paint if end_rev % 20 == 0 else mean_frame_time
      anomaly_entity = anomaly.Anomaly(
          start_revision=end_rev - 5, end_revision=end_rev, test=test_key,
          median_before_anomaly=100, median_after_anomaly=200,
          sheriff=sheriff_key)
      anomaly_entity.SetIsImprovement()
      anomaly_key = anomaly_entity.put()
      key_map[end_rev] = anomaly_key.urlsafe()

    # Add some (2) already-triaged alerts.
    for end_rev in range(10120, 10140, 10):
      test_key = first_paint if end_rev % 20 == 0 else mean_frame_time
      bug_id = -1 if end_rev % 20 == 0 else 12345
      anomaly_entity = anomaly.Anomaly(
          start_revision=end_rev - 5, end_revision=end_rev, test=test_key,
          median_before_anomaly=100, median_after_anomaly=200,
          bug_id=bug_id, sheriff=sheriff_key)
      anomaly_entity.SetIsImprovement()
      anomaly_key = anomaly_entity.put()
      key_map[end_rev] = anomaly_key.urlsafe()
      if bug_id > 0:
        bug_data.Bug(id=bug_id).put()

    # Add some (6) non-triaged improvements.
    for end_rev in range(10140, 10200, 10):
      test_key = mean_frame_time
      anomaly_entity = anomaly.Anomaly(
          start_revision=end_rev - 5, end_revision=end_rev, test=test_key,
          median_before_anomaly=200, median_after_anomaly=100,
          sheriff=sheriff_key)
      anomaly_entity.SetIsImprovement()
      anomaly_key = anomaly_entity.put()
      self.assertTrue(anomaly_entity.is_improvement)
      key_map[end_rev] = anomaly_key.urlsafe()

    return key_map

  def testGet(self):
    response = self.testapp.get('/alerts')
    self.assertEqual('text/html', response.content_type)
    self.assertIn('Chrome Performance Alerts', response.body)

  def testPost_NoParametersSet_UntriagedAlertsListed(self):
    key_map = self._AddAlertsToDataStore()
    response = self.testapp.post('/alerts')
    anomaly_list = self.GetJsonValue(response, 'anomaly_list')
    self.assertEqual(12, len(anomaly_list))
    # The test below depends on the order of the items, but the order is not
    # guaranteed; it depends on the timestamps, which depend on put order.
    anomaly_list.sort(key=lambda a: -a['end_revision'])
    expected_end_rev = 10110
    for alert in anomaly_list:
      self.assertEqual(expected_end_rev, alert['end_revision'])
      self.assertEqual(expected_end_rev - 5, alert['start_revision'])
      self.assertEqual(key_map[expected_end_rev], alert['key'])
      self.assertEqual('ChromiumGPU', alert['master'])
      self.assertEqual('linux-release', alert['bot'])
      self.assertEqual('scrolling-benchmark', alert['testsuite'])
      if expected_end_rev % 20 == 0:
        self.assertEqual('first_paint', alert['test'])
      else:
        self.assertEqual('mean_frame_time', alert['test'])
      self.assertEqual('100.0%', alert['percent_changed'])
      self.assertIsNone(alert['bug_id'])
      expected_end_rev -= 10
    self.assertEqual(expected_end_rev, 9990)

  def testPost_TriagedParameterSet_TriagedListed(self):
    self._AddAlertsToDataStore()
    response = self.testapp.post('/alerts', {'triaged': 'true'})
    anomaly_list = self.GetJsonValue(response, 'anomaly_list')
    # The alerts listed should contain those added above, including alerts
    # that have a bug ID that is not None.
    self.assertEqual(14, len(anomaly_list))
    expected_end_rev = 10130
    # The test below depends on the order of the items, but the order is not
    # guaranteed; it depends on the timestamps, which depend on put order.
    anomaly_list.sort(key=lambda a: -a['end_revision'])
    for alert in anomaly_list:
      if expected_end_rev == 10130:
        self.assertEqual(12345, alert['bug_id'])
      elif expected_end_rev == 10120:
        self.assertEqual(-1, alert['bug_id'])
      else:
        self.assertIsNone(alert['bug_id'])
      expected_end_rev -= 10
    self.assertEqual(expected_end_rev, 9990)

  def testPost_ImprovementsParameterSet_ListsImprovements(self):
    self._AddAlertsToDataStore()
    response = self.testapp.post('/alerts', {'improvements': 'true'})
    anomaly_list = self.GetJsonValue(response, 'anomaly_list')
    self.assertEqual(18, len(anomaly_list))

  def testPost_SheriffParameterSet_OtherSheriffAlertsListed(self):
    self._AddAlertsToDataStore()
    # Add another sheriff to the mock datastore, and set the sheriff of some
    # anomalies to be this new sheriff.
    sheriff2_key = sheriff.Sheriff(
        id='Sheriff2', email='sullivan@google.com').put()
    mean_frame_time = utils.TestKey(
        'ChromiumGPU/linux-release/scrolling-benchmark/mean_frame_time')
    anomalies = anomaly.Anomaly.query(
        anomaly.Anomaly.test == mean_frame_time).fetch()
    for anomaly_entity in anomalies:
      anomaly_entity.sheriff = sheriff2_key
      anomaly_entity.put()

    response = self.testapp.post('/alerts', {'sheriff': 'Sheriff2'})
    anomaly_list = self.GetJsonValue(response, 'anomaly_list')
    sheriff_list = self.GetJsonValue(response, 'sheriff_list')
    for alert in anomaly_list:
      self.assertEqual('mean_frame_time', alert['test'])
    self.assertEqual(2, len(sheriff_list))
    self.assertEqual('Chromium Perf Sheriff', sheriff_list[0])
    self.assertEqual('Sheriff2', sheriff_list[1])

  def testPost_StoppageAlerts_EmbedsStoppageAlertListAndOneTable(self):
    sheriff.Sheriff(id='Sheriff', patterns=['M/b/*/*']).put()
    testing_common.AddTests(['M'], ['b'], {'foo': {'bar': {}}})
    test_key = utils.TestKey('M/b/foo/bar')
    rows = testing_common.AddRows('M/b/foo/bar', {9800, 9802})
    for row in rows:
      stoppage_alert.CreateStoppageAlert(test_key.get(), row).put()
    response = self.testapp.post('/alerts?sheriff=Sheriff')
    stoppage_alert_list = self.GetJsonValue(response, 'stoppage_alert_list')
    self.assertEqual(2, len(stoppage_alert_list))

  @mock.patch('logging.error')
  def testPost_StoppageAlertWithBogusRow_LogsErrorAndShowsTable(
      self, mock_logging_error):
    sheriff.Sheriff(id='Sheriff', patterns=['M/b/*/*']).put()
    testing_common.AddTests(['M'], ['b'], {'foo': {'bar': {}}})
    test_key = utils.TestKey('M/b/foo/bar')
    row_parent = utils.GetTestContainerKey(test_key)
    row = graph_data.Row(parent=row_parent, id=1234)
    stoppage_alert.CreateStoppageAlert(test_key.get(), row).put()
    response = self.testapp.post('/alerts?sheriff=Sheriff')
    stoppage_alert_list = self.GetJsonValue(response, 'stoppage_alert_list')
    self.assertEqual(1, len(stoppage_alert_list))
    self.assertEqual(1, mock_logging_error.call_count)

  def testPost_WithBogusSheriff_HasErrorMessage(self):
    response = self.testapp.post('/alerts?sheriff=Foo')
    error = self.GetJsonValue(response, 'error')
    self.assertIsNotNone(error)

  def testPost_ExternalUserRequestsInternalOnlySheriff_ErrorMessage(self):
    sheriff.Sheriff(id='Foo', internal_only=True).put()
    self.assertFalse(utils.IsInternalUser())
    response = self.testapp.post('/alerts?sheriff=Foo')
    error = self.GetJsonValue(response, 'error')
    self.assertIsNotNone(error)


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