First jutsus

Datasets

Driving a test with a dataset is straightforward with Ocarina:

multi_login_dataset: Sequence[MappingProxyType[ImmutableCredentialsKeys, str]] = [
    MappingProxyType(
        {
            "login": "any",
            "password": "figatellu",
        }
    ),
    MappingProxyType(
        {
            "login": "Napoleon",
            "password": "figatellu",
        }
    ),
    MappingProxyType(
        {
            "login": "NoSicilianAllowed",
            "password": "figatellu",
        }
    ),
    MappingProxyType(
        {
            "login": "anonymous",
            "password": "figatellu",
        }
    ),
    MappingProxyType(
        {
            "login": "TheEmpire",
            "password": "figatellu",
        }
    ),
]

def _create_login_scenario(credentials: ImmutableCredentials) -> SeleniumTestScenario:
    """Welcome to functional factories."""

    def _scenario(driver: WebDriver, logger: ILogger):
        dashboard_creds = credentials  # <- [!] Provided by the closure

        on_dashboard_login_page = DashboardLoginPage(driver=driver)
        on_dashboard_welcome_page = DashboardWelcomePage(driver=driver)

        # * ...
        return Scenario(
            test_chain=[
                drive_page(
                    # * ...
                    act(
                        on_dashboard_login_page,
                        login_without_otp_and_with_retries(
                            dashboard_creds,  # <- [!]
                            retries_amount,
                            logger=logger,
                        ),
                    )
                    .failure(
                        just_log_error(
                            "Failed to connect to the dashboard without OTP...",
                        )
                    )
                    .success(
                        just_log_success(
                            f"Connected to the dashboard as {dashboard_creds['login']}!"
                            # ⬆️ [!]
                        )
                    ),
                ),
                drive_page(
                    act(on_dashboard_welcome_page, ...)
                    .failure(...)
                    .success(...)
                ),
            ]
        )

    return _scenario


multi_login_tests = [
    create_selenium_test(
        name=f"Login - {creds['login']}",
        test_scenario=_create_login_scenario(creds),
    )
    for creds in multi_login_dataset
]

A closure is all it takes.
Note that Scenario is declared from the inside here. It makes sense, since the whole point is to encapsulate it.

multi_login_tests is therefore a list of Test, which we unpack into a TestSuite, like so:

def create_suite(
    *,
    drivers_pool: SeleniumWebDriversPool,
) -> TestSuite:
    return TestSuite(
        name="Login (data-driven PoC)",
        tests=[*multi_login_tests],
        drivers_pool=drivers_pool,
    )

Smoke tests

📖 A smoke test is a quick, shallow verification of a software system to ensure its core functions work correctly, before running deeper tests. The goal is to catch obvious blocking failures early: if things go up in smoke right away, there's no point going further.

To run smoke tests at the start of a cycle with Ocarina:

E2E_CYCLE_NAME = "My very first cycle with Ocarina"

def create_e2e_test_cycle(drivers_pool: SeleniumWebDriversPool):
    """e2e test cycle."""
    return TestCycle(
        name=E2E_CYCLE_NAME,
        campaigns=[
            create_my_first_campaign(drivers_pool=drivers_pool),
        ],
        smoke_tests_campaigns=[
            create_my_first_smoke_campaign(drivers_pool=drivers_pool),
            create_my_second_smoke_campaign(drivers_pool=drivers_pool),
        ],
        mode="wait-for-all-smoke-tests",
    )

mode accepts two values (default: "fail-fast-on-first-smoke-campaigns-sequence-fail"):

• "fail-fast-on-first-smoke-campaigns-sequence-fail": as soon as one smoke-test campaign fails, the remaining ones are skipped.
• "wait-for-all-smoke-tests": all smoke-test campaigns run to completion, even if one fails along the way.

In both cases, main tests are skipped if any smoke test has failed.

Setup and teardown

Scenario accepts two optional callbacks: setup and teardown.

Scenario(
    setup=seed_test_user,
    test_chain=[
        drive_page(
            act(page, open_page)
                .failure(log_error("Failed to open..."))
                .success(log_success("Opened!")),
        ),
    ],
    teardown=delete_test_user,
)

Lifecycle

1. setup(): runs before the test_chain. On failure, the test_chain is skipped and teardown still runs. If every attempt fails due to setup, the test is marked SKIPPED (not FAILED),
2. test_chain: the actual test steps,
3. teardown(): always runs, even on failure. Errors are logged and ignored.

setup and teardown are Effect.
They are meant for infrastructure concerns: seeding a database, calling an API, cleaning up state...

If encapsulation is needed: closure.

Proxy pattern

One use case from the canonical example is HumanizedDriver:

class HumanizedDriver(WebDriver):
    def __init__(
        self, driver: WebDriver, **keyboard_config: Unpack[KeyboardConfig]
    ) -> None:
        object.__init__(self)
        self._driver = driver
        self._config = keyboard_config

    def find_element(
        self,
        by: str | RelativeBy = "id",
        value: str | None = None,
    ) -> _HumanizedWebElement:
        element = self._driver.find_element(by, value)
        return _HumanizedWebElement(element, self._config)

    def find_elements(
        self,
        by: str | RelativeBy = "id",
        value: str | None = None,
    ) -> list[WebElement]:
        elements = self._driver.find_elements(by, value)
        return [_HumanizedWebElement(el, self._config) for el in elements]

    def __getattr__(self, name: str):
        return getattr(self._driver, name)

The idea: return Web Elements that behave differently for user-like interactions. Keystrokes, in this case.
Transparent to the type system, transparent to the runtime.

Which then allows:

create_selenium_test(
    name="Send the form",
    test_scenario=lambda driver, logger: Scenario(
        test_chain=_send_form(
            HumanizedDriver(  # <- [!]
                driver,
                wpm=125,
                typo_rate=0.14,
                hesitation_rate=0.02,
                burst_rate=0.35,
                late_correction_rate=0.6,
            ),
            logger,
        ),
    ),
)

Or, with a closure:

def _scenario(driver: WebDriver, logger: ILogger):
    humanized_driver = HumanizedDriver(  # <- [!]
        driver,
        wpm=125,
        typo_rate=0.14,
        hesitation_rate=0.02,
        burst_rate=0.35,
        late_correction_rate=0.6,
    ),

    on_some_form_page = SomeFormPage(driver=humanized_driver)  # <- [!]

The same principle applies to the logger, routing it toward a sink, for instance. That case isn't canonically covered by Ocarina.

Reactive programming: NO

Ocarina test scenarios are intentionally static.
Yet a web application is dynamic, and sometimes capturing a value on the fly to pass it to a later step is perfectly legitimate.

Ocarina doesn't answer that. It doesn't need to.

Architectural answer

What we're after here is an in-memory cache.

We generate keys just before the test chain kicks off, and pass them to the POM actions. Actions write and read through a unique key.
The scenario just hands them out:

# * ...
cache = in_memory_cache_with_30m_ttl
username_key = reserve_free_cache_key(cache)
otp_send_date_key = reserve_free_cache_key(cache)

return [
    drive_page(
        # * ...
        act(
            on_dashboard_login_page,
            start_to_login_with_otp_and_with_retries(
                dashboard_creds,
                retries_amount,
                cache=cache,
                logger=logger,
                username_key=username_key,
                otp_send_date_key=otp_send_date_key,
            ),
        )
        .failure(
            just_log_error(
                "Failed to fill and confirm the login form with OTP...",
            )
        )
        .success(
            just_log_success(
                "Filled and confirmed the login form with OTP!"
            )
        ),
        act(
            on_dashboard_login_page,
            verify_otp_screen,
        )
        .failure(
            just_log_error(
                "Failed to verify the OTP screen...",
            )
        )
        .success(just_log_success("Verified the OTP screen!")),
        act(
            on_dashboard_login_page,
            type_otp_with_retries(
                retries_amount,
                cache=cache,
                logger=logger,
                username_key=username_key,
                otp_send_date_key=otp_send_date_key,
            ),
        )
        .failure(
            just_log_error(
                "Failed to confirm the OTP code...",
            )
        )
        .success(just_log_success("Confirmed the OTP code!")),
    ),
    # * ...
]

API calls and locks

API and locks have to be handled in the POMs.

⚠️ Ocarina doesn't support async/await and never will.

API calls: synchronous requests is enough.
Locks: threading.Lock if a single process at a time, otherwise Redis distributed locks are enough (redis.StrictRedis + redis.lock).

Browser profile

Some cases require passing a profile with --profile-path:

• Proxy authentication,
• Pre-loaded extensions,
• Local settings (language, timezone, certificates...),
• Etc.

---

Excellent work!
See you soon, Mojo reader.

---

"Muddy water is best cleared by leaving it alone."

― Alan Watts