Coverage for src / ocarina / dsl / testing / oc_test_suite.py: 23.53%

88 statements  

« prev     ^ index     » next       coverage.py v7.13.5, created at 2026-06-04 16:22 +0200

1"""Test suite orchestration with parallel execution. 

2 

3Supports single/multi-threaded execution, worker saturation, per-test logging, 

4screenshot capture on failure, and automatic retry on transient errors. 

5 

6Execution and retry logic are delegated to TestExecutor and TestFlow. 

7TestSuite is responsible only for validation, worker saturation, and 

8dispatching tests across threads. 

9""" 

10 

11import copy 

12import logging 

13import random 

14import time 

15from concurrent.futures import FIRST_COMPLETED, ThreadPoolExecutor, wait 

16from typing import TYPE_CHECKING 

17 

18from ocarina.custom_invariants.testing.oc_test_runners_ids import ( 

19 validate_test_runners_ids, 

20) 

21from ocarina.custom_invariants.testing.oc_test_runners_names import ( 

22 validate_test_runners_names, 

23) 

24from ocarina.custom_invariants.testing.workers import validate_workers_amount 

25from ocarina.dsl.testing.filter_tests_by_ids import filter_tests_by_ids 

26from ocarina.dsl.testing.internals.test_executor import TestExecutor 

27from ocarina.dsl.testing.internals.test_flow import TestFlow 

28from ocarina.infra.drivers_pool import WarmupTimeoutError 

29from ocarina.opinionated.infra.act_counter import ActCounter as ThreadsBasedActCounter 

30 

31if TYPE_CHECKING: 

32 from collections.abc import Iterable, Sequence 

33 

34 from ocarina.custom_types.oc_test_layers import ( 

35 TestId, 

36 TestSuiteResults, 

37 ) 

38 from ocarina.custom_types.thunk import Thunk 

39 from ocarina.dsl.testing.oc_test import Test 

40 from ocarina.infra.act_counter import ActCounter 

41 from ocarina.infra.drivers_pool import WebDriversPool 

42 from ocarina.ports.ilogger import ILogger 

43 from ocarina.ports.itake_screenshot import ITakeScreenshot 

44 

45_DEFAULT_MAX_RETRIES_PER_TEST = 8 

46 

47 

48class TestSuite[Driver]: 

49 """Orchestrator for a sequence of related tests. 

50 

51 Handles parallel execution, worker saturation, and result aggregation. 

52 Per-test retry logic is owned by TestFlow. 

53 Per-attempt execution is owned by TestExecutor. 

54 

55 Transient errors trigger automatic retry with linear backoff (see TestFlow). 

56 Assertion/business failures fail immediately without retry. 

57 """ 

58 

59 def _guards_with_mounted_tests(self, tests: Sequence[Test[Driver]]) -> None: 

60 """Validate test names are unique (including [COPY N] duplicates).""" 

61 validate_test_runners_names( 

62 tests=tests, name="tests" 

63 ).execute().raise_if_invalid() 

64 

65 def _guards_on_invoke(self, tests: Sequence[Test[Driver]]) -> None: 

66 """Validate original test IDs are unique.""" 

67 validate_test_runners_ids( 

68 tests=tests, name="tests" 

69 ).execute().raise_if_invalid() 

70 

71 def __init__( # noqa: PLR0913 

72 self, 

73 *, 

74 name: str, 

75 tests: Sequence[Test[Driver]], 

76 create_logger: Thunk[ILogger], 

77 drivers_pool: WebDriversPool[Driver], 

78 take_screenshot: ITakeScreenshot[Driver], 

79 act_counter: ActCounter | None = None, 

80 transient_errors: tuple[type[Exception], ...] = (), 

81 copy_indicator: str = "COPY", 

82 put_space_after_copy_indicator: bool = True, 

83 max_retries_per_test: int | None = None, 

84 autoscreen_on_fail: bool = False, 

85 saturate_workers: bool | None = None, 

86 only_ids: Iterable[str] = (), 

87 exclude_ids: Iterable[str] = (), 

88 ) -> None: 

89 """Initialize test suite. 

90 

91 Args: 

92 name: Suite name used in logs and reports. 

93 tests: Tests to execute. Must have unique test_id values. 

94 create_logger: Factory called per test to create an isolated logger. 

95 drivers_pool: Thread-safe pool of driver instances. 

96 take_screenshot: Callable (driver, logger, category) -> None. 

97 act_counter: Tracks act() call count per test execution. 

98 transient_errors: Exception types that trigger retry. Default: (). 

99 copy_indicator: Prefix for duplicated tests. Default: "COPY" -> "[COPY 1]". 

100 put_space_after_copy_indicator: "[COPY 1]" vs "[COPY1]". Default: True. 

101 max_retries_per_test: Max retry attempts on transient failures. Default: 8. 

102 autoscreen_on_fail: Capture screenshot automatically on test failure. 

103 saturate_workers: Duplicate tests to reach max_workers. Default: None. 

104 only_ids: If non-empty, keep only tests whose test_id is in this set. 

105 Mutually exclusive with exclude_ids. 

106 exclude_ids: If non-empty, drop tests whose test_id is in this set. 

107 Mutually exclusive with only_ids. 

108 

109 Raises: 

110 ValueError: If both only_ids and exclude_ids are non-empty. 

111 

112 """ 

113 self._guards_on_invoke(tests) 

114 

115 self.name = name 

116 self._create_logger = create_logger 

117 self._drivers_pool = drivers_pool 

118 self._take_screenshot = take_screenshot 

