xref: /aosp_15_r20/external/cronet/testing/flake_suppressor_common/expectations_unittest.py (revision 6777b5387eb2ff775bb5750e3f5d96f37fb7352b)
1#!/usr/bin/env vpython3
2# Copyright 2021 The Chromium Authors
3# Use of this source code is governed by a BSD-style license that can be
4# found in the LICENSE file.
5
6# pylint: disable=protected-access
7
8import datetime
9import os
10import sys
11import tempfile
12import unittest
13import unittest.mock as mock
14
15from pyfakefs import fake_filesystem_unittest  # pylint:disable=import-error
16
17from flake_suppressor_common import common_typing as ct
18from flake_suppressor_common import expectations
19from flake_suppressor_common import unittest_utils as uu
20
21
22# Note for all tests in this class: We can safely check the contents of the file
23# at the end despite potentially having multiple added lines because Python 3.7+
24# guarantees that dictionaries remember insertion order, so there is no risk of
25# the order of modification changing.
26@unittest.skipIf(sys.version_info[0] != 3, 'Python 3-only')
27class IterateThroughResultsForUserUnittest(fake_filesystem_unittest.TestCase):
28  def setUp(self) -> None:
29    self._new_stdout = open(os.devnull, 'w')
30    self.setUpPyfakefs()
31    self._expectations = uu.UnitTestExpectationProcessor()
32    # Redirect stdout since the tested function prints a lot.
33    self._old_stdout = sys.stdout
34    sys.stdout = self._new_stdout
35
36    self._input_patcher = mock.patch.object(expectations.ExpectationProcessor,
37                                            'PromptUserForExpectationAction')
38    self._input_mock = self._input_patcher.start()
39    self.addCleanup(self._input_patcher.stop)
40
41    self.result_map = {
42        'pixel_integration_test': {
43            'foo_test': {
44                tuple(['win']): ['a'],
45                tuple(['mac']): ['b'],
46            },
47            'bar_test': {
48                tuple(['win']): ['c'],
49            },
50        },
51    }
52
53    self.expectation_file = os.path.join(uu.ABSOLUTE_EXPECTATION_FILE_DIRECTORY,
54                                         'pixel_expectations.txt')
55    uu.CreateFile(self, self.expectation_file)
56    expectation_file_contents = uu.TAG_HEADER + """\
57[ win ] some_test [ Failure ]
58[ mac ] some_test [ Failure ]
59[ android ] some_test [ Failure ]
60"""
61    with open(self.expectation_file, 'w') as outfile:
62      outfile.write(expectation_file_contents)
63
64    self._expectation_file_patcher = mock.patch.object(
65        uu.UnitTestExpectationProcessor, 'GetExpectationFileForSuite')
66    self._expectation_file_mock = self._expectation_file_patcher.start()
67    self._expectation_file_mock.return_value = self.expectation_file
68    self.addCleanup(self._expectation_file_patcher.stop)
69
70  def tearDown(self) -> None:
71    sys.stdout = self._old_stdout
72    self._new_stdout.close()
73
74  def testIterateThroughResultsForUserIgnoreNoGroupByTags(self) -> None:
75    """Tests that everything appears to function with ignore and no group."""
76    self._input_mock.return_value = (None, None)
77    self._expectations.IterateThroughResultsForUser(self.result_map, False,
78                                                    True)
79    expected_contents = uu.TAG_HEADER + """\
80[ win ] some_test [ Failure ]
81[ mac ] some_test [ Failure ]
82[ android ] some_test [ Failure ]
83"""
84    with open(self.expectation_file) as infile:
85      self.assertEqual(infile.read(), expected_contents)
86
87  def testIterateThroughResultsForUserIgnoreGroupByTags(self) -> None:
88    """Tests that everything appears to function with ignore and grouping."""
89    self._input_mock.return_value = (None, None)
90    self._expectations.IterateThroughResultsForUser(self.result_map, True, True)
91    expected_contents = uu.TAG_HEADER + """\
92[ win ] some_test [ Failure ]
93[ mac ] some_test [ Failure ]
94[ android ] some_test [ Failure ]
95"""
96    with open(self.expectation_file) as infile:
97      self.assertEqual(infile.read(), expected_contents)
98
99  def testIterateThroughResultsForUserRetryNoGroupByTags(self) -> None:
100    """Tests that everything appears to function with retry and no group."""
101    self._input_mock.return_value = ('RetryOnFailure', '')
102    self._expectations.IterateThroughResultsForUser(self.result_map, False,
103                                                    True)
104    expected_contents = uu.TAG_HEADER + """\
105[ win ] some_test [ Failure ]
106[ mac ] some_test [ Failure ]
107[ android ] some_test [ Failure ]
108[ win ] foo_test [ RetryOnFailure ]
109[ mac ] foo_test [ RetryOnFailure ]
110[ win ] bar_test [ RetryOnFailure ]
111"""
112    with open(self.expectation_file) as infile:
113      self.assertEqual(infile.read(), expected_contents)
114
115  def testIterateThroughResultsForUserRetryGroupByTags(self) -> None:
116    """Tests that everything appears to function with retry and grouping."""
117    self._input_mock.return_value = ('RetryOnFailure', 'crbug.com/1')
118    self._expectations.IterateThroughResultsForUser(self.result_map, True, True)
119    expected_contents = uu.TAG_HEADER + """\
120[ win ] some_test [ Failure ]
121crbug.com/1 [ win ] foo_test [ RetryOnFailure ]
122crbug.com/1 [ win ] bar_test [ RetryOnFailure ]
123[ mac ] some_test [ Failure ]
124crbug.com/1 [ mac ] foo_test [ RetryOnFailure ]
125[ android ] some_test [ Failure ]
126"""
127    with open(self.expectation_file) as infile:
128      self.assertEqual(infile.read(), expected_contents)
129
130  def testIterateThroughResultsForUserFailNoGroupByTags(self) -> None:
131    """Tests that everything appears to function with failure and no group."""
132    self._input_mock.return_value = ('Failure', 'crbug.com/1')
133    self._expectations.IterateThroughResultsForUser(self.result_map, False,
134                                                    True)
135    expected_contents = uu.TAG_HEADER + """\
136[ win ] some_test [ Failure ]
137[ mac ] some_test [ Failure ]
138[ android ] some_test [ Failure ]
139crbug.com/1 [ win ] foo_test [ Failure ]
140crbug.com/1 [ mac ] foo_test [ Failure ]
141crbug.com/1 [ win ] bar_test [ Failure ]
142"""
143    with open(self.expectation_file) as infile:
144      self.assertEqual(infile.read(), expected_contents)
145
146  def testIterateThroughResultsForUserFailGroupByTags(self) -> None:
147    """Tests that everything appears to function with failure and grouping."""
148    self._input_mock.return_value = ('Failure', '')
149    self._expectations.IterateThroughResultsForUser(self.result_map, True, True)
150    expected_contents = uu.TAG_HEADER + """\
151[ win ] some_test [ Failure ]
152[ win ] foo_test [ Failure ]
153[ win ] bar_test [ Failure ]
154[ mac ] some_test [ Failure ]
155[ mac ] foo_test [ Failure ]
156[ android ] some_test [ Failure ]
157"""
158    with open(self.expectation_file) as infile:
159      self.assertEqual(infile.read(), expected_contents)
160
161  def testIterateThroughResultsForUserNoIncludeAllTags(self) -> None:
162    """Tests that everything appears to function without including all tags"""
163    self.result_map = {
164        'pixel_integration_test': {
165            'foo_test': {
166                tuple(['win', 'win10']): ['a'],
167                tuple(['mac']): ['b'],
168            },
169            'bar_test': {
170                tuple(['win']): ['c'],
171            },
172        },
173    }
174    self._input_mock.return_value = ('RetryOnFailure', '')
175    self._expectations.IterateThroughResultsForUser(self.result_map, False,
176                                                    False)
177    expected_contents = uu.TAG_HEADER + """\
178[ win ] some_test [ Failure ]
179[ mac ] some_test [ Failure ]
180[ android ] some_test [ Failure ]
181[ win10 ] foo_test [ RetryOnFailure ]
182[ mac ] foo_test [ RetryOnFailure ]
183[ win ] bar_test [ RetryOnFailure ]
184"""
185    with open(self.expectation_file) as infile:
186      self.assertEqual(infile.read(), expected_contents)
187
188
189@unittest.skipIf(sys.version_info[0] != 3, 'Python 3-only')
190class IterateThroughResultsWithThresholdsUnittest(
191    fake_filesystem_unittest.TestCase):
192  def setUp(self) -> None:
193    self.setUpPyfakefs()
194    self._expectations = uu.UnitTestExpectationProcessor()
195    self.result_map = {
196        'pixel_integration_test': {
197            'foo_test': {
198                tuple(['win']): ['a'],
199                tuple(['mac']): ['b'],
200            },
201            'bar_test': {
202                tuple(['win']): ['c'],
203            },
204        },
205    }
206
207    self.expectation_file = os.path.join(uu.ABSOLUTE_EXPECTATION_FILE_DIRECTORY,
208                                         'pixel_expectations.txt')
209    uu.CreateFile(self, self.expectation_file)
210    expectation_file_contents = uu.TAG_HEADER + """\
211[ win ] some_test [ Failure ]
212[ mac ] some_test [ Failure ]
213[ android ] some_test [ Failure ]
214"""
215    with open(self.expectation_file, 'w') as outfile:
216      outfile.write(expectation_file_contents)
217
218    self._expectation_file_patcher = mock.patch.object(
219        uu.UnitTestExpectationProcessor, 'GetExpectationFileForSuite')
220    self._expectation_file_mock = self._expectation_file_patcher.start()
221    self._expectation_file_mock.return_value = self.expectation_file
222    self.addCleanup(self._expectation_file_patcher.stop)
223
224  def testGroupByTags(self) -> None:
225    """Tests that threshold-based expectations work when grouping by tags."""
226    result_counts = {
227        tuple(['win']): {
228            # We expect this to be ignored since it has a 1% flake rate.
229            'foo_test': 100,
230            # We expect this to be RetryOnFailure since it has a 25% flake rate.
231            'bar_test': 4,
232        },
233        tuple(['mac']): {
234            # We expect this to be Failure since it has a 50% flake rate.
235            'foo_test': 2
236        }
237    }
238    self._expectations.IterateThroughResultsWithThresholds(
239        self.result_map, True, result_counts, 0.02, 0.5, True)
240    expected_contents = uu.TAG_HEADER + """\
241[ win ] some_test [ Failure ]
242[ win ] bar_test [ RetryOnFailure ]
243[ mac ] some_test [ Failure ]
244[ mac ] foo_test [ Failure ]
245[ android ] some_test [ Failure ]
246"""
247    with open(self.expectation_file) as infile:
248      self.assertEqual(infile.read(), expected_contents)
249
250  def testNoGroupByTags(self) -> None:
251    """Tests that threshold-based expectations work when not grouping by tags"""
252    result_counts = {
253        tuple(['win']): {
254            # We expect this to be ignored since it has a 1% flake rate.
255            'foo_test': 100,
256            # We expect this to be RetryOnFailure since it has a 25% flake rate.
257            'bar_test': 4,
258        },
259        tuple(['mac']): {
260            # We expect this to be Failure since it has a 50% flake rate.
261            'foo_test': 2
262        }
263    }
264    self._expectations.IterateThroughResultsWithThresholds(
265        self.result_map, False, result_counts, 0.02, 0.5, True)
266    expected_contents = uu.TAG_HEADER + """\
267[ win ] some_test [ Failure ]
268[ mac ] some_test [ Failure ]
269[ android ] some_test [ Failure ]
270[ mac ] foo_test [ Failure ]
271[ win ] bar_test [ RetryOnFailure ]
272"""
273    with open(self.expectation_file) as infile:
274      self.assertEqual(infile.read(), expected_contents)
275
276  def testNoIncludeAllTags(self) -> None:
277    """Tests that threshold-based expectations work when filtering tags."""
278    self.result_map = {
279        'pixel_integration_test': {
280            'foo_test': {
281                tuple(['win', 'win10']): ['a'],
282                tuple(['mac']): ['b'],
283            },
284            'bar_test': {
285                tuple(['win', 'win10']): ['c'],
286            },
287        },
288    }
289
290    result_counts = {
291        tuple(['win', 'win10']): {
292            # We expect this to be ignored since it has a 1% flake rate.
293            'foo_test': 100,
294            # We expect this to be RetryOnFailure since it has a 25% flake rate.
295            'bar_test': 4,
296        },
297        tuple(['mac']): {
298            # We expect this to be Failure since it has a 50% flake rate.
299            'foo_test': 2
300        }
301    }
302    self._expectations.IterateThroughResultsWithThresholds(
303        self.result_map, False, result_counts, 0.02, 0.5, False)
304    expected_contents = uu.TAG_HEADER + """\
305[ win ] some_test [ Failure ]
306[ mac ] some_test [ Failure ]
307[ android ] some_test [ Failure ]
308[ mac ] foo_test [ Failure ]
309[ win10 ] bar_test [ RetryOnFailure ]
310"""
311    with open(self.expectation_file) as infile:
312      self.assertEqual(infile.read(), expected_contents)
313
314
315@unittest.skipIf(sys.version_info[0] != 3, 'Python 3-only')
316class CreateExpectationsForAllResultsUnittest(fake_filesystem_unittest.TestCase
317                                              ):
318  def setUp(self) -> None:
319    self.setUpPyfakefs()
320    self._expectations = uu.UnitTestExpectationProcessor()
321    self.result_map = {
322        'pixel_integration_test': {
323            'foo_test': {
324                tuple(['win']): [
325                    ct.ResultTupleType(
326                        ct.ResultStatus.FAIL, 'http://ci.chromium.org/b/1111',
327                        datetime.date.today() - datetime.timedelta(days=2),
328                        False, ['Pass']),
329                    ct.ResultTupleType(
330                        ct.ResultStatus.FAIL, 'http://ci.chromium.org/b/2222',
331                        datetime.date.today() - datetime.timedelta(days=3),
332                        False, ['Pass']),
333                    ct.ResultTupleType(ct.ResultStatus.FAIL,
334                                       'http://ci.chromium.org/b/3333',
335                                       datetime.date.today(), False, ['Pass']),
336                ],
337                tuple(['mac']): [
338                    ct.ResultTupleType(
339                        ct.ResultStatus.FAIL, 'http://ci.chromium.org/b/1111',
340                        datetime.date.today() - datetime.timedelta(days=1),
341                        False, ['Pass']),
342                    ct.ResultTupleType(ct.ResultStatus.FAIL,
343                                       'http://ci.chromium.org/b/2222',
344                                       datetime.date.today(), False, ['Pass']),
345                    ct.ResultTupleType(
346                        ct.ResultStatus.FAIL, 'http://ci.chromium.org/b/3333',
347                        datetime.date.today() - datetime.timedelta(days=3),
348                        False, ['Pass']),
349                ],
350            },
351            'bar_test': {
352                tuple(['win']): [
353                    ct.ResultTupleType(ct.ResultStatus.FAIL,
354                                       'http://ci.chromium.org/b/4444',
355                                       datetime.date.today(), False, ['Pass']),
356                    ct.ResultTupleType(
357                        ct.ResultStatus.FAIL, 'http://ci.chromium.org/b/5555',
358                        datetime.date.today() - datetime.timedelta(days=1),
359                        False, ['Pass']),
360                    ct.ResultTupleType(
361                        ct.ResultStatus.FAIL, 'http://ci.chromium.org/b/6666',
362                        datetime.date.today() - datetime.timedelta(days=2),
363                        False, ['Pass']),
364                ],
365            },
366            'baz_test': {
367                # This test config causes build fail on less than 2 consecutive
368                # days, and thus should not exist in the output.
369                tuple(['win']): [
370                    ct.ResultTupleType(ct.ResultStatus.FAIL,
371                                       'http://ci.chromium.org/b/7777',
372                                       datetime.date.today(), False, ['Pass']),
373                    ct.ResultTupleType(ct.ResultStatus.FAIL,
374                                       'http://ci.chromium.org/b/8888',
375                                       datetime.date.today(), False, ['Pass']),
376                    ct.ResultTupleType(ct.ResultStatus.FAIL,
377                                       'http://ci.chromium.org/b/9999',
378                                       datetime.date.today(), False, ['Pass']),
379                ],
380                tuple(['mac']): [
381                    ct.ResultTupleType(ct.ResultStatus.FAIL,
382                                       'http://ci.chromium.org/b/7777',
383                                       datetime.date.today(), False, ['Pass']),
384                    ct.ResultTupleType(ct.ResultStatus.FAIL,
385                                       'http://ci.chromium.org/b/8888',
386                                       datetime.date.today(), False, ['Pass']),
387                ],
388            },
389            'wpt_test': {
390                # Test for same test in all builders over threshold.
391                tuple(['win']): [
392                    ct.ResultTupleType(ct.ResultStatus.FAIL,
393                                       'http://ci.chromium.org/b/1234',
394                                       datetime.date.today(), False, ['Pass']),
395                ],
396                tuple(['mac']): [
397                    ct.ResultTupleType(
398                        ct.ResultStatus.FAIL, 'http://ci.chromium.org/b/2345',
399                        datetime.date.today() - datetime.timedelta(days=1),
400                        False, ['Pass']),
401                    ct.ResultTupleType(ct.ResultStatus.FAIL,
402                                       'http://ci.chromium.org/b/3456',
403                                       datetime.date.today(), False, ['Pass']),
404                ],
405            },
406        },
407    }
408    self.build_fail_total_number_threshold = 3
409    self.build_fail_consecutive_day_threshold = 2
410    self.build_fail_recent_day_threshold = 1
411
412    self.expectation_file = os.path.join(uu.ABSOLUTE_EXPECTATION_FILE_DIRECTORY,
413                                         'pixel_expectations.txt')
414    uu.CreateFile(self, self.expectation_file)
415    expectation_file_contents = uu.TAG_HEADER + """\
416[ win ] some_test [ Failure Pass ]
417[ mac ] some_test [ Failure Pass ]
418[ android ] some_test [ Failure Pass ]
419"""
420    with open(self.expectation_file, 'w') as outfile:
421      outfile.write(expectation_file_contents)
422
423    self._expectation_file_patcher = mock.patch.object(
424        uu.UnitTestExpectationProcessor, 'GetExpectationFileForSuite')
425    self._expectation_file_mock = self._expectation_file_patcher.start()
426    self._expectation_file_mock.return_value = self.expectation_file
427    self.addCleanup(self._expectation_file_patcher.stop)
428
429  def testGroupByTags(self) -> None:
430    """Tests that threshold-based expectations work when grouping by tags."""
431    self._expectations.CreateExpectationsForAllResults(
432        self.result_map, True, True, self.build_fail_total_number_threshold,
433        self.build_fail_consecutive_day_threshold,
434        self.build_fail_recent_day_threshold)
435    expected_contents = uu.TAG_HEADER + """\
436[ win ] some_test [ Failure Pass ]
437[ win ] foo_test [ Failure Pass ]
438[ win ] bar_test [ Failure Pass ]
439[ win ] wpt_test [ Failure Pass ]
440[ mac ] some_test [ Failure Pass ]
441[ mac ] foo_test [ Failure Pass ]
442[ mac ] wpt_test [ Failure Pass ]
443[ android ] some_test [ Failure Pass ]
444"""
445    with open(self.expectation_file) as infile:
446      self.assertEqual(infile.read(), expected_contents)
447
448  def testNoGroupByTags(self) -> None:
449    """Tests that threshold-based expectations work when not grouping by tags"""
450    self._expectations.CreateExpectationsForAllResults(
451        self.result_map, False, True, self.build_fail_total_number_threshold,
452        self.build_fail_consecutive_day_threshold,
453        self.build_fail_recent_day_threshold)
454    expected_contents = uu.TAG_HEADER + """\
455[ win ] some_test [ Failure Pass ]
456[ mac ] some_test [ Failure Pass ]
457[ android ] some_test [ Failure Pass ]
458[ win ] foo_test [ Failure Pass ]
459[ mac ] foo_test [ Failure Pass ]
460[ win ] bar_test [ Failure Pass ]
461[ win ] wpt_test [ Failure Pass ]
462[ mac ] wpt_test [ Failure Pass ]
463"""
464    with open(self.expectation_file) as infile:
465      self.assertEqual(infile.read(), expected_contents)
466
467  def testNoIncludeAllTags(self) -> None:
468    """Tests that threshold-based expectations work when filtering tags."""
469    self.result_map = {
470        'pixel_integration_test': {
471            'foo_test': {
472                tuple(['win', 'win10']): [
473                    ct.ResultTupleType(
474                        ct.ResultStatus.FAIL, 'http://ci.chromium.org/b/1111',
475                        datetime.date.today() - datetime.timedelta(days=2),
476                        False, ['Pass']),
477                    ct.ResultTupleType(
478                        ct.ResultStatus.FAIL, 'http://ci.chromium.org/b/2222',
479                        datetime.date.today() - datetime.timedelta(days=3),
480                        False, ['Pass']),
481                    ct.ResultTupleType(ct.ResultStatus.FAIL,
482                                       'http://ci.chromium.org/b/3333',
483                                       datetime.date.today(), False, ['Pass']),
484                ],
485                tuple(['mac']): [
486                    ct.ResultTupleType(
487                        ct.ResultStatus.FAIL, 'http://ci.chromium.org/b/1111',
488                        datetime.date.today() - datetime.timedelta(days=1),
489                        False, ['Pass']),
490                    ct.ResultTupleType(ct.ResultStatus.FAIL,
491                                       'http://ci.chromium.org/b/2222',
492                                       datetime.date.today(), False, ['Pass']),
493                    ct.ResultTupleType(
494                        ct.ResultStatus.FAIL, 'http://ci.chromium.org/b/3333',
495                        datetime.date.today() - datetime.timedelta(days=3),
496                        False, ['Pass']),
497                ],
498            },
499            'bar_test': {
500                tuple(['win', 'win10']): [
501                    ct.ResultTupleType(ct.ResultStatus.FAIL,
502                                       'http://ci.chromium.org/b/4444',
503                                       datetime.date.today(), False, ['Pass']),
504                    ct.ResultTupleType(
505                        ct.ResultStatus.FAIL, 'http://ci.chromium.org/b/5555',
506                        datetime.date.today() - datetime.timedelta(days=1),
507                        False, ['Pass']),
508                    ct.ResultTupleType(
509                        ct.ResultStatus.FAIL, 'http://ci.chromium.org/b/6666',
510                        datetime.date.today() - datetime.timedelta(days=2),
511                        False, ['Pass']),
512                ],
513            },
514            'baz_test': {
515                # This test config causes build fail on less than 2 consecutive
516                # days, and thus should not exist in the output.
517                tuple(['win']): [
518                    ct.ResultTupleType(ct.ResultStatus.FAIL,
519                                       'http://ci.chromium.org/b/7777',
520                                       datetime.date.today(), False, ['Pass']),
521                    ct.ResultTupleType(ct.ResultStatus.FAIL,
522                                       'http://ci.chromium.org/b/8888',
523                                       datetime.date.today(), False, ['Pass']),
524                    ct.ResultTupleType(ct.ResultStatus.FAIL,
525                                       'http://ci.chromium.org/b/9999',
526                                       datetime.date.today(), False, ['Pass']),
527                ],
528                tuple(['mac']): [
529                    ct.ResultTupleType(ct.ResultStatus.FAIL,
530                                       'http://ci.chromium.org/b/7777',
531                                       datetime.date.today(), False, ['Pass']),
532                    ct.ResultTupleType(ct.ResultStatus.FAIL,
533                                       'http://ci.chromium.org/b/8888',
534                                       datetime.date.today(), False, ['Pass']),
535                ],
536            },
537        },
538    }
539    self._expectations.CreateExpectationsForAllResults(
540        self.result_map, False, False, self.build_fail_total_number_threshold,
541        self.build_fail_consecutive_day_threshold,
542        self.build_fail_recent_day_threshold)
543    expected_contents = uu.TAG_HEADER + """\
544[ win ] some_test [ Failure Pass ]
545[ mac ] some_test [ Failure Pass ]
546[ android ] some_test [ Failure Pass ]
547[ win10 ] foo_test [ Failure Pass ]
548[ mac ] foo_test [ Failure Pass ]
549[ win10 ] bar_test [ Failure Pass ]
550"""
551    with open(self.expectation_file) as infile:
552      self.assertEqual(infile.read(), expected_contents)
553
554
555@unittest.skipIf(sys.version_info[0] != 3, 'Python 3-only')
556class FindFailuresInSameConditionUnittest(unittest.TestCase):
557  def setUp(self) -> None:
558    self._expectations = uu.UnitTestExpectationProcessor()
559    self.result_map = {
560        'pixel_integration_test': {
561            'foo_test': {
562                tuple(['win']): ['a'],
563                tuple(['mac']): ['a', 'b'],
564            },
565            'bar_test': {
566                tuple(['win']): ['a', 'b', 'c'],
567                tuple(['mac']): ['a', 'b', 'c', 'd'],
568            },
569        },
570        'webgl_conformance_integration_test': {
571            'foo_test': {
572                tuple(['win']): ['a', 'b', 'c', 'd', 'e'],
573                tuple(['mac']): ['a', 'b', 'c', 'd', 'e', 'f'],
574            },
575            'bar_test': {
576                tuple(['win']): ['a', 'b', 'c', 'd', 'e', 'f', 'g'],
577                tuple(['mac']): ['a', 'b', 'c', 'd', 'e', 'f', 'g', 'h'],
578            },
579        },
580    }
581
582  def testFindFailuresInSameTest(self) -> None:
583    other_failures = self._expectations.FindFailuresInSameTest(
584        self.result_map, 'pixel_integration_test', 'foo_test', tuple(['win']))
585    self.assertEqual(other_failures, [(tuple(['mac']), 2)])
586
587  def testFindFailuresInSameConfig(self) -> None:
588    typ_tag_ordered_result_map = self._expectations._ReorderMapByTypTags(
589        self.result_map)
590    other_failures = self._expectations.FindFailuresInSameConfig(
591        typ_tag_ordered_result_map, 'pixel_integration_test', 'foo_test',
592        tuple(['win']))
593    expected_other_failures = [
594        ('pixel_integration_test.bar_test', 3),
595        ('webgl_conformance_integration_test.foo_test', 5),
596        ('webgl_conformance_integration_test.bar_test', 7),
597    ]
598    self.assertEqual(len(other_failures), len(expected_other_failures))
599    self.assertEqual(set(other_failures), set(expected_other_failures))
600
601
602@unittest.skipIf(sys.version_info[0] != 3, 'Python 3-only')
603class ModifyFileForResultUnittest(fake_filesystem_unittest.TestCase):
604  def setUp(self) -> None:
605    self.setUpPyfakefs()
606    self._expectations = uu.UnitTestExpectationProcessor()
607    self.expectation_file = os.path.join(uu.ABSOLUTE_EXPECTATION_FILE_DIRECTORY,
608                                         'expectation.txt')
609    uu.CreateFile(self, self.expectation_file)
610    self._expectation_file_patcher = mock.patch.object(
611        uu.UnitTestExpectationProcessor, 'GetExpectationFileForSuite')
612    self._expectation_file_mock = self._expectation_file_patcher.start()
613    self.addCleanup(self._expectation_file_patcher.stop)
614    self._expectation_file_mock.return_value = self.expectation_file
615
616  def testNoGroupByTags(self) -> None:
617    """Tests that not grouping by tags appends to the end."""
618    expectation_file_contents = uu.TAG_HEADER + """\
619[ win ] some_test [ Failure ]
620
621[ mac ] some_test [ Failure ]
622"""
623    with open(self.expectation_file, 'w') as outfile:
624      outfile.write(expectation_file_contents)
625    self._expectations.ModifyFileForResult('some_file', 'some_test',
626                                           ('win', 'win10'), '', 'Failure',
627                                           False, True)
628    expected_contents = uu.TAG_HEADER + """\
629[ win ] some_test [ Failure ]
630
631[ mac ] some_test [ Failure ]
632[ win win10 ] some_test [ Failure ]
633"""
634    with open(self.expectation_file) as infile:
635      self.assertEqual(infile.read(), expected_contents)
636
637  def testGroupByTagsNoMatch(self) -> None:
638    """Tests that grouping by tags but finding no match appends to the end."""
639    expectation_file_contents = uu.TAG_HEADER + """\
640[ mac ] some_test [ Failure ]
641"""
642    with open(self.expectation_file, 'w') as outfile:
643      outfile.write(expectation_file_contents)
644    self._expectations.ModifyFileForResult('some_file', 'some_test',
645                                           ('win', 'win10'), '', 'Failure',
646                                           True, True)
647    expected_contents = uu.TAG_HEADER + """\
648[ mac ] some_test [ Failure ]
649[ win win10 ] some_test [ Failure ]
650"""
651    with open(self.expectation_file) as infile:
652      self.assertEqual(infile.read(), expected_contents)
653
654  def testGroupByTagsMatch(self) -> None:
655    """Tests that grouping by tags and finding a match adds mid-file."""
656    expectation_file_contents = uu.TAG_HEADER + """\
657[ win ] some_test [ Failure ]
658
659[ mac ] some_test [ Failure ]
660"""
661    with open(self.expectation_file, 'w') as outfile:
662      outfile.write(expectation_file_contents)
663    self._expectations.ModifyFileForResult('some_file', 'foo_test',
664                                           ('win', 'win10'), '', 'Failure',
665                                           True, True)
666    expected_contents = uu.TAG_HEADER + """\
667[ win ] some_test [ Failure ]
668[ win ] foo_test [ Failure ]
669
670[ mac ] some_test [ Failure ]
671"""
672    with open(self.expectation_file) as infile:
673      self.assertEqual(infile.read(), expected_contents)
674
675
676@unittest.skipIf(sys.version_info[0] != 3, 'Python 3-only')
677class FilterToMostSpecificTagTypeUnittest(fake_filesystem_unittest.TestCase):
678  def setUp(self) -> None:
679    self._expectations = uu.UnitTestExpectationProcessor()
680    self.setUpPyfakefs()
681    with tempfile.NamedTemporaryFile(delete=False) as tf:
682      self.expectation_file = tf.name
683
684  def testBasic(self):
685    """Tests that only the most specific tags are kept."""
686    expectation_file_contents = """\
687# tags: [ tag1_least_specific tag1_middle_specific tag1_most_specific ]
688# tags: [ tag2_least_specific tag2_middle_specific tag2_most_specific ]"""
689    with open(self.expectation_file, 'w') as outfile:
690      outfile.write(expectation_file_contents)
691
692    tags = ('tag1_least_specific', 'tag1_most_specific', 'tag2_middle_specific',
693            'tag2_least_specific')
694    filtered_tags = self._expectations.FilterToMostSpecificTypTags(
695        tags, self.expectation_file)
696    self.assertEqual(filtered_tags,
697                     ('tag1_most_specific', 'tag2_middle_specific'))
698
699  def testSingleTags(self) -> None:
700    """Tests that functionality works as expected with single tags."""
701    expectation_file_contents = """\
702# tags: [ tag1_most_specific ]
703# tags: [ tag2_most_specific ]"""
704    with open(self.expectation_file, 'w') as outfile:
705      outfile.write(expectation_file_contents)
706
707    tags = ('tag1_most_specific', 'tag2_most_specific')
708    filtered_tags = self._expectations.FilterToMostSpecificTypTags(
709        tags, self.expectation_file)
710    self.assertEqual(filtered_tags, tags)
711
712  def testUnusedTags(self) -> None:
713    """Tests that functionality works as expected with extra/unused tags."""
714    expectation_file_contents = """\
715# tags: [ tag1_least_specific tag1_middle_specific tag1_most_specific ]
716# tags: [ tag2_least_specific tag2_middle_specific tag2_most_specific ]
717# tags: [ some_unused_tag ]"""
718    with open(self.expectation_file, 'w') as outfile:
719      outfile.write(expectation_file_contents)
720
721    tags = ('tag1_least_specific', 'tag1_most_specific', 'tag2_middle_specific',
722            'tag2_least_specific')
723    filtered_tags = self._expectations.FilterToMostSpecificTypTags(
724        tags, self.expectation_file)
725    self.assertEqual(filtered_tags,
726                     ('tag1_most_specific', 'tag2_middle_specific'))
727
728  def testMultiline(self) -> None:
729    """Tests that functionality works when tags cover multiple lines."""
730    expectation_file_contents = """\
731# tags: [ tag1_least_specific
732#         tag1_middle_specific
733#         tag1_most_specific ]
734# tags: [ tag2_least_specific
735#         tag2_middle_specific tag2_most_specific ]"""
736    with open(self.expectation_file, 'w') as outfile:
737      outfile.write(expectation_file_contents)
738
739    tags = ('tag1_least_specific', 'tag1_middle_specific', 'tag1_most_specific',
740            'tag2_middle_specific', 'tag2_least_specific')
741    filtered_tags = self._expectations.FilterToMostSpecificTypTags(
742        tags, self.expectation_file)
743    self.assertEqual(filtered_tags,
744                     ('tag1_most_specific', 'tag2_middle_specific'))
745
746  def testMissingTags(self) -> None:
747    """Tests that a file not having all tags is an error."""
748    expectation_file_contents = """\
749# tags: [ tag1_least_specific tag1_middle_specific ]
750# tags: [ tag2_least_specific tag2_middle_specific tag2_most_specific ]"""
751    with open(self.expectation_file, 'w') as outfile:
752      outfile.write(expectation_file_contents)
753
754    tags = ('tag1_least_specific', 'tag1_most_specific', 'tag2_middle_specific',
755            'tag2_least_specific')
756    with self.assertRaises(RuntimeError):
757      self._expectations.FilterToMostSpecificTypTags(tags,
758                                                     self.expectation_file)
759
760
761@unittest.skipIf(sys.version_info[0] != 3, 'Python 3-only')
762class FindBestInsertionLineForExpectationUnittest(
763    fake_filesystem_unittest.TestCase):
764  def setUp(self) -> None:
765    self.setUpPyfakefs()
766    self._expectations = uu.UnitTestExpectationProcessor()
767    self.expectation_file = os.path.join(uu.ABSOLUTE_EXPECTATION_FILE_DIRECTORY,
768                                         'expectation.txt')
769    uu.CreateFile(self, self.expectation_file)
770    expectation_file_contents = uu.TAG_HEADER + """\
771[ win ] some_test [ Failure ]
772
773[ mac ] some_test [ Failure ]
774
775[ win release ] bar_test [ Failure ]
776[ win ] foo_test [ Failure ]
777
778[ chromeos ] some_test [ Failure ]
779"""
780    with open(self.expectation_file, 'w') as outfile:
781      outfile.write(expectation_file_contents)
782
783  def testNoMatchingTags(self) -> None:
784    """Tests behavior when there are no expectations with matching tags."""
785    insertion_line, tags = (
786        self._expectations.FindBestInsertionLineForExpectation(
787            tuple(['android']), self.expectation_file))
788    self.assertEqual(insertion_line, -1)
789    self.assertEqual(tags, set())
790
791  def testMatchingTagsLastEntryChosen(self) -> None:
792    """Tests that the last matching line is chosen."""
793    insertion_line, tags = (
794        self._expectations.FindBestInsertionLineForExpectation(
795            tuple(['win']), self.expectation_file))
796    # We expect "[ win ] foo_test [ Failure ]" to be chosen
797    expected_line = len(uu.TAG_HEADER.splitlines()) + 6
798    self.assertEqual(insertion_line, expected_line)
799    self.assertEqual(tags, set(['win']))
800
801  def testMatchingTagsClosestMatchChosen(self) -> None:
802    """Tests that the closest tag match is chosen."""
803    insertion_line, tags = (
804        self._expectations.FindBestInsertionLineForExpectation(
805            ('win', 'release'), self.expectation_file))
806    # We expect "[ win release ] bar_test [ Failure ]" to be chosen
807    expected_line = len(uu.TAG_HEADER.splitlines()) + 5
808    self.assertEqual(insertion_line, expected_line)
809    self.assertEqual(tags, set(['win', 'release']))
810
811
812class AssertCheckoutIsUpToDateUnittest(unittest.TestCase):
813  def setUp(self) -> None:
814    self._expectations = uu.UnitTestExpectationProcessor()
815    self._origin_patcher = mock.patch(
816        'flake_suppressor_common.expectations.ExpectationProcessor.'
817        'GetOriginExpectationFileContents')
818    self._origin_mock = self._origin_patcher.start()
819    self.addCleanup(self._origin_patcher.stop)
820    self._local_patcher = mock.patch(
821        'flake_suppressor_common.expectations.' +
822        'ExpectationProcessor.GetLocalCheckoutExpectationFileContents')
823    self._local_mock = self._local_patcher.start()
824    self.addCleanup(self._local_patcher.stop)
825
826  def testContentsMatch(self) -> None:
827    """Tests the happy path where the contents match."""
828    self._origin_mock.return_value = {
829        'foo.txt': 'foo_content',
830        'bar.txt': 'bar_content',
831    }
832    self._local_mock.return_value = {
833        'bar.txt': 'bar_content',
834        'foo.txt': 'foo_content',
835    }
836    self._expectations.AssertCheckoutIsUpToDate()
837
838  def testContentsDoNotMatch(self) -> None:
839    """Tests that mismatched contents results in a failure."""
840    self._origin_mock.return_value = {
841        'foo.txt': 'foo_content',
842        'bar.txt': 'bar_content',
843    }
844    # Differing keys.
845    self._local_mock.return_value = {
846        'bar.txt': 'bar_content',
847        'foo2.txt': 'foo_content',
848    }
849    with self.assertRaises(RuntimeError):
850      self._expectations.AssertCheckoutIsUpToDate()
851
852    # Differing values.
853    self._local_mock.return_value = {
854        'bar.txt': 'bar_content',
855        'foo.txt': 'foo_content2',
856    }
857    with self.assertRaises(RuntimeError):
858      self._expectations.AssertCheckoutIsUpToDate()
859
860
861class OverFailedBuildThresholdUnittest(unittest.TestCase):
862  def setUp(self) -> None:
863    self.build_fail_total_number_threshold = 3
864
865  def testOverThreshold(self) -> None:
866    """Tests functionality when |result_tuple_list| passes
867    |build_fail_total_number_threshold|.
868
869    True is expected output on these inputs.
870    """
871    result_tuple_list = [
872        ct.ResultTupleType(ct.ResultStatus.FAIL,
873                           'http://ci.chromium.org/b/1111',
874                           datetime.date(2021, 1, 1), False, ['Pass']),
875        ct.ResultTupleType(ct.ResultStatus.FAIL,
876                           'http://ci.chromium.org/b/2222',
877                           datetime.date(2022, 1, 1), False, ['Pass']),
878        ct.ResultTupleType(ct.ResultStatus.FAIL,
879                           'http://ci.chromium.org/b/3333',
880                           datetime.date(2023, 1, 1), False, ['Pass']),
881    ]
882    self.assertTrue(
883        expectations.OverFailedBuildThreshold(
884            result_tuple_list, self.build_fail_total_number_threshold))
885
886  def testUnderThreshold(self) -> None:
887    """Tests functionality when |result_tuple_list| cannot pass
888       |build_fail_total_number_threshold|.
889
890    False is expected output on these inputs.
891    """
892    result_tuple_list = [
893        ct.ResultTupleType(ct.ResultStatus.FAIL,
894                           'http://ci.chromium.org/b/1111',
895                           datetime.date(2022, 1, 1), False, ['Pass']),
896        ct.ResultTupleType(ct.ResultStatus.FAIL,
897                           'http://ci.chromium.org/b/2222',
898                           datetime.date(2022, 1, 2), False, ['Pass']),
899    ]
900    self.assertFalse(
901        expectations.OverFailedBuildThreshold(
902            result_tuple_list, self.build_fail_total_number_threshold))
903
904    result_tuple_list = [
905        ct.ResultTupleType(ct.ResultStatus.FAIL,
906                           'http://ci.chromium.org/b/1111',
907                           datetime.date(2022, 1, 1), False, ['Pass']),
908        ct.ResultTupleType(ct.ResultStatus.FAIL,
909                           'http://ci.chromium.org/b/2222',
910                           datetime.date(2022, 1, 2), False, ['Pass']),
911        ct.ResultTupleType(ct.ResultStatus.FAIL,
912                           'http://ci.chromium.org/b/2222',
913                           datetime.date(2022, 1, 3), False, ['Pass']),
914    ]
915    self.assertFalse(
916        expectations.OverFailedBuildThreshold(
917            result_tuple_list, self.build_fail_total_number_threshold))
918
919
920class OverFailedBuildByConsecutiveDayThresholdUnittest(unittest.TestCase):
921  def setUp(self) -> None:
922    self.build_fail_consecutive_day_threshold = 3
923
924  def testOverThreshold(self) -> None:
925    """Tests functionality when |result_tuple_list| passes
926       |build_fail_consecutive_day_threshold|.
927
928    True is expected output on these inputs.
929    """
930    result_tuple_list = [
931        ct.ResultTupleType(ct.ResultStatus.FAIL,
932                           'http://ci.chromium.org/b/1111',
933                           datetime.date(2022, 1, 2), False, ['Pass']),
934        ct.ResultTupleType(ct.ResultStatus.FAIL,
935                           'http://ci.chromium.org/b/2222',
936                           datetime.date(2022, 1, 1), False, ['Pass']),
937        ct.ResultTupleType(ct.ResultStatus.FAIL,
938                           'http://ci.chromium.org/b/3333',
939                           datetime.date(2022, 1, 3), False, ['Pass']),
940    ]
941    self.assertTrue(
942        expectations.OverFailedBuildByConsecutiveDayThreshold(
943            result_tuple_list, self.build_fail_consecutive_day_threshold))
944
945  def testUnderThreshold(self) -> None:
946    """Tests functionality when |result_tuple_list| cannot pass
947       |build_fail_consecutive_day_threshold|.
948
949    False is expected output on these inputs.
950    """
951    result_tuple_list = [
952        ct.ResultTupleType(ct.ResultStatus.FAIL,
953                           'http://ci.chromium.org/b/1111',
954                           datetime.date(2022, 1, 1), False, ['Pass']),
955        ct.ResultTupleType(ct.ResultStatus.FAIL,
956                           'http://ci.chromium.org/b/2222',
957                           datetime.date(2022, 1, 1), False, ['Pass']),
958        ct.ResultTupleType(ct.ResultStatus.FAIL,
959                           'http://ci.chromium.org/b/3333',
960                           datetime.date(2022, 1, 1), False, ['Pass']),
961    ]
962    self.assertFalse(
963        expectations.OverFailedBuildByConsecutiveDayThreshold(
964            result_tuple_list, self.build_fail_consecutive_day_threshold))
965
966    result_tuple_list = [
967        ct.ResultTupleType(ct.ResultStatus.FAIL,
968                           'http://ci.chromium.org/b/1111',
969                           datetime.date(2022, 1, 1), False, ['Pass']),
970        ct.ResultTupleType(ct.ResultStatus.FAIL,
971                           'http://ci.chromium.org/b/2222',
972                           datetime.date(2022, 1, 2), False, ['Pass']),
973    ]
974    self.assertFalse(
975        expectations.OverFailedBuildByConsecutiveDayThreshold(
976            result_tuple_list, self.build_fail_consecutive_day_threshold))
977
978    result_tuple_list = [
979        ct.ResultTupleType(ct.ResultStatus.FAIL,
980                           'http://ci.chromium.org/b/1111',
981                           datetime.date(2022, 1, 1), False, ['Pass']),
982        ct.ResultTupleType(ct.ResultStatus.FAIL,
983                           'http://ci.chromium.org/b/2222',
984                           datetime.date(2022, 1, 2), False, ['Pass']),
985        ct.ResultTupleType(ct.ResultStatus.FAIL,
986                           'http://ci.chromium.org/b/3333',
987                           datetime.date(2022, 1, 4), False, ['Pass']),
988    ]
989    self.assertFalse(
990        expectations.OverFailedBuildByConsecutiveDayThreshold(
991            result_tuple_list, self.build_fail_consecutive_day_threshold))
992
993
994class FailedBuildWithinRecentDayThresholdUnittest(unittest.TestCase):
995  def setUp(self) -> None:
996    self.build_fail_recent_day_threshold = 3
997
998  def testWithinThreshold(self) -> None:
999    """Tests functionality when |result_tuple_list| has build fail within
1000       |build_fail_recent_day_threshold|.
1001
1002    True is expected output on these inputs.
1003    """
1004    result_tuple_list = [
1005        ct.ResultTupleType(ct.ResultStatus.FAIL,
1006                           'http://ci.chromium.org/b/1111',
1007                           datetime.date.today(), False, ['Pass']),
1008        ct.ResultTupleType(ct.ResultStatus.FAIL,
1009                           'http://ci.chromium.org/b/2222',
1010                           datetime.date.today(), False, ['Pass']),
1011        ct.ResultTupleType(ct.ResultStatus.FAIL,
1012                           'http://ci.chromium.org/b/3333',
1013                           datetime.date.today(), False, ['Pass']),
1014    ]
1015    self.assertTrue(
1016        expectations.FailedBuildWithinRecentDayThreshold(
1017            result_tuple_list, self.build_fail_recent_day_threshold))
1018
1019  def testBeyondThreshold(self) -> None:
1020    """Tests functionality when |result_tuple_list| has no build fail within
1021       |build_fail_recent_day_threshold|.
1022
1023    False is expected output on these inputs.
1024    """
1025    result_tuple_list = [
1026        ct.ResultTupleType(ct.ResultStatus.FAIL,
1027                           'http://ci.chromium.org/b/1111',
1028                           datetime.date(2022, 1, 1), False, ['Pass']),
1029        ct.ResultTupleType(ct.ResultStatus.FAIL,
1030                           'http://ci.chromium.org/b/2222',
1031                           datetime.date(2022, 1, 1), False, ['Pass']),
1032        ct.ResultTupleType(ct.ResultStatus.FAIL,
1033                           'http://ci.chromium.org/b/3333',
1034                           datetime.date(2022, 1, 1), False, ['Pass']),
1035    ]
1036    self.assertFalse(
1037        expectations.FailedBuildWithinRecentDayThreshold(
1038            result_tuple_list, self.build_fail_recent_day_threshold))
1039
1040
1041if __name__ == '__main__':
1042  unittest.main(verbosity=2)
1043