# 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 base64
import json
import unittest

import mock

from google.appengine.ext import ndb

from dashboard import testing_common
from dashboard import utils
from dashboard.models import graph_data


class UtilsTest(testing_common.TestCase):

  def setUp(self):
    super(UtilsTest, self).setUp()
    testing_common.SetIsInternalUser('internal@chromium.org', True)
    testing_common.SetIsInternalUser('foo@chromium.org', False)

  def _AssertMatches(self, test_path, pattern):
    """Asserts that a test path matches a pattern with MatchesPattern."""
    test_key = utils.TestKey(test_path)
    self.assertTrue(utils.TestMatchesPattern(test_key, pattern))

  def _AssertDoesntMatch(self, test_path, pattern):
    """Asserts that a test path doesn't match a pattern with MatchesPattern."""
    test_key = utils.TestKey(test_path)
    self.assertFalse(utils.TestMatchesPattern(test_key, pattern))

  def testMatchesPattern_AllWildcards(self):
    self._AssertMatches(
        'ChromiumPerf/cros-one/dromaeo.top25/Total', '*/*/*/*')
    self._AssertDoesntMatch(
        'ChromiumPerf/cros-one/dromaeo.top25/Total', '*/*/*')

  def testMatchesPattern_SomeWildcards(self):
    self._AssertMatches(
        'ChromiumPerf/cros-one/dromaeo.top25/Total',
        'ChromiumPerf/*/dromaeo.top25/*')
    self._AssertDoesntMatch(
        'ChromiumPerf/cros-one/dromaeo.top25/Total',
        'ChromiumPerf/*/dromaeo.another_page_set/*')

  def testMatchesPattern_SomePartialWildcards(self):
    self._AssertMatches(
        'ChromiumPerf/cros-one/dromaeo.top25/Total',
        'ChromiumPerf/cros-*/dromaeo.*/Total')
    self._AssertDoesntMatch(
        'ChromiumPerf/cros-one/dromaeoXtop25/Total',
        'ChromiumPerf/cros-*/dromaeo.*/Total')
    self._AssertDoesntMatch(
        'ChromiumPerf/cros-one/dromaeo.top25/Total',
        'OtherMaster/cros-*/dromaeo.*/Total')

  def testMatchesPattern_MorePartialWildcards(self):
    # Note that the wildcard matches zero or more characters.
    self._AssertMatches(
        'ChromiumPerf/cros-one/dromaeo.top25/Total',
        'Chromium*/cros-one*/*.*/To*al')
    self._AssertDoesntMatch(
        'ChromiumPerf/cros-one/dromaeo.top25/Total',
        'Chromium*/linux-*/*.*/To*al')

  def testMatchesPattern_RequiresFullMatchAtEnd(self):
    # If there is no wildcard at the beginning or end of the
    # test path part, then a part will only match if it matches
    # right up to the beginning or end.
    self._AssertDoesntMatch(
        'ChromiumPerf/cros-one/dromaeo.top25/Total',
        'ChromiumPerf/cros-one/dromaeo.top25/*Tot')
    self._AssertDoesntMatch(
        'ChromiumPerf/cros-one/dromaeo.top25/Total',
        'ChromiumPerf/cros-one/dromaeo.top25/otal*')

  def _PutEntitiesAllExternal(self):
    """Puts entities (none internal-only) and returns the keys."""
    master = graph_data.Master(id='M').put()
    graph_data.Bot(parent=master, id='b').put()
    keys = [
        graph_data.TestMetadata(id='M/b/a', internal_only=False).put(),
        graph_data.TestMetadata(id='M/b/b', internal_only=False).put(),
        graph_data.TestMetadata(id='M/b/c', internal_only=False).put(),
        graph_data.TestMetadata(id='M/b/d', internal_only=False).put(),
    ]
    return keys

  def _PutEntitiesHalfInternal(self):
    """Puts entities (half internal-only) and returns the keys."""
    master = graph_data.Master(id='M').put()
    graph_data.Bot(parent=master, id='b').put()
    keys = [
        graph_data.TestMetadata(id='M/b/ax', internal_only=True).put(),
        graph_data.TestMetadata(id='M/b/a', internal_only=False).put(),
        graph_data.TestMetadata(id='M/b/b', internal_only=False).put(),
        graph_data.TestMetadata(id='M/b/bx', internal_only=True).put(),
        graph_data.TestMetadata(id='M/b/c', internal_only=False).put(),
        graph_data.TestMetadata(id='M/b/cx', internal_only=True).put(),
        graph_data.TestMetadata(id='M/b/d', internal_only=False).put(),
        graph_data.TestMetadata(id='M/b/dx', internal_only=True).put(),
    ]
    return keys

  def testGetMulti_ExternalUser_ReturnsSomeEntities(self):
    keys = self._PutEntitiesHalfInternal()
    self.SetCurrentUser('foo@chromium.org')
    self.assertEqual(len(keys) / 2, len(utils.GetMulti(keys)))

  def testGetMulti_InternalUser_ReturnsAllEntities(self):
    keys = self._PutEntitiesHalfInternal()
    self.SetCurrentUser('internal@chromium.org')
    self.assertEqual(len(keys), len(utils.GetMulti(keys)))

  def testGetMulti_AllExternalEntities_ReturnsAllEntities(self):
    keys = self._PutEntitiesAllExternal()
    self.SetCurrentUser('internal@chromium.org')
    self.assertEqual(len(keys), len(utils.GetMulti(keys)))

  def testTestPath_Test(self):
    key = ndb.Key('Master', 'm', 'Bot', 'b', 'Test', 'suite', 'Test', 'metric')
    self.assertEqual('m/b/suite/metric', utils.TestPath(key))

  def testTestPath_TestMetadata(self):
    key = ndb.Key('TestMetadata', 'm/b/suite/metric')
    self.assertEqual('m/b/suite/metric', utils.TestPath(key))

  def testTestPath_Container(self):
    key = ndb.Key('TestContainer', 'm/b/suite/metric')
    self.assertEqual('m/b/suite/metric', utils.TestPath(key))

  def testTestMetadataKey_None(self):
    key = utils.TestMetadataKey(None)
    self.assertIsNone(key)

  def testTestMetadataKey_Test(self):
    key = utils.TestMetadataKey(
        ndb.Key('Master', 'm', 'Bot', 'b', 'Test', 'suite', 'Test', 'metric'))
    self.assertEqual('TestMetadata', key.kind())
    self.assertEqual('m/b/suite/metric', key.id())
    self.assertEqual(('TestMetadata', 'm/b/suite/metric'), key.flat())

  def testTestMetadataKey_TestMetadata(self):
    original_key = ndb.Key('TestMetadata', 'm/b/suite/metric')
    key = utils.TestMetadataKey(original_key)
    self.assertEqual(original_key, key)

  def testTestMetadataKey_String(self):
    key = utils.TestMetadataKey('m/b/suite/metric/page')
    self.assertEqual('TestMetadata', key.kind())
    self.assertEqual('m/b/suite/metric/page', key.id())
    self.assertEqual(('TestMetadata', 'm/b/suite/metric/page'), key.flat())

  def testOldStyleTestKey_None(self):
    key = utils.OldStyleTestKey(None)
    self.assertIsNone(key)

  def testOldStyleTestKey_Test(self):
    original_key = ndb.Key(
        'Master', 'm', 'Bot', 'b', 'Test', 'suite', 'Test', 'metric')
    key = utils.OldStyleTestKey(original_key)
    self.assertEqual(original_key, key)

  def testOldStyleTestKey_TestMetadata(self):
    key = utils.OldStyleTestKey(ndb.Key('TestMetadata', 'm/b/suite/metric'))
    self.assertEqual('Test', key.kind())
    self.assertEqual('metric', key.id())
    self.assertEqual(
        ('Master', 'm', 'Bot', 'b', 'Test', 'suite', 'Test', 'metric'),
        key.flat())

  def testOldStyleTestKey_String(self):
    key = utils.OldStyleTestKey('m/b/suite/metric')
    self.assertEqual('Test', key.kind())
    self.assertEqual('metric', key.id())
    self.assertEqual(
        ('Master', 'm', 'Bot', 'b', 'Test', 'suite', 'Test', 'metric'),
        key.flat())

  def testTestSuiteName_Basic(self):
    key = utils.TestKey('Master/bot/suite-foo/sub/x/y/z')
    self.assertEqual('suite-foo', utils.TestSuiteName(key))

  def testMinimumRange_Empty_ReturnsNone(self):
    self.assertIsNone(utils.MinimumRange([]))

  def testMinimumRange_NotOverlapping_ReturnsNone(self):
    self.assertIsNone(utils.MinimumRange([(5, 10), (15, 20)]))

  def testMinimumRange_OneRange_ReturnsSameRange(self):
    self.assertEqual((5, 10), utils.MinimumRange([(5, 10)]))

  def testMinimumRange_OverlapsForOneNumber_ReturnsRangeWithOneNumber(self):
    self.assertEqual((5, 5), utils.MinimumRange([(2, 5), (5, 10)]))

  def testMinimumRange_MoreThanTwoRanges_ReturnsIntersection(self):
    self.assertEqual((6, 14), utils.MinimumRange(
        [(3, 20), (5, 15), (6, 25), (3, 14)]))

  def testValidate_StringNotInOptionList_Fails(self):
    with self.assertRaises(ValueError):
      utils.Validate(
          ['completed', 'pending', 'failed'], 'running')

  def testValidate_InvalidType_Fails(self):
    with self.assertRaises(ValueError):
      utils.Validate(int, 'a string')

  def testValidate_MissingProperty_Fails(self):
    with self.assertRaises(ValueError):
      utils.Validate(
          {'status': str, 'try_job_id': int, 'required_property': int},
          {'status': 'completed', 'try_job_id': 1234})

  def testValidate_InvalidTypeInDict_Fails(self):
    with self.assertRaises(ValueError):
      utils.Validate(
          {'status': int, 'try_job_id': int},
          {'status': 'completed', 'try_job_id': 1234})

  def testValidate_StringNotInNestedOptionList_Fails(self):
    with self.assertRaises(ValueError):
      utils.Validate(
          {'values': {'nested_values': ['orange', 'banana']}},
          {'values': {'nested_values': 'apple'}})

  def testValidate_MissingPropertyInNestedDict_Fails(self):
    with self.assertRaises(ValueError):
      utils.Validate(
          {'values': {'nested_values': ['orange', 'banana']}},
          {'values': {}})

  def testValidate_ExpectedValueIsNone_Passes(self):
    utils.Validate(None, 'running')

  def testValidate_StringInOptionList_Passes(self):
    utils.Validate(str, 'a string')

  def testValidate_HasExpectedProperties_Passes(self):
    utils.Validate(
        {'status': str, 'try_job_id': int},
        {'status': 'completed', 'try_job_id': 1234})

  def testValidate_StringInNestedOptionList_Passes(self):
    utils.Validate(
        {'values': {'nested_values': ['orange', 'banana']}},
        {'values': {'nested_values': 'orange'}})

  def testValidate_TypeConversion_Passes(self):
    utils.Validate([1], '1')

  @mock.patch.object(utils, 'ServiceAccountHttp', mock.MagicMock())
  @mock.patch('utils.discovery.build')
  def testIsGroupMember_PositiveCase(self, mock_discovery_build):
    mock_request = mock.MagicMock()
    mock_request.execute = mock.MagicMock(return_value={'is_member': True})
    mock_service = mock.MagicMock()
    mock_service.membership = mock.MagicMock(
        return_value=mock_request)
    mock_discovery_build.return_value = mock_service
    self.assertTrue(utils.IsGroupMember('foo@bar.com', 'group'))
    mock_service.membership.assert_called_once_with(
        identity='foo@bar.com', group='group')

  @mock.patch.object(utils, 'ServiceAccountHttp', mock.MagicMock())
  @mock.patch('logging.error')
  @mock.patch('utils.discovery.build')
  def testIsGroupMember_RequestFails_LogsErrorAndReturnsFalse(
      self, mock_discovery_build, mock_logging_error):
    mock_service = mock.MagicMock()
    mock_service.membership = mock.MagicMock(
        return_value={'error': 'Some error'})
    mock_discovery_build.return_value = mock_service
    self.assertFalse(utils.IsGroupMember('foo@bar.com', 'group'))
    self.assertEqual(1, mock_logging_error.call_count)


def _MakeMockFetch(base64_encoded=True, status=200):
  """Returns a mock fetch object that returns a canned response."""
  def _MockFetch(_):
    response_text = json.dumps({'key': 'this is well-formed JSON.'})
    if base64_encoded:
      response_text = base64.b64encode(response_text)
    return testing_common.FakeResponseObject(status, response_text)
  return _MockFetch


class DownloadChromiumFileTest(testing_common.TestCase):

  @mock.patch('google.appengine.api.urlfetch.fetch',
              _MakeMockFetch())
  def testDownloadChromiumFile_BasicCase(self):
    self.assertEqual(
        json.dumps({'key': 'this is well-formed JSON.'}),
        utils.DownloadChromiumFile('some/file'))

  @mock.patch('google.appengine.api.urlfetch.fetch',
              _MakeMockFetch(base64_encoded=False))
  @mock.patch('logging.error')
  def testDownloadChromiumFile_BadEncoding(self, mock_logging_error):
    self.assertIsNone(utils.DownloadChromiumFile('some/file'))
    self.assertEqual(1, mock_logging_error.call_count)

  @mock.patch('google.appengine.api.urlfetch.fetch',
              _MakeMockFetch(status=400))
  @mock.patch('logging.error')
  def testDownloadChromiumFile_Non200Status(self, mock_logging_error):
    self.assertIsNone(utils.DownloadChromiumFile('some/file'))
    self.assertEqual(1, mock_logging_error.call_count)


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