119 self._act_counter = act_counter or ThreadsBasedActCounter() 

120 self._transient_errors = transient_errors 

121 self._results: TestSuiteResults = {} 

122 self._copy_indicator = copy_indicator 

123 self._put_space_after_copy_indicator = put_space_after_copy_indicator 

124 self._max_retries_per_test = ( 

125 _DEFAULT_MAX_RETRIES_PER_TEST 

126 if max_retries_per_test is None 

127 else max_retries_per_test 

128 ) 

129 self._autoscreen_on_fail = autoscreen_on_fail 

130 self._saturate_workers = saturate_workers 

131 

132 self._campaign_name: str = "" 

133 self._cycle_name: str = "" 

134 

135 self._tests = filter_tests_by_ids( 

136 tests, 

137 only=only_ids, 

138 exclude=exclude_ids, 

139 logger=create_logger().set_prefix( 

140 lambda: f"{self.name}: filtering tests..." 

141 ), 

142 ) 

143 

144 def _build_flow(self) -> TestFlow[Driver]: 

145 """Construct a TestFlow wired to this suite's dependencies.""" 

146 return TestFlow( 

147 executor=TestExecutor( 

148 create_logger=self._create_logger, 

149 take_screenshot=self._take_screenshot, 

150 act_counter=self._act_counter, 

151 transient_errors=self._transient_errors, 

152 autoscreen_on_fail=self._autoscreen_on_fail, 

153 ), 

154 drivers_pool=self._drivers_pool, 

155 create_logger=self._create_logger, 

156 act_counter=self._act_counter, 

157 cycle_name=self._cycle_name, 

158 campaign_name=self._campaign_name, 

159 suite_name=self.name, 

160 max_retries=self._max_retries_per_test, 

161 ) 

162 

163 def _prepare_tests_for_saturated_threading( 

164 self, max_workers: int 

165 ) -> Sequence[Test[Driver]]: 

166 """Duplicate tests randomly until count reaches max_workers. 

167 

168 Copies are named "[COPY N] original_name". 

169 """ 

170 base_tests = list(self._tests) 

171 

172 if len(base_tests) == 0: 

173 return base_tests 

174 

175 extra_tests: list[Test[Driver]] = [] 

176 copy_counters: dict[str, int] = {} 

177 

178 for _ in range(max_workers - len(base_tests)): 

179 original = random.choice(base_tests) # noqa: S311 

180 copy_counters.setdefault(original.name, 0) 

181 copy_counters[original.name] += 1 

182 

183 cloned = copy.copy(original) 

184 space = " " if self._put_space_after_copy_indicator else "" 

185 cloned.name = ( 

186 f"[{self._copy_indicator}{space}{copy_counters[original.name]}]" 

187 f" {original.name}" 

188 ) 

189 extra_tests.append(cloned) 

190 

191 return base_tests + extra_tests 

192 

193 @property 

194 def test_names_and_ids(self) -> Sequence[tuple[str, TestId]]: 

195 """Return (name, test_id) pairs for all tests in the suite.""" 

196 return [(test.name, test.test_id) for test in self._tests] 

197 

198 def run( 

199 self, 

200 *, 

201 max_workers: int, 

202 saturate_workers: bool = True, 

203 ) -> TestSuiteResults: 

204 """Execute all tests. Returns dict mapping test name -> TestSuiteResult. 

205 

206 Args: 

207 max_workers: Concurrent workers. 1 = sequential execution. 

208 saturate_workers: Duplicate tests to reach max_workers. Default: True. 

209 

210 Raises: 

211 AggregateInvariantViolationError: If max_workers < 1 or duplicate names. 

212 

213 """ 

214 self._results.clear() 

215 

216 validate_workers_amount( 

217 workers_amount=max_workers, name="max_workers" 

218 ).execute().raise_if_invalid() 

219 

220 resolved_saturate_workers = ( 

221 self._saturate_workers 

222 if self._saturate_workers is not None 

223 else saturate_workers 

224 ) 

225 

226 flow = self._build_flow() 

227 

228 if max_workers == 1: 

229 self._guards_with_mounted_tests(self._tests) 

230 for test in self._tests: 

231 self._results[test.name] = flow.run(test) 

232 else: 

233 if resolved_saturate_workers and len(self._tests) > 0: 

234 try: 

235 self._drivers_pool.warmup() 

236 except WarmupTimeoutError: 

237 logging.basicConfig(level=logging.INFO) 

238 logger = logging.getLogger(__name__) 

239 logger.info( 

240 "Warmup stalled: killing the program. Expect it to raise soon." 

241 " " 

242 "Cleanup will be attempted" 

243 " " 

244 "but some browser processes may survive." 

245 " " 

246 "Check Activity Monitor / Dock for remaining instances." 

247 ) 

248 time.sleep(30) 

249 self._drivers_pool.shutdown() 

250 time.sleep(5) 

251 raise 

252 tests = self._prepare_tests_for_saturated_threading( 

253 max_workers=max_workers 

254 ) 

255 else: 

256 tests = self._tests 

257 

258 self._guards_with_mounted_tests(tests) 

259 

260 with ThreadPoolExecutor(max_workers=max_workers) as executor: 

261 futures = { 

262 executor.submit(flow.run, test): (test.name, test.test_id) 

263 for test in tests 

264 } 

265 

266 while futures: 

267 done, _ = wait(futures, return_when=FIRST_COMPLETED) 

268 for future in done: 

269 name, _ = futures.pop(future) 

270 self._results[name] = future.result() 

271 

272 return self._results