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


import datetime
import os
import sys
import tempfile
import unittest
import unittest.mock as mock

# vpython-provided modules.
from pyfakefs import fake_filesystem_unittest

# //testing imports.
from flake_suppressor_common import common_typing as ct
from flake_suppressor_common import expectations
from flake_suppressor_common import unittest_utils as uu


# Note for all tests in this class: We can safely check the contents of the file
# at the end despite potentially having multiple added lines because Python 3.7+
# guarantees that dictionaries remember insertion order, so there is no risk of
# the order of modification changing.
@unittest.skipIf(sys.version_info[0] != 3, 'Python 3-only')
class IterateThroughResultsForUserUnittest(fake_filesystem_unittest.TestCase):
  def setUp(self) -> None:
    self._new_stdout = open(os.devnull, 'w')
    self.setUpPyfakefs()
    self._expectations = uu.UnitTestExpectationProcessor()
    # Redirect stdout since the tested function prints a lot.
    self._old_stdout = sys.stdout
    sys.stdout = self._new_stdout

    self._input_patcher = mock.patch.object(
      expectations.ExpectationProcessor, 'PromptUserForExpectationAction'
    )
    self._input_mock = self._input_patcher.start()
    self.addCleanup(self._input_patcher.stop)

    self.result_map = {
      'pixel_integration_test': {
        'foo_test': {
          tuple(['win']): ['a'],
          tuple(['mac']): ['b'],
        },
        'bar_test': {
          tuple(['win']): ['c'],
        },
      },
    }

    self.expectation_file = os.path.join(
      uu.ABSOLUTE_EXPECTATION_FILE_DIRECTORY, 'pixel_expectations.txt'
    )
    uu.CreateFile(self, self.expectation_file)
    expectation_file_contents = (
      uu.TAG_HEADER
      + """\
[ win ] some_test [ Failure ]
[ mac ] some_test [ Failure ]
[ android ] some_test [ Failure ]
"""
    )
    with open(self.expectation_file, 'w') as outfile:
      outfile.write(expectation_file_contents)

    self._expectation_file_patcher = mock.patch.object(
      uu.UnitTestExpectationProcessor, 'GetExpectationFileForSuite'
    )
    self._expectation_file_mock = self._expectation_file_patcher.start()
    self._expectation_file_mock.return_value = self.expectation_file
    self.addCleanup(self._expectation_file_patcher.stop)

  def tearDown(self) -> None:
    sys.stdout = self._old_stdout
    self._new_stdout.close()

  def testIterateThroughResultsForUserIgnoreNoGroupByTags(self) -> None:
    """Tests that everything appears to function with ignore and no group."""
    self._input_mock.return_value = (None, None)
    self._expectations.IterateThroughResultsForUser(
      self.result_map, False, True
    )
    expected_contents = (
      uu.TAG_HEADER
      + """\
[ win ] some_test [ Failure ]
[ mac ] some_test [ Failure ]
[ android ] some_test [ Failure ]
"""
    )
    with open(self.expectation_file) as infile:
      self.assertEqual(infile.read(), expected_contents)

  def testIterateThroughResultsForUserIgnoreGroupByTags(self) -> None:
    """Tests that everything appears to function with ignore and grouping."""
    self._input_mock.return_value = (None, None)
    self._expectations.IterateThroughResultsForUser(self.result_map, True, True)
    expected_contents = (
      uu.TAG_HEADER
      + """\
[ win ] some_test [ Failure ]
[ mac ] some_test [ Failure ]
[ android ] some_test [ Failure ]
"""
    )
    with open(self.expectation_file) as infile:
      self.assertEqual(infile.read(), expected_contents)

  def testIterateThroughResultsForUserRetryNoGroupByTags(self) -> None:
    """Tests that everything appears to function with retry and no group."""
    self._input_mock.return_value = ('RetryOnFailure', '')
    self._expectations.IterateThroughResultsForUser(
      self.result_map, False, True
    )
    expected_contents = (
      uu.TAG_HEADER
      + """\
[ win ] some_test [ Failure ]
[ mac ] some_test [ Failure ]
[ android ] some_test [ Failure ]
[ win ] foo_test [ RetryOnFailure ]
[ mac ] foo_test [ RetryOnFailure ]
[ win ] bar_test [ RetryOnFailure ]
"""
    )
    with open(self.expectation_file) as infile:
      self.assertEqual(infile.read(), expected_contents)

  def testIterateThroughResultsForUserRetryGroupByTags(self) -> None:
    """Tests that everything appears to function with retry and grouping."""
    self._input_mock.return_value = ('RetryOnFailure', 'crbug.com/1')
    self._expectations.IterateThroughResultsForUser(self.result_map, True, True)
    expected_contents = (
      uu.TAG_HEADER
      + """\
[ win ] some_test [ Failure ]
crbug.com/1 [ win ] foo_test [ RetryOnFailure ]
crbug.com/1 [ win ] bar_test [ RetryOnFailure ]
[ mac ] some_test [ Failure ]
crbug.com/1 [ mac ] foo_test [ RetryOnFailure ]
[ android ] some_test [ Failure ]
"""
    )
    with open(self.expectation_file) as infile:
      self.assertEqual(infile.read(), expected_contents)

  def testIterateThroughResultsForUserFailNoGroupByTags(self) -> None:
    """Tests that everything appears to function with failure and no group."""
    self._input_mock.return_value = ('Failure', 'crbug.com/1')
    self._expectations.IterateThroughResultsForUser(
      self.result_map, False, True
    )
    expected_contents = (
      uu.TAG_HEADER
      + """\
[ win ] some_test [ Failure ]
[ mac ] some_test [ Failure ]
[ android ] some_test [ Failure ]
crbug.com/1 [ win ] foo_test [ Failure ]
crbug.com/1 [ mac ] foo_test [ Failure ]
crbug.com/1 [ win ] bar_test [ Failure ]
"""
    )
    with open(self.expectation_file) as infile:
      self.assertEqual(infile.read(), expected_contents)

  def testIterateThroughResultsForUserFailGroupByTags(self) -> None:
    """Tests that everything appears to function with failure and grouping."""
    self._input_mock.return_value = ('Failure', '')
    self._expectations.IterateThroughResultsForUser(self.result_map, True, True)
    expected_contents = (
      uu.TAG_HEADER
      + """\
[ win ] some_test [ Failure ]
[ win ] foo_test [ Failure ]
[ win ] bar_test [ Failure ]
[ mac ] some_test [ Failure ]
[ mac ] foo_test [ Failure ]
[ android ] some_test [ Failure ]
"""
    )
    with open(self.expectation_file) as infile:
      self.assertEqual(infile.read(), expected_contents)

  def testIterateThroughResultsForUserNoIncludeAllTags(self) -> None:
    """Tests that everything appears to function without including all tags"""
    self.result_map = {
      'pixel_integration_test': {
        'foo_test': {
          tuple(['win', 'win10']): ['a'],
          tuple(['mac']): ['b'],
        },
        'bar_test': {
          tuple(['win']): ['c'],
        },
      },
    }
    self._input_mock.return_value = ('RetryOnFailure', '')
    self._expectations.IterateThroughResultsForUser(
      self.result_map, False, False
    )
    expected_contents = (
      uu.TAG_HEADER
      + """\
[ win ] some_test [ Failure ]
[ mac ] some_test [ Failure ]
[ android ] some_test [ Failure ]
[ win10 ] foo_test [ RetryOnFailure ]
[ mac ] foo_test [ RetryOnFailure ]
[ win ] bar_test [ RetryOnFailure ]
"""
    )
    with open(self.expectation_file) as infile:
      self.assertEqual(infile.read(), expected_contents)


@unittest.skipIf(sys.version_info[0] != 3, 'Python 3-only')
class IterateThroughResultsWithThresholdsUnittest(
  fake_filesystem_unittest.TestCase
):
  def setUp(self) -> None:
    self.setUpPyfakefs()
    self._expectations = uu.UnitTestExpectationProcessor()
    self.result_map = {
      'pixel_integration_test': {
        'foo_test': {
          tuple(['win']): ['a'],
          tuple(['mac']): ['b'],
        },
        'bar_test': {
          tuple(['win']): ['c'],
        },
      },
    }

    self.expectation_file = os.path.join(
      uu.ABSOLUTE_EXPECTATION_FILE_DIRECTORY, 'pixel_expectations.txt'
    )
    uu.CreateFile(self, self.expectation_file)
    expectation_file_contents = (
      uu.TAG_HEADER
      + """\
[ win ] some_test [ Failure ]
[ mac ] some_test [ Failure ]
[ android ] some_test [ Failure ]
"""
    )
    with open(self.expectation_file, 'w') as outfile:
      outfile.write(expectation_file_contents)

    self._expectation_file_patcher = mock.patch.object(
      uu.UnitTestExpectationProcessor, 'GetExpectationFileForSuite'
    )
    self._expectation_file_mock = self._expectation_file_patcher.start()
    self._expectation_file_mock.return_value = self.expectation_file
    self.addCleanup(self._expectation_file_patcher.stop)

  def testGroupByTags(self) -> None:
    """Tests that threshold-based expectations work when grouping by tags."""
    result_counts = {
      tuple(['win']): {
        # We expect this to be ignored since it has a 1% flake rate.
        'foo_test': 100,
        # We expect this to be RetryOnFailure since it has a 25% flake rate.
        'bar_test': 4,
      },
      tuple(['mac']): {
        # We expect this to be Failure since it has a 50% flake rate.
        'foo_test': 2
      },
    }
    self._expectations.IterateThroughResultsWithThresholds(
      self.result_map, True, result_counts, 0.02, 0.5, True
    )
    expected_contents = (
      uu.TAG_HEADER
      + """\
[ win ] some_test [ Failure ]
[ win ] bar_test [ RetryOnFailure ]
[ mac ] some_test [ Failure ]
[ mac ] foo_test [ Failure ]
[ android ] some_test [ Failure ]
"""
    )
    with open(self.expectation_file) as infile:
      self.assertEqual(infile.read(), expected_contents)

  def testNoGroupByTags(self) -> None:
    """Tests that threshold-based expectations work when not grouping by tags"""
    result_counts = {
      tuple(['win']): {
        # We expect this to be ignored since it has a 1% flake rate.
        'foo_test': 100,
        # We expect this to be RetryOnFailure since it has a 25% flake rate.
        'bar_test': 4,
      },
      tuple(['mac']): {
        # We expect this to be Failure since it has a 50% flake rate.
        'foo_test': 2
      },
    }
    self._expectations.IterateThroughResultsWithThresholds(
      self.result_map, False, result_counts, 0.02, 0.5, True
    )
    expected_contents = (
      uu.TAG_HEADER
      + """\
[ win ] some_test [ Failure ]
[ mac ] some_test [ Failure ]
[ android ] some_test [ Failure ]
[ mac ] foo_test [ Failure ]
[ win ] bar_test [ RetryOnFailure ]
"""
    )
    with open(self.expectation_file) as infile:
      self.assertEqual(infile.read(), expected_contents)

  def testNoIncludeAllTags(self) -> None:
    """Tests that threshold-based expectations work when filtering tags."""
    self.result_map = {
      'pixel_integration_test': {
        'foo_test': {
          tuple(['win', 'win10']): ['a'],
          tuple(['mac']): ['b'],
        },
        'bar_test': {
          tuple(['win', 'win10']): ['c'],
        },
      },
    }

    result_counts = {
      tuple(['win', 'win10']): {
        # We expect this to be ignored since it has a 1% flake rate.
        'foo_test': 100,
        # We expect this to be RetryOnFailure since it has a 25% flake rate.
        'bar_test': 4,
      },
      tuple(['mac']): {
        # We expect this to be Failure since it has a 50% flake rate.
        'foo_test': 2
      },
    }
    self._expectations.IterateThroughResultsWithThresholds(
      self.result_map, False, result_counts, 0.02, 0.5, False
    )
    expected_contents = (
      uu.TAG_HEADER
      + """\
[ win ] some_test [ Failure ]
[ mac ] some_test [ Failure ]
[ android ] some_test [ Failure ]
[ mac ] foo_test [ Failure ]
[ win10 ] bar_test [ RetryOnFailure ]
"""
    )
    with open(self.expectation_file) as infile:
      self.assertEqual(infile.read(), expected_contents)


@unittest.skipIf(sys.version_info[0] != 3, 'Python 3-only')
class CreateExpectationsForAllResultsUnittest(
  fake_filesystem_unittest.TestCase
):
  def setUp(self) -> None:
    self.setUpPyfakefs()
    self._expectations = uu.UnitTestExpectationProcessor()
    self.result_map = {
      'pixel_integration_test': {
        'foo_test': {
          tuple(['win']): [
            ct.ResultTupleType(
              ct.ResultStatus.FAIL,
              'http://ci.chromium.org/b/1111',
              datetime.date.today() - datetime.timedelta(days=2),
              False,
              ['Pass'],
            ),
            ct.ResultTupleType(
              ct.ResultStatus.FAIL,
              'http://ci.chromium.org/b/2222',
              datetime.date.today() - datetime.timedelta(days=3),
              False,
              ['Pass'],
            ),
            ct.ResultTupleType(
              ct.ResultStatus.FAIL,
              'http://ci.chromium.org/b/3333',
              datetime.date.today(),
              False,
              ['Pass'],
            ),
          ],
          tuple(['mac']): [
            ct.ResultTupleType(
              ct.ResultStatus.FAIL,
              'http://ci.chromium.org/b/1111',
              datetime.date.today() - datetime.timedelta(days=1),
              False,
              ['Pass'],
            ),
            ct.ResultTupleType(
              ct.ResultStatus.FAIL,
              'http://ci.chromium.org/b/2222',
              datetime.date.today(),
              False,
              ['Pass'],
            ),
            ct.ResultTupleType(
              ct.ResultStatus.FAIL,
              'http://ci.chromium.org/b/3333',
              datetime.date.today() - datetime.timedelta(days=3),
              False,
              ['Pass'],
            ),
          ],
        },
        'bar_test': {
          tuple(['win']): [
            ct.ResultTupleType(
              ct.ResultStatus.FAIL,
              'http://ci.chromium.org/b/4444',
              datetime.date.today(),
              False,
              ['Pass'],
            ),
            ct.ResultTupleType(
              ct.ResultStatus.FAIL,
              'http://ci.chromium.org/b/5555',
              datetime.date.today() - datetime.timedelta(days=1),
              False,
              ['Pass'],
            ),
            ct.ResultTupleType(
              ct.ResultStatus.FAIL,
              'http://ci.chromium.org/b/6666',
              datetime.date.today() - datetime.timedelta(days=2),
              False,
              ['Pass'],
            ),
          ],
        },
        'baz_test': {
          # This test config causes build fail on less than 2 consecutive
          # days, and thus should not exist in the output.
          tuple(['win']): [
            ct.ResultTupleType(
              ct.ResultStatus.FAIL,
              'http://ci.chromium.org/b/7777',
              datetime.date.today(),
              False,
              ['Pass'],
            ),
            ct.ResultTupleType(
              ct.ResultStatus.FAIL,
              'http://ci.chromium.org/b/8888',
              datetime.date.today(),
              False,
              ['Pass'],
            ),
            ct.ResultTupleType(
              ct.ResultStatus.FAIL,
              'http://ci.chromium.org/b/9999',
              datetime.date.today(),
              False,
              ['Pass'],
            ),
          ],
          tuple(['mac']): [
            ct.ResultTupleType(
              ct.ResultStatus.FAIL,
              'http://ci.chromium.org/b/7777',
              datetime.date.today(),
              False,
              ['Pass'],
            ),
            ct.ResultTupleType(
              ct.ResultStatus.FAIL,
              'http://ci.chromium.org/b/8888',
              datetime.date.today(),
              False,
              ['Pass'],
            ),
          ],
        },
        'wpt_test': {
          # Test for same test in all builders over threshold.
          tuple(['win']): [
            ct.ResultTupleType(
              ct.ResultStatus.FAIL,
              'http://ci.chromium.org/b/1234',
              datetime.date.today(),
              False,
              ['Pass'],
            ),
          ],
          tuple(['mac']): [
            ct.ResultTupleType(
              ct.ResultStatus.FAIL,
              'http://ci.chromium.org/b/2345',
              datetime.date.today() - datetime.timedelta(days=1),
              False,
              ['Pass'],
            ),
            ct.ResultTupleType(
              ct.ResultStatus.FAIL,
              'http://ci.chromium.org/b/3456',
              datetime.date.today(),
              False,
              ['Pass'],
            ),
          ],
        },
      },
    }
    self.build_fail_total_number_threshold = 3
    self.build_fail_consecutive_day_threshold = 2
    self.build_fail_recent_day_threshold = 1

    self.expectation_file = os.path.join(
      uu.ABSOLUTE_EXPECTATION_FILE_DIRECTORY, 'pixel_expectations.txt'
    )
    uu.CreateFile(self, self.expectation_file)
    expectation_file_contents = (
      uu.TAG_HEADER
      + """\
[ win ] some_test [ Failure Pass ]
[ mac ] some_test [ Failure Pass ]
[ android ] some_test [ Failure Pass ]
"""
    )
    with open(self.expectation_file, 'w') as outfile:
      outfile.write(expectation_file_contents)

    self._expectation_file_patcher = mock.patch.object(
      uu.UnitTestExpectationProcessor, 'GetExpectationFileForSuite'
    )
    self._expectation_file_mock = self._expectation_file_patcher.start()
    self._expectation_file_mock.return_value = self.expectation_file
    self.addCleanup(self._expectation_file_patcher.stop)

  def testGroupByTags(self) -> None:
    """Tests that threshold-based expectations work when grouping by tags."""
    self._expectations.CreateExpectationsForAllResults(
      self.result_map,
      True,
      True,
      self.build_fail_total_number_threshold,
      self.build_fail_consecutive_day_threshold,
      self.build_fail_recent_day_threshold,
    )
    expected_contents = (
      uu.TAG_HEADER
      + """\
[ win ] some_test [ Failure Pass ]
[ win ] foo_test [ Failure Pass ]
[ win ] bar_test [ Failure Pass ]
[ win ] wpt_test [ Failure Pass ]
[ mac ] some_test [ Failure Pass ]
[ mac ] foo_test [ Failure Pass ]
[ mac ] wpt_test [ Failure Pass ]
[ android ] some_test [ Failure Pass ]
"""
    )
    with open(self.expectation_file) as infile:
      self.assertEqual(infile.read(), expected_contents)

  def testNoGroupByTags(self) -> None:
    """Tests that threshold-based expectations work when not grouping by tags"""
    self._expectations.CreateExpectationsForAllResults(
      self.result_map,
      False,
      True,
      self.build_fail_total_number_threshold,
      self.build_fail_consecutive_day_threshold,
      self.build_fail_recent_day_threshold,
    )
    expected_contents = (
      uu.TAG_HEADER
      + """\
[ win ] some_test [ Failure Pass ]
[ mac ] some_test [ Failure Pass ]
[ android ] some_test [ Failure Pass ]
[ win ] foo_test [ Failure Pass ]
[ mac ] foo_test [ Failure Pass ]
[ win ] bar_test [ Failure Pass ]
[ win ] wpt_test [ Failure Pass ]
[ mac ] wpt_test [ Failure Pass ]
"""
    )
    with open(self.expectation_file) as infile:
      self.assertEqual(infile.read(), expected_contents)

  def testNoIncludeAllTags(self) -> None:
    """Tests that threshold-based expectations work when filtering tags."""
    self.result_map = {
      'pixel_integration_test': {
        'foo_test': {
          tuple(['win', 'win10']): [
            ct.ResultTupleType(
              ct.ResultStatus.FAIL,
              'http://ci.chromium.org/b/1111',
              datetime.date.today() - datetime.timedelta(days=2),
              False,
              ['Pass'],
            ),
            ct.ResultTupleType(
              ct.ResultStatus.FAIL,
              'http://ci.chromium.org/b/2222',
              datetime.date.today() - datetime.timedelta(days=3),
              False,
              ['Pass'],
            ),
            ct.ResultTupleType(
              ct.ResultStatus.FAIL,
              'http://ci.chromium.org/b/3333',
              datetime.date.today(),
              False,
              ['Pass'],
            ),
          ],
          tuple(['mac']): [
            ct.ResultTupleType(
              ct.ResultStatus.FAIL,
              'http://ci.chromium.org/b/1111',
              datetime.date.today() - datetime.timedelta(days=1),
              False,
              ['Pass'],
            ),
            ct.ResultTupleType(
              ct.ResultStatus.FAIL,
              'http://ci.chromium.org/b/2222',
              datetime.date.today(),
              False,
              ['Pass'],
            ),
            ct.ResultTupleType(
              ct.ResultStatus.FAIL,
              'http://ci.chromium.org/b/3333',
              datetime.date.today() - datetime.timedelta(days=3),
              False,
              ['Pass'],
            ),
          ],
        },
        'bar_test': {
          tuple(['win', 'win10']): [
            ct.ResultTupleType(
              ct.ResultStatus.FAIL,
              'http://ci.chromium.org/b/4444',
              datetime.date.today(),
              False,
              ['Pass'],
            ),
            ct.ResultTupleType(
              ct.ResultStatus.FAIL,
              'http://ci.chromium.org/b/5555',
              datetime.date.today() - datetime.timedelta(days=1),
              False,
              ['Pass'],
            ),
            ct.ResultTupleType(
              ct.ResultStatus.FAIL,
              'http://ci.chromium.org/b/6666',
              datetime.date.today() - datetime.timedelta(days=2),
              False,
              ['Pass'],
            ),
          ],
        },
        'baz_test': {
          # This test config causes build fail on less than 2 consecutive
          # days, and thus should not exist in the output.
          tuple(['win']): [
            ct.ResultTupleType(
              ct.ResultStatus.FAIL,
              'http://ci.chromium.org/b/7777',
              datetime.date.today(),
              False,
              ['Pass'],
            ),
            ct.ResultTupleType(
              ct.ResultStatus.FAIL,
              'http://ci.chromium.org/b/8888',
              datetime.date.today(),
              False,
              ['Pass'],
            ),
            ct.ResultTupleType(
              ct.ResultStatus.FAIL,
              'http://ci.chromium.org/b/9999',
              datetime.date.today(),
              False,
              ['Pass'],
            ),
          ],
          tuple(['mac']): [
            ct.ResultTupleType(
              ct.ResultStatus.FAIL,
              'http://ci.chromium.org/b/7777',
              datetime.date.today(),
              False,
              ['Pass'],
            ),
            ct.ResultTupleType(
              ct.ResultStatus.FAIL,
              'http://ci.chromium.org/b/8888',
              datetime.date.today(),
              False,
              ['Pass'],
            ),
          ],
        },
      },
    }
    self._expectations.CreateExpectationsForAllResults(
      self.result_map,
      False,
      False,
      self.build_fail_total_number_threshold,
      self.build_fail_consecutive_day_threshold,
      self.build_fail_recent_day_threshold,
    )
    expected_contents = (
      uu.TAG_HEADER
      + """\
[ win ] some_test [ Failure Pass ]
[ mac ] some_test [ Failure Pass ]
[ android ] some_test [ Failure Pass ]
[ win10 ] foo_test [ Failure Pass ]
[ mac ] foo_test [ Failure Pass ]
[ win10 ] bar_test [ Failure Pass ]
"""
    )
    with open(self.expectation_file) as infile:
      self.assertEqual(infile.read(), expected_contents)


@unittest.skipIf(sys.version_info[0] != 3, 'Python 3-only')
class FindFailuresInSameConditionUnittest(unittest.TestCase):
  def setUp(self) -> None:
    self._expectations = uu.UnitTestExpectationProcessor()
    self.result_map = {
      'pixel_integration_test': {
        'foo_test': {
          tuple(['win']): ['a'],
          tuple(['mac']): ['a', 'b'],
        },
        'bar_test': {
          tuple(['win']): ['a', 'b', 'c'],
          tuple(['mac']): ['a', 'b', 'c', 'd'],
        },
      },
      'webgl_conformance_integration_test': {
        'foo_test': {
          tuple(['win']): ['a', 'b', 'c', 'd', 'e'],
          tuple(['mac']): ['a', 'b', 'c', 'd', 'e', 'f'],
        },
        'bar_test': {
          tuple(['win']): ['a', 'b', 'c', 'd', 'e', 'f', 'g'],
          tuple(['mac']): ['a', 'b', 'c', 'd', 'e', 'f', 'g', 'h'],
        },
      },
    }

  def testFindFailuresInSameTest(self) -> None:
    other_failures = self._expectations.FindFailuresInSameTest(
      self.result_map, 'pixel_integration_test', 'foo_test', tuple(['win'])
    )
    self.assertEqual(other_failures, [(tuple(['mac']), 2)])

  def testFindFailuresInSameConfig(self) -> None:
    typ_tag_ordered_result_map = self._expectations._ReorderMapByTypTags(
      self.result_map
    )
    other_failures = self._expectations.FindFailuresInSameConfig(
      typ_tag_ordered_result_map,
      'pixel_integration_test',
      'foo_test',
      tuple(['win']),
    )
    expected_other_failures = [
      ('pixel_integration_test.bar_test', 3),
      ('webgl_conformance_integration_test.foo_test', 5),
      ('webgl_conformance_integration_test.bar_test', 7),
    ]
    self.assertEqual(len(other_failures), len(expected_other_failures))
    self.assertEqual(set(other_failures), set(expected_other_failures))


@unittest.skipIf(sys.version_info[0] != 3, 'Python 3-only')
class ModifyFileForResultUnittest(fake_filesystem_unittest.TestCase):
  def setUp(self) -> None:
    self.setUpPyfakefs()
    self._expectations = uu.UnitTestExpectationProcessor()
    self.expectation_file = os.path.join(
      uu.ABSOLUTE_EXPECTATION_FILE_DIRECTORY, 'expectation.txt'
    )
    uu.CreateFile(self, self.expectation_file)
    self._expectation_file_patcher = mock.patch.object(
      uu.UnitTestExpectationProcessor, 'GetExpectationFileForSuite'
    )
    self._expectation_file_mock = self._expectation_file_patcher.start()
    self.addCleanup(self._expectation_file_patcher.stop)
    self._expectation_file_mock.return_value = self.expectation_file

  def testNoGroupByTags(self) -> None:
    """Tests that not grouping by tags appends to the end."""
    expectation_file_contents = (
      uu.TAG_HEADER
      + """\
[ win ] some_test [ Failure ]

[ mac ] some_test [ Failure ]
"""
    )
    with open(self.expectation_file, 'w') as outfile:
      outfile.write(expectation_file_contents)
    self._expectations.ModifyFileForResult(
      'some_file', 'some_test', ('win', 'win10'), '', 'Failure', False, True
    )
    expected_contents = (
      uu.TAG_HEADER
      + """\
[ win ] some_test [ Failure ]

[ mac ] some_test [ Failure ]
[ win win10 ] some_test [ Failure ]
"""
    )
    with open(self.expectation_file) as infile:
      self.assertEqual(infile.read(), expected_contents)

  def testGroupByTagsNoMatch(self) -> None:
    """Tests that grouping by tags but finding no match appends to the end."""
    expectation_file_contents = (
      uu.TAG_HEADER
      + """\
[ mac ] some_test [ Failure ]
"""
    )
    with open(self.expectation_file, 'w') as outfile:
      outfile.write(expectation_file_contents)
    self._expectations.ModifyFileForResult(
      'some_file', 'some_test', ('win', 'win10'), '', 'Failure', True, True
    )
    expected_contents = (
      uu.TAG_HEADER
      + """\
[ mac ] some_test [ Failure ]
[ win win10 ] some_test [ Failure ]
"""
    )
    with open(self.expectation_file) as infile:
      self.assertEqual(infile.read(), expected_contents)

  def testGroupByTagsMatch(self) -> None:
    """Tests that grouping by tags and finding a match adds mid-file."""
    expectation_file_contents = (
      uu.TAG_HEADER
      + """\
[ win ] some_test [ Failure ]

[ mac ] some_test [ Failure ]
"""
    )
    with open(self.expectation_file, 'w') as outfile:
      outfile.write(expectation_file_contents)
    self._expectations.ModifyFileForResult(
      'some_file', 'foo_test', ('win', 'win10'), '', 'Failure', True, True
    )
    expected_contents = (
      uu.TAG_HEADER
      + """\
[ win ] some_test [ Failure ]
[ win ] foo_test [ Failure ]

[ mac ] some_test [ Failure ]
"""
    )
    with open(self.expectation_file) as infile:
      self.assertEqual(infile.read(), expected_contents)


@unittest.skipIf(sys.version_info[0] != 3, 'Python 3-only')
class FilterToMostSpecificTagTypeUnittest(fake_filesystem_unittest.TestCase):
  def setUp(self) -> None:
    self._expectations = uu.UnitTestExpectationProcessor()
    self.setUpPyfakefs()
    with tempfile.NamedTemporaryFile(delete=False) as tf:
      self.expectation_file = tf.name

  def testBasic(self):
    """Tests that only the most specific tags are kept."""
    expectation_file_contents = """\
# tags: [ tag1_least_specific tag1_middle_specific tag1_most_specific ]
# tags: [ tag2_least_specific tag2_middle_specific tag2_most_specific ]"""
    with open(self.expectation_file, 'w') as outfile:
      outfile.write(expectation_file_contents)

    tags = (
      'tag1_least_specific',
      'tag1_most_specific',
      'tag2_middle_specific',
      'tag2_least_specific',
    )
    filtered_tags = self._expectations.FilterToMostSpecificTypTags(
      tags, self.expectation_file
    )
    self.assertEqual(
      filtered_tags, ('tag1_most_specific', 'tag2_middle_specific')
    )

  def testSingleTags(self) -> None:
    """Tests that functionality works as expected with single tags."""
    expectation_file_contents = """\
# tags: [ tag1_most_specific ]
# tags: [ tag2_most_specific ]"""
    with open(self.expectation_file, 'w') as outfile:
      outfile.write(expectation_file_contents)

    tags = ('tag1_most_specific', 'tag2_most_specific')
    filtered_tags = self._expectations.FilterToMostSpecificTypTags(
      tags, self.expectation_file
    )
    self.assertEqual(filtered_tags, tags)

  def testUnusedTags(self) -> None:
    """Tests that functionality works as expected with extra/unused tags."""
    expectation_file_contents = """\
# tags: [ tag1_least_specific tag1_middle_specific tag1_most_specific ]
# tags: [ tag2_least_specific tag2_middle_specific tag2_most_specific ]
# tags: [ some_unused_tag ]"""
    with open(self.expectation_file, 'w') as outfile:
      outfile.write(expectation_file_contents)

    tags = (
      'tag1_least_specific',
      'tag1_most_specific',
      'tag2_middle_specific',
      'tag2_least_specific',
    )
    filtered_tags = self._expectations.FilterToMostSpecificTypTags(
      tags, self.expectation_file
    )
    self.assertEqual(
      filtered_tags, ('tag1_most_specific', 'tag2_middle_specific')
    )

  def testMultiline(self) -> None:
    """Tests that functionality works when tags cover multiple lines."""
    expectation_file_contents = """\
# tags: [ tag1_least_specific
#         tag1_middle_specific
#         tag1_most_specific ]
# tags: [ tag2_least_specific
#         tag2_middle_specific tag2_most_specific ]"""
    with open(self.expectation_file, 'w') as outfile:
      outfile.write(expectation_file_contents)

    tags = (
      'tag1_least_specific',
      'tag1_middle_specific',
      'tag1_most_specific',
      'tag2_middle_specific',
      'tag2_least_specific',
    )
    filtered_tags = self._expectations.FilterToMostSpecificTypTags(
      tags, self.expectation_file
    )
    self.assertEqual(
      filtered_tags, ('tag1_most_specific', 'tag2_middle_specific')
    )

  def testMissingTags(self) -> None:
    """Tests that a file not having all tags is an error."""
    expectation_file_contents = """\
# tags: [ tag1_least_specific tag1_middle_specific ]
# tags: [ tag2_least_specific tag2_middle_specific tag2_most_specific ]"""
    with open(self.expectation_file, 'w') as outfile:
      outfile.write(expectation_file_contents)

    tags = (
      'tag1_least_specific',
      'tag1_most_specific',
      'tag2_middle_specific',
      'tag2_least_specific',
    )
    with self.assertRaises(RuntimeError):
      self._expectations.FilterToMostSpecificTypTags(
        tags, self.expectation_file
      )


@unittest.skipIf(sys.version_info[0] != 3, 'Python 3-only')
class FindBestInsertionLineForExpectationUnittest(
  fake_filesystem_unittest.TestCase
):
  def setUp(self) -> None:
    self.setUpPyfakefs()
    self._expectations = uu.UnitTestExpectationProcessor()
    self.expectation_file = os.path.join(
      uu.ABSOLUTE_EXPECTATION_FILE_DIRECTORY, 'expectation.txt'
    )
    uu.CreateFile(self, self.expectation_file)
    expectation_file_contents = (
      uu.TAG_HEADER
      + """\
[ win ] some_test [ Failure ]

[ mac ] some_test [ Failure ]

[ win release ] bar_test [ Failure ]
[ win ] foo_test [ Failure ]

[ chromeos ] some_test [ Failure ]
"""
    )
    with open(self.expectation_file, 'w') as outfile:
      outfile.write(expectation_file_contents)

  def testNoMatchingTags(self) -> None:
    """Tests behavior when there are no expectations with matching tags."""
    insertion_line, tags = (
      self._expectations.FindBestInsertionLineForExpectation(
        tuple(['android']), self.expectation_file
      )
    )
    self.assertEqual(insertion_line, -1)
    self.assertEqual(tags, set())

  def testMatchingTagsLastEntryChosen(self) -> None:
    """Tests that the last matching line is chosen."""
    insertion_line, tags = (
      self._expectations.FindBestInsertionLineForExpectation(
        tuple(['win']), self.expectation_file
      )
    )
    # We expect "[ win ] foo_test [ Failure ]" to be chosen
    expected_line = len(uu.TAG_HEADER.splitlines()) + 6
    self.assertEqual(insertion_line, expected_line)
    self.assertEqual(tags, set(['win']))

  def testMatchingTagsClosestMatchChosen(self) -> None:
    """Tests that the closest tag match is chosen."""
    insertion_line, tags = (
      self._expectations.FindBestInsertionLineForExpectation(
        ('win', 'release'), self.expectation_file
      )
    )
    # We expect "[ win release ] bar_test [ Failure ]" to be chosen
    expected_line = len(uu.TAG_HEADER.splitlines()) + 5
    self.assertEqual(insertion_line, expected_line)
    self.assertEqual(tags, set(['win', 'release']))


class AssertCheckoutIsUpToDateUnittest(unittest.TestCase):
  def setUp(self) -> None:
    self._expectations = uu.UnitTestExpectationProcessor()
    self._origin_patcher = mock.patch(
      'flake_suppressor_common.expectations.ExpectationProcessor.'
      'GetOriginExpectationFileContents'
    )
    self._origin_mock = self._origin_patcher.start()
    self.addCleanup(self._origin_patcher.stop)
    self._local_patcher = mock.patch(
      'flake_suppressor_common.expectations.'
      + 'ExpectationProcessor.GetLocalCheckoutExpectationFileContents'
    )
    self._local_mock = self._local_patcher.start()
    self.addCleanup(self._local_patcher.stop)

  def testContentsMatch(self) -> None:
    """Tests the happy path where the contents match."""
    self._origin_mock.return_value = {
      'foo.txt': 'foo_content',
      'bar.txt': 'bar_content',
    }
    self._local_mock.return_value = {
      'bar.txt': 'bar_content',
      'foo.txt': 'foo_content',
    }
    self._expectations.AssertCheckoutIsUpToDate()

  def testContentsDoNotMatch(self) -> None:
    """Tests that mismatched contents results in a failure."""
    self._origin_mock.return_value = {
      'foo.txt': 'foo_content',
      'bar.txt': 'bar_content',
    }
    # Differing keys.
    self._local_mock.return_value = {
      'bar.txt': 'bar_content',
      'foo2.txt': 'foo_content',
    }
    with self.assertRaises(RuntimeError):
      self._expectations.AssertCheckoutIsUpToDate()

    # Differing values.
    self._local_mock.return_value = {
      'bar.txt': 'bar_content',
      'foo.txt': 'foo_content2',
    }
    with self.assertRaises(RuntimeError):
      self._expectations.AssertCheckoutIsUpToDate()


class OverFailedBuildThresholdUnittest(unittest.TestCase):
  def setUp(self) -> None:
    self.build_fail_total_number_threshold = 3

  def testOverThreshold(self) -> None:
    """Tests functionality when |result_tuple_list| passes
    |build_fail_total_number_threshold|.

    True is expected output on these inputs.
    """
    result_tuple_list = [
      ct.ResultTupleType(
        ct.ResultStatus.FAIL,
        'http://ci.chromium.org/b/1111',
        datetime.date(2021, 1, 1),
        False,
        ['Pass'],
      ),
      ct.ResultTupleType(
        ct.ResultStatus.FAIL,
        'http://ci.chromium.org/b/2222',
        datetime.date(2022, 1, 1),
        False,
        ['Pass'],
      ),
      ct.ResultTupleType(
        ct.ResultStatus.FAIL,
        'http://ci.chromium.org/b/3333',
        datetime.date(2023, 1, 1),
        False,
        ['Pass'],
      ),
    ]
    self.assertTrue(
      expectations.OverFailedBuildThreshold(
        result_tuple_list, self.build_fail_total_number_threshold
      )
    )

  def testUnderThreshold(self) -> None:
    """Tests functionality when |result_tuple_list| cannot pass
       |build_fail_total_number_threshold|.

    False is expected output on these inputs.
    """
    result_tuple_list = [
      ct.ResultTupleType(
        ct.ResultStatus.FAIL,
        'http://ci.chromium.org/b/1111',
        datetime.date(2022, 1, 1),
        False,
        ['Pass'],
      ),
      ct.ResultTupleType(
        ct.ResultStatus.FAIL,
        'http://ci.chromium.org/b/2222',
        datetime.date(2022, 1, 2),
        False,
        ['Pass'],
      ),
    ]
    self.assertFalse(
      expectations.OverFailedBuildThreshold(
        result_tuple_list, self.build_fail_total_number_threshold
      )
    )

    result_tuple_list = [
      ct.ResultTupleType(
        ct.ResultStatus.FAIL,
        'http://ci.chromium.org/b/1111',
        datetime.date(2022, 1, 1),
        False,
        ['Pass'],
      ),
      ct.ResultTupleType(
        ct.ResultStatus.FAIL,
        'http://ci.chromium.org/b/2222',
        datetime.date(2022, 1, 2),
        False,
        ['Pass'],
      ),
      ct.ResultTupleType(
        ct.ResultStatus.FAIL,
        'http://ci.chromium.org/b/2222',
        datetime.date(2022, 1, 3),
        False,
        ['Pass'],
      ),
    ]
    self.assertFalse(
      expectations.OverFailedBuildThreshold(
        result_tuple_list, self.build_fail_total_number_threshold
      )
    )


class OverFailedBuildByConsecutiveDayThresholdUnittest(unittest.TestCase):
  def setUp(self) -> None:
    self.build_fail_consecutive_day_threshold = 3

  def testOverThreshold(self) -> None:
    """Tests functionality when |result_tuple_list| passes
       |build_fail_consecutive_day_threshold|.

    True is expected output on these inputs.
    """
    result_tuple_list = [
      ct.ResultTupleType(
        ct.ResultStatus.FAIL,
        'http://ci.chromium.org/b/1111',
        datetime.date(2022, 1, 2),
        False,
        ['Pass'],
      ),
      ct.ResultTupleType(
        ct.ResultStatus.FAIL,
        'http://ci.chromium.org/b/2222',
        datetime.date(2022, 1, 1),
        False,
        ['Pass'],
      ),
      ct.ResultTupleType(
        ct.ResultStatus.FAIL,
        'http://ci.chromium.org/b/3333',
        datetime.date(2022, 1, 3),
        False,
        ['Pass'],
      ),
    ]
    self.assertTrue(
      expectations.OverFailedBuildByConsecutiveDayThreshold(
        result_tuple_list, self.build_fail_consecutive_day_threshold
      )
    )

  def testUnderThreshold(self) -> None:
    """Tests functionality when |result_tuple_list| cannot pass
       |build_fail_consecutive_day_threshold|.

    False is expected output on these inputs.
    """
    result_tuple_list = [
      ct.ResultTupleType(
        ct.ResultStatus.FAIL,
        'http://ci.chromium.org/b/1111',
        datetime.date(2022, 1, 1),
        False,
        ['Pass'],
      ),
      ct.ResultTupleType(
        ct.ResultStatus.FAIL,
        'http://ci.chromium.org/b/2222',
        datetime.date(2022, 1, 1),
        False,
        ['Pass'],
      ),
      ct.ResultTupleType(
        ct.ResultStatus.FAIL,
        'http://ci.chromium.org/b/3333',
        datetime.date(2022, 1, 1),
        False,
        ['Pass'],
      ),
    ]
    self.assertFalse(
      expectations.OverFailedBuildByConsecutiveDayThreshold(
        result_tuple_list, self.build_fail_consecutive_day_threshold
      )
    )

    result_tuple_list = [
      ct.ResultTupleType(
        ct.ResultStatus.FAIL,
        'http://ci.chromium.org/b/1111',
        datetime.date(2022, 1, 1),
        False,
        ['Pass'],
      ),
      ct.ResultTupleType(
        ct.ResultStatus.FAIL,
        'http://ci.chromium.org/b/2222',
        datetime.date(2022, 1, 2),
        False,
        ['Pass'],
      ),
    ]
    self.assertFalse(
      expectations.OverFailedBuildByConsecutiveDayThreshold(
        result_tuple_list, self.build_fail_consecutive_day_threshold
      )
    )

    result_tuple_list = [
      ct.ResultTupleType(
        ct.ResultStatus.FAIL,
        'http://ci.chromium.org/b/1111',
        datetime.date(2022, 1, 1),
        False,
        ['Pass'],
      ),
      ct.ResultTupleType(
        ct.ResultStatus.FAIL,
        'http://ci.chromium.org/b/2222',
        datetime.date(2022, 1, 2),
        False,
        ['Pass'],
      ),
      ct.ResultTupleType(
        ct.ResultStatus.FAIL,
        'http://ci.chromium.org/b/3333',
        datetime.date(2022, 1, 4),
        False,
        ['Pass'],
      ),
    ]
    self.assertFalse(
      expectations.OverFailedBuildByConsecutiveDayThreshold(
        result_tuple_list, self.build_fail_consecutive_day_threshold
      )
    )


class FailedBuildWithinRecentDayThresholdUnittest(unittest.TestCase):
  def setUp(self) -> None:
    self.build_fail_recent_day_threshold = 3

  def testWithinThreshold(self) -> None:
    """Tests functionality when |result_tuple_list| has build fail within
       |build_fail_recent_day_threshold|.

    True is expected output on these inputs.
    """
    result_tuple_list = [
      ct.ResultTupleType(
        ct.ResultStatus.FAIL,
        'http://ci.chromium.org/b/1111',
        datetime.date.today(),
        False,
        ['Pass'],
      ),
      ct.ResultTupleType(
        ct.ResultStatus.FAIL,
        'http://ci.chromium.org/b/2222',
        datetime.date.today(),
        False,
        ['Pass'],
      ),
      ct.ResultTupleType(
        ct.ResultStatus.FAIL,
        'http://ci.chromium.org/b/3333',
        datetime.date.today(),
        False,
        ['Pass'],
      ),
    ]
    self.assertTrue(
      expectations.FailedBuildWithinRecentDayThreshold(
        result_tuple_list, self.build_fail_recent_day_threshold
      )
    )

  def testBeyondThreshold(self) -> None:
    """Tests functionality when |result_tuple_list| has no build fail within
       |build_fail_recent_day_threshold|.

    False is expected output on these inputs.
    """
    result_tuple_list = [
      ct.ResultTupleType(
        ct.ResultStatus.FAIL,
        'http://ci.chromium.org/b/1111',
        datetime.date(2022, 1, 1),
        False,
        ['Pass'],
      ),
      ct.ResultTupleType(
        ct.ResultStatus.FAIL,
        'http://ci.chromium.org/b/2222',
        datetime.date(2022, 1, 1),
        False,
        ['Pass'],
      ),
      ct.ResultTupleType(
        ct.ResultStatus.FAIL,
        'http://ci.chromium.org/b/3333',
        datetime.date(2022, 1, 1),
        False,
        ['Pass'],
      ),
    ]
    self.assertFalse(
      expectations.FailedBuildWithinRecentDayThreshold(
        result_tuple_list, self.build_fail_recent_day_threshold
      )
    )


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