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

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

Three lines in custom_types/ + one in railway/. The whole DSL rides on them.

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
TypeFP semanticsDefinition
EffectSide effect (deferred)() -> None
Thunk[T]Deferred computation with value() -> T (laziness + value)
Result[T]Computation that can failDiscriminated union Ok[T] | Fail

Effect vs Thunk distinction#

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

Function returns something consumed elsewhere → Thunk[T].
Otherwise → Effect.

03.02 — Closures as the primitive of inversion of control

03.02 — Closures as the primitive of inversion of control#

No DI container. No injection framework. A closure is Ocarina’s injection primitive.

The idea#

A closure captures values from its enclosing scope. Return a function from another function and the outer parameters are available inside, even after the outer function returns.

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

Also currying: make_handler(logger)(msg)(exc). Each call captures a new layer of environment.

03.03 — Lazy evaluation

03.03 — Lazy evaluation#

In Ocarina, nothing is executed until you explicitly trigger it. That’s what makes scenarios composable as values.

Zzz#

LocationFormTrigger
ChainRunner[T]Thunk[ActionChain[T]]runner.run()
validate(...)ValidationStartBlock / ValidationAssertBlock.execute()
match_page(...)returns a ChainRunner[Any].run() (via the surrounding chain)
Watcher.callbackCallable[[Watcher], None]_loop when start() is called
logger.set_prefix(thunk)Thunk[str]recomputed at each log call
Scenario.setup / teardownEffectcalled by TestExecutor
bootstrap(post_exec=...)Callable[[TestCycleResults], None]called after run_plugins
CliBuilder(effects_factory=lambda ns: (...))Effectscalled after the argparse parse
test_scenario: TestScenario[Driver]Callable[[Driver, ILogger], Scenario[Driver]]called by Test.spawn
dispatch[mode]() in TestCycle.run_alldict of Thunk[bool]called on lookup

ChainRunner#

runner = drive_page(act1, act2, act3)        # ⚠️  nothing executed
# … later …
chain = runner.run()                         # ▶︎  execution
CapabilityWithout lazinessWith laziness
Store a scenario in a variableimpossible (already executed)trivial
Multiply [runner] * 5executes once, you get 5 refs to the resultexecutes 5 times
Pass a runner to another runner (composition)impossibletrivial
Reorder acts in a test refactorhardtrivial

validate(...).execute()#

v = validate(value, name="x").assert_that(is_positive).assert_that(is_not_zero)
# … you can compose …
combined = chain_validations(v, other_validation)
# … nothing executed so far …
combined.execute().raise_if_invalid()        # ▶︎  execution + aggregation

That’s how _ValidationChain collects every error before raising a single one (AggregateInvariantViolationError).

03.04 — reduce / fold in Ocarina

03.04 — reduce / fold in Ocarina#

One reduce in the whole code base. But it’s the central reducer — it’s what makes chain_actions work.

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)

Anatomy#

FP conceptPython realization
Accumulator typeActionChain[T]
Element typeActionSuccess[T]
Reducerreducer(chain, step) -> ActionChain[T]
Initial valuefirst.execute()
Iterablerest (the *rest tuple)
Short-circuitif chain.has_failed(): return chain
ResultActionChain[T] (the last one)

Execution flow#

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 raises → 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 of the reduce = <failed ActionChain>

Why reduce#

# Imperative approach (equivalent)
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
ReduceImperative loop
A single value flows (the chain)A state + a mutation
Reducer = pure function (testable in isolation)Inline logic
Explicit “accumulator” semanticsImplicit logic
Functional standardImperative standard

An expression of intent: we fold a list onto an accumulator. Semantics jump out on the first read. This is what’s called a catamorphism.

03.05 — Declarative programming

03.05 — Declarative programming#

An Ocarina scenario describes, it doesn’t execute. That’s what makes it factorable, multipliable, readable.

Demonstration#

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!")),
    ),
]

This code executes nothing. It describes:

03.06 — PEP 695 generics in Ocarina

03.06 — PEP 695 generics in Ocarina#

Ocarina leans hard on the generics syntax from PEP 695 (Python 3.12+). It’s what keeps the whole typed ecosystem from being a mess.

Three forms of PEP 695#

1. Generic classes#

@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]:  # with bound
    ...

Before PEP 695:

from typing import Generic, TypeVar

T = TypeVar("T")

class Ok(_BaseResult, Generic[T]):
    value: T
    error: None = None

2. Parameterized type aliases#

type Result[T] = Ok[T] | Fail
type Thunk[T] = Callable[[], T]
type Action[T] = Thunk[Result[T]]
type FailureHandler = Callable[[Exception], None]
type Predicate[T] = Callable[[T], None]
type Effect = Callable[[], None]
type TestScenario[Driver] = Callable[[Driver, ILogger], Scenario[Driver]]
type TestScenarioFragment[Driver] = Callable[[Driver, ILogger], TestChain]
type TestChain = Sequence[ChainRunner[Any]]
type TestWatchers[Driver] = Sequence[Watcher[Driver]] | None

Before PEP 695:

03.07 — Discriminated unions + TypeGuard + @final = sealed unions

03.07 — Discriminated unions + TypeGuard + @final = “sealed” unions#

The combo that makes Result[T] and TestResult both typed and usable without casts.

Discriminated union#

@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
  • Two constructors: Ok[T] and Fail.
  • Discriminant is the isinstance check (not a field like tag: Literal["ok", "fail"]).
  • Both are @final — no possible subtype. The union is exhaustive, i.e. “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 is Ok[int], so .value exists
    if is_fail(result):
        return f"Error: {result.error}"        # mypy: result is Fail, so .error exists
    # mypy can detect this branch is unreachable (exhaustiveness)

_BaseResult#

class _BaseResult:
    error: Exception | None

This base lets you access result.error without narrowing: