• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1# Copyright 2021 The Chromium Authors
2# Use of this source code is governed by a BSD-style license that can be
3# found in the LICENSE file.
4
5import os
6
7from typing import List, Tuple, Iterable
8
9from pyfakefs import fake_filesystem_unittest
10
11from flake_suppressor_common import common_typing as ct
12from flake_suppressor_common import expectations as expectations_module
13from flake_suppressor_common import queries
14from flake_suppressor_common import results as results_module
15from flake_suppressor_common import tag_utils
16
17
18CHROMIUM_SRC_DIR = os.path.realpath(
19    os.path.join(os.path.dirname(__file__), '..', '..'))
20RELATIVE_EXPECTATION_FILE_DIRECTORY = os.path.join('content', 'test', 'gpu',
21                                                   'gpu_tests',
22                                                   'test_expectations')
23ABSOLUTE_EXPECTATION_FILE_DIRECTORY = os.path.join(
24    CHROMIUM_SRC_DIR, RELATIVE_EXPECTATION_FILE_DIRECTORY)
25
26TAG_HEADER = """\
27# OS
28# tags: [ android android-lollipop android-marshmallow android-nougat
29#             android-pie android-r android-s android-t
30#         chromeos
31#         fuchsia
32#         linux ubuntu
33#         mac highsierra mojave catalina bigsur monterey
34#         win win8 win10 ]
35# Browser
36# tags: [ android-chromium android-webview-instrumentation
37#         debug debug-x64
38#         release release-x64
39#         fuchsia-chrome web-engine-shell ]
40# results: [ Failure RetryOnFailure Pass Skip Slow ]
41"""
42
43
44def CreateFile(test: fake_filesystem_unittest.TestCase, *args,
45               **kwargs) -> None:
46  # TODO(crbug.com/40160566): Remove this and just use fs.create_file() when
47  # Catapult is updated to a newer version of pyfakefs that is compatible with
48  # Chromium's version.
49  if hasattr(test.fs, 'create_file'):
50    test.fs.create_file(*args, **kwargs)
51  else:
52    test.fs.CreateFile(*args, **kwargs)
53
54
55class FakeProcess():
56  def __init__(self, stdout: str):
57    self.stdout = stdout or ''
58
59
60class UnitTest_BigQueryQuerier(queries.BigQueryQuerier):
61  def GetResultCountCIQuery(self) -> str:
62    return """SELECT * FROM foo"""
63
64  def GetResultCountTryQuery(self) -> str:
65    return """submitted_builds SELECT * FROM bar"""
66
67  def GetFlakyOrFailingCiQuery(self) -> str:
68    return """SELECT * FROM foo"""
69
70  def GetFlakyOrFailingTryQuery(self) -> str:
71    return """submitted_builds SELECT * FROM bar"""
72
73  def GetFailingBuildCulpritFromCiQuery(self) -> str:
74    raise NotImplementedError()
75
76
77class UnitTestResultProcessor(results_module.ResultProcessor):
78  def GetTestSuiteAndNameFromResultDbName(self, result_db_name: str
79                                          ) -> Tuple[str, str]:
80    _, suite, __, test_name = result_db_name.split('.', 3)
81    return suite, test_name
82
83
84class UnitTestTagUtils(tag_utils.BaseTagUtils):
85  def RemoveIgnoredTags(self, tags: Iterable[str]) -> ct.TagTupleType:
86    tags = list(set(tags) - set(['win-laptop']))
87    tags.sort()
88    return tuple(tags)
89
90
91# pylint: disable=unused-argument
92class UnitTestExpectationProcessor(expectations_module.ExpectationProcessor):
93  def GetExpectationFileForSuite(self, suite: str,
94                                 typ_tags: ct.TagTupleType) -> str:
95    filename = suite.replace('integration_test', 'expectations.txt')
96    return os.path.join(ABSOLUTE_EXPECTATION_FILE_DIRECTORY, filename)
97
98  def IsSuiteUnsupported(self, suite) -> bool:
99    return False
100
101  def GetExpectedResult(self, fraction: float, flaky_threshold: float) -> str:
102    if fraction < flaky_threshold:
103      return 'RetryOnFailure'
104    return 'Failure'
105
106  def ListLocalCheckoutExpectationFiles(self) -> List[str]:
107    raise NotImplementedError()
108
109  def ListOriginExpectationFiles(self) -> List[str]:
110    raise NotImplementedError()
111
112# pylint: enable=unused-argument
113