02.03.01 — The Result[T] type#

Source file: src/ocarina/railway/result.py — zero dependencies.

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 design decisions, dissected#

1. _BaseResult as a common parent#

class _BaseResult:
    error: Exception | None

One job: let mypy read result.error without narrowing first. Without the base, you’d have to write:

02.03.02 — The builder state machine#

Source file: src/ocarina/dsl/testing_with_railway/internals/action_chain.py

Here’s where the entire ROP mechanic lives: four successive typed states that literally (as in: the type-checker literally) refuse any syntactic improvisation. The DSL doubles as its own immune system.

Overview#

                      ┌─────────────────┐
   start(action)  →   │  ActionStart[T] │     an Action[T] = Thunk[Result[T]]
                      └────────┬────────┘
                               │ .failure(failure_handler)
                               ▼
                      ┌─────────────────┐
                      │ ActionFailure[T]│
                      └────────┬────────┘
                               │ .success(success_handler)
                               ▼
                      ┌─────────────────┐
                      │ ActionSuccess[T]│
                      └────────┬────────┘
                               │ .execute()   ── runs action(), fires the handler
                               ▼
                      ┌─────────────────┐
                      │ ActionChain[T]  │     holds (has_failed, result)
                      └──────┬──────────┘
                             │ .then(next_action_or_start)
              ┌──────────────┴──────────────┐
              │                             │
   has_failed=True                   has_failed=False
              │                             │
              ▼                             ▼
   ┌──────────────────┐         ┌───────────────────┐
   │ NeutralAction-   │         │ ActionStart[T]    │  → cycle starts over
   │ Start[T]         │         │  (the next action │
   │ (failure rail:   │         │   will execute)   │
   │  the whole chain │         └───────────────────┘
   │  becomes no-op,  │
   │  while keeping   │
   │  the fluent API) │
   └──────────────────┘
              │
              ▼ (.failure → .success → .execute, all no-op)
   ┌──────────────────┐
   │  ActionChain[T]  │   has_failed=True, result=<accumulated Fail>
   └──────────────────┘

The types Action, FailureHandler, SuccHandler#

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

• Action[T] is a Thunk. Calling ActionStart(action) does not run anything. The action just gets captured.
• FailureHandler gets the exception, not the Fail or the Result. Deliberate: 99% of handlers want to log the exception, take a screenshot, and move on.
• SuccHandler is an Effect (no argument). Same idea — a success handler logs a static message, snaps a screenshot, doesn’t need the result.

Typestate Pattern: why you can’t get it wrong#

The chain is strictly linear. Each method returns a different type that only exposes the next legal step:

02.03.03 — The failure rail: NeutralAction*#

Edge case in the builder: what happens when you chain an action onto an ActionChain that has already failed?

Why Neutral classes rather than a user-side if#

When you chain an action onto an already-failed ActionChain, you slide into three Neutral classes that honor the same API but do nothing.

Picture the DSL without this. User code would look like:

chain = act1.failure(h).success(h).execute()

if chain.is_ok():
    chain = chain.then(act2).failure(h).success(h).execute()

if chain.is_ok():
    chain = chain.then(act3).failure(h).success(h).execute()

Imperative, unreadable, and the whole point of the DSL is gone. The neutral rail keeps the writing flat:

02.03.04 — chain_actions and ChainRunner#

Source file: src/ocarina/dsl/testing_with_railway/chain_actions.py

The primitive that makes the DSL “flat.” Without it, an N-step scenario is a staircase of .then(...).failure(...).success(...).execute(). With it, it’s a list.

“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

(Excerpt from the docstring.)

02.03.05 — create_act and its hooks#

Source file: src/ocarina/dsl/testing_with_railway/constructors/create_act.py

The low-level primitive that builds a test step constructor. You call it once to mint a project-specific act verb. In theory a complex project might want several distinct act verbs with different hooks, but no real-world need has shown up yet — recommendation: create one. Not a singleton, just a convention.

Signature#

def create_act(
    pom: TPOM,
    action: Callable[[TPOM], TPOM],
    *,
    on_failure: Callable[[TPOM, Exception], Fail] | None = None,
    on_run_effect: Effect | None = None,
    act_counter_effect: Effect | None = None,
) -> ActionStart[TPOM]:

Code#

def run_action() -> Result[TPOM]:
    try:
        if act_counter_effect:
            act_counter_effect()
        else:
            ThreadsBasedActCounter().incr_act_call_count()

        if on_run_effect:
            on_run_effect()

        result_pom = action(pom)
        return Ok(result_pom)
    except Exception as exc:  # noqa: BLE001
        if on_failure:
            return on_failure(pom, exc)
        return Fail(error=exc)

return ActionStart(run_action)

1. The try wraps EVERYTHING: act_counter_effect, on_run_effect, and the action. If any side effect raises (say on_run_effect hits a dead resource), it counts as a test-step failure. Everything that runs “during the test step” IS the test step.
2. Order is deterministic: counter → run effect → action. The counter ticks even if the action ends up failing. Intentional — we want to know how many steps were attempted, not how many passed.
3. on_failure can return an enriched Fail: that’s where a WebDriverException becomes an HttpErrorPageReachedError.

Default ActCounter: ThreadsBasedActCounter#

from ocarina.opinionated.infra.act_counter import ActCounter as ThreadsBasedActCounter
# ...
if act_counter_effect:
    act_counter_effect()
else:
    ThreadsBasedActCounter().incr_act_call_count()

Source: src/ocarina/opinionated/infra/act_counter.py

02.03.06 — drive_page#

Source file: src/ocarina/opinionated/dsl/drive_page.py

Code#

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

Literally chain_actions.

Why this alias?#

1. Semantics: “I’m taking control of a page”#

Holy Book quote (“First scenarios”):

drive_page expresses the fact that you are taking control of one page. Any transition becomes explicit through the opening of a new drive_page.

return [
    drive_page(
        act(on_homepage, open_homepage)...,
        act(on_homepage, verify_homepage)...,
        act(on_homepage, click_cta)...,
    ),                                          # ⬅ closing control of the homepage
    drive_page(                                 # ⬅ opening control of the next page
        act(on_target_page, verify_target_page)...,
    ),
]

2. Discipline#

Mixing acts from two different POMs inside one drive_page is a mypy error (see 02-action-chain-states.md, “narrowing via act()” section). The semantics get enforced concretely.

02.03.07 — match_page / when#

Source file: src/ocarina/dsl/testing_with_railway/match_page.py

Bolted onto the framework after the fact. The Holy Book puts it bluntly: “match_page and when were added after the fact; the Igoristan was so random that the use case forced itself onto us. Their implementation was simple, proof of the grammar’s flexibility: other analogous structures could follow just as easily.”

The problem#

Some pages render differently:

• Cookies banner showing or not.
• A/B test serving two UI variants.
• Degraded mode (maintenance page) instead of the real page.
• Anti-bot captcha.
• …

The scenario needs to branch without breaking the ROP chain.