host.local_host¶

class otto.host.local_host.LocalFileTransfer(name: str, max_filename_len: int = 255)¶

Bases: BaseFileTransfer

File transfer for LocalHost — a local copy via shutil.copy2().

Concrete BaseFileTransfer so the ABC’s progress contract holds uniformly across every backend in the host fleet (Unix’s UnixFileTransfer, embedded’s EmbeddedFileTransfer, and this one). Per-file completion is the granularity — shutil.copy2 is a single blocking C call with no progress hook, the analogue of an embedded fs read.

class otto.host.local_host.LocalHost(log: bool = True, resources: set[str] = <factory>, products: list[otto.host.product.Product] = <factory>, power_control: 'PowerController | None' = None)¶

Bases: PosixPrivilege, PosixFileOps, BaseHost

name : str¶

id : str¶

Stable identifier for the local host — always "local".

log : bool¶

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

resources : set[str]¶

Resources required to reserve this host — always empty for LocalHost.

products : list[Product]¶

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

power_control : PowerController | None¶

Always None — LocalHost/DockerContainerHost are not power-controlled.

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

Run a command in a fresh subprocess (stateless, concurrent-safe).

Each call spawns an independent process — no state persists between calls, and multiple oneshot() calls can run concurrently via asyncio.gather().

async open_session(name: str) → HostSession¶

Open a named persistent shell 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]¶

Copy files to dest_dir on the local filesystem.

Delegates to LocalFileTransfer so progress reporting flows through the same BaseFileTransfer machinery as Unix and embedded backends.

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]¶

Copy files to dest_dir on the local filesystem.

Delegates to LocalFileTransfer; see get().

async is_reachable(timeout: float = 10.0) → bool¶

Return True — the local machine is always reachable.

async close() → None¶