123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190 |
- # Copyright 2020 The Chromium Authors. All rights reserved.
- # Use of this source code is governed by a BSD-style license that can be
- # found in the LICENSE file.
- """Helper methods for unittests."""
- from __future__ import print_function
- from typing import Any, Callable, Iterable, List, Optional, Tuple, Type
- import unittest.mock as mock
- from unexpected_passes_common import builders
- from unexpected_passes_common import expectations
- from unexpected_passes_common import data_types
- from unexpected_passes_common import queries as queries_module
- def CreateStatsWithPassFails(passes: int, fails: int) -> data_types.BuildStats:
- stats = data_types.BuildStats()
- for _ in range(passes):
- stats.AddPassedBuild()
- for i in range(fails):
- stats.AddFailedBuild('build_id%d' % i)
- return stats
- def _CreateSimpleQueries(clauses: Iterable[str]) -> List[str]:
- queries = []
- # Not actually a valid query since we don't specify the table, but it works.
- for c in clauses:
- queries.append("""\
- SELECT *
- WHERE %s
- """ % c)
- return queries
- class SimpleFixedQueryGenerator(queries_module.FixedQueryGenerator):
- def GetQueries(self) -> List[str]:
- return _CreateSimpleQueries(self.GetClauses())
- class SimpleSplitQueryGenerator(queries_module.SplitQueryGenerator):
- def GetQueries(self) -> List[str]:
- return _CreateSimpleQueries(self.GetClauses())
- class SimpleBigQueryQuerier(queries_module.BigQueryQuerier):
- def _GetQueryGeneratorForBuilder(self, builder: data_types.BuilderEntry
- ) -> queries_module.BaseQueryGenerator:
- if not self._large_query_mode:
- return SimpleFixedQueryGenerator(builder, 'AND True')
- return SimpleSplitQueryGenerator(builder, ['test_id'], 200)
- def _GetRelevantExpectationFilesForQueryResult(self, _) -> None:
- return None
- def _StripPrefixFromTestId(self, test_id: str) -> str:
- return test_id.split('.')[-1]
- def _GetActiveBuilderQuery(self, _, __) -> str:
- return ''
- def CreateGenericQuerier(
- suite: Optional[str] = None,
- project: Optional[str] = None,
- num_samples: Optional[int] = None,
- large_query_mode: Optional[bool] = None,
- cls: Optional[Type[queries_module.BigQueryQuerier]] = None
- ) -> queries_module.BigQueryQuerier:
- suite = suite or 'pixel'
- project = project or 'project'
- num_samples = num_samples or 5
- large_query_mode = large_query_mode or False
- cls = cls or SimpleBigQueryQuerier
- return cls(suite, project, num_samples, large_query_mode)
- def GetArgsForMockCall(call_args_list: List[tuple],
- call_number: int) -> Tuple[tuple, dict]:
- """Helper to more sanely get call args from a mocked method.
- Args:
- call_args_list: The call_args_list member from the mock in question.
- call_number: The call number to pull args from, starting at 0 for the first
- call to the method.
- Returns:
- A tuple (args, kwargs). |args| is a list of arguments passed to the method.
- |kwargs| is a dict containing the keyword arguments padded to the method.
- """
- args = call_args_list[call_number][0]
- kwargs = call_args_list[call_number][1]
- return args, kwargs
- class FakePool():
- """A fake pathos.pools.ProcessPool instance.
- Real pools don't like being given MagicMocks, so this allows testing of
- code that uses pathos.pools.ProcessPool by returning this from
- multiprocessing_utils.GetProcessPool().
- """
- def map(self, f: Callable[[Any], Any], inputs: Iterable[Any]) -> List[Any]:
- retval = []
- for i in inputs:
- retval.append(f(i))
- return retval
- def apipe(self, f: Callable[[Any], Any],
- inputs: Iterable[Any]) -> 'FakeAsyncResult':
- return FakeAsyncResult(f(inputs))
- class FakeAsyncResult():
- """A fake AsyncResult like the one from multiprocessing or pathos."""
- def __init__(self, result: Any):
- self._result = result
- def ready(self) -> bool:
- return True
- def get(self) -> Any:
- return self._result
- class FakeProcess():
- """A fake subprocess Process object."""
- def __init__(self,
- returncode: Optional[int] = None,
- stdout: Optional[str] = None,
- stderr: Optional[str] = None,
- finish: bool = True):
- if finish:
- self.returncode = returncode or 0
- else:
- self.returncode = None
- self.stdout = stdout or ''
- self.stderr = stderr or ''
- self.finish = finish
- def communicate(self, _) -> Tuple[str, str]:
- return self.stdout, self.stderr
- def terminate(self) -> None:
- if self.finish:
- raise OSError('Tried to terminate a finished process')
- class GenericBuilders(builders.Builders):
- #pylint: disable=useless-super-delegation
- def __init__(self,
- suite: Optional[str] = None,
- include_internal_builders: bool = False):
- super().__init__(suite, include_internal_builders)
- #pylint: enable=useless-super-delegation
- def _BuilderRunsTestOfInterest(self, _test_map) -> bool:
- return True
- def GetIsolateNames(self) -> dict:
- return {}
- def GetFakeCiBuilders(self) -> dict:
- return {}
- def GetNonChromiumBuilders(self) -> dict:
- return {}
- def RegisterGenericBuildersImplementation() -> None:
- builders.RegisterInstance(GenericBuilders())
- class GenericExpectations(expectations.Expectations):
- def GetExpectationFilepaths(self) -> list:
- return []
- def _GetExpectationFileTagHeader(self, _) -> str:
- return """\
- # tags: [ linux mac win ]
- # results: [ Failure RetryOnFailure Skip Pass ]
- """
- def CreateGenericExpectations() -> GenericExpectations:
- return GenericExpectations()
|