02.03.03 — The failure rail: NeutralAction*

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.04.03 — .otherwise(...) and _any_of

02.04.03 — .otherwise(...) and _any_of#

How Ocarina expresses a logical OR between two predicates without breaking error aggregation.

The problem#

Concrete case: you want “age == 18 OR age <= 65.” The ROP/invariants syntax is a strict chain of assertions, and every .assert_that() is an AND. How do you express an OR?

Code#

def otherwise(self, fallback: Predicate[T], *, msg: str | None = None) -> ValidationAssertBlock[T]:
    if self._last_predicate is None:
        raise RuntimeError("otherwise() must follow assert_that().")

    fallback_with_msg = _with_msg(fallback, msg, self._name)
    combined = _any_of(
        self._last_predicate, *([*self._otherwise_predicates, fallback_with_msg])
    )
    self._chain._steps.pop()                    # replace the last step
    self._chain.add_assertion(self._value, combined, self._name)
    self._last_predicate = combined
    self._otherwise_predicates.append(fallback_with_msg)
    return self
  1. Guard: otherwise() has to follow an assert_that(). With no prior predicate it raises. (In practice the type-checker already rules it out — ValidationStartBlock.otherwise doesn’t exist.)
  2. Combination: combine previous predicate + all accumulated otherwise predicates + the new one, via _any_of. Result: a composite predicate.
  3. Step replacement: pop the last step from _ValidationChain and push the combined one. Otherwise the original predicate AND the combined one would both be present, and the original error would resurface.

_any_of#

def _any_of[T](*predicates: _PredicateWithMsg[T]) -> _PredicateWithMsg[T]:
    def _try_predicate(p: _PredicateWithMsg[T], value: T) -> InvariantViolationError | None:
        try:
            p(value)
        except InvariantViolationError as exc:
            return exc
        else:
            return None

    def combined(value: T) -> None:
        errors = []
        for p in predicates:
            error = _try_predicate(p, value)
            if error is None:
                return                              # ✅ one predicate passes → all pass
            errors.append(error)

        formatted_error_messages = "  | " + "\n  | ".join(str(e) for e in errors)
        msg = (
            f"All predicates failed for value {value!r}.\n"
            "» At least one of the following conditions must be satisfied:\n"
            f"{formatted_error_messages}"
        )
        raise InvariantViolationError(msg)

    return _PredicateWithMsg(combined)
  1. First success wins: the moment one predicate passes, combined returns None.
  2. All fail: build an aggregated message listing every failure, separated by |.
  3. Result is a _PredicateWithMsg: plug it back into the chain like any other predicate.
  4. Aggregation is recursive: 3 successive otherwise() calls fold (P_orig OR P_otherwise1 OR P_otherwise2 OR P_otherwise3) into one predicate.

Example#

validate(age, name="age")
    .assert_that(is_equal_to(18), msg="Must be 18 or at most 65")
    .otherwise(is_less_than_or_equal_to(65), msg="Must be 18 or at most 65")
    .execute()
    .raise_if_invalid()
  1. If age == 18: is_equal_to(18) passes → OK.
  2. If age == 30: is_equal_to(18) raises, is_less_than_or_equal_to(65) passes → OK.
  3. If age == 70: both raise → InvariantViolationError with:
age: All predicates failed for value 70.
» At least one of the following conditions must be satisfied:
  | age: Must be 18 or at most 65
  | age: Must be 18 or at most 65

(The duplicated message comes from passing the same msg= twice. Intentional here — we want a single statement of intent.)

02.05.03 — TestFlow[Driver] — retry policy

02.05.03 — TestFlow[Driver] — retry policy#

Source file: src/ocarina/dsl/testing/internals/test_flow.py

One job: for a single Test, run the retry loop, acquire a fresh driver per attempt, apply the backoff policy.

Why an intermediate level between TestSuite and TestExecutor#

LevelDoesn’t know about
TestExecutorRetries, the driver pool
TestFlowInter-test concurrency, suite-level aggregation
TestSuiteOne attempt’s execution, retries

A genuinely strict separation. TestFlow is the only level that knows both the pool and the retry policy — combining them is its whole job.

02.10.03 — Screenshotter[TDriver]

02.10.03 — Screenshotter[TDriver]#

Source file: src/ocarina/infra/screenshotter.py

Generic, thread-safe screenshot utility. Driver-agnostic via a Protocol, configurable via a dataclass. Supports burst.

Protocol#

class ScreenshotDriver(Protocol):
    def save_screenshot(self, path: str) -> bool:
        """Save standard viewport screenshot."""

That’s it. Any object with save_screenshot(path: str) -> bool qualifies. Selenium WebDriver has it natively (driver.save_screenshot), Playwright can be wrapped, a fake driver in tests exposes it trivially.

ScreenshotterConfig[TDriver]#

@dataclass(frozen=True)
class ScreenshotterConfig[TDriver: ScreenshotDriver]:
    output_dir: Path
    file_ext: str = ".png"
    health_check: HealthCheck[TDriver] | None = None
    save_full_page: SaveFullPageScreenshot[TDriver] | None = None
    default_burst_delay: float = 0.5
    max_filename_retries: int = 500
    uuid_length: int = 8
FieldTypeRoleDefault
output_dirPathScreenshot directory — 
file_extstrExtension (.png, .jpg).png
health_checkHealthCheck[TDriver] | None(driver) -> None that raises if the driver is deadNone
save_full_pageSaveFullPageScreenshot[TDriver] | None(driver, path) -> bool for full-page screenshots (Firefox only)None
default_burst_delayfloatDelay between 2 shots in burst mode0.5s
max_filename_retriesintMax attempts to generate a unique name500
uuid_lengthintUUID suffix length in the name8

Immutable. Configure once, ship.

02.11.03 — Auto CLI store + flags + validation

02.11.03 — Auto CLI store + flags + validation#

Source file: src/ocarina/opinionated/cli/selenium/create_cli_store.py

Flags#

FlagDefaultTypeValidation
--driver-path""stris_file (except --browser safari)
--profile-pathNonestris_none OR is_dir
--browserNone (required)strargparse choices (per OS)
--not-headlessFalse (= headless by default)bool (store_true)phantom
--workers5intis_positive + is_not_zero
--logger"terminal+file"strchoices=LOGGERS_CHOICES
--wait-timeout10int≤ 60, > 0
--dont-force-delete-tmp-dirsFalsebool (store_true)phantom
--only[]list[str] (nargs="+")phantom + mutex
--exclude[]list[str] (nargs="+")phantom + mutex
_DEFAULT_WORKERS_AMOUNT = 5
_DEFAULT_LOGGER: SupportedLogger = "terminal+file"
_DEFAULT_BROWSER_AUTOMATION_TIMEOUT = 10
_MAX_BROWSER_AUTOMATION_TIMEOUT = 60

Literal[...]#

type SeleniumCliStoreKeys = Literal[
    "driver_path", "profile_path", "browser", "headless", "workers",
    "logger", "wait_timeout", "force_delete_tmp_dirs", "only", "exclude",
]

These keys are the store’s whole vocabulary. Call-sites use them: store.get("workers"), store.get("browser"), etc.

Chapter 02 — Ocarina, the framework

Chapter 02 — Ocarina, the framework#

Unpacks the whole ocarina framework (Python 3.14+, v1.1.10) and Railway Oriented Programming straight through to the reporting plugins. Structured as a layered walk: deepest (the Result[T] type) to most visible (the bootstrap that boots everything).

Chapter plan#

#SubjectFile or folder
01Technical identity, dependencies, toolchain01-identity.md
02Full module tree + layered diagram02-module-tree.md
03Railway Oriented Programming03-railway/
04Invariants (validate, assertions, chain_validations)04-invariants/
05Orchestration (Test → TestSuite → TestCampaign → TestCycle)05-orchestration/
06Lifecycle of a Scenario06-scenario.md
07Watcher07-watcher.md
08POMBase08-pom-base.md
09Ports & adapters (ILogger, ITakeScreenshot)09-ports.md
10Infrastructure (pool, builder, screenshotter, act counter, Selenium adapters)10-infra/
11Opinionated layer (CLI, loggers, plugins, bootstrap)11-opinionated/
12Custom types & custom errors12-custom-types-errors.md

Chapter 02.03 — Railway Oriented Programming

Chapter 02.03 — Railway Oriented Programming#

Ocarina’s beating heart. The whole DSL rides on this mechanic: success and failure as two parallel rails. A failure switches the train onto the failure rail and locks it there (short-circuit), but composition stays linear so the scenario still reads top-to-bottom.

Plan#

#FileSubject
0101-result.mdResult[T] = Ok[T] | Fail, is_ok/is_fail
0202-action-chain-states.mdState machine ActionStart → ActionFailure → ActionSuccess → ActionChain.
0303-neutral-rail.mdFailure rail: NeutralActionStart / NeutralActionFailure / NeutralActionSuccess.
0404-chain-actions-fold.mdChainRunner[T] (thunk) + chain_actions (fold).
0505-create-act-hooks.mdcreate_act and its hooks (on_failure, on_run_effect, act_counter_effect).
0606-drive-page.mddrive_page, a semantic alias.
0707-match-page-when.mdmatch_page / when + exception policy.

Lineage#

Direct inspiration: Railway Oriented Programming, popularized by Scott Wlaschin (F#).

02.03.04 — chain_actions and ChainRunner

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.04.04 — .then(...) and chain_validations(...)

02.04.04 — .then(...) and chain_validations(...)#

Two distinct mechanisms to compose multiple validations into one.

.then(new_value)#

def then[U](self, new_value: U, *, name: str | None = None) -> ValidationStartBlock[U]:
    return ValidationStartBlock(new_value, self._chain, name)
  1. Switches the type from T to U. The checker accepts the change.
  2. Keeps the chain (self._chain). Past assertions stay in it; new ones stack on top.
  3. Swaps the name: the name you pass to .then() is the one used for the next assertions, not the previous block’s.

Canonical use case#

validate(user, name="user")
    .assert_that(is_not_none)
    .then(user.email, name="user.email")
    .assert_that(is_email)
    .then(user.age, name="user.age")
    .assert_that(is_positive)
    .assert_that(is_less_than_or_equal_to(120))
    .execute()
    .raise_if_invalid()

Reads as: validate user, then user.email, then user.age. All pass → OK. One fails → every error rolls up into the AggregateInvariantViolationError.

02.05.04 — TestSuite[Driver] — parallelized engine

02.05.04 — TestSuite[Driver] — parallelized engine#

Source file: src/ocarina/dsl/testing/oc_test_suite.py

One job: run a sequence of Tests in parallel, handle saturation, ID filtering, and pre-execution invariant validation.

Constructor#

def __init__(
    self,
    *,
    name: str,
    tests: Sequence[Test[Driver]],
    create_logger: Thunk[ILogger],
    drivers_pool: WebDriversPool[Driver],
    take_screenshot: ITakeScreenshot[Driver],
    act_counter: ActCounter | None = None,
    transient_errors: tuple[type[Exception], ...] = (),
    copy_indicator: str = "COPY",
    put_space_after_copy_indicator: bool = True,
    max_retries_per_test: int | None = None,
    autoscreen_on_fail: bool = False,
    saturate_workers: bool | None = None,
    only_ids: Iterable[str] = (),
    exclude_ids: Iterable[str] = (),
) -> None:
    self._guards_on_invoke(tests)
    # ... assignments ...
    self._tests = filter_tests_by_ids(
        tests,
        only=only_ids,
        exclude=exclude_ids,
        logger=create_logger().set_prefix(lambda: f"{self.name}: filtering tests..."),
    )
ParameterTypeRoleDefault
namestrSuite name (appears in logs & reports) — 
testsSequence[Test[Driver]]List of tests to run — 
create_loggerThunk[ILogger]Logger factory — 
drivers_poolWebDriversPool[Driver]Shared pool — 
take_screenshotITakeScreenshot[Driver]Callable (driver, logger, category) -> None — 
act_counterActCounter | NoneIf NoneThreadsBasedActCounter()None
transient_errorstuple[type[Exception], ...]Exceptions that trigger a retry()
copy_indicatorstrPrefix for saturation-cloned tests"COPY"
put_space_after_copy_indicatorbool[COPY 1] (True) vs [COPY1] (False)True
max_retries_per_testint | NoneMax retries for a potentially flaky testNone_DEFAULT_MAX_RETRIES_PER_TEST (8)
autoscreen_on_failboolAutomatically capture one or several screenshots (burst) on failureFalse
saturate_workersbool | NoneClone tests to saturate every worker if the pool is large enoughNone → cascade: suite > campaign > bootstrap
only_idsIterable[str]--only filter()
exclude_idsIterable[str]--exclude filter()

Guardrails#

At __init__ time#

def _guards_on_invoke(self, tests: Sequence[Test[Driver]]) -> None:
    validate_test_runners_ids(tests=tests, name="tests").execute().raise_if_invalid()

→ Every test_id must be unique. Otherwise: AggregateInvariantViolationError. Raised at construction, before any execution attempt. Duplicate a test_id by accident and you find out instantly.