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.
3Supports single/multi-threaded execution, worker saturation, per-test logging,
4screenshot capture on failure, and automatic retry on transient errors.
6Execution and retry logic are delegated to TestExecutor and TestFlow.
7TestSuite is responsible only for validation, worker saturation, and
8dispatching tests across threads.
9"""
11import copy
12import logging
13import random
14import time
15from concurrent.futures import FIRST_COMPLETED, ThreadPoolExecutor, wait
16from typing import TYPE_CHECKING
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
31if TYPE_CHECKING:
32 from collections.abc import Iterable, Sequence
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
45_DEFAULT_MAX_RETRIES_PER_TEST = 8
48class TestSuite[Driver]:
49 """Orchestrator for a sequence of related tests.
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.
55 Transient errors trigger automatic retry with linear backoff (see TestFlow).
56 Assertion/business failures fail immediately without retry.
57 """
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()
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()
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.
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.
109 Raises:
110 ValueError: If both only_ids and exclude_ids are non-empty.
112 """
113 self._guards_on_invoke(tests)
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
132 self._campaign_name: str = ""
133 self._cycle_name: str = ""
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 )
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 )
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.
168 Copies are named "[COPY N] original_name".
169 """
170 base_tests = list(self._tests)
172 if len(base_tests) == 0:
173 return base_tests
175 extra_tests: list[Test[Driver]] = []
176 copy_counters: dict[str, int] = {}
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
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)
191 return base_tests + extra_tests
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]
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.
206 Args:
207 max_workers: Concurrent workers. 1 = sequential execution.
208 saturate_workers: Duplicate tests to reach max_workers. Default: True.
210 Raises:
211 AggregateInvariantViolationError: If max_workers < 1 or duplicate names.
213 """
214 self._results.clear()
216 validate_workers_amount(
217 workers_amount=max_workers, name="max_workers"
218 ).execute().raise_if_invalid()
220 resolved_saturate_workers = (
221 self._saturate_workers
222 if self._saturate_workers is not None
223 else saturate_workers
224 )
226 flow = self._build_flow()
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
258 self._guards_with_mounted_tests(tests)
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 }
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()
272 return self._results