host.unix_host

Unix host class.

Unix hosts (Linux being the concrete kernel today; macOS/BSD trivially compatible) accessed over the network via SSH or Telnet, with bash as the remote shell. Manages two responsibilities:

  • Command execution

    • SSH (via asyncssh)

    • telnet (via telnetlib3)

  • File transfers

    • SCP (via asyncssh)

    • SFTP (via asyncssh)

    • FTP (via aioftp)

    • netcat (via commands on the client and host)

All kinds of host connections, for command execution and for file transfers, should be able to establish a connection first, keep it open, and then use it for multiple commands and transfers. Host connections can be explictly closed by calling the .close() method or in the destructor of the host object.

The .run() method runs a single command (str) or a list of commands. Depending on the .term value the correct connection type (ssh or telnet) is used without being specified as an argument.

The .put() and .get() methods both take a single file or a list of files. Depending on the .transfer value the correct connection type (scp, sftp, ftp, or netcat) is used without being specified as an argument.

History: this class was originally named RemoteHost. With the introduction of EmbeddedHost for bare-metal/RTOS targets, RemoteHost is now an abstract base for any network-reached host and the bash-on-SSH/Telnet concrete class lives here as UnixHost.

class otto.host.unix_host.UnixHost(ip: str, creds: dict[str, str], element: str, os_type: str = 'unix', os_name: str | None = 'Linux', os_version: str | None = None, name: str = None, user: str | None = None, element_id: int | None = None, board: str | None = None, slot: int | None = None, hw_version: str | None = None, sw_version: str | None = None, term: str = 'ssh', is_virtual: bool = False, docker_capable: bool = False, transfer: str = 'scp', valid_terms: list[str] = <factory>, valid_transfers: list[str] = <factory>, default_dest_dir: ~pathlib.Path = <factory>, max_filename_len: int = 255, ssh_options: ~otto.host.options.SshOptions = <factory>, telnet_options: ~otto.host.options.TelnetOptions = <factory>, sftp_options: ~otto.host.options.SftpOptions = <factory>, scp_options: ~otto.host.options.ScpOptions = <factory>, ftp_options: ~otto.host.options.FtpOptions = <factory>, nc_options: ~otto.host.options.NcOptions = <factory>, command_frame: ~otto.host.command_frame.CommandFrame | None = None, snmp: ~otto.host.options.SnmpOptions | None = None, hop: str | None = None, resources: set[str] = <factory>, interfaces: dict[str, str] = <factory>, products: list[Product] = <factory>, power_control: PowerController | None = None, log: bool = True, log_stdout: bool = True, toolchain: ~otto.host.toolchain.Toolchain = <factory>, _connection_factory: type[~otto.host.connections.ConnectionManager] | None = None, *, _lab: Lab | None = None)

Bases: PosixPrivilege, PosixFileOps, RemoteHost

Unix host accessed via SSH or Telnet, with bash as the remote shell.

ip : str

IP address of the host.

creds : dict[str, str]

Users and their respective passwords for this host.

element : str

Network element to which this host belongs.

os_type : str

Default profile selector for a bare UnixHost. A custom unix-based profile (e.g. ubuntu-22.04) records its own name here.

os_name : str | None

Kernel/OS name. Defaults to Linux (the concrete Unix kernel today).

os_version : str | None

OS/kernel version string, or None if unspecified.

name : str

Human readable name to represent the host. Automatically generated if not provided.

user : str | None

User with which to log in. If not provided, the first user in the creds dict will be used.

element_id : int | None

Network element identifier to which this host belongs. None indicates there are no other NEs of this type and a number is not needed.

board : str | None

Name of the board type to which this host belongs.

slot : int | None

Phyiscal slot number of the board to which this host belongs.

hw_version : str | None

Hardware version description.

sw_version : str | None

Software version description.

term : str

Protocol used to issue terminal commands.

is_virtual : bool

Determines whether a host is a VM or not.

docker_capable : bool

Whether this host can run Docker containers (i.e., has a docker daemon and the configured user can talk to it). Containers declared by projects are scheduled onto docker-capable hosts; non-capable hosts are skipped.

transfer : str

Protocol used to transfer files.

valid_terms : list[str]

Closed menu of term backends this host supports (active is term).

valid_transfers : list[str]

Closed menu of transfer backends this host supports (active is transfer).

default_dest_dir : Path

Default landing directory for put / get when the caller supplies an empty or relative dest_dir. Defaults to Path(), which preserves the existing behavior — SCP/SFTP resolve a relative destination against the SSH user’s home directory. Override per-host to land transfers in a fixed location regardless of the caller’s argument. See default_dest_dir.

max_filename_len : int

Upper bound on the basename length (including extension) accepted by the target’s filesystem. Defaults to 255 — the Linux NAME_MAX, also the cap for ext4 / XFS / Btrfs / NTFS. Lower it for hosts on a tighter filesystem; see max_filename_len for details. Over-limit names are rejected by put() / get() with a self-explaining error instead of an opaque File name too long midway through the transfer.

ssh_options : SshOptions

Connection options for SSH sessions (port, timeout, known_hosts, port-forwarding rules, etc.).

telnet_options : TelnetOptions

Connection options for telnet sessions (port, cols/rows, auto-resize, etc.).

sftp_options : SftpOptions

Connection options for SFTP file transfers.

scp_options : ScpOptions

Connection options for SCP file transfers.

ftp_options : FtpOptions

Connection options for FTP file transfers (port, encoding, FTPS, etc.).

nc_options : NcOptions

Connection options for netcat file transfers (nc executable, port strategy, listener check, etc.).

command_frame : CommandFrame | None

Shell-framing dialect for this host’s bash console. None (the default) lets the SessionManager use its built-in BashFrame, preserving the historical behavior exactly. Lab data may name a registered frame by string (resolved in __post_init__); a profile or subclass may supply an instance. Promoted to a common field in Phase A so any host can declare its dialect — see command_frame.

snmp : SnmpOptions | None

Optional SNMP polling config (lab snmp block). When set, otto’s monitor collects this host’s metrics over SNMP instead of running shell commands. SNMP monitoring is not embedded-only — a Unix host may use it to poll a real SNMP agent. See SnmpOptions.

hop : str | None

Host ID of the intermediate hop used to reach this host, or None for direct connection.

resources : set[str]

Names of resources required to use this host.

interfaces : dict[str, str]

Named secondary interface addresses (see interfaces). Resolve with address_for().

products : list[Product]

Software-under-test deployed to this host. Default empty. See products.

power_control : PowerController | None

Pluggable power backend. Lab data declares it by string (a config-free controller type) or a [power] table ({type, on_cmd, off_cmd, ...}); __post_init__ coerces it to an instance. None → power()/reboot(hard=True) fail loud. See power_control.

log : bool

Determines whether this host should log its output to stdout and log files. Setting this field to False effectively sets log_stdout to False as well.

log_stdout : bool

Determines whether this host should log its output to stdout. Commands and their output are still logged to log files if log is True.

toolchain : Toolchain

Toolchain associated with this host’s products. Used by the coverage pipeline to select the correct gcov and lcov binaries. Defaults to system-installed tools.

id : str

Unique identifier for this host.

rebuild_connections() None

Recreate the ConnectionManager and dependents.

Useful after changing hop or when the host must reconnect on a new event loop (e.g. after pytest.main() returns and coverage collection starts in a fresh asyncio.run()).

async verify_connection() CommandStatus

Attempt to connect without running any commands. Used by dry-run mode.

async close() None
async oneshot(cmd: str, timeout: float | None = None, log: bool = True) CommandStatus

Run a single command concurrent-safely, independent of the persistent shell.

Unlike run(), this method is concurrent-safe: multiple oneshot() calls can run simultaneously via asyncio.gather() or asyncio.create_task() without corrupting each other or the persistent shell session.

Key differences from run():

Property

run()

oneshot()

Shell state Concurrency Expect support Connection cost Best for

Persistent (cd, env persist) Sequential only Yes Reuses existing session Multi-step workflows, state

Stateless (fresh each call) Safe for asyncio.gather() No Reuses cached oneshot pool One-off / parallel cmds

Implementation details:

  • SSH: runs via SSHClientConnection.create_process() — a lightweight exec channel on the existing TCP connection. No new TCP handshake or authentication needed.

  • Telnet: telnet has no stateless exec primitive, so otto keeps a free-list pool of dedicated internal shell sessions. Serial callers reuse one session (one TCP+auth handshake amortized over all calls); concurrent callers each pull their own session off the free-list, opening a new one if none are free. This preserves the independence guarantee while avoiding the 1–2 s handshake on every call.

Parameters:
  • cmd – Shell command to run. Shell operators (<, >, |) work on SSH because asyncssh wraps the command in a shell; on telnet the command runs through the login shell of the new session.

  • timeout – Seconds before the command is considered hung. None (the default) disables the timeout — appropriate for long-running commands such as a netcat listener waiting for a connection.

Returns:

CommandStatus with the command, captured output, Status enum, and exit code.

See also

run(): stateful, sequential alternative with expect support.

async open_session(name: str) HostSession

Open a named persistent shell session.

Unlike run(), which uses a single default session, this method creates an additional named session that can run commands concurrently with the default session (or other named sessions).

The session is established eagerly — any connection errors surface here. Call close() when done, or use the async context manager protocol:

async with (await host.open_session("monitor")) as mon:
    result = await mon.run("stat /tmp/file.bin")
Parameters:

name – Identifier for this session. Reusing an existing name returns the existing session if it is still alive, or replaces it if dead.

Returns:

A HostSession proxy exposing run, send, expect, and close.

See also

oneshot(): stateless one-shot alternative. run(): default persistent session.

async send(text: str, log: bool = True) None

Send raw text to the host’s persistent session.

async expect(pattern: str | Pattern[str], timeout: float = 10.0) str

Wait for a pattern in the host’s session output stream.

async get(src_files: ~types.Annotated[list[~pathlib.Path] | ~pathlib.Path, ~otto.utils.Arg(variadic=True, elem_type=~pathlib.Path, name=None, help=Remote file(s) to download.)], dest_dir: ~pathlib.Path, show_progress: ~typing.Annotated[bool, <otto.utils._Exclude object at 0x7fcc41e8fea0>] = True) tuple[Status, str]

Transfer files from remote host to the local machine.

async put(src_files: ~types.Annotated[list[~pathlib.Path] | ~pathlib.Path, ~otto.utils.Arg(variadic=True, elem_type=~pathlib.Path, name=None, help=Local file(s) to upload.)], dest_dir: ~pathlib.Path, show_progress: ~typing.Annotated[bool, <otto.utils._Exclude object at 0x7fcc41e8fea0>] = True) tuple[Status, str]

Transfer files from local machine to remote host.

async shutdown() tuple[Status, str]

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