configmodule¶

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

Fleet host access and host-dispatch helpers for the active lab.

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.

Raises FileNotFoundError (the historical OttoEnv() startup contract).

Lab dataclass and lab-loading utilities for assembling a host registry from lab data.

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

Bases: object

Container for a named lab environment and its registered hosts.

A Lab aggregates the Host objects parsed from lab data files under a single name. Multiple labs can be merged via + to build a composite environment that spans several lab data sources.

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.

Repo settings loading, parsing, and test-collection helpers for SUT repositories.

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

Runtime representation of a single SUT (system-under-test) repository.

Parsed from .otto/settings.toml at construction time (via __post_init__). Holds the resolved paths for lab data, test directories, and init modules, plus Docker and OS-profile settings contributed by that repo. Multiple Repo instances are managed by the configmodule package when OTTO_SUT_DIRS lists more than one directory.

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

logging_capture : list[str]¶: Explicit top-level logger prefixes from [logging] capture whose logging.getLogger(__name__) records otto should route into its sinks, in addition to the package prefixes auto-derived from init/libs. See product_log_prefixes().

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¶: Build a Rich panel listing all lab names available from this repo’s host source.

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¶: Read and return the raw text of this repo’s .otto/settings.toml file.

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

product_log_prefixes() → set[str]¶

Top-level package names whose getLogger(__name__) records otto captures.

The set is declared init module roots, immediate sub-packages of each libs dir, and explicit [logging] capture entries.

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 each module path listed in self.init.

Importing these modules triggers any registration side effects they perform at module level — e.g. registering custom hosts, products, or term/transfer backends, or defining @instruction commands.

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() → None¶

Apply all repo settings.

Extends sys.path with configured lib directories, imports init modules, and imports test files to trigger suite registration.

async set_git_description() → None¶

Populate _git_description from git describe output.

Sets _git_description to the parenthesised tag description on success, or to an empty string when git describe fails (e.g. no tags exist in the repo).

async set_commit_hash() → None¶: Populate _git_hash with the full SHA of the current HEAD commit.

property commit : str | None¶: Return the full HEAD commit SHA, fetching it on first access if needed.

property description : str | None¶

Return the cached git describe string, fetching it on first access.

The value is the parenthesised tag "(<tag>)" on success, or "" when no tags exist (None before the first access).

property commit_name : str¶: Return a display string combining the commit SHA and the git description.

async run_git_command(cmd: str) → CommandStatus¶

Run a git sub-command in this repo’s sut_dir and return the result.

Parameters:¶: cmd – The git sub-command and its arguments (e.g. "log -1 --format=%H").
Returns:¶: A CommandStatus containing the command’s exit status and output.

otto.configmodule.repo.apply_repo_settings(repos: list[Repo]) → None¶: Call apply_settings() on each Repo in repos in order.

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.

Semantic version parsing for product version strings declared in settings.toml.

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

Bases: object

Parsed semantic version (major.minor.patch) from a product version string.

Constructed by passing a "major.minor.patch" string to the constructor, which validates the format and populates the three integer fields. repr returns the original dotted string.

major : int¶: Product major version.

minor : int¶: Product minor version.

patch : int¶: Product patch version.