unittest_utils.py 5.5 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190
  1. # Copyright 2020 The Chromium Authors. All rights reserved.
  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. from __future__ import print_function
  6. from typing import Any, Callable, Iterable, List, Optional, Tuple, Type
  7. import unittest.mock as mock
  8. from unexpected_passes_common import builders
  9. from unexpected_passes_common import expectations
  10. from unexpected_passes_common import data_types
  11. from unexpected_passes_common import queries as queries_module
  12. def CreateStatsWithPassFails(passes: int, fails: int) -> data_types.BuildStats:
  13. stats = data_types.BuildStats()
  14. for _ in range(passes):
  15. stats.AddPassedBuild()
  16. for i in range(fails):
  17. stats.AddFailedBuild('build_id%d' % i)
  18. return stats
  19. def _CreateSimpleQueries(clauses: Iterable[str]) -> List[str]:
  20. queries = []
  21. # Not actually a valid query since we don't specify the table, but it works.
  22. for c in clauses:
  23. queries.append("""\
  24. SELECT *
  25. WHERE %s
  26. """ % c)
  27. return queries
  28. class SimpleFixedQueryGenerator(queries_module.FixedQueryGenerator):
  29. def GetQueries(self) -> List[str]:
  30. return _CreateSimpleQueries(self.GetClauses())
  31. class SimpleSplitQueryGenerator(queries_module.SplitQueryGenerator):
  32. def GetQueries(self) -> List[str]:
  33. return _CreateSimpleQueries(self.GetClauses())
  34. class SimpleBigQueryQuerier(queries_module.BigQueryQuerier):
  35. def _GetQueryGeneratorForBuilder(self, builder: data_types.BuilderEntry
  36. ) -> queries_module.BaseQueryGenerator:
  37. if not self._large_query_mode:
  38. return SimpleFixedQueryGenerator(builder, 'AND True')
  39. return SimpleSplitQueryGenerator(builder, ['test_id'], 200)
  40. def _GetRelevantExpectationFilesForQueryResult(self, _) -> None:
  41. return None
  42. def _StripPrefixFromTestId(self, test_id: str) -> str:
  43. return test_id.split('.')[-1]
  44. def _GetActiveBuilderQuery(self, _, __) -> str:
  45. return ''
  46. def CreateGenericQuerier(
  47. suite: Optional[str] = None,
  48. project: Optional[str] = None,
  49. num_samples: Optional[int] = None,
  50. large_query_mode: Optional[bool] = None,
  51. cls: Optional[Type[queries_module.BigQueryQuerier]] = None
  52. ) -> queries_module.BigQueryQuerier:
  53. suite = suite or 'pixel'
  54. project = project or 'project'
  55. num_samples = num_samples or 5
  56. large_query_mode = large_query_mode or False
  57. cls = cls or SimpleBigQueryQuerier
  58. return cls(suite, project, num_samples, large_query_mode)
  59. def GetArgsForMockCall(call_args_list: List[tuple],
  60. call_number: int) -> Tuple[tuple, dict]:
  61. """Helper to more sanely get call args from a mocked method.
  62. Args:
  63. call_args_list: The call_args_list member from the mock in question.
  64. call_number: The call number to pull args from, starting at 0 for the first
  65. call to the method.
  66. Returns:
  67. A tuple (args, kwargs). |args| is a list of arguments passed to the method.
  68. |kwargs| is a dict containing the keyword arguments padded to the method.
  69. """
  70. args = call_args_list[call_number][0]
  71. kwargs = call_args_list[call_number][1]
  72. return args, kwargs
  73. class FakePool():
  74. """A fake pathos.pools.ProcessPool instance.
  75. Real pools don't like being given MagicMocks, so this allows testing of
  76. code that uses pathos.pools.ProcessPool by returning this from
  77. multiprocessing_utils.GetProcessPool().
  78. """
  79. def map(self, f: Callable[[Any], Any], inputs: Iterable[Any]) -> List[Any]:
  80. retval = []
  81. for i in inputs:
  82. retval.append(f(i))
  83. return retval
  84. def apipe(self, f: Callable[[Any], Any],
  85. inputs: Iterable[Any]) -> 'FakeAsyncResult':
  86. return FakeAsyncResult(f(inputs))
  87. class FakeAsyncResult():
  88. """A fake AsyncResult like the one from multiprocessing or pathos."""
  89. def __init__(self, result: Any):
  90. self._result = result
  91. def ready(self) -> bool:
  92. return True
  93. def get(self) -> Any:
  94. return self._result
  95. class FakeProcess():
  96. """A fake subprocess Process object."""
  97. def __init__(self,
  98. returncode: Optional[int] = None,
  99. stdout: Optional[str] = None,
  100. stderr: Optional[str] = None,
  101. finish: bool = True):
  102. if finish:
  103. self.returncode = returncode or 0
  104. else:
  105. self.returncode = None
  106. self.stdout = stdout or ''
  107. self.stderr = stderr or ''
  108. self.finish = finish
  109. def communicate(self, _) -> Tuple[str, str]:
  110. return self.stdout, self.stderr
  111. def terminate(self) -> None:
  112. if self.finish:
  113. raise OSError('Tried to terminate a finished process')
  114. class GenericBuilders(builders.Builders):
  115. #pylint: disable=useless-super-delegation
  116. def __init__(self,
  117. suite: Optional[str] = None,
  118. include_internal_builders: bool = False):
  119. super().__init__(suite, include_internal_builders)
  120. #pylint: enable=useless-super-delegation
  121. def _BuilderRunsTestOfInterest(self, _test_map) -> bool:
  122. return True
  123. def GetIsolateNames(self) -> dict:
  124. return {}
  125. def GetFakeCiBuilders(self) -> dict:
  126. return {}
  127. def GetNonChromiumBuilders(self) -> dict:
  128. return {}
  129. def RegisterGenericBuildersImplementation() -> None:
  130. builders.RegisterInstance(GenericBuilders())
  131. class GenericExpectations(expectations.Expectations):
  132. def GetExpectationFilepaths(self) -> list:
  133. return []
  134. def _GetExpectationFileTagHeader(self, _) -> str:
  135. return """\
  136. # tags: [ linux mac win ]
  137. # results: [ Failure RetryOnFailure Skip Pass ]
  138. """
  139. def CreateGenericExpectations() -> GenericExpectations:
  140. return GenericExpectations()