02.11.01 — CliBuilder + CliArg + _SilentArgumentParser

02.11.01 — CliBuilder + CliArg + _SilentArgumentParser#

Source file: src/ocarina/opinionated/cli/builder.py

Declarative overlay on top of argparse. Aggregates validation errors, rewrites the help output on error, and registers post-parse effects.

_SilentArgumentParser#

class _SilentArgumentParser(ArgumentParser):
    def error(self, message: str) -> Never:
        """Raise an error."""
        raise ValueError(message)

The standard ArgumentParser calls sys.exit(2) directly on error. With that turd of a default, you can’t intercept and you can’t aggregate multiple errors.

_SilentArgumentParser reroutes errors as ValueError. So CliBuilder.parse can catch and aggregate them.

02.11.02 — CliStore[TKeys] + _CliField[T] + phantom_validate

02.11.02 — CliStore[TKeys] + _CliField[T] + phantom_validate#

Source files: src/ocarina/opinionated/cli/store.py, src/ocarina/opinionated/cli/phantoms.py

Write-once store for parsed CLI values. Each field validates on write. TKeys: Literal[...] gives key autocomplete.

_CliField[T] — write-once field#

class _CliField[T]:
    def __init__(self, *, validate: ValidationChainBuilder[T]) -> None:
        self._value: T | _Unset = _UNSET
        self._validate = validate

    def set(self, value: T) -> None:
        if not isinstance(self._value, _Unset):
            raise RuntimeError("Value already set.")
        self._validate(_validate(value)).execute().raise_if_invalid()
        self._value = value

    def get(self) -> T:
        if isinstance(self._value, _Unset):
            raise RuntimeError("Value not set yet.")
        return self._value

1. _Unset sentinel#

class _Unset:
    pass

_UNSET = _Unset()

Why not None? Because None can be a legitimate value (--profile-path isn’t mandatory; its post-parse value can legitimately be None). A dedicated sentinel separates “not set yet” from “set to None”.

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.

02.11.04 — Loggers: PrintLogger, FileLogger, PrintAndFileLogger, MutedLogger

02.11.04 — Loggers: PrintLogger, FileLogger, PrintAndFileLogger, MutedLogger#

Source folder: src/ocarina/opinionated/loggers/

Four canonical ILogger implementations + a create_matching_logger factory. All opt-in.

create_matching_logger#

LOGGERS_CHOICES = ("terminal", "file", "terminal+file", "muted")

def create_matching_logger(mode: SupportedLogger) -> ILogger:
    if mode == "terminal":
        return PrintLogger()
    if mode == "file":
        return FileLogger(base_dir=get_default_log_dir())
    if mode == "terminal+file":
        return PrintAndFileLogger(base_dir=get_default_log_dir())
    if mode == "muted":
        return MutedLogger()
    raise ValueError(f"Unsupported logger mode: {mode}")

And:

def get_default_log_dir() -> Path:
    return Path.cwd() / ".ocarina_logs"

→ By default, logs land in .ocarina_logs/ at the user project root. Gitignored by convention.

PrintLogger#

class PrintLogger(ILogger):
    def __init__(self) -> None:
        self._prefix_thunk: Thunk[str] = lambda: ""
        self._domain_taxonomy: tuple[str, ...] = ("",)

    def set_prefix(self, prefix_thunk: Thunk[str]) -> Self:
        self._prefix_thunk = prefix_thunk
        return self

    def set_domain_taxonomy(self, taxonomy: tuple[str, ...]) -> Self:
        self._domain_taxonomy = taxonomy
        return self

    def raw(self, *args, stream=None, **kwargs) -> None:
        if stream is None:
            stream = sys.stdout
        print(*args, file=stream, **kwargs)

    def info(self, msg, *args, exc=None, **kwargs):
        self._print_log("[INFO]", msg, *args, exc=exc, stream=sys.stdout)
    # ... id. for critical / error / warning / debug / success / test_name ...
  • State: _prefix_thunk (Thunk) + _domain_taxonomy (tuple).
  • Output: sys.stdout (info, debug, success, test_name) or sys.stderr (critical, error, warning, exception).
  • Format: typically <prefix> [LEVEL] <message>, with ANSI colors if terminal.
  • cleanup(): no-op.

FileLogger#

@final
class FileLogger(PrintLogger):
    def __init__(
        self, *,
        base_dir: Path,
        with_flush_effect: bool = True,
        with_fallback_on_print_logger_when_no_taxonomy_effect: bool = True,
    ) -> None:
        self._base_dir = base_dir
        self._with_flush_effect = with_flush_effect
        self._with_fallback_on_print_logger_when_no_taxonomy_effect = (
            with_fallback_on_print_logger_when_no_taxonomy_effect
        )
        super().__init__()

    @property
    def _log_file_path(self) -> Path:
        if len(self._domain_taxonomy) == 1:
            folder_path = self._base_dir
            file_name = f"{self._domain_taxonomy[0]}.log"
        else:
            folder_path = self._base_dir.joinpath(*self._domain_taxonomy[:-1])
            file_name = f"{self._domain_taxonomy[-1]}.log"
        folder_path.mkdir(parents=True, exist_ok=True)
        return folder_path / file_name
TaxonomyFolderFile
("") — → Fallback to PrintLogger if enabled
("annoncements",)base_dirannoncements.log
("e2e", "Login")base_dir/e2eLogin.log
("e2e", "Login", "happy paths", "Login without OTP")base_dir/e2e/Login/happy pathsLogin without OTP.log

The taxonomy is the folder tree. The DOCX reports (see 05-plugins-reports.md) walk it recursively and emit one .docx per test case.

02.11.05 — Report plugins

02.11.05 — Report plugins#

Source folder: src/ocarina/opinionated/plugins/reports/

pretty_print_results, results_to_json, generate_docx_proof, timing. All opt-in. Run sequentially or in parallel via run_plugins.

pretty_print_results#

# src/ocarina/opinionated/plugins/reports/pretty_print_results.py
def pretty_print_results(results: TestCycleResults, *, with_colors: bool = False) -> None:
    """Display the full test results."""
Campaign
• Suite
  > Test case 1
    » PASSED
  > Test case 2
    » FAILED
      → Error message
        ⫸ At step 3
  > Test case 3
    » SKIPPED

Test results:
1 FAILED | 1 PASSED | 1 SKIPPED
AspectDetail
Hierarchy3 levels: campaign , suite >, case »
LevelsPASSED (green), FAILED (red), SKIPPED (gray)
ErrorMessage + step number where it failed ("At step 3")
SummaryAggregated count at the bottom

results_to_json#

def generate_json_results(*, results: TestCycleResults, output_dir: Path, logger: ILogger) -> None:
    """Write test results to a JSON file in results_dir."""
    output_dir.mkdir(parents=True, exist_ok=True)
    max_stem_length = 8
    max_attempts = 500
    for _ in range(max_attempts):
        filename = f"{uuid.uuid4().hex[:max_stem_length]}.json"
        file_path = output_dir / filename
        if not file_path.exists():
            break
    else:
        raise RuntimeError(f"Can't generate unique JSON filename in: {output_dir}.")
{
  "Campaign": {
    "Suite": {
      "test_case": [
        {"status": "success" | "fail", "error": "..."},
        counter,
        metadata
      ]
    }
  }
}
def _result_to_serializable(v: Any) -> dict[str, str]:
    if is_ok(v):
        return {"status": "success"}
    if is_fail(v):
        return {"status": "fail", "error": str(v.error)}
    raise TypeError(f"Expected Ok or Fail instances, but got: {type(v)}")

Filename is an 8-char hex UUID → 16⁸ ≈ 4 billion options. No collisions in practice. 500 safety retries.

02.11.06 — bootstrap + run_plugins

02.11.06 — bootstrap + run_plugins#

Source file: src/ocarina/opinionated/launcher/bootstrap.py

Entry point of an Ocarina project.
Three steps: cycle.run_all → run_plugins → post_exec.

Code#

def bootstrap[T](
    *,
    test_cycle: TestCycle[T],
    run_plugins: Callable[[TestCycleResults], None],
    post_exec: Callable[[TestCycleResults], None] | None = None,
    saturate_workers: bool = True,
) -> None:
    results = test_cycle.run_all(saturate_workers=saturate_workers)
    run_plugins(results)
    if post_exec:
        post_exec(results)

Order#

1. cycle.run_all()  →  results
2. run_plugins(results)
3. post_exec(results) (optional)
StepEffectWhy this order
1All tests runWe need results for the rest
2Plugins fire (DOCX, JSON, etc.)Must run before post_exec since it can take generation time
3Post-exec (pretty_print + sys.exit)Last thing — after this, the process is dead

Typing#

run_plugins: Callable[[TestCycleResults], None]
post_exec: Callable[[TestCycleResults], None] | None = None
  • Both receive the results as an argument: