02.10.01 — WebDriversPool[Driver]#

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

Thread-safe driver pool, max concurrency enforced by semaphore, monitored async warmup, clean shutdown. No reuse — every acquire() disposes its driver on the way out (clean state).

Constructor#

@final
class WebDriversPool[Driver]:
    def __init__(
        self,
        create_driver: Thunk[BuiltWebDriver[Driver]],
        max_size: int,
        warmup_timeout: float | None = None,
    ) -> None:
        self._create_driver = create_driver
        self._pool: Queue[BuiltWebDriver[Driver]] = Queue(max_size)
        self._semaphore = Semaphore(max_size)
        self._warmup_timeout = (
            warmup_timeout if warmup_timeout is not None and warmup_timeout > 0.1
            else 60.0 * 5
        )

acquire()#

@contextmanager
def acquire(self) -> Iterator[Driver]:
    try:
        driver, dispose = self._pool.get_nowait()
    except Empty:
        self._semaphore.acquire()
        try:
            driver, dispose = self._create_driver()
        except Exception:
            self._semaphore.release()
            raise

    try:
        yield driver
    finally:
        with suppress(Exception):
            dispose()
        self._semaphore.release()

              acquire()
                  │
                  ▼
   ┌──────────────────────────────────┐
   │ pool.get_nowait()                │── OK ─► driver, dispose         ← warmup case
   └──────────────┬───────────────────┘
                  │ Empty
                  ▼
   ┌──────────────────────────────────┐
   │ sem.acquire()                    │  ← blocks if N drivers alive
   └──────────────┬───────────────────┘
                  ▼
   ┌──────────────────────────────────┐
   │ create_driver()                  │── raises ─► sem.release(); raise
   └──────────────┬───────────────────┘
                  ▼
   ┌──────────────────────────────────┐
   │ yield driver                     │  ← caller uses it
   └──────────────┬───────────────────┘
                  ▼
                finally :
                  with suppress : dispose()       ← driver disposed
                  sem.release()                   ← release a slot

1. No reuse#

Queue.get_nowait() consumes the entry. Once pulled, the driver never goes back. At the end (finally), it’s disposed.

02.10.02 — DriverBuilder[Driver]#

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

Encapsulates profile handling (often a tmp copy of a user directory) and produces the (driver, dispose) pair the pool expects.

Why a builder?#

Building a Selenium driver with a profile requires:

1. Copy the profile to a temp folder (otherwise Firefox/Chrome locks the original).
2. Start the driver pointing at the temp folder.
3. At the end: driver.quit(), then delete the temp folder.

DriverBuilder factors out those three steps:

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

Immutable. Configure once, ship.

02.10.04 — ActCounter + ThreadsBasedActCounter#

Thread-local counter of act calls per attempt. Lets the report say “this test made 17 steps before dying at step 18.”

Interface#

# src/ocarina/infra/act_counter.py
class ActCounter:
    def get(self) -> int: ...
    def reset(self) -> None: ...
    def incr_act_call_count(self) -> None: ...

Default implementation#

# src/ocarina/opinionated/infra/act_counter.py
from threading import local

_thread_local = local()
_COUNTER_KEY: Final[str] = "ocarina_counter"


class ActCounter(_ActCounter):
    def get(self) -> int:
        return getattr(_thread_local, _COUNTER_KEY, 0)

    def reset(self) -> None:
        setattr(_thread_local, _COUNTER_KEY, 0)

    def incr_act_call_count(self) -> None:
        if not hasattr(_thread_local, _COUNTER_KEY):
            self.reset()
        setattr(_thread_local, _COUNTER_KEY, getattr(_thread_local, _COUNTER_KEY) + 1)

• threading.local(): namespace where attributes are per thread. Each thread has its own value.
• _COUNTER_KEY = "ocarina_counter": attribute key.
• No lock: each thread only writes its own value, so no race.

Why thread-local rather than a shared counter + lock?#

Ocarina’s architecture says one thread = one test. We’ve got no appetite for a whole state monad — might as well park this “impurity” (still a piece of state) right here. That’s it.

02.10.05 — Selenium adapters#

Source folder: src/ocarina/infra/selenium/

Everything that materializes the abstractions (POMBase, WebDriversPool, Screenshotter) in Selenium flavor. Lives outside the pure DSL — and since 1.1.3 it has a twin: src/ocarina/infra/playwright/ ships the exact same contracts in Playwright flavor. Swapping backend is now picking a folder, not writing one.

Files in the folder#

create_driver.py#

def _build_firefox(*, profile_path, driver_path, headless, wait_timeout) -> WebDriver:
    service = FirefoxService(executable_path=driver_path)
    options = FirefoxOptions()
    if headless:
        options.add_argument("-headless")
    if profile_path:
        options.profile = FirefoxProfile(profile_path)
    driver = Firefox(service=service, options=options)
    driver.implicitly_wait(wait_timeout)
    return driver


def _build_chrome(*, profile_path, driver_path, headless, wait_timeout) -> WebDriver:
    service = ChromeService(executable_path=driver_path)
    options = ChromeOptions()
    if headless:
        options.add_argument("--headless=new")
    if profile_path:
        options.add_argument(f"--user-data-dir={profile_path}")
    driver = Chrome(service=service, options=options)
    driver.implicitly_wait(wait_timeout)
    return driver


def _build_edge(*, profile_path, driver_path, headless, wait_timeout) -> WebDriver:
    # … same as Chrome with EdgeService/EdgeOptions/Edge …


def _build_safari(*, profile_path, driver_path, headless, wait_timeout) -> WebDriver:
    # … Safari supports neither driver_path (uses macOS' native safaridriver),
    #   nor profile_path (no custom profile), nor headless …

1. implicitly_wait set only once#

Each builder calls driver.implicitly_wait(wait_timeout) once. Any find_element that doesn’t find the element immediately waits up to wait_timeout seconds before raising NoSuchElementException.

02.10.06 — The Playwright actor: one owner thread#

Source folder: src/ocarina/infra/playwright/

The Playwright adapter mirrors the Selenium one file for file (create_driver, create_drivers_pool, create_screenshotter, driver_healthcheck, mixins), with one extra file: driver.py. That file deserves a chapter of its own. It is what reconciles Playwright’s sync API — inherently bound to one thread — with Ocarina’s threaded model (pool, warmup, Watcher). The model has been stressed and stabilized; it holds.

The problem: Playwright’s sync API is thread-affine#

Playwright’s sync API binds every object it hands out — Playwright, Browser, BrowserContext, Page, Locator, … — to the thread that called sync_playwright().start(). Under the hood it is a greenlet pinned to that thread. Touching any of them from another thread raises: