02.05.01 — Test[Driver]#

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

Signature#

@final
class Test[Driver]:
    def __init__(
        self,
        *,
        name: TestName,
        test_id: str | None = None,
        test_scenario: TestScenario[Driver],
        pre_test_scenarios_fragments: Sequence[TestScenarioFragment[Driver]] | None = None,
        post_test_scenarios_fragments: Sequence[TestScenarioFragment[Driver]] | None = None,
        skipped: bool = False,
    ) -> None:
        if test_id is None:
            test_id = name
        self.name = name
        self.test_id = test_id
        self._test_scenario = test_scenario
        self._pre_test_scenarios_fragments = pre_test_scenarios_fragments or []
        self._post_test_scenarios_fragments = post_test_scenarios_fragments or []
        self._skipped = skipped

Six parameters:

1. @final#

No user inheritance. Want a “special” test? Compose via fragments or a scenario — don’t subclass.

04.01 — "From the outside like a user" strategy#

The posture lives in the scenario tests’ conftest.py. No test cheats by peeking at internals.

tests/scenarios/conftest.py#

FakeDriver#

@dataclass
class FakeDriver:
    title: str = "fake"
    disposed: bool = False

That’s all. Ocarina’s core calls no Selenium API.

07.01 — Tree#

src/
├── main.py                                                     # entry point (see README)
│
├── api/                                                        # external HTTP clients
│   ├── retrieve_dashboard_otp_code.py                          # filter + sort + pick OTP
│   ├── get_otp_history.py                                      # HTTP GET /api/otp-history
│   └── constants/endpoints.py                                  # URLs of tests-workers endpoints
│
├── caches/                                                     # in-memory L1 cache
│   ├── l1.py                                                   # dogpile.cache.memory TTL 30m
│   └── reserve_free_cache_key.py                               # UUID + threading.Lock
│
├── constants/                                                  # constants
│   ├── pages/
│   │   ├── homepage.py                                         # HOMEPAGE_URL
│   │   ├── dashboard.py                                        # DASHBOARD_URL + input ids
│   │   ├── random_loaders.py
│   │   ├── sacred_upload.py
│   │   ├── corsicamon.py
│   │   ├── chaotic_form.py
│   │   ├── madness.py
│   │   ├── random_error_page.py
│   │   └── donkey_sausage_eater_detector.py
│   └── sys/
│       ├── redis_keys.py                                       # OTP_SEND_LOCK_KEY, ...
│       └── transient_errors.py                                 # (WebDriverException, HttpError...)
│
├── lib/
│   ├── connectors/test_steps/actions/                          # functions (TPOM) -> TPOM
│   │   ├── homepage.py
│   │   ├── dashboard_login.py                                  # ~10 connectors (with/without retries)
│   │   ├── dashboard_welcome.py
│   │   ├── dashboard_protected_page.py
│   │   ├── sacred_upload.py
│   │   ├── corsicamon_enter_api_key.py
│   │   ├── corsicamon_main.py
│   │   ├── chaotic_form.py
│   │   ├── random_error.py
│   │   ├── random_loaders.py
│   │   ├── madness.py
│   │   ├── this_is_bastia.py
│   │   ├── cors_errors.py
│   │   ├── bsod.py
│   │   ├── dsed.py
│   │   └── ids_bypassed.py
│   ├── custom_errors/
│   │   ├── http.py                                             # HttpErrorPageReachedError
│   │   └── transient_error.py                                  # TransientError
│   └── ext/                                                    # extensions / adapters
│       ├── ocarina/
│       │   ├── adapters/
│       │   │   ├── agnostic/
│       │   │   │   ├── act.py                                  # ⭐ act adapter (on_failure hook)
│       │   │   │   ├── env_getters.py                          # ⭐ typed EnvGetters adapter
│       │   │   │   └── match_page.py                           # ⭐ match_page adapter
│       │   │   └── selenium/
│       │   │       ├── test_suite.py                           # ⭐ TestSuite adapter
│       │   │       ├── test_campaign.py                        # ⭐ TestCampaign adapter
│       │   │       ├── cli_getters.py                          # typed getters from CliStore
│       │   │       ├── logs.py                                 # create_just_log_error, etc.
│       │   │       └── screenshotter.py                        # take_screenshot helper
│       │   └── regex/error_page.py                             # ERROR_PAGE_REGEX
│       ├── redis/
│       │   └── client.py                                       # Redis Singleton client
│       └── selenium/
│           ├── humanize/                                       # ⭐ HumanizedDriver + keyboard
│           │   ├── proxy.py
│           │   └── keyboard.py
│           ├── pages/verify_elements_presence.py
│           └── watchers/catch_me_if_you_can_watcher.py         # ⭐ Watcher callback
│
├── pages/                                                      # POMs (Selenium + SeleniumTitleMixin)
│   ├── homepage.py
│   ├── random_loaders.py
│   ├── chaotic_form.py
│   ├── random_error.py
│   ├── sacred_upload/
│   │   ├── sacred_upload.py
│   │   └── fixtures/                                           # files to upload
│   ├── dashboard/
│   │   ├── login.py                                            # ⭐ with OTP, retries, redis locks
│   │   ├── welcome_page.py
│   │   └── protected_page.py
│   ├── corsicamon/
│   │   ├── enter_api_key.py
│   │   └── main.py
│   ├── madness/
│   │   ├── base.py
│   │   ├── matchers.py                                         # has_cors, has_this_is_bastia
│   │   ├── cors.py
│   │   └── this_is_bastia.py
│   └── donkey_sausage_detector/
│       └── ids_bypassed.py
│
└── tests/
    ├── cycles/e2e.py                                           # ⭐ TestCycle (smoke + main)
    ├── campaigns/
    │   ├── global_smoke_tests.py                               # smoke
    │   ├── corsicamon.py                                       # main (+ smoke variant)
    │   ├── dashboard_login.py                                  # main
    │   ├── randomness.py                                       # main
    │   └── sacred_upload.py                                    # main
    ├── suites/
    │   ├── global_smoke_tests.py
    │   ├── randomness.py
    │   ├── dashboard/
    │   │   ├── access/
    │   │   │   ├── happy_paths.py
    │   │   │   └── unhappy_paths.py
    │   │   └── data_driven/multi_login.py
    │   ├── sacred_upload/{happy_paths,unhappy_paths}.py
    │   └── corsicamon/{smoke_tests,happy_paths,unhappy_paths}.py
    └── scenarios/
        ├── homepage/verify_homepage.py
        ├── dashboard/
        │   ├── access/{happy_paths,unhappy_paths}.py
        │   ├── back_to_igoristan.py
        │   └── data_driven/{multi_login,datasets/multi_login}.py
        ├── sacred_upload/{upload_files,just_go_back_to_igoristan}.py
        ├── corsicamon/{enter_api_key,new_draw,add_corsicamon,back_to_igoristan}.py
        └── randomness/
            ├── level_1/{random_error_page,random_loaders_page}.py
            ├── level_2/{dsed,madness}.py
            ├── level_3/chaotic_form.py
            └── level_4/walkthrough.py

