02.05.02 — TestExecutor[Driver]#
Source file: src/ocarina/dsl/testing/internals/test_executor.py
One job: run one attempt of a test with one driver. Knows nothing about retries, the driver pool, or suite-level aggregation.
ExecutionOutcome#
@final
@dataclass(frozen=True, slots=True)
class ExecutionOutcome:
result: TestResult
skipped: bool
setup_failed: bool
should_retry: bool
steps_count: int
| Field | Type | Meaning |
|---|
result | TestResult = Result[Any] | None | The chain’s result (Ok, Fail), or None if skip / setup_failed. |
skipped | bool | True if test_runner.skipped (i.e. Test(skipped=True)). |
setup_failed | bool | True if the scenario’s setup() raised. |
should_retry | bool | True if the retry rule applies (transient_error detected). |
steps_count | int | Number of act calls; -1 if skip or setup_failed. |
slots=True: blocks dynamic attribute addition, saves memory. This object rides the hot path — slots earns its keep.frozen=True: immutable, safe to share across threads.@final: no inheritance.
Execution order for one attempt#
┌──────────────────────────────────────────────────────────────────────┐
│ test_runner = test.spawn(driver, logger_with_taxonomy) │
└──────────────────────────────┬───────────────────────────────────────┘
│
▼
┌─────────────────────────────────┐
│ test_runner.skipped ? │── True ─► return Outcome(skipped=True, ...)
└────────────────┬────────────────┘
│ False
▼
┌─────────────────────────────────┐
│ logger.test_name(test.name) │ (annotation)
└────────────────┬────────────────┘
│
▼
┌─────────────────────────────────┐
│ setup() (if not None) │── raises ─► teardown() (if not None)
└────────────────┬────────────────┘ ↓
│ return Outcome(setup_failed=True, should_retry=True, ...)
▼
┌─────────────────────────────────┐
│ watchers.start(driver, logger, │
│ take_screenshot)│ (1 daemon thread per watcher)
└────────────────┬────────────────┘
│
▼
┌─────────────────────────────────┐
│ _run_chain(chain_runners, ...) │── returns (result, should_retry)
└────────────────┬────────────────┘
│
▼
┌─────────────────────────────────┐
│ watchers.stop() │ (always)
└────────────────┬────────────────┘
│
▼
┌─────────────────────────────────┐
│ teardown() (if not None) │ (ALWAYS, even if chain failed)
└────────────────┬────────────────┘ (exceptions are logged & swallowed)
│
▼
┌─────────────────────────────────┐
│ steps_count = act_counter.get()│
│ return Outcome(...) │
└─────────────────────────────────┘
execute#
def execute(
self, test: Test[Driver], *,
driver: Driver,
taxonomy: tuple[str, ...],
logger_with_taxonomy: ILogger,
logger_without_taxonomy: ILogger,
attempt: int, max_attempts: int,
) -> ExecutionOutcome:
test_runner = test.spawn(driver, logger_with_taxonomy)
if test_runner.skipped:
return ExecutionOutcome(result=None, skipped=True, setup_failed=False,
should_retry=False, steps_count=-1)
logger_without_taxonomy.test_name(test.name)
if test_runner.setup is not None:
try:
test_runner.setup()
except Exception as exc:
msg = f"{test.name} -- Setup failed (attempt {attempt}/{max_attempts}): {exc}"
logger_with_taxonomy.warning(msg)
if test_runner.teardown is not None:
self._run_teardown(test_runner.teardown, test_name=test.name, logger=logger_with_taxonomy)
return ExecutionOutcome(result=None, skipped=False, setup_failed=True,
should_retry=True, steps_count=-1)
watchers: Sequence[Watcher[Driver]] = test_runner.watchers or []
self._start_watchers(watchers, driver=driver, test_name=test.name, taxonomy=taxonomy)
result, should_retry = self._run_chain(
test_runner.chain_runners, test_name=test.name, attempt=attempt,
driver=driver, logger=logger_without_taxonomy,
logger_with_taxonomy=logger_with_taxonomy, max_attempts=max_attempts,
)
self._stop_watchers(watchers)
if test_runner.teardown is not None:
self._run_teardown(test_runner.teardown, test_name=test.name, logger=logger_with_taxonomy)
steps_count = self._act_counter.get()
return ExecutionOutcome(result=result, skipped=False, setup_failed=False,
should_retry=should_retry, steps_count=steps_count)
Two loggers: why#
logger_with_taxonomy vs logger_without_taxonomy:
