02.05.02 — TestExecutor[Driver]#
Fichier source : src/ocarina/dsl/testing/internals/test_executor.py
Responsabilité unique : exécuter une seule tentative d’un test, avec un seul driver. Ne connaît ni le rejeu, ni la pool de drivers, ni l’agrégation au niveau suite.
ExecutionOutcome#
@final
@dataclass(frozen=True, slots=True)
class ExecutionOutcome:
result: TestResult
skipped: bool
setup_failed: bool
should_retry: bool
steps_count: int
| Champ | Type | Sens |
|---|
result | TestResult = Result[Any] | None | Le résultat de la chaîne (Ok, Fail), ou None si skip / setup_failed. |
skipped | bool | True si test_runner.skipped (i.e. Test(skipped=True)). |
setup_failed | bool | True si la fonction setup() du scénario a levé. |
should_retry | bool | True si la règle de rejeu s’applique (transient_error détecté). |
steps_count | int | Nombre d’act appelés ; -1 si skip ou setup_failed. |
slots=True : empêche l’ajout dynamique d’attributs et économise mémoire. C’est un objet qui circule en hot-path, le slots est justifié.frozen=True : immutable, sûr à partager entre threads.@final : pas d’héritage.
Ordre d’exécution d’une tentative#
┌──────────────────────────────────────────────────────────────────────┐
│ test_runner = test.spawn(driver, logger_with_taxonomy) │
└──────────────────────────────┬───────────────────────────────────────┘
│
▼
┌─────────────────────────────────┐
│ test_runner.skipped ? │── True ─► return Outcome(skipped=True, ...)
└────────────────┬────────────────┘
│ False
▼
┌─────────────────────────────────┐
│ logger.test_name(test.name) │ (annotation)
└────────────────┬────────────────┘
│
▼
┌─────────────────────────────────┐
│ setup() (si non-None) │── leve ─► teardown() (si non-None)
└────────────────┬────────────────┘ ↓
│ return Outcome(setup_failed=True, should_retry=True, ...)
▼
┌─────────────────────────────────┐
│ watchers.start(driver, logger, │
│ take_screenshot)│ (1 daemon thread par watcher)
└────────────────┬────────────────┘
│
▼
┌─────────────────────────────────┐
│ _run_chain(chain_runners, ...) │── retourne (result, should_retry)
└────────────────┬────────────────┘
│
▼
┌─────────────────────────────────┐
│ watchers.stop() │ (toujours)
└────────────────┬────────────────┘
│
▼
┌─────────────────────────────────┐
│ teardown() (si non-None) │ (TOUJOURS, même si chain a fail)
└────────────────┬────────────────┘ (les exceptions sont logguées & avalées)
│
▼
┌─────────────────────────────────┐
│ 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)
Deux loggers : pourquoi#
logger_with_taxonomy vs logger_without_taxonomy :