Folder semantics#

lib/ vs pages/#

1. pages/: POMs — encapsulate a page’s state (locators, action methods).
2. lib/connectors/test_steps/actions/: connectors — functions that call into POMs.

act(on_homepage, open_homepage)
#                ^^^^^^^^^^^^^
#                connector defined in lib/connectors/test_steps/actions/homepage.py

def open_homepage(p: Homepage) -> Homepage:
    return p.open()

The AI project’s CLAUDE.md enforces this as a discipline:

03.03 — Lazy evaluation#

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

Zzz#

ChainRunner#

runner = drive_page(act1, act2, act3)        # ⚠️  nothing executed
# … later …
chain = runner.run()                         # ▶︎  execution

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).

07.03 — Dashboard login scenarios#

Three families: happy paths, unhappy paths, data-driven multi-login. All exercise the 10%-fail useAuth and OTP coordination.

dashboard_login#

# src/tests/campaigns/dashboard_login.py
def create_igoristan_login_campaign(*, drivers_pool: SeleniumWebDriversPool) -> TestCampaign:
    return TestCampaign(
        name="Dashboard login",
        suites=[
            create_igoristan_login_happy_paths_test_suite(drivers_pool=drivers_pool),
            create_igoristan_login_unhappy_paths_test_suite(drivers_pool=drivers_pool),
            create_igoristan_login_data_driven_test_suite(drivers_pool=drivers_pool),
        ],
    )

Happy paths#

# src/tests/suites/dashboard/access/happy_paths.py
def create_igoristan_login_happy_paths_test_suite(*, drivers_pool) -> TestSuite:
    return TestSuite(
        name="Login happy paths",
        tests=[
            test_dashboard_login_without_otp_happy_path,
            test_dashboard_login_with_otp_happy_path,
            test_dashboard_login_page_back_to_igoristan_button,
        ],
        drivers_pool=drivers_pool,
    )

Test 1: login without OTP#

def dashboard_login_without_otp_happy_path(driver, logger):
    dashboard_creds = create_env_getters().get_credentials("dashboard")
    on_dashboard_login_page = DashboardLoginPage(driver=driver)
    on_dashboard_welcome_page = DashboardWelcomePage(driver=driver)

    just_log_error = create_just_log_error(logger=logger)
    log_error_with_current_url = create_log_error_with_current_url(logger=logger, driver=driver)
    just_log_success = create_just_log_success(logger=logger)
    log_success_with_current_url_and_take_screenshot = create_log_success_with_current_url_and_take_screenshot(...)

    retries_amount = max(get_max_workers(), 10)

    return [
        drive_page(
            act(on_dashboard_login_page, open_dashboard_login_page)
                .failure(just_log_error("Failed to open the dashboard login page..."))
                .success(just_log_success("Opened the dashboard login page!")),
            act(on_dashboard_login_page, verify_dashboard_login_page)
                .failure(log_error_with_current_url("Failed to verify..."))
                .success(log_success_with_current_url_and_take_screenshot("Verified!")),
            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("Connected to the dashboard!")),
        ),
        drive_page(
            act(on_dashboard_welcome_page, verify_dashboard_welcome_page)
                .failure(log_error_with_current_url("Failed to verify..."))
                .success(log_success_with_current_url_and_take_screenshot("Verified!")),
        ),
    ]

1. retries_amount = max(get_max_workers(), 10): at least 10 internal login retries to beat the 10% random failure.
2. login_without_otp_and_with_retries(creds, retries_amount, logger=logger): closure capturing creds, retry count, and logger. Returns Callable[[DashboardLoginPage], DashboardLoginPage].
3. 2 drive_pages: login page then welcome page. Each opens, verifies, screenshots.
4. Log factories everywhere: no inline lambdas.
5. SeleniumTitleMixin: get_current_title() feeds act’s on_failure hook.

Test 2: login with OTP#

def dashboard_login_with_otp_happy_path(driver, logger):
    # ... same setup ...

    cache = in_memory_cache_with_30m_ttl
    fresh_cache_key_for_username = reserve_free_cache_key(cache)
    fresh_cache_key_for_otp_send_button_click_date = reserve_free_cache_key(cache)

    return [
        drive_page(
            act(on_dashboard_login_page, open_dashboard_login_page)...,
            act(on_dashboard_login_page, verify_dashboard_login_page)...,
            act(on_dashboard_login_page,
                start_to_login_with_otp_and_with_retries(
                    dashboard_creds, retries_amount,
                    cache=cache, logger=logger,
                    username_cache_key=fresh_cache_key_for_username,
                    otp_send_button_click_date_cache_key=fresh_cache_key_for_otp_send_button_click_date,
                ))...,
            act(on_dashboard_login_page, verify_otp_screen)...,
            act(on_dashboard_login_page,
                type_otp_with_retries(
                    retries_amount,
                    cache=cache, logger=logger,
                    username_cache_key=fresh_cache_key_for_username,
                    otp_send_button_click_date_cache_key=fresh_cache_key_for_otp_send_button_click_date,
                ))...,
        ),
        drive_page(
            act(on_dashboard_welcome_page, verify_dashboard_welcome_page)...,
            act(on_dashboard_welcome_page, click_on_go_to_nested_page_btn)...,
        ),
        drive_page(
            act(on_dashboard_protected_page, verify_dashboard_protected_page)...,
        ),
    ]

Three drive_pages, five acts in the first. L1 cache + reserved keys share values between acts (see 08-caches-locks.md).

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:

Chapter 04 — Framework internal tests#

How Ocarina tests itself. Five test families, a clear-eyed coverage policy, an Allure report archived on GitHub Pages.

Outline#

Summary table#

Approach#

In conftest.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.06 — Scenario[Driver]#

Source file: src/ocarina/custom_types/scenario.py

Dataclass#

@final
@dataclass(frozen=True)
class Scenario[Driver]:
    test_chain: TestChain
    setup: TestSetup = field(default=None)
    teardown: TestTeardown = field(default=None)
    watchers: TestWatchers[Driver] | None = field(default=None)

# src/ocarina/custom_types/test_components.py
type TestChain = Sequence[ChainRunner[Any]]
type TestSetup = Effect | None
type TestTeardown = Effect | None
type TestWatchers[Driver] = Sequence[Watcher[Driver]] | None

Lifecycle (per attempt)#

1. setup()           — optional, free-form Effect (DB, API, …)
       → raises    :  test_chain skipped, jump to teardown,
                      return Outcome(setup_failed=True, should_retry=True)
       → ok        :  continue to test_chain

2. test_chain        — the actual chain (Sequence[ChainRunner])
       (watchers run during this time)

3. teardown()        — optional, always executed
       → raises    :  log warning + ignore (does not affect the verdict)

If EVERY attempt raises during setup:
    → test marked SKIPPED (not FAILED)
    → warning log « setup keeps failing »

setup and teardown#

setup and teardown are driver-free and injection-free by design.

09.03.07 — Refactor skills#

Skills that refactor the existing automated-test base.

Listing (potentially non-exhaustive)#

refactor-fragmentation#

input  : the test base
output : DRY refactor suggestions:
            - blocks repeated in 3+ scenarios → extract into a fragment
            - near-identical connectors → extract a parameterized connector
            - repeated log+screenshot sequences → extract a helper
         + per-case recommendation

CLAUDE.md: