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

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.

02.05.05 — Worker saturation#

Ocarina-specific trick: if a suite has fewer tests than workers, tests get randomly cloned until the worker count is filled. Copies are renamed [COPY 1] <name>, [COPY 2] <name>, etc.

Why#

The Holy Book formalizes the rationale in the “First real-world hurdles” chapter:

Its saturate_workers option lets you force random cloning of tests inside a suite.

As soon as there are more available workers in the DriversPool than there are tests to run in a suite, Ocarina will randomly clone tests, start every driver, and assign each one a test to perform.

Chapter 02.05 — Orchestration#

Test → TestSuite → TestCampaign → TestCycle: how each level slots into the next, who owns parallelization, who owns retries, who decides skipping, and where the pre-execution invariants live.

Plan#

Diagram#

            ┌────────────────────────────────────────────────┐
            │                  TestCycle                     │
            │                                                │
            │  ┌──────────────────────────────────────────┐  │
            │  │ smoke_tests_campaigns                    │  │ ◄── first, gate
            │  │  └─ TestCampaign                         │  │     mode: fail-fast |
            │  │      └─ TestSuite                        │  │           wait-for-all
            │  │          └─ Test                         │  │
            │  └──────────────────────────────────────────┘  │
            │                                                │
            │  ┌──────────────────────────────────────────┐  │
            │  │ campaigns (main)                         │  │ ◄── skipped if a
            │  │  └─ TestCampaign                         │  │     smoke fails
            │  │      └─ TestSuite                        │  │
            │  │          └─ Test                         │  │
            │  └──────────────────────────────────────────┘  │
            └────────────────────────────────────────────────┘

Who does what?#

The strict separation is deliberate. TestExecutor knows nothing about retries. TestFlow knows nothing about concurrency. TestSuite knows nothing about campaign-level aggregation. One axis of responsibility per class.

02.05.06 — TestCampaign[Driver]#

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

One job: run a sequence of suites in order, share a workers config, and declare campaign_has_failed.

Code#

class TestCampaign[Driver]:
    def __init__(
        self,
        *,
        name: str,
        suites: Sequence[TestSuite[Driver]],
        max_workers: int,
        saturate_workers: bool | None = None,
    ) -> None:
        validate_test_suites_names(suites=suites, name="suites").execute().raise_if_invalid()

        self.name = name
        self._suites = suites
        self._results: TestCampaignResults = {}
        self._max_workers = max_workers
        self._saturate_workers = saturate_workers

        for suite in self._suites:
            suite._campaign_name = self.name

    def run_all(
        self, *, skip_all: bool = False, saturate_workers: bool = True
    ) -> TestCampaignResults:
        self._results.clear()

        resolved_saturate_workers = (
            self._saturate_workers
            if self._saturate_workers is not None
            else saturate_workers
        )

        if skip_all:
            for suite in self._suites:
                self._results[suite.name] = {
                    name: (None, -1, test_id)
                    for name, test_id in suite.test_names_and_ids
                }
            return self._results

        for suite in self._suites:
            self._results[suite.name] = suite.run(
                max_workers=self._max_workers,
                saturate_workers=resolved_saturate_workers,
            )

        return self._results


def campaign_has_failed(results: TestCampaignResults) -> bool:
    return any(
        is_test_result_fail(outcome)
        for campaign_results in results.values()
        for outcome, _, _ in campaign_results.values()
    )

Anatomy#

Constructor guard#

validate_test_suites_names(suites=suites, name="suites").execute().raise_if_invalid()

→ Suite names inside a campaign must be unique — case-insensitively since 1.1.10 (NFC + casefold). Raised at construction.