configmodule¶

The configmodule package handles environment variables, repository discovery, settings parsing, and lab loading.

otto.configmodule.configmodule.all_hosts(pattern: Pattern[str] | None = None, *, include_containers: bool = False, term: str | None = None, transfer: str | None = None, ssh_options: SshOptions | None = None, telnet_options: TelnetOptions | None = None, sftp_options: SftpOptions | None = None, scp_options: ScpOptions | None = None, ftp_options: FtpOptions | None = None, nc_options: NcOptions | None = None) → Generator[RemoteHost, Any, Any]¶

Yield the active lab’s real remote hosts, optionally filtered by regex.

This is the fleet generator: it yields every network-reached RemoteHost in the active lab — both UnixHost (SSH/telnet to a shell) and EmbeddedHost (telnet to an RTOS console). DockerContainerHost entries are skipped by default because containers aren’t operated on as part of the host fleet (e.g. otto monitor, coverage collection); containers remain reachable for targeted use via tab completion and get_host. Pass include_containers=True to yield container hosts as well.

Parameters:¶

pattern – Compiled regex matched against each host’s id via pattern.search(). When None (the default), all hosts are yielded.
include_containers – When True, also yield DockerContainerHost entries. Defaults to False.
term, transfer – optional active-protocol override; see _apply_option_overrides.
ssh_options, telnet_options, sftp_options, scp_options,
ftp_options, nc_options – Optional per-call option overrides. When supplied, each yielded host is a fresh dataclasses.replace()-style copy whose corresponding *_options field is replaced by the caller’s instance (wholesale replacement, not per-key merge). The new host has a fresh ConnectionManager constructed with the override options, so the override values shape whichever connection opens first. Stored hosts in lab.hosts are untouched. Override keys that don’t correspond to a field on a given host are silently dropped — e.g. ssh_options is ignored for an EmbeddedHost, which only carries telnet_options. When no applicable overrides remain, the stored instance is yielded as-is so identity is preserved. Hop resolution is internal and is not affected by overrides.

Yields:¶

RemoteHost – Each matching UnixHost or EmbeddedHost from the lab configuration.

Examples

Filter the active lab’s hosts by id pattern (see Using otto as a library for a runnable, in-memory example):

import re
seeds = list(all_hosts(re.compile(r"tomato")))

Call an async host method on every matching host.

Parameters:¶

method – Unbound async method (e.g. UnixHost.oneshot).
*args – Positional arguments forwarded to method after the host.
pattern – Compiled regex filter passed to all_hosts().
concurrent – When True (default), run all calls via asyncio.gather with return_exceptions=True. When False, execute serially.
include_containers – Forwarded to all_hosts(). When False (default), container hosts are excluded.
term, transfer – optional active-protocol override; see _apply_option_overrides.
ssh_options, telnet_options, sftp_options, scp_options,
ftp_options, nc_options – Optional per-call option overrides forwarded to all_hosts(). See its docstring for semantics.
**kwargs – Keyword arguments forwarded to method.

Returns:¶

A dict keyed by host ID. Values are the return of method, or a BaseException if that host’s call failed.

Examples

Call an unbound async method on every matching host:

import re
from otto.host import UnixHost
results = await do_for_all_hosts(
    UnixHost.oneshot, "uname -a",
    pattern=re.compile(r"router"),
)

async otto.configmodule.configmodule.run_on_all_hosts(cmds: list[str] | str, pattern: Pattern[str] | None = None, concurrent: bool = True, timeout: float | None = None, *, include_containers: bool = False, term: str | None = None, transfer: str | None = None, ssh_options: SshOptions | None = None, telnet_options: TelnetOptions | None = None, sftp_options: SftpOptions | None = None, scp_options: ScpOptions | None = None, ftp_options: FtpOptions | None = None, nc_options: NcOptions | None = None) → dict[str, RunResult | BaseException]¶

Run commands on every matching host via run().

Convenience wrapper around do_for_all_hosts() for the most common use case.

Parameters:¶

cmds – Command string or list of command strings.
pattern – Compiled regex filter passed to all_hosts().
concurrent – When True (default), run all calls via asyncio.gather. When False, execute serially.
timeout – Per-host timeout forwarded to run.
include_containers – Forwarded to do_for_all_hosts(). When False (default), container hosts are excluded.
term, transfer – optional active-protocol override; see _apply_option_overrides.
ssh_options, telnet_options, sftp_options, scp_options,
ftp_options, nc_options – Optional per-call option overrides forwarded to do_for_all_hosts().

Returns:¶

A dict keyed by host ID. Values are RunResult instances, or a BaseException if that host’s call failed.

Examples

Run a command on every matching host:

results = await run_on_all_hosts("uname -a")

otto.configmodule.configmodule.get_host(host_id: str, *, term: str | None = None, transfer: str | None = None, ssh_options: SshOptions | None = None, telnet_options: TelnetOptions | None = None, sftp_options: SftpOptions | None = None, scp_options: ScpOptions | None = None, ftp_options: FtpOptions | None = None, nc_options: NcOptions | None = None) → UnixHost¶

Return the host registered under host_id in the active lab.

Parameters:¶

host_id – Unique host id (as produced by UnixHost.id).
term, transfer – optional active-protocol override; see _apply_option_overrides.
ssh_options, telnet_options, sftp_options, scp_options,
ftp_options, nc_options – Optional per-call option overrides. Each non-None argument replaces the corresponding *_options field on a returned copy wholesale; the copy is built via dataclasses.replace() so the new host’s ConnectionManager is constructed with the override options from the start. The stored host (and any connection it owns) is untouched. With no overrides, the stored instance is returned unchanged so get_host('x') is get_host('x') still holds. Hop resolution is internal and is not affected by overrides.

otto.configmodule.configmodule.get_lab() → Lab¶: Return the active lab from the current OttoContext.

Environment variables that are needed before parsing CLI arguments

otto.configmodule.env.validate_path(path: Path | None, must_exist: bool = True) → None¶

Validate that path exists when must_exist is True.

Raises FileNotFoundError if the path is set but does not exist on disk. A None path is always accepted (the env var was not set).

otto.configmodule.env.load_otto_env() → OttoEnvSettings¶: Construct the OTTO_* env settings and validate that every sut_dir exists, raising FileNotFoundError (the historical OttoEnv() startup contract).

class otto.configmodule.lab.Lab(name: 'str', resources: 'set[str]' = <factory>, hosts: 'dict[str, Host]' = <factory>)¶

Bases: object

name : str¶: Name of this lab.

resources : set[str]¶: Resources required to reserve this lab.

hosts : dict[str, Host]¶: Host objects, keyed by unique host id.

add_host(host: Host) → None¶

Add a Host object to the Lab’s dictionary of hosts.

Parameters¶

host : Host to add to the dictionary of hosts

otto.configmodule.lab.load_lab(labnames: str | list[str], search_paths: list[Path] | None = None, preferences: dict[str, dict[str, Any]] | None = None, repository: LabRepository | None = None) → Lab¶

Build a Lab object from one or more lab names.

Parameters¶

labnamesstr | list[str]: Name(s) of lab data to retrieve (a comma-separated string is split).
search_pathslist[Path] | None: Directories searched by the default json backend. Ignored when repository is supplied.
preferencesdict[str, dict[str, Any]] | None: The unified {selector: {capability: [...] | option_table: {key: val}}} product-preference table applied to every host in the resulting lab. None reproduces today’s behavior.
repositoryLabRepository | None: A pre-built host-source backend (e.g. from otto.storage.build_lab_repository()). When None, a built-in json backend over search_paths is used — preserving library/script behavior.

Returns¶

Lab: Fully defined lab instance.

class otto.configmodule.repo.DockerImage(name: str, dockerfile: Path, context: Path, target: str | None = None, build_args: tuple[tuple[str, str], ...] = ())¶

Bases: object

A Dockerfile-built image declared by a project.

name : str¶: Short logical name used in tags and CLI selection.

dockerfile : Path¶: Absolute path to the Dockerfile.

context : Path¶: Absolute path to the build context directory.

target : str | None = None¶: Optional multi-stage build target.

build_args : tuple[tuple[str, str], ...] = ()¶: Frozen list of (name, value) build args. Tuples (not dicts) so the container is hashable and order is preserved for context-hash inputs.

class otto.configmodule.repo.DockerCompose(path: Path, default_host: str | None = None, services: tuple[str, ...] = ())¶

Bases: object

A docker-compose file contributed by a project.

path : Path¶: Absolute path to the compose YAML file.

default_host : str | None = None¶: Lab host id where this stack should run by default. Overridden by otto docker up --on <host>.

services : tuple[str, ...] = ()¶: Service names declared in the compose file. Used to synthesize container host ids for tab-completion without parsing YAML at completion-fast-path time. The runtime is the source of truth and will warn on mismatch with docker compose config --services.

class otto.configmodule.repo.DockerSettings(registry_url: str = 'docker.io', images: tuple[DockerImage, ...] = (), composes: tuple[DockerCompose, ...] = ())¶

Bases: object

Per-repo docker configuration parsed from [docker] in settings.toml.

registry_url : str = 'docker.io'¶: Default registry. Overridable per-image via the image’s tag prefix.

images : tuple[DockerImage, ...] = ()¶: Images this project knows how to build.

composes : tuple[DockerCompose, ...] = ()¶: Compose files this project contributes.

class otto.configmodule.repo.CollectedTest(nodeid: str, name: str, path: Path, cls_name: str | None)¶

Bases: object

A single test item collected from a SUT repo’s test directories.

Attributes¶

nodeid :: Full pytest node ID, e.g. dir/test_x.py::ClassName::test_fn. Suitable for use directly as the SUITE argument to otto test.
name :: Test function name only, e.g. test_fn.
path :: Absolute path to the test file.
cls_name :: Class name if the test belongs to a class, else None.

nodeid : str¶

name : str¶

path : Path¶

cls_name : str | None¶

class otto.configmodule.repo.Repo(sut_dir: pathlib.Path, settings: dict[str, typing.Any] = <factory>)¶

Bases: object

sut_dir : Path¶: SUT directory from which the settings came.

name : str¶: Product/repo name

version : Version¶: Product version

labs : list[Path]¶: Paths to lab data

valid_labs : list[str]¶

Lab names this repo supports (by labs membership), e.g. an embedded product that only runs in an embedded lab. Empty when the key is unset.

Parsed here; enforcement — rejecting a selected --lab that is not in this list, and treating an empty list as “the repo must declare its labs” rather than allow-all — is intentionally deferred to lab-selection time and not yet wired in. Parsing must not silently treat unset as allow-all.

libs : list[Path]¶: Extra paths to add to the PYTHONPATH

init : list[str]¶

Module paths that need to be imported during otto init.

Modules containing instructions are an example of modules that need to be imported eagerly.

tests : list[Path]¶: Directories that contain test suites.

host_preferences : dict[str, dict[str, Any]]¶: Unified per-selector product preferences: {regex_selector: {capability: [ordered backends] | option_table: {key: val}}}. The factory matches each host’s id against the selectors (definition-order cascade) and partitions the result into capability selections (forwarded to the resolver) and option-value defaults (applied per-key, product-wins).

os_profiles : dict[str, OsProfile]¶: Named OS profiles declared by this repo’s [os_profiles] settings, keyed by profile name. Each is also registered into the global os-profile registry at parse time so lab-data entries can select it by name in the os_type field. See otto.host.os_profile.register_os_profile().

settings : dict[str, Any]¶: Repo settings dict as parsed from the settings.toml file

docker_settings : DockerSettings¶: Parsed [docker] table — image build definitions, compose files, and registry URL. Defaults to an empty DockerSettings when the section is absent.

get_lab_panel() → Panel¶

get_instructions_panel() → Panel¶

Build a Rich panel listing all instructions contributed by this repo.

Instructions are attributed to this repo by matching each registered instruction’s module against the module prefixes in init.

collect_tests() → list[CollectedTest]¶

Collect all tests from this repo’s configured test directories.

Performs a single pytest collection pass (no tests are executed). The returned list can be passed to any of the get*Panel methods so that multiple listing options share one collection run.

Returns¶

list[CollectedTest]: One entry per discovered test item, in collection order.

get_tests_panel(items: list[CollectedTest]) → Panel¶

Rich panel listing every individual test with its full run syntax.

Each line shows otto test <absolute-path>::[Class::]test_fn which can be copy-pasted directly to run that specific test regardless of the current working directory.

Parameters¶

items :: Pre-collected tests from collect_tests().

get_test_files_panel(items: list[CollectedTest]) → Panel¶

Rich panel listing unique test files with their run syntax.

Each line shows otto test <absolute-path> which runs all tests in that file.

Parameters¶

items :: Pre-collected tests from collect_tests().

get_test_suites_panel(items: list[CollectedTest]) → Panel¶

Rich panel listing unique test suites with their run syntax.

Only class-based tests are listed, using just ClassName — the subcommand name passed directly to otto test ClassName. Bare functions (not part of a class) are omitted since they have no corresponding otto test subcommand. Entries are de-duplicated and preserve collection order.

Parameters¶

items :: Pre-collected tests from collect_tests().

get_otto_settings_path() → Path¶

Create the path to the otto settings TOML file.

Returns¶

Path to the otto settings TOML file.

Raises¶

FileNotFoundError: If the TOML file is not found.

read_settings() → str¶

parse_settings() → None¶: Parse + validate the repo’s .otto/settings.toml via SettingsModel.

property reservation_settings : dict[str, Any]¶

Return the [reservations] settings sub-dict with ${sut_dir} expanded.

Returns an empty dict when the section is absent. Every string value (including nested tables) has ${sut_dir} substituted so the reservation backend can use the same path-expansion convention as the other repo settings.

property lab_settings : dict[str, Any]¶

Return the [lab] settings sub-dict with ${sut_dir} expanded.

Returns an empty dict when the section is absent, so the host-source factory falls back to the built-in json backend over this repo’s labs search paths.

add_libs_to_pythonpath() → None¶: Add configured library directories to the PYTHONPATH

import_init_modules() → None¶

import_test_files() → None¶

Import test_*.py files from each configured tests directory.

This triggers @register_suite() decorators, which populate otto.suite.register._SUITE_REGISTRY at import time. The registry is later consumed by cli/test.py to add sub-Typers to testing_app.

apply_settings()¶

async set_git_description()¶

async set_commit_hash()¶

property commit¶

property description¶

property commit_name : str¶

async run_git_command(cmd: str) → CommandStatus¶

otto.configmodule.repo.apply_repo_settings(repos: list[Repo]) → None¶

otto.configmodule.repo.get_repos(repos: list[Path]) → list[Repo]¶

Create Repo objects from the list of provided repo paths.

Parameters¶

repos : List of paths to repos under test.

Returns¶

List of Repo objects

Raises¶

FileNotFoundError: If a repo’s settings TOML file is not found.

class otto.configmodule.version.Version(version: str)¶

Bases: object

major : int¶: Product major version.

minor : int¶: Product minor version.

patch : int¶: Product patch version.