02.03.04 — chain_actions et le ChainRunner#

Fichier source : src/ocarina/dsl/testing_with_railway/chain_actions.py

C’est la primitive qui rend le DSL « plat ». Sans elle, un scénario à N pas serait un escalier de .then(...).failure(...).success(...).execute(). Avec elle, c’est une liste.

« Parenthesis hell »#

Solution :

# Flat, readable syntax
runner = chain_actions(
    action1.failure(h1).success(h2),
    action2.failure(h3).success(h4),
    action3.failure(h5).success(h6)
)
chain = runner.run()  # Execute when ready

Benefits :

• Lazy evaluation : Build chain without executing
• Flat syntax : No deep nesting
• Automatic short-circuiting : Stops on first failure
• Composability : ChainRunner is a value

(Extrait de docstring.)

ChainRunner[T]#

@final
class ChainRunner[T]:
    def __init__(self, *, thunk: Thunk[ActionChain[T]]) -> None:
        self._thunk = thunk

    def run(self) -> ActionChain[T]:
        return self._thunk()

1. @final, pas d’héritage utilisateur.
2. Constructeur en kwargs-only (*, thunk: ...) empêche l’erreur classique : ChainRunner(no_idea) qui aurait été ambigu.
3. Le thunk est un Thunk[ActionChain[T]] : c’est-à-dire Callable[[], ActionChain[T]]. Aucun effet de bord n’arrive avant .run().
4. .run() retourne l’ActionChain[T] : on récupère donc la machine à états (cf. 02-action-chain-states.md) sur laquelle on peut interroger has_failed, is_ok, result.

chain_actions[T](first, *rest)#

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)

C’est un fold left :

Pourquoi first et *rest séparés ?#

1. Typage strict : on garantit au moins un ActionSuccess[T]. chain_actions() sans argument ne compile pas.
2. Initial du fold : il faut une valeur initiale (ActionChain[T]), donc on doit avoir exécuté le premier acte avant la boucle. Si l’on faisait *all, on devrait construire un ActionChain[T] neutre initial (« le rail vide »), ce qui forcerait à reconnaître un état « nothing done ».
3. Lisibilité : chain_actions(action1.failure...success(...), action2.failure...success(...), ...) se lit naturellement.

Pourquoi les fonctions internes#

def thunk() -> ActionChain[T]:
    def reducer(...): ...
    return reduce(reducer, rest, first.execute())
return ChainRunner(thunk=thunk)

• thunk capture en closure first et rest. chain_actions est donc une closure sur ses arguments. Quand on appelle plus tard runner.run(), le thunk() réutilise ces captures.
• reducer est défini à l’intérieur de thunk : il n’a pas besoin de capturer de variables externes, et le définir ici plutôt qu’au niveau module évite d’exposer un symbole privé au reste du package.

Composition : ChainRunner#

• On peut stocker un ChainRunner dans une variable et le réutiliser.
• On peut le passer en argument.
• On peut multiplier une liste de ChainRunner : [runner] * 5 répète l’exécution 5 fois quand la séquence est jouée. Le Holy Book le mentionne explicitement (chapitre « Scenarios composability », section « Répétitions »).
• L’aliasing est trivial :

click_confirm_cookies = drive_page(
    act(on_homepage, confirm_cookie_banner)
        .failure(log_error_with_current_url("Failed to dismiss banner..."))
        .success(log_success_with_current_url_and_take_screenshot("Banner dismissed!"))
)

# … plus tard …
return [
    match_page(branches=[
        when(check_that_page.has_cookies_banner, name="cookies", then=[click_confirm_cookies]),
        when(check_that_page.has_not_cookies_banner, name="no cookies", then=[]),
    ]),
    drive_page(...),
]

drive_page est un monomorphisme de chain_actions#

Le seul drive_page du framework est :

def drive_page(
    first: ActionSuccess[TPOM], *rest: ActionSuccess[TPOM]
) -> ChainRunner[TPOM]:
    return chain_actions(first, *rest)

Voir 06-drive-page.md pour la raison d’être de cet alias.

Trace d’exécution annotée#

Cas concret : trois acts, le second échoue.

runner = drive_page(act1, act2, act3)   # ⚠️  rien exécuté

chain = runner.run()                    # ▶︎  exécution :
  ┌─ first.execute()                      = ActionChain(ok=True, result=Ok(...))   ← état initial du fold
  │
  ├─ reduce(reducer, [act2, act3], <ActionChain ci-dessus>)
  │      │
  │      ├─ reducer(<ok chain>, act2) :
  │      │     chain.then(act2.__action__)        → ActionStart        (rail succès)
  │      │     .failure(act2.__failure_handler__) → ActionFailure
  │      │     .success(act2.__success_handler__) → ActionSuccess
  │      │     .execute()                         → action lève → Fail
  │      │                                        → failure_handler(exc)  ❌
  │      │                                        → ActionChain(ok=False, result=Fail(exc))
  │      │
  │      └─ reducer(<failed chain>, act3) :
  │            chain.has_failed() → True → return chain      ⚠️  act3 jamais exécuté
  │
  └─ return ActionChain(ok=False, result=Fail(exc))

functools#

from functools import reduce

Pas de toolz, pas de funcy, pas de wrapper maison. Juste functools.reduce, c’est la signature canonique d’un fold en Python.

Tests dédiés#

tests/scenarios/test_railway_and_action_chain.py contient des tests sur :

• chaînage de 3+ actes en succession,
• court-circuit après 2e échec (le 3e n’est jamais appelé),
• comptage d’act via ActCounter (le compteur n’est pas incrémenté pour les pas de test court-circuités).