02.03.01 — Le type Result[T]#

Fichier source : src/ocarina/railway/result.py — zéro dépendance.

Code#

from dataclasses import dataclass, field
from typing import TypeGuard, final


class _BaseResult:
    """Base class ensuring both Ok and Fail have error attribute for type narrowing."""
    error: Exception | None


@final
@dataclass(frozen=True)
class Ok[T](_BaseResult):
    value: T
    error: None = None


@final
@dataclass(frozen=True)
class Fail(_BaseResult):
    error: Exception = field(default_factory=lambda: Exception("Unknown error"))


type Result[T] = Ok[T] | Fail


def is_ok[T](result: Result[T]) -> TypeGuard[Ok[T]]:
    return isinstance(result, Ok)


def is_fail[T](result: Result[T]) -> TypeGuard[Fail]:
    return isinstance(result, Fail)

Six décisions de design à décortiquer#

1. _BaseResult comme parent commun#

class _BaseResult:
    error: Exception | None

But unique : permettre à mypy d’accéder à result.error sans narrowing préalable. Sans cette base, on devrait écrire :

02.04.01 — Le flot validate → assert_that → execute → raise_if_invalid#

Fichier source : src/ocarina/dsl/invariants/validate.py (entry-point) + src/ocarina/dsl/invariants/internals/validation_chain.py

La machine à états#

Diagramme#

validate(value: T)
       │
       ▼
ValidationStartBlock[T]
       │
       └─ assert_that(predicate, *, msg=None)
              │
              ▼
       ValidationAssertBlock[T]   ← état qui accepte plusieurs continuations
              │
              ├─► assert_that(P)     [boucle, AND chaînable, voir le diagramme dédié]
              ├─► otherwise(P)       [OR, voir 03-otherwise-any-of.md]
              ├─► then(U, name=...)  [change de valeur, voir 04-then-chain-of-validations.md]
              └─► execute()          → _ValidationResult
                                          │
                                          ├─ is_valid: bool
                                          ├─ errors: Sequence[InvariantViolationError]
                                          ├─ validated_values: Sequence[Any]
                                          └─ raise_if_invalid() : None | raise AggregateInvariantViolationError

Boucle assert_that#

assert_that formule une assertion sur la valeur courante. Plusieurs assert_that peuvent se suivre, chacun ajoutant une condition supplémentaire à la chaîne (AND logique, ordre préservé).

02.05.01 — Test[Driver]#

Fichier source : src/ocarina/dsl/testing/oc_test.py

Signature#

@final
class Test[Driver]:
    def __init__(
        self,
        *,
        name: TestName,
        test_id: str | None = None,
        test_scenario: TestScenario[Driver],
        pre_test_scenarios_fragments: Sequence[TestScenarioFragment[Driver]] | None = None,
        post_test_scenarios_fragments: Sequence[TestScenarioFragment[Driver]] | None = None,
        skipped: bool = False,
    ) -> None:
        if test_id is None:
            test_id = name
        self.name = name
        self.test_id = test_id
        self._test_scenario = test_scenario
        self._pre_test_scenarios_fragments = pre_test_scenarios_fragments or []
        self._post_test_scenarios_fragments = post_test_scenarios_fragments or []
        self._skipped = skipped

Six paramètres :

1. @final#

Pas d’héritage utilisateur. Si l’on veut un test « spécial », on compose via les fragments ou via un scénario, on ne sous-classe pas.

02.10.01 — WebDriversPool[Driver]#

Fichier source : src/ocarina/infra/drivers_pool.py

Pool thread-safe de drivers, max concurrence garantie par sémaphore, warmup asynchrone surveillé, shutdown propre. Pas de réutilisation : chaque acquire() détruit son driver à la fin (état propre).

Constructeur#

@final
class WebDriversPool[Driver]:
    def __init__(
        self,
        create_driver: Thunk[BuiltWebDriver[Driver]],
        max_size: int,
        warmup_timeout: float | None = None,
    ) -> None:
        self._create_driver = create_driver
        self._pool: Queue[BuiltWebDriver[Driver]] = Queue(max_size)
        self._semaphore = Semaphore(max_size)
        self._warmup_timeout = (
            warmup_timeout if warmup_timeout is not None and warmup_timeout > 0.1
            else 60.0 * 5
        )

acquire()#

@contextmanager
def acquire(self) -> Iterator[Driver]:
    try:
        driver, dispose = self._pool.get_nowait()
    except Empty:
        self._semaphore.acquire()
        try:
            driver, dispose = self._create_driver()
        except Exception:
            self._semaphore.release()
            raise

    try:
        yield driver
    finally:
        with suppress(Exception):
            dispose()
        self._semaphore.release()

              acquire()
                  │
                  ▼
   ┌──────────────────────────────────┐
   │ pool.get_nowait()                │── OK ─► driver, dispose         ← cas warmup
   └──────────────┬───────────────────┘
                  │ Empty
                  ▼
   ┌──────────────────────────────────┐
   │ sem.acquire()                    │  ← bloque si N drivers vivants
   └──────────────┬───────────────────┘
                  ▼
   ┌──────────────────────────────────┐
   │ create_driver()                  │── leve ─► sem.release(); raise
   └──────────────┬───────────────────┘
                  ▼
   ┌──────────────────────────────────┐
   │ yield driver                     │  ← caller utilise
   └──────────────┬───────────────────┘
                  ▼
                finally :
                  with suppress : dispose()       ← driver détruit
                  sem.release()                   ← rend une place

1. Pas de réutilisation#

Queue.get_nowait() consomme l’entrée. Une fois sorti, le driver n’est jamais remis dans la queue. À la fin (finally), il est disposé.

02.11.01 — CliBuilder + CliArg + _SilentArgumentParser#

Fichier source : src/ocarina/opinionated/cli/builder.py

Surcouche déclarative au-dessus d’argparse. Permet d’agréger les erreurs de validation, ré-écrire la sortie d’aide en cas d’erreur, et enregistrer des effets post-parse.

_SilentArgumentParser#

class _SilentArgumentParser(ArgumentParser):
    def error(self, message: str) -> Never:
        """Raise an error."""
        raise ValueError(message)

L’ArgumentParser standard appelle sys.exit(2) directement en cas d’erreur. Avec cette merde, on ne peut pas intercepter, on ne peut pas agréger plusieurs erreurs.

_SilentArgumentParser re-route les erreurs en ValueError. Donc CliBuilder.parse peut les attraper et les agréger.

02.03.02 — La machine à états du builder#

Fichier source : src/ocarina/dsl/testing_with_railway/internals/action_chain.py

C’est ici qu’est implémentée toute la mécanique ROP : le passage par quatre états successifs typés, qui interdit littéralement (au sens du type-checker) toute fantaisie syntaxique. Le DSL est aussi son propre système immunitaire.

Vue d’ensemble#

                      ┌─────────────────┐
   start(action)  →   │  ActionStart[T] │     un Action[T] = Thunk[Result[T]]
                      └────────┬────────┘
                               │ .failure(failure_handler)
                               ▼
                      ┌─────────────────┐
                      │ ActionFailure[T]│
                      └────────┬────────┘
                               │ .success(success_handler)
                               ▼
                      ┌─────────────────┐
                      │ ActionSuccess[T]│
                      └────────┬────────┘
                               │ .execute()   ── exécute action(), fire le handler
                               ▼
                      ┌─────────────────┐
                      │ ActionChain[T]  │     contient (has_failed, result)
                      └──────┬──────────┘
                             │ .then(next_action_or_start)
              ┌──────────────┴──────────────┐
              │                             │
   has_failed=True                   has_failed=False
              │                             │
              ▼                             ▼
   ┌──────────────────┐         ┌──────────────────┐
   │ NeutralAction-   │         │ ActionStart[T]   │  → cycle recommence
   │ Start[T]         │         │  (l'action suiv. │
   │ (rail d'échec :  │         │   sera exécutée) │
   │  toute la chaîne │         └──────────────────┘
   │  devient no-op,  │
   │ tout en gardant  │
   │ l'API fluide)    │
   └──────────────────┘
              │
              ▼ (.failure → .success → .execute, tout en no-op)
   ┌──────────────────┐
   │  ActionChain[T]  │   has_failed=True, result=<le Fail accumulé>
   └──────────────────┘

Les types Action, FailureHandler, SuccHandler#

type Action[T]        = Thunk[Result[T]]      # () -> Result[T]
type FailureHandler   = Callable[[Exception], None]
type SuccHandler      = Effect                # () -> None

• Action[T] est un Thunk. Cela veut dire qu’au moment où ActionStart(action) est appelé, rien n’est exécuté. L’action est juste capturée.
• FailureHandler reçoit l’exception, mais ne reçoit pas le Fail ni le Result. C’est volontaire : 99% des handlers veulent tracer l’exception, prendre un screenshot, et c’est tout.
• SuccHandler est un Effect (sans argument). Idem : un handler de succès log un message statique, fait un screenshot, pas besoin du résultat.

Typestate Pattern : pourquoi on ne peut pas se tromper#

L’enchaînement est strictement linéaire. Chaque méthode renvoie un type différent, qui n’expose que la suite logique de l’enchaînement :

02.04.02 — Catalogue des assertions builtin#

Fichier source : src/ocarina/dsl/invariants/assertions.py ~25 prédicats.

Toutes les assertions suivent le même contrat :

• Soit un prédicat direct (value: T) -> None qui lève InvariantViolationError si le contrat n’est pas respecté.
• Soit une closure quand il faut passer un argument de configuration (is_equal_to(cmp), etc.).
• Soit, exceptionnellement, une higher-order function (HOF) comme each, placée ici par pragmatisme (créer un fichier entier pour ce seul cas serait overkill).

Tableau complet#

Quelques implémentations en détail#

is_email#

def is_email(value: str) -> None:
    """Assert that the string is a valid email address (fast check)."""
    if " " in value:
        raise InvariantViolationError(f"'{value}' must not contain whitespace.")
    if value.count("@") != 1:
        raise InvariantViolationError(f"'{value}' must contain exactly one '@' character.")
    local_part, domain_part = value.split("@")
    if not local_part or not domain_part:
        raise InvariantViolationError(f"'{value}' must have non-empty local and domain parts.")
    if "." not in domain_part:
        raise InvariantViolationError(f"Domain part of '{value}' must contain at least one '.'.")

Quatre vérifications minimales. Pas de regex RFC5322. C’est volontaire : l’auteur fait le pari que la validation exhaustive d’un email se fait en envoyant un email, pas en regex.

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

• 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 :

02.10.02 — DriverBuilder[Driver]#

Fichier source : src/ocarina/infra/driver_builder.py

Encapsule la gestion du profil (souvent une copie tmp d’un répertoire utilisateur) et produit la paire (driver, dispose) attendue par la pool.

Pourquoi un builder ?#

Quand on construit un driver Selenium avec un profil, il faut :

1. Copier le profil dans un dossier temporaire (sinon Firefox/Chrome locked sur le profil original).
2. Lancer le driver en pointant vers le dossier temporaire.
3. À la fin : driver.quit() puis supprimer le dossier temporaire.

DriverBuilder factorise ces trois étapes :

02.11.02 — CliStore[TKeys] + _CliField[T] + phantom_validate#

Fichiers source : src/ocarina/opinionated/cli/store.py, src/ocarina/opinionated/cli/phantoms.py

Store write-once des valeurs CLI parsées. Chaque champ valide à l’écriture. Le TKeys: Literal[...] apporte de l’autocomplétion sur les clés.

_CliField[T] — champ write-once#

class _CliField[T]:
    def __init__(self, *, validate: ValidationChainBuilder[T]) -> None:
        self._value: T | _Unset = _UNSET
        self._validate = validate

    def set(self, value: T) -> None:
        if not isinstance(self._value, _Unset):
            raise RuntimeError("Value already set.")
        self._validate(_validate(value)).execute().raise_if_invalid()
        self._value = value

    def get(self) -> T:
        if isinstance(self._value, _Unset):
            raise RuntimeError("Value not set yet.")
        return self._value

1. Sentinel _Unset#

class _Unset:
    pass

_UNSET = _Unset()

Pourquoi pas None ? Parce que None peut être une valeur légitime (--profile-path n’est pas obligatoire ; sa valeur après parse peut légitimement être None). Une sentinelle dédiée distingue « pas encore set » de « set à None ».