03.01 — Effect, Thunk[T], Result[T]

03.01 — Effect, Thunk[T], Result[T]#

Trois lignes de code dans custom_types/ + une dans railway/. Tout le DSL repose dessus.

Triplet#

# src/ocarina/custom_types/effect.py
type Effect = Callable[[], None]
type Effects = tuple[Effect, ...]

# src/ocarina/custom_types/thunk.py
type Thunk[T] = Callable[[], T]

# src/ocarina/railway/result.py
type Result[T] = Ok[T] | Fail
TypeSémantique FPDéfinition
EffectEffet de bord (déféré)() -> None
Thunk[T]Computation déférée avec valeur() -> T (paresse + valeur)
Result[T]Computation qui peut échouerUnion discriminée Ok[T] | Fail

Distinction Effect vs Thunk#

log_msg: Effect = lambda: print("hello")        # () -> None
get_42: Thunk[int] = lambda: 42                 # () -> int
fetch:  Thunk[Result[int]] = lambda: Ok(42)     # () -> Result[int]

Si la fonction retourne quelque chose qu’on va consommer ailleurs → Thunk[T].
Sinon → Effect.

03.02 — Closures comme primitive d'inversion de contrôle

03.02 — Closures comme primitive d’inversion de contrôle#

Pas de DI container. Pas de framework d’injection. Une closure est la primitive d’injection d’Ocarina.

L’idée#

Une closure capture des valeurs dans son scope englobant. Quand on retourne une fonction depuis une autre fonction, les paramètres de l’extérieur sont disponibles à l’intérieur, même après que la fonction extérieure ait rendu la main.

def make_handler(logger: ILogger) -> Callable[[str], FailureHandler]:
    def make_failure_handler(msg: str) -> FailureHandler:
        def actual_handler(exc: Exception) -> None:
            logger.error(msg, exc=exc)
        return actual_handler
    return make_failure_handler

C’est aussi du currying : make_handler(logger)(msg)(exc). Chaque appel capture une nouvelle couche d’environnement.

03.03 — Évaluation paresseuse

03.03 — Évaluation paresseuse#

Dans Ocarina, rien n’est exécuté tant qu’on ne l’a pas explicitement déclenché. C’est ce qui rend les scénarios composables comme des valeurs.

Zzz#

EndroitFormeDéclencheur
ChainRunner[T]Thunk[ActionChain[T]]runner.run()
validate(...)ValidationStartBlockValidationAssertBlock.execute()
match_page(...)retourne un ChainRunner[Any].run() (via la chaîne englobante)
Watcher.callbackCallable[[Watcher], None]_loop quand start() est appelé
logger.set_prefix(thunk)Thunk[str]recalculé à chaque appel de log
Scenario.setupteardownEffectappelé par TestExecutor
bootstrap(post_exec=...)Callable[[TestCycleResults], None]appelé après run_plugins
CliBuilder(effects_factory=lambda ns: (...))Effectsappelés après le parse argparse
test_scenario: TestScenario[Driver]Callable[[Driver, ILogger], Scenario[Driver]]appelé par Test.spawn
dispatch[mode]() dans TestCycle.run_alldict de Thunk[bool]appelé en lookup

ChainRunner#

runner = drive_page(act1, act2, act3)        # ⚠️  rien exécuté
# … plus tard …
chain = runner.run()                         # ▶︎  exécution
CapacitéSans paresseAvec paresse
Stocker un scénario dans une variableimpossible (déjà exécuté)trivial
Multiplier [runner] * 5exécute 1 fois, on a 5 références au résultatexécute 5 fois
Passer un runner à un autre runner (composition)impossibletrivial
Réordonner les act dans un test refactordifficiletrivial

validate(...).execute()#

v = validate(value, name="x").assert_that(is_positive).assert_that(is_not_zero)
# … on peut composer …
combined = chain_validations(v, other_validation)
# … rien d'exécuté jusqu'ici …
combined.execute().raise_if_invalid()        # ▶︎  exécution + agrégation

C’est ce qui permet à _ValidationChain de collecter toutes les erreurs avant d’en lever une seule (AggregateInvariantViolationError).

03.04 — reduce / fold dans Ocarina

03.04 — reduce / fold dans Ocarina#

Un seul reduce dans toute la base de code. Mais c’est le réducteur central : il fait fonctionner chain_actions.

Code#

# src/ocarina/dsl/testing_with_railway/chain_actions.py
from functools import reduce

def chain_actions[T](
    first: ActionSuccess[T], *rest: ActionSuccess[T]
) -> ChainRunner[T]:

    def thunk() -> ActionChain[T]:
        def reducer(chain: ActionChain[T], step: ActionSuccess[T]) -> ActionChain[T]:
            if chain.has_failed():
                return chain
            return (
                chain.then(step.__action__)
                .failure(step.__failure_handler__)
                .success(step.__success_handler__)
                .execute()
            )
        return reduce(reducer, rest, first.execute())

    return ChainRunner(thunk=thunk)

Anatomie#

Concept FPRéalisation Python
Type accumulatorActionChain[T]
Type elementActionSuccess[T]
Reducerreducer(chain, step) -> ActionChain[T]
Initial valuefirst.execute()
Iterablerest (le tuple de *rest)
Short-circuitif chain.has_failed(): return chain
ResultActionChain[T] (le dernier)

Flot d’exécution#

initial = first.execute()                                          # → ActionChain(ok=True, result=Ok(...))

step 1 : reducer(<initial>, act2) :
  chain.has_failed() = False
  chain.then(act2.__action__).failure(...).success(...).execute()
  → action lève → Fail
  → failure_handler(exc)  ❌ FIRE
  → ActionChain(ok=False, result=Fail(exc))

step 2 : reducer(<failed>, act3) :
  chain.has_failed() = True
  return chain                                                     ⚠️  short-circuit

result du reduce = <failed ActionChain>

Pourquoi reduce#

# Approche impérative (équivalente)
def thunk() -> ActionChain[T]:
    chain = first.execute()
    for step in rest:
        if chain.has_failed():
            break
        chain = (
            chain.then(step.__action__)
            .failure(step.__failure_handler__)
            .success(step.__success_handler__)
            .execute()
        )
    return chain
ReduceBoucle impérative
Une seule valeur circule (le chain)Un état + une mutation
Reducer = fonction pure (testable en isolation)Logique inline
Sémantique « accumulateur » expliciteLogique implicite
Standard fonctionnelStandard impératif

C’est une expression d’intention : on replie une liste sur un accumulator. La sémantique est claire dès la première lecture. C’est ce qu’on appelle un catamorphisme.

03.05 — Programmation déclarative

03.05 — Programmation déclarative#

Un scénario d’Ocarina décrit, il n’exécute pas. C’est ce qui le rend factorisable, multipliable, et lisible.

Démonstration#

return [
    drive_page(
        act(on_homepage, open_then_verify_homepage)
            .failure(just_log_error("Failed to reach the homepage..."))
            .success(log_success_with_current_url_and_take_screenshot("On the homepage!")),
        act(on_homepage, click_book_call_page_cta)
            .failure(just_log_error("Failed to click on the 'Book a call' CTA..."))
            .success(just_log_success("Clicked on the 'Book a call' CTA!")),
    ),
    drive_page(
        act(on_book_a_call_page, verify_book_call_page)
            .failure(just_log_error("Failed to verify the 'Book a call' page..."))
            .success(log_success_with_current_url_and_take_screenshot("On the 'Book a call' page!")),
    ),
]

Ce code n’exécute rien. Il décrit :

03.06 — Generics PEP 695 dans Ocarina

03.06 — Generics PEP 695 dans Ocarina#

Ocarina exploite à fond la syntaxe générique introduite par PEP 695 (Python 3.12+). C’est ce qui rend tout l’écosystème typé sans que ce soit le foutoir.

Trois formes de PEP 695#

1. Classes génériques#

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

class TestSuite[Driver]:
    ...

class Watcher[Driver]:
    ...

class ChainRunner[T]:
    ...

class ValidationStartBlock[T]:
    ...

class CliStore[TKeys: str]:  # avec bound
    ...

Avant PEP 695 :

03.07 — Unions discriminées + TypeGuard + @final = unions « sealed »

03.07 — Unions discriminées + TypeGuard + @final = unions « sealed »#

Le combo qui rend Result[T] et TestResult à la fois typés et utilisables sans cast.

Union discriminée#

@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
  • Deux constructeurs : Ok[T] et Fail.
  • Le discriminant est l’isinstance check (pas un champ comme tag: Literal["ok", "fail"]).
  • Les deux sont @final : aucun sous-type possible. → l’union est exhaustive, donc « sealed ».

Narrowing, TypeGuard#

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)
def handle_result(result: Result[int]) -> str:
    if is_ok(result):
        return f"Got: {result.value}"          # mypy : result est Ok[int], donc .value existe
    if is_fail(result):
        return f"Error: {result.error}"        # mypy : result est Fail, donc .error existe
    # mypy peut détecter que cette branche est inatteignable (exhaustivité)

_BaseResult#

class _BaseResult:
    error: Exception | None

Cette base permet d’accéder à result.error sans narrower :