host.host

otto.host.host.get_logging_command_output_enabled()

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

Return type:

bool

otto.host.host.is_dry_run()

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

Return type:

bool

class otto.host.host.ShellCommand(cmd, expects=None, timeout=None, log=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 : --is-rst--:py:class:`str`

Command string to execute.

expects : --is-rst--:py:class:`tuple`\ \[:py:class:`str` | :py:class:`~re.Pattern`\ \[:py:class:`str`], :py:class:`str`] | :py:class:`list`\ \[:py:class:`tuple`\ \[:py:class:`str` | :py:class:`~re.Pattern`\ \[:py:class:`str`], :py:class:`str`]] | :py:obj:`None`

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

timeout : --is-rst--:py:class:`float` | :py:obj:`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 : --is-rst--:py:class:`bool` | :py:obj:`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, statuses)

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 : --is-rst--:py:class:`~otto.utils.Status`

Aggregate status across all commands.

statuses : --is-rst--:py:class:`list`\ \[:py:class:`~otto.utils.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 : --is-rst--:py:class:`bool`

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

id : --is-rst--:py:class:`str`

Unique identifier for this host.

name : --is-rst--:py:class:`str`

Human-readable name for this host.

resources : --is-rst--:py:class:`set`\ \[:py:class:`str`]

Resources required to reserve this host.

products : --is-rst--:py:class:`list`\ \[Product]

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

power_control : --is-rst--:py:class:`~otto.host.power.PowerController` | :py:obj:`None`

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

async interact()
Return type:

None

async run(cmds, expects=None, timeout=None, log=True, sudo=False)
Return type:

RunResult

async oneshot(cmd, timeout=None, log=True)
Return type:

CommandStatus

async open_session(name)
Return type:

HostSession

async send(text, log=True)
Return type:

None

async expect(pattern, timeout=30.0)
Return type:

str

async get(src_files, dest_dir)
Return type:

tuple[Status, str]

async put(src_files, dest_dir)
Return type:

tuple[Status, str]

async power(state=None)
Return type:

tuple[Status, str]

async reboot(hard=False, wait=False, timeout=600.0)
Return type:

tuple[Status, str]

async shutdown()
Return type:

tuple[Status, str]

async is_reachable(timeout=10.0)
Return type:

bool

async wait_until_up(timeout, interval=2.0)
Return type:

bool

async wait_until_down(timeout, interval=2.0)
Return type:

bool

async close()
Return type:

None

async stage()
Return type:

tuple[Status, str]

async install(stage_only=False)
Return type:

tuple[Status, str]

async uninstall()
Return type:

tuple[Status, str]

async is_installed()
Return type:

bool

async is_uninstalled()
Return type:

bool

class otto.host.host.BaseHost

Bases: ABC

id : --is-rst--:py:class:`str`
name : --is-rst--:py:class:`str`
log : --is-rst--:py:class:`bool`
resources : --is-rst--:py:class:`set`\ \[:py:class:`str`]
products : --is-rst--:py:class:`list`\ \[Product]
power_control : --is-rst--:py:class:`~otto.host.power.PowerController` | :py:obj:`None`
async switch_user(user='', password=None)

Switch the persistent session to another user via su.

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

Return type:

None

as_user(user='root', password=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()

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.

Return type:

None

async run(cmds, expects=None, timeout=None, log=True, sudo=False)

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

Return type:

RunResult

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, timeout=None, log=True)
Return type:

CommandStatus

async open_session(name)
Return type:

HostSession

async send(text, log=True)
Return type:

None

async expect(pattern, timeout=30.0)
Return type:

str

async get(src_files, dest_dir)
Return type:

tuple[Status, str]

async put(src_files, dest_dir)
Return type:

tuple[Status, str]

async close()
Return type:

None

async stage()

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.

Return type:

tuple[Status, str]

async install(stage_only=False)

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.

Return type:

tuple[Status, str]

async uninstall()

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

Return type:

tuple[Status, str]

async is_installed()

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

Return type:

bool

async is_uninstalled()

Inverse of is_installed().

Return type:

bool

async power(state=None)

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.

Return type:

tuple[Status, str]

async reboot(hard=False, wait=False, timeout=600.0)

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.

Return type:

tuple[Status, str]

async shutdown()

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

Return type:

tuple[Status, str]

async is_reachable(timeout=10.0)

Whether this host answers a lightweight connection probe.

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

Return type:

bool

async wait_until_up(timeout, interval=2.0)

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

Return type:

bool

async wait_until_down(timeout, interval=2.0)

Poll is_reachable() until not reachable or timeout.

Return type:

bool

start_repeat(name, cmds, interval, times=-1, duration=datetime.timedelta(days=999999999, seconds=86399, microseconds=999999), until=datetime.datetime(9999, 12, 31, 23, 59, 59, 999999), on_result=None, max_history=1000)
Return type:

None

async stop_repeat(name)
Return type:

None

async stop_all_repeats()
Return type:

None

repeat_results(name)
Return type:

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 : --is-rst--:py:class:`~otto.host.host.Host` | :py:obj:`None`
filter(record)

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.

Return type:

bool

class otto.host.host.SuppressCommandOutput(host=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 : --is-rst--:py:data:`~typing.Optional`\ \[:py:class:`~otto.host.host.Host`] = None

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