host.host

otto.host.host.get_logging_command_output_enabled() → bool

Return True if command-output logging is enabled on the active context.

otto.host.host.is_dry_run() → bool

Return True if dry-run mode is enabled on the active context.

class otto.host.host.ShellCommand(cmd: str, expects: tuple[str | Pattern[str], str] | list[tuple[str | Pattern[str], str]] | None = None, timeout: float | None = None, log: bool | None = None)

Bases: object

A command plus the per-command options that should be used to run it.

Fields left as None inherit from the run-level kwargs on Host.run(). A scalar Expect value is accepted for expects for ergonomics; it is normalized to a one-element list before execution.

cmd : str

Command string to execute.

expects : tuple[str | Pattern[str], str] | list[tuple[str | Pattern[str], str]] | None

Per-command expects. None inherits the run-level expects value.

timeout : float | None

Per-command timeout cap. None inherits the run-level timeout value.

In list form, the effective timeout is always bounded by the remaining cumulative budget.

log : bool | None

Per-command logging switch. None inherits the run-level log value. Set False to suppress this command’s echo and output from the console and log file (e.g. a multi-KB inline payload); the returned CommandStatus is unaffected.

class otto.host.host.RunResult(status: Status, statuses: list[CommandStatus])

Bases: object

Unified result of Host.run() regardless of how many commands ran.

statuses always has one entry per issued command. status is the aggregate: Status.Success when every entry is ok, otherwise the first non-ok status encountered (matching the old tuple-form semantics).

status : Status

Aggregate status across all commands.

statuses : list[CommandStatus]

Per-command statuses in execution order.

property only : CommandStatus

Return the sole CommandStatus when exactly one command ran.

Raises ValueError otherwise — useful for single-command call sites that want to read fields directly without unpacking.

class otto.host.host.Host(*args, **kwargs)

Bases: Protocol

log : bool

Determines whether this host should log its output to stdout and log files.

id : str

Unique identifier for this host.

name : str

Human-readable name for this host.

resources : set[str]

Resources required to reserve this host.

products : list['Product']

Software-under-test deployed to this host (default empty).

power_control : PowerController | None

Pluggable power backend, or None when this host can’t be power-controlled.

async interact() → None
async run(cmds: str | ShellCommand | Sequence[str | ShellCommand], expects: tuple[str | Pattern[str], str] | list[tuple[str | Pattern[str], str]] | None = None, timeout: float | None = None, log: bool = True, sudo: bool = False) → RunResult
async oneshot(cmd: str, timeout: float | None = None, log: bool = True) → CommandStatus
async open_session(name: str) → HostSession
async send(text: str, log: bool = True) → None
async expect(pattern: str | Pattern[str], timeout: float = 30.0) → str
async get(src_files: list[Path] | Path, dest_dir: Path) → tuple[Status, str]
async put(src_files: list[Path] | Path, dest_dir: Path) → tuple[Status, str]
async power(state: str | None = None) → tuple[Status, str]
async reboot(hard: bool = False, wait: bool = False, timeout: float = 600.0) → tuple[Status, str]
async shutdown() → tuple[Status, str]
async is_reachable(timeout: float = 10.0) → bool
async wait_until_up(timeout: float, interval: float = 2.0) → bool
async wait_until_down(timeout: float, interval: float = 2.0) → bool
async close() → None
async stage() → tuple[Status, str]
async install(stage_only: bool = False) → tuple[Status, str]
async uninstall() → tuple[Status, str]
async is_installed() → bool
async is_uninstalled() → bool
class otto.host.host.BaseHost

Bases: ABC

id : str
name : str
log : bool
resources : set[str]
products : list['Product']
power_control : PowerController | None
async switch_user(user: str = '', password: str | None = None) → None

Switch the persistent session to another user via su.

Default raises — only posix-shell hosts (via PosixPrivilege) support su.

as_user(user: str = 'root', password: str | None = None)

Async context manager to run a block as user.

Default raises — only posix-shell hosts (via PosixPrivilege) support su-based user switching.

async interact() → None

Open an interactive shell bridged to the local terminal.

Subclasses implement _interact to do the actual protocol work. This wrapper exists so CLI and SDK callers have a single public entry point.

stdin and stdout are bridged directly to the remote terminal and the session is recorded to the otto log. Press Ctrl+] to disconnect locally without ending the remote session; type exit or logout to end the session normally.

async run(cmds: ~types.Annotated[str | ~otto.host.host.ShellCommand | ~collections.abc.Sequence[str | ~otto.host.host.ShellCommand], ~otto.utils.Arg(variadic=True, elem_type=str, name=None, help=Command(s) to run.)], expects: ~types.Annotated[tuple[str | ~re.Pattern[str], str] | list[tuple[str | ~re.Pattern[str], str]] | None, <otto.utils._Exclude object at 0x7fcc41e8fea0>] | None = None, timeout: ~types.Annotated[float | None, ~otto.utils.Opt(elem_type=None, name=None, help=Per-command/cumulative timeout (seconds).)] | None = None, log: ~typing.Annotated[bool, <otto.utils._Exclude object at 0x7fcc41e8fea0>] = True, sudo: bool = False) → RunResult

Execute one or more commands on the host via the persistent shell session.

The session is stateful: working directory changes (cd), exported environment variables, and other shell state persist between calls, just as they would in an interactive terminal.

Parameters:

  • cmds – A single command (str or ShellCommand) or a sequence of commands. Strings and ShellCommand objects may be mixed. For single-command calls, read the result via result.only (or result.statuses[0]).

  • expects – Default (pattern, response) pair(s) for interactive prompts. Accepts a single Expect tuple or a list of them. Each command inherits this value unless its own ShellCommand.expects is set.

  • timeout – For a single command, the per-command timeout. For a sequence, a cumulative timeout shared across all commands — each command receives the remaining budget; when exhausted, remaining commands are skipped with Status.Error. ShellCommand.timeout caps the per-command value but is still bounded by the remaining budget.

  • sudo – If True, each command is rewritten through _elevate before execution. Hosts that do not support elevation (e.g. embedded/RTOS) raise NotImplementedError — see _elevate.

Returns:

RunResult with the aggregate Status and a list of per-command CommandStatus entries.

See also

oneshot(): stateless, concurrent-safe alternative for one-off commands.

async oneshot(cmd: str, timeout: float | None = None, log: bool = True) → CommandStatus
async open_session(name: str) → HostSession
async send(text: str, log: bool = True) → None
async expect(pattern: str | Pattern[str], timeout: float = 30.0) → str
async get(src_files: list[Path] | Path, dest_dir: Path) → tuple[Status, str]
async put(src_files: list[Path] | Path, dest_dir: Path) → tuple[Status, str]
async close() → None
async stage() → tuple[Status, str]

Stage every product onto this host (transfer/place, no install).

Iterates products in declaration order, returning the first non-ok (Status, str); an empty list is a successful no-op.

async install(stage_only: bool = False) → tuple[Status, str]

Stage, then install every product.

Calls stage() first; returns early if stage_only is set or the stage step failed. Otherwise installs each product in declaration order, short-circuiting on the first failure. Projects may override for cross-product ordering/dependencies.

async uninstall() → tuple[Status, str]

Uninstall every product (best-effort).

Attempts every product even if one fails, returning the first non-ok result seen (so cleanup is not abandoned halfway).

async is_installed() → bool

True iff there is at least one product and all report installed.

An empty products list is not installed (avoids the vacuous-truth surprise of all([])).

async is_uninstalled() → bool

Inverse of is_installed().

async power(state: Annotated[str | None, Arg(variadic=False, elem_type=None, name=None, help=None)] | None = None) → tuple[Status, str]

Power this host 'on'/'off', or toggle when state is None.

Toggling reads the controller’s status(); if the controller can’t report state, pass an explicit state.

async reboot(hard: bool = False, wait: bool = False, timeout: float = 600.0) → tuple[Status, str]

Reboot this host.

hard=False (default) issues the in-shell reboot command (_soft_reboot); hard=True power-cycles via the PowerController. When wait, block on wait_until_up() (up to timeout, default 10 minutes); if the host is still unreachable when timeout expires, the result is downgraded to Failed.

async shutdown() → tuple[Status, str]

Power this host off from its own shell (distinct from external power('off')). Per-family override; default raises.

async is_reachable(timeout: float = 10.0) → bool

Whether this host answers a lightweight connection probe.

Per-family override; default raises (no generic probe).

async wait_until_up(timeout: float, interval: float = 2.0) → bool

Poll is_reachable() until reachable or timeout. Returns success.

async wait_until_down(timeout: float, interval: float = 2.0) → bool

Poll is_reachable() until not reachable or timeout.

start_repeat(name: str, cmds: list[str] | str, interval: timedelta, times: int = -1, duration: timedelta = datetime.timedelta(days=999999999, seconds=86399, microseconds=999999), until: datetime = datetime.datetime(9999, 12, 31, 23, 59, 59, 999999), on_result: Callable[[str, datetime, list[CommandStatus]], None] | None = None, max_history: int = 1000) → None
async stop_repeat(name: str) → None
async stop_all_repeats() → None
repeat_results(name: str) → list[tuple[datetime, list[CommandStatus]]]
class otto.host.host.HostFilter(name='')

Bases: Filter

Filter log records based on whether command output logging is globally enabled.

host : Host | None
filter(record: LogRecord) → bool

Determine if the specified record is to be logged.

Returns True if the record should be logged, or False otherwise. If deemed appropriate, the record may be modified in-place.

class otto.host.host.SuppressCommandOutput(host: Host | None = None)

Bases: object

Suppress command/output logging for one host or globally.

On enter, the prior state is snapshotted; on exit it is restored. That makes nesting safe — an inner context cannot clobber an outer one — and makes concurrent per-host suppressions race-free, since each context only touches its own host’s log attribute.

The no-host (global) path mutates log_command_output on the active OttoContext when one is present. When no context is active the call is a no-op (there is nothing to suppress). Prefer the per-host form when suppressing work that runs concurrently.

host : Host | None = None

Host object to suppress. If not provided, all host output is affected.