#!/usr/bin/env vpython3
# Copyright 2020 The Chromium Authors
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.

from __future__ import print_function

import json
import os
import sys
from typing import Any, Dict, Set, Tuple
import unittest

if sys.version_info[0] == 2:
  import mock
else:
  import unittest.mock as mock

from pyfakefs import fake_filesystem_unittest

from unexpected_passes_common import builders
from unexpected_passes_common import constants
from unexpected_passes_common import data_types
from unexpected_passes_common import unittest_utils


class FakeFilesystemTestCaseWithFileCreation(fake_filesystem_unittest.TestCase):
  def CreateFile(self, *args, **kwargs):
    # TODO(crbug.com/1156806): Remove this and just use fs.create_file() when
    # Catapult is updated to a newer version of pyfakefs that is compatible with
    # Chromium's version.
    if hasattr(self.fs, 'create_file'):
      self.fs.create_file(*args, **kwargs)
    else:
      self.fs.CreateFile(*args, **kwargs)


class GetCiBuildersUnittest(FakeFilesystemTestCaseWithFileCreation):
  def setUp(self) -> None:
    self._builders_instance = unittest_utils.GenericBuilders(
        suite='webgl_conformance')
    self._isolate_patcher = mock.patch.object(
        self._builders_instance,
        'GetIsolateNames',
        return_value={'telemetry_gpu_integration_test'})
    self._isolate_mock = self._isolate_patcher.start()
    self.addCleanup(self._isolate_patcher.stop)

  def testJsonContentLoaded(self) -> None:
    """Tests that the correct JSON data is loaded in."""
    self.setUpPyfakefs()
    gpu_json = {
        'AAAAA1 AUTOGENERATED FILE DO NOT EDIT': {},
        'Android Release (Nexus 5X)': {
            'isolated_scripts': [{
                'args': [
                    'webgl_conformance',
                ],
                'isolate_name':
                'telemetry_gpu_integration_test',
            }],
        },
        'GPU Linux Builder': {},
    }
    gpu_fyi_json = {
        'AAAAA1 AUTOGENERATED FILE DO NOT EDIT': {},
        'ANGLE GPU Android Release (Nexus 5X)': {
            'isolated_scripts': [{
                'args': [
                    'webgl_conformance',
                ],
                'isolate_name':
                'telemetry_gpu_integration_test',
            }],
        },
        'GPU FYI Linux Builder': {},
    }
    # Should be ignored.
    tryserver_json = {
        'AAAAA1 AUTOGENERATED FILE DO NOT EDIT': {},
        'Trybot': {
            'isolated_scripts': [{
                'args': [
                    'webgl_conformance',
                ],
                'isolate_name':
                'telemetry_gpu_integration_test',
            }],
        },
    }
    # Also should be ignored.
    not_buildbot_json = {
        'Not buildbot': {
            'isolated_scripts': [{
                'args': [
                    'webgl_conformance',
                ],
                'isolate_name':
                'telemetry_gpu_integration_test',
            }],
        },
    }

    self.CreateFile(os.path.join(builders.TESTING_BUILDBOT_DIR,
                                 'chromium.gpu.json'),
                    contents=json.dumps(gpu_json))
    self.CreateFile(os.path.join(builders.TESTING_BUILDBOT_DIR,
                                 'chromium.gpu.fyi.json'),
                    contents=json.dumps(gpu_fyi_json))
    self.CreateFile(os.path.join(builders.TESTING_BUILDBOT_DIR,
                                 'tryserver.gpu.json'),
                    contents=json.dumps(tryserver_json))
    self.CreateFile(os.path.join(builders.TESTING_BUILDBOT_DIR,
                                 'not_buildbot.json'),
                    contents=json.dumps(not_buildbot_json))

    gpu_builders = self._builders_instance.GetCiBuilders()
    self.assertEqual(
        gpu_builders,
        set([
            data_types.BuilderEntry('Android Release (Nexus 5X)',
                                    constants.BuilderTypes.CI, False),
            data_types.BuilderEntry('ANGLE GPU Android Release (Nexus 5X)',
                                    constants.BuilderTypes.CI, False),
            data_types.BuilderEntry('GPU Linux Builder',
                                    constants.BuilderTypes.CI, False),
            data_types.BuilderEntry('GPU FYI Linux Builder',
                                    constants.BuilderTypes.CI, False),
        ]))

  def testPublicInternalBuilders(self) -> None:
    """Tests that public internal builders are treated as internal."""
    self.setUpPyfakefs()
    gpu_json = {
        'AAAAA1 AUTOGENERATED FILE DO NOT EDIT': {},
        'Android Release (Nexus 5X)': {
            'isolated_scripts': [{
                'args': [
                    'webgl_conformance',
                ],
                'isolate_name':
                'telemetry_gpu_integration_test',
            }],
        },
        'GPU Linux Builder': {},
    }
    gpu_internal_json = {
        'AAAAA1 AUTOGENERATED FILE DO NOT EDIT': {},
        'Android Chrome Release (Nexus 5X)': {
            'isolated_scripts': [{
                'args': [
                    'webgl_conformance',
                ],
                'isolate_name':
                'telemetry_gpu_integration_test',
            }],
        },
        'GPU Chrome Linux Builder': {},
    }
    internal_json = {
        'AAAAA1 AUTOGENERATED FILE DO NOT EDIT': {},
        'Android Internal Release (Nexus 5X)': {
            'isolated_scripts': [{
                'args': [
                    'webgl_conformance',
                ],
                'isolate_name':
                'telemetry_gpu_integration_test',
            }],
        },
        'GPU Internal Linux Builder': {},
    }

    self.CreateFile(os.path.join(builders.TESTING_BUILDBOT_DIR,
                                 'chromium.gpu.json'),
                    contents=json.dumps(gpu_json))
    self.CreateFile(os.path.join(builders.TESTING_BUILDBOT_DIR,
                                 'chrome.gpu.fyi.json'),
                    contents=json.dumps(gpu_internal_json))
    self.CreateFile(os.path.join(builders.INTERNAL_TESTING_BUILDBOT_DIR,
                                 'internal.json'),
                    contents=json.dumps(internal_json))

    gpu_builders = self._builders_instance.GetCiBuilders()
    self.assertEqual(
        gpu_builders,
        set([
            data_types.BuilderEntry('Android Release (Nexus 5X)',
                                    constants.BuilderTypes.CI, False),
            data_types.BuilderEntry('GPU Linux Builder',
                                    constants.BuilderTypes.CI, False),
        ]))

    internal_instance = unittest_utils.GenericBuilders(
        suite='webgl_conformance', include_internal_builders=True)
    with mock.patch.object(internal_instance,
                           'GetIsolateNames',
                           return_value={'telemetry_gpu_integration_test'}):
      gpu_builders = internal_instance.GetCiBuilders()
    self.assertEqual(
        gpu_builders,
        set([
            data_types.BuilderEntry('Android Release (Nexus 5X)',
                                    constants.BuilderTypes.CI, False),
            data_types.BuilderEntry('Android Chrome Release (Nexus 5X)',
                                    constants.BuilderTypes.CI, True),
            data_types.BuilderEntry('Android Internal Release (Nexus 5X)',
                                    constants.BuilderTypes.CI, True),
            data_types.BuilderEntry('GPU Linux Builder',
                                    constants.BuilderTypes.CI, False),
            data_types.BuilderEntry('GPU Chrome Linux Builder',
                                    constants.BuilderTypes.CI, True),
            data_types.BuilderEntry('GPU Internal Linux Builder',
                                    constants.BuilderTypes.CI, True),
        ]))

  def testFilterBySuite(self) -> None:
    """Tests that only builders that run the given suite are returned."""

    def SideEffect(tm: Dict[str, Any]) -> bool:
      tests = tm.get('isolated_scripts', [])
      for t in tests:
        if t.get('isolate_name') == 'foo_integration_test':
          if 'webgl_conformance' in t.get('args', []):
            return True
      return False

    self.setUpPyfakefs()
    gpu_json = {
        'AAAAA1 AUTOGENERATED FILE DO NOT EDIT': {},
        'Android Tester': {
            'isolated_scripts': [
                {
                    'args': [
                        'webgl_conformance',
                    ],
                    'isolate_name': 'not_telemetry',
                },
            ],
        },
        'Linux Tester': {
            'isolated_scripts': [
                {
                    'args': [
                        'not_a_suite',
                    ],
                    'isolate_name': 'foo_integration_test',
                },
            ],
        },
        'Windows Tester': {
            'isolated_scripts': [
                {
                    'args': [
                        'webgl_conformance',
                    ],
                    'isolate_name': 'foo_integration_test',
                },
            ],
        },
    }

    self.CreateFile(os.path.join(builders.TESTING_BUILDBOT_DIR,
                                 'chromium.json'),
                    contents=json.dumps(gpu_json))

    with mock.patch.object(self._builders_instance,
                           '_BuilderRunsTestOfInterest',
                           side_effect=SideEffect):
      gpu_builders = self._builders_instance.GetCiBuilders()
    self.assertEqual(
        gpu_builders,
        set([
            data_types.BuilderEntry('Windows Tester', constants.BuilderTypes.CI,
                                    False)
        ]))

  def testRealContentCanBeLoaded(self) -> None:
    """Tests that *something* from the real JSON files can be loaded."""
    # This directory is not available on swarming, so if it doesn't exist, just
    # skip the test.
    if not os.path.exists(builders.TESTING_BUILDBOT_DIR):
      return
    self.assertNotEqual(len(self._builders_instance.GetCiBuilders()), 0)


class GetMirroredBuildersForCiBuilderUnittest(unittest.TestCase):
  def setUp(self) -> None:
    self._builders_instance = builders.Builders('suite', False)
    self._bb_patcher = mock.patch.object(self._builders_instance,
                                         '_GetBuildbucketOutputForCiBuilder')
    self._bb_mock = self._bb_patcher.start()
    self.addCleanup(self._bb_patcher.stop)
    self._fake_ci_patcher = mock.patch.object(self._builders_instance,
                                              'GetFakeCiBuilders',
                                              return_value={})
    self._fake_ci_mock = self._fake_ci_patcher.start()
    self.addCleanup(self._fake_ci_patcher.stop)
    self._non_chromium_patcher = mock.patch.object(
        self._builders_instance,
        'GetNonChromiumBuilders',
        return_value={'foo_non_chromium'})
    self._non_chromium_mock = self._non_chromium_patcher.start()
    self.addCleanup(self._non_chromium_patcher.stop)

  def testFakeCiBuilder(self) -> None:
    """Tests that a fake CI builder gets properly mapped."""
    self._fake_ci_mock.return_value = {
        data_types.BuilderEntry('foo_ci', constants.BuilderTypes.CI, False):
        {data_types.BuilderEntry('foo_try', constants.BuilderTypes.TRY, False)}
    }
    try_builder, found_mirror = (
        self._builders_instance._GetMirroredBuildersForCiBuilder(
            data_types.BuilderEntry('foo_ci', constants.BuilderTypes.CI,
                                    False)))
    self.assertTrue(found_mirror)
    self.assertEqual(
        try_builder,
        set([
            data_types.BuilderEntry('foo_try', constants.BuilderTypes.TRY,
                                    False)
        ]))
    self._bb_mock.assert_not_called()

  def testNoBuildbucketOutput(self) -> None:
    """Tests that a failure to get Buildbucket output is surfaced."""
    self._bb_mock.return_value = ''
    builder_entry = data_types.BuilderEntry('nonexistent',
                                            constants.BuilderTypes.CI, False)
    try_builder, found_mirror = (
        self._builders_instance._GetMirroredBuildersForCiBuilder(builder_entry))
    self.assertFalse(found_mirror)
    self.assertEqual(try_builder, set([builder_entry]))

  def testBuildbucketOutput(self):
    """Tests that Buildbucket output is parsed correctly."""
    self._bb_mock.return_value = json.dumps({
        'output': {
            'properties': {
                'mirrored_builders': [
                    'try:foo_try',
                    'try:bar_try',
                ]
            }
        }
    })
    try_builders, found_mirror = (
        self._builders_instance._GetMirroredBuildersForCiBuilder(
            data_types.BuilderEntry('foo_ci', constants.BuilderTypes.CI,
                                    False)))
    self.assertTrue(found_mirror)
    self.assertEqual(
        try_builders,
        set([
            data_types.BuilderEntry('foo_try', constants.BuilderTypes.TRY,
                                    False),
            data_types.BuilderEntry('bar_try', constants.BuilderTypes.TRY,
                                    False)
        ]))

  def testBuildbucketOutputInternal(self) -> None:
    """Tests that internal Buildbucket output is parsed correctly."""
    self._bb_mock.return_value = json.dumps({
        'output': {
            'properties': {
                'mirrored_builders': [
                    'try:foo_try',
                    'try:bar_try',
                ]
            }
        }
    })
    try_builders, found_mirror = (
        self._builders_instance._GetMirroredBuildersForCiBuilder(
            data_types.BuilderEntry('foo_ci', constants.BuilderTypes.CI, True)))
    self.assertTrue(found_mirror)
    self.assertEqual(
        try_builders,
        set([
            data_types.BuilderEntry('foo_try', constants.BuilderTypes.TRY,
                                    True),
            data_types.BuilderEntry('bar_try', constants.BuilderTypes.TRY, True)
        ]))


class GetTryBuildersUnittest(FakeFilesystemTestCaseWithFileCreation):
  def setUp(self) -> None:
    self._builders_instance = builders.Builders('suite', False)
    self._get_patcher = mock.patch.object(self._builders_instance,
                                          '_GetMirroredBuildersForCiBuilder')
    self._get_mock = self._get_patcher.start()
    self.addCleanup(self._get_patcher.stop)
    self._runs_test_patcher = mock.patch.object(self._builders_instance,
                                                '_BuilderRunsTestOfInterest')
    self._runs_test_mock = self._runs_test_patcher.start()
    self.addCleanup(self._runs_test_patcher.stop)

    self.setUpPyfakefs()
    # Make sure the directory exists.
    self.CreateFile(
        os.path.join(builders.TESTING_BUILDBOT_DIR, 'placeholder.txt'))

  def testMirrorNoOutputCausesFailure(self) -> None:
    """Tests that a failure to get Buildbot output raises an exception."""
    builder = data_types.BuilderEntry('foo_ci', constants.BuilderTypes.CI,
                                      False)
    self._get_mock.return_value = (set([builder]), False)
    self._runs_test_mock.return_value = True
    with self.assertRaises(RuntimeError):
      self._builders_instance.GetTryBuilders([builder])

  def testMirrorOutputReturned(self) -> None:
    """Tests that parsed, mirrored builders get returned on success."""

    def SideEffect(ci_builder: data_types.BuilderEntry
                   ) -> Tuple[Set[data_types.BuilderEntry], bool]:
      b = [
          data_types.BuilderEntry(ci_builder.name.replace('ci', 'try'),
                                  constants.BuilderTypes.TRY, False),
          data_types.BuilderEntry(ci_builder.name.replace('ci', 'try2'),
                                  constants.BuilderTypes.TRY, False),
      ]
      return set(b), True

    self._get_mock.side_effect = SideEffect
    self._runs_test_mock.return_value = False
    mirrored_builders = self._builders_instance.GetTryBuilders([
        data_types.BuilderEntry('foo_ci', constants.BuilderTypes.CI, False),
        data_types.BuilderEntry('bar_ci', constants.BuilderTypes.CI, False),
    ])
    self.assertEqual(
        mirrored_builders,
        set([
            data_types.BuilderEntry('foo_try', constants.BuilderTypes.TRY,
                                    False),
            data_types.BuilderEntry('foo_try2', constants.BuilderTypes.TRY,
                                    False),
            data_types.BuilderEntry('bar_try', constants.BuilderTypes.TRY,
                                    False),
            data_types.BuilderEntry('bar_try2', constants.BuilderTypes.TRY,
                                    False),
        ]))

  def testDedicatedJsonContentLoaded(self) -> None:
    """Tests that tryserver JSON content is loaded."""

    def SideEffect(test_spec: Dict[str, Any]) -> bool:
      # Treat non-empty test specs as valid.
      return bool(test_spec)

    self._runs_test_mock.side_effect = SideEffect
    # Should be ignored.
    gpu_json = {
        'AAAAA1 AUTOGENERATED FILE DO NOT EDIT': {},
        'Android Release (Nexus 5X)': {
            'isolated_scripts': [{
                'args': [
                    'webgl_conformance',
                ],
                'isolate_name':
                'telemetry_gpu_integration_test',
            }],
        },
        'GPU Linux Builder': {},
    }
    # Should be ignored.
    gpu_fyi_json = {
        'AAAAA1 AUTOGENERATED FILE DO NOT EDIT': {},
        'ANGLE GPU Android Release (Nexus 5X)': {
            'isolated_scripts': [{
                'args': [
                    'webgl_conformance',
                ],
                'isolate_name':
                'telemetry_gpu_integration_test',
            }],
        },
        'GPU FYI Linux Builder': {},
    }
    tryserver_json = {
        'AAAAA1 AUTOGENERATED FILE DO NOT EDIT': {},
        'Trybot': {
            'isolated_scripts': [{
                'args': [
                    'webgl_conformance',
                ],
                'isolate_name':
                'telemetry_gpu_integration_test',
            }],
        },
        'Trybot Empty': {},
    }
    # Also should be ignored.
    not_buildbot_json = {
        'Not buildbot': {
            'isolated_scripts': [{
                'args': [
                    'webgl_conformance',
                ],
                'isolate_name':
                'telemetry_gpu_integration_test',
            }],
        },
    }

    self.CreateFile(os.path.join(builders.TESTING_BUILDBOT_DIR,
                                 'chromium.gpu.json'),
                    contents=json.dumps(gpu_json))
    self.CreateFile(os.path.join(builders.TESTING_BUILDBOT_DIR,
                                 'chromium.gpu.fyi.json'),
                    contents=json.dumps(gpu_fyi_json))
    self.CreateFile(os.path.join(builders.TESTING_BUILDBOT_DIR,
                                 'tryserver.gpu.json'),
                    contents=json.dumps(tryserver_json))
    self.CreateFile(os.path.join(builders.TESTING_BUILDBOT_DIR,
                                 'not_buildbot.json'),
                    contents=json.dumps(not_buildbot_json))

    gpu_builders = self._builders_instance.GetTryBuilders({})
    self.assertEqual(
        gpu_builders,
        set([
            data_types.BuilderEntry('Trybot', constants.BuilderTypes.TRY,
                                    False),
        ]))

  def testDedicatedFilterBySuite(self) -> None:
    """Tests that only builders that run the given suite are returned."""

    def SideEffect(tm: Dict[str, Any]) -> bool:
      tests = tm.get('isolated_scripts', [])
      for t in tests:
        if t.get('isolate_name') == 'foo_integration_test':
          if 'webgl_conformance' in t.get('args', []):
            return True
      return False

    self._runs_test_mock.side_effect = SideEffect
    gpu_json = {
        'AAAAA1 AUTOGENERATED FILE DO NOT EDIT': {},
        'Android Tester': {
            'isolated_scripts': [
                {
                    'args': [
                        'webgl_conformance',
                    ],
                    'isolate_name': 'not_telemetry',
                },
            ],
        },
        'Linux Tester': {
            'isolated_scripts': [
                {
                    'args': [
                        'not_a_suite',
                    ],
                    'isolate_name': 'foo_integration_test',
                },
            ],
        },
        'Windows Tester': {
            'isolated_scripts': [
                {
                    'args': [
                        'webgl_conformance',
                    ],
                    'isolate_name': 'foo_integration_test',
                },
            ],
        },
    }

    self.CreateFile(os.path.join(builders.TESTING_BUILDBOT_DIR,
                                 'tryserver.chromium.json'),
                    contents=json.dumps(gpu_json))

    gpu_builders = self._builders_instance.GetTryBuilders({})
    self.assertEqual(
        gpu_builders,
        set([
            data_types.BuilderEntry('Windows Tester',
                                    constants.BuilderTypes.TRY, False)
        ]))

  def testDedicatedAndMirroredCombined(self) -> None:
    """Tests that both dedicated and mirrored trybots are returned."""

    def SideEffect(_: Any) -> Tuple[Set[data_types.BuilderEntry], bool]:
      return set({
          data_types.BuilderEntry('mirrored_trybot', constants.BuilderTypes.TRY,
                                  False)
      }), True

    self._get_mock.side_effect = SideEffect
    self._runs_test_mock.return_value = True
    tryserver_json = {
        'AAAAA1 AUTOGENERATED FILE DO NOT EDIT': {},
        'Trybot': {
            'isolated_scripts': [{
                'args': [
                    'webgl_conformance',
                ],
                'isolate_name':
                'telemetry_gpu_integration_test',
            }],
        },
    }

    self.CreateFile(os.path.join(builders.TESTING_BUILDBOT_DIR,
                                 'tryserver.chromium.json'),
                    contents=json.dumps(tryserver_json))

    try_builders = self._builders_instance.GetTryBuilders({
        data_types.BuilderEntry('ci_builder', constants.BuilderTypes.CI, False)
    })
    self.assertEqual(
        try_builders, {
            data_types.BuilderEntry('mirrored_trybot',
                                    constants.BuilderTypes.TRY, False),
            data_types.BuilderEntry('Trybot', constants.BuilderTypes.TRY, False)
        })


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