• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
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 Any, Callable, 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 FakePool():
101  """A fake pathos.pools.ProcessPool instance.
102
103  Real pools don't like being given MagicMocks, so this allows testing of
104  code that uses pathos.pools.ProcessPool by returning this from
105  multiprocessing_utils.GetProcessPool().
106  """
107
108  def map(self, f: Callable[[Any], Any], inputs: Iterable[Any]) -> List[Any]:
109    retval = []
110    for i in inputs:
111      retval.append(f(i))
112    return retval
113
114  def apipe(self, f: Callable[[Any], Any],
115            inputs: Iterable[Any]) -> 'FakeAsyncResult':
116    return FakeAsyncResult(f(inputs))
117
118  def close(self) -> None:
119    pass
120
121  def join(self) -> None:
122    pass
123
124
125class FakeAsyncResult():
126  """A fake AsyncResult like the one from multiprocessing or pathos."""
127
128  def __init__(self, result: Any):
129    self._result = result
130
131  def ready(self) -> bool:
132    return True
133
134  def get(self) -> Any:
135    return self._result
136
137
138class FakeProcess():
139  """A fake subprocess Process object."""
140
141  def __init__(self,
142               returncode: Optional[int] = None,
143               stdout: Optional[str] = None,
144               stderr: Optional[str] = None,
145               finish: bool = True):
146    if finish:
147      self.returncode = returncode or 0
148    else:
149      self.returncode = None
150    self.stdout = stdout or ''
151    self.stderr = stderr or ''
152    self.finish = finish
153
154  def communicate(self, _) -> Tuple[str, str]:
155    return self.stdout, self.stderr
156
157  def terminate(self) -> None:
158    if self.finish:
159      raise OSError('Tried to terminate a finished process')
160
161
162class GenericBuilders(builders.Builders):
163  #pylint: disable=useless-super-delegation
164  def __init__(self,
165               suite: Optional[str] = None,
166               include_internal_builders: bool = False):
167    super().__init__(suite, include_internal_builders)
168  #pylint: enable=useless-super-delegation
169
170  def _BuilderRunsTestOfInterest(self, _test_map) -> bool:
171    return True
172
173  def GetIsolateNames(self) -> dict:
174    return {}
175
176  def GetFakeCiBuilders(self) -> dict:
177    return {}
178
179  def GetNonChromiumBuilders(self) -> dict:
180    return {}
181
182
183def RegisterGenericBuildersImplementation() -> None:
184  builders.RegisterInstance(GenericBuilders())
185
186
187class GenericExpectations(expectations.Expectations):
188  def GetExpectationFilepaths(self) -> list:
189    return []
190
191  def _GetExpectationFileTagHeader(self, _) -> str:
192    return """\
193# tags: [ linux mac win ]
194# tags: [ amd intel nvidia ]
195# results: [ Failure RetryOnFailure Skip Pass ]
196"""
197
198  def _GetKnownTags(self) -> Set[str]:
199    return set(['linux', 'mac', 'win', 'amd', 'intel', 'nvidia'])
200
201
202def CreateGenericExpectations() -> GenericExpectations:
203  return GenericExpectations()
204
205
206def RegisterGenericExpectationsImplementation() -> None:
207  expectations.RegisterInstance(CreateGenericExpectations())
208