xref: /aosp_15_r20/external/cronet/testing/unexpected_passes_common/unittest_utils.py (revision 6777b5387eb2ff775bb5750e3f5d96f37fb7352b) 
1# Copyright 2020 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"""Helper methods for unittests."""
5
6from __future__ import print_function
7
8from typing import Iterable, List, Optional, Set, Tuple, Type
9import unittest.mock as mock
10
11from unexpected_passes_common import builders
12from unexpected_passes_common import expectations
13from unexpected_passes_common import data_types
14from unexpected_passes_common import queries as queries_module
15
16
17def CreateStatsWithPassFails(passes: int, fails: int) -> data_types.BuildStats:
18  stats = data_types.BuildStats()
19  for _ in range(passes):
20    stats.AddPassedBuild(frozenset())
21  for i in range(fails):
22    stats.AddFailedBuild('build_id%d' % i, frozenset())
23  return stats
24
25
26def _CreateSimpleQueries(clauses: Iterable[str]) -> List[str]:
27  queries = []
28  # Not actually a valid query since we don't specify the table, but it works.
29  for c in clauses:
30    queries.append("""\
31SELECT *
32WHERE %s
33""" % c)
34  return queries
35
36
37class SimpleFixedQueryGenerator(queries_module.FixedQueryGenerator):
38  def GetQueries(self) -> List[str]:
39    return _CreateSimpleQueries(self.GetClauses())
40
41
42class SimpleSplitQueryGenerator(queries_module.SplitQueryGenerator):
43  def GetQueries(self) -> List[str]:
44    return _CreateSimpleQueries(self.GetClauses())
45
46
47class SimpleBigQueryQuerier(queries_module.BigQueryQuerier):
48  def _GetQueryGeneratorForBuilder(self, builder: data_types.BuilderEntry
49                                   ) -> queries_module.BaseQueryGenerator:
50    if not self._large_query_mode:
51      return SimpleFixedQueryGenerator(builder, 'AND True')
52    return SimpleSplitQueryGenerator(builder, ['test_id'], 200)
53
54  def _GetRelevantExpectationFilesForQueryResult(self, _) -> None:
55    return None
56
57  def _StripPrefixFromTestId(self, test_id: str) -> str:
58    return test_id.split('.')[-1]
59
60  def _GetActiveBuilderQuery(self, _, __) -> str:
61    return ''
62
63
64def CreateGenericQuerier(
65    suite: Optional[str] = None,
66    project: Optional[str] = None,
67    num_samples: Optional[int] = None,
68    large_query_mode: Optional[bool] = None,
69    num_jobs: Optional[int] = None,
70    use_batching: bool = True,
71    cls: Optional[Type[queries_module.BigQueryQuerier]] = None
72) -> queries_module.BigQueryQuerier:
73  suite = suite or 'pixel'
74  project = project or 'project'
75  num_samples = num_samples or 5
76  large_query_mode = large_query_mode or False
77  cls = cls or SimpleBigQueryQuerier
78  return cls(suite, project, num_samples, large_query_mode, num_jobs,
79             use_batching)
80
81
82def GetArgsForMockCall(call_args_list: List[tuple],
83                       call_number: int) -> Tuple[tuple, dict]:
84  """Helper to more sanely get call args from a mocked method.
85
86  Args:
87    call_args_list: The call_args_list member from the mock in question.
88    call_number: The call number to pull args from, starting at 0 for the first
89        call to the method.
90
91  Returns:
92    A tuple (args, kwargs). |args| is a list of arguments passed to the method.
93    |kwargs| is a dict containing the keyword arguments padded to the method.
94  """
95  args = call_args_list[call_number][0]
96  kwargs = call_args_list[call_number][1]
97  return args, kwargs
98
99
100class FakeProcess():
101  """A fake subprocess Process object."""
102
103  def __init__(self,
104               returncode: Optional[int] = None,
105               stdout: Optional[str] = None,
106               stderr: Optional[str] = None,
107               finish: bool = True):
108    if finish:
109      self.returncode = returncode or 0
110    else:
111      self.returncode = None
112    self.stdout = stdout or ''
113    self.stderr = stderr or ''
114    self.finish = finish
115
116  def communicate(self, _) -> Tuple[str, str]:
117    return self.stdout, self.stderr
118
119  def terminate(self) -> None:
120    if self.finish:
121      raise OSError('Tried to terminate a finished process')
122
123
124class GenericBuilders(builders.Builders):
125  #pylint: disable=useless-super-delegation
126  def __init__(self,
127               suite: Optional[str] = None,
128               include_internal_builders: bool = False):
129    super().__init__(suite, include_internal_builders)
130  #pylint: enable=useless-super-delegation
131
132  def _BuilderRunsTestOfInterest(self, _test_map) -> bool:
133    return True
134
135  def GetIsolateNames(self) -> dict:
136    return {}
137
138  def GetFakeCiBuilders(self) -> dict:
139    return {}
140
141  def GetNonChromiumBuilders(self) -> dict:
142    return {}
143
144
145def RegisterGenericBuildersImplementation() -> None:
146  builders.RegisterInstance(GenericBuilders())
147
148
149class GenericExpectations(expectations.Expectations):
150  def GetExpectationFilepaths(self) -> list:
151    return []
152
153  def _GetExpectationFileTagHeader(self, _) -> str:
154    return """\
155# tags: [ linux mac win ]
156# tags: [ amd intel nvidia ]
157# results: [ Failure RetryOnFailure Skip Pass ]
158"""
159
160  def _GetKnownTags(self) -> Set[str]:
161    return set(['linux', 'mac', 'win', 'amd', 'intel', 'nvidia'])
162
163
164def CreateGenericExpectations() -> GenericExpectations:
165  return GenericExpectations()
166
167
168def RegisterGenericExpectationsImplementation() -> None:
169  expectations.RegisterInstance(CreateGenericExpectations())
170