host.transfer.base

Base classes and shared utilities for file-transfer backends.

Defines the abstract BaseFileTransfer that every transfer backend must subclass, the TransferContext frozen data class (the uniform construction seam for registered backends), validate_filename_lengths() (guards against filesystem NAME_MAX violations before any bytes move), and the TransferProgressHandler / TransferProgressFactory type aliases consumed by the progress-bar wiring layer.

class otto.host.transfer.base.TransferContext(transfer: str, host_name: str, max_filename_len: int = 255, exec_cmd: collections.abc.Callable[..., Coroutine[Any, Any, CommandResult]] | None = None, connections: ConnectionManager | None = None, nc_options: NcOptions | None = None, scp_options: ScpOptions | None = None, get_local_ip: collections.abc.Callable[[], str] | None = None, filesystem: EmbeddedFileSystem | None = None)

Bases: object

Construction inputs a host provides to build its file transfer backend.

The frozen public seam for custom transfer backends. Carries the union of what any family’s built-ins receive at their call sites; a unix backend reads the unix fields, an embedded backend the embedded ones. Selector validation (host-family applicability) runs before construction, so a backend never sees a ctx missing the fields it needs.

transfer : str
host_name : str
max_filename_len : int = 255
exec_cmd : Callable[..., Coroutine[Any, Any, CommandResult]] | None = None
connections : ConnectionManager | None = None
nc_options : NcOptions | None = None
scp_options : ScpOptions | None = None
get_local_ip : Callable[[], str] | None = None
filesystem : EmbeddedFileSystem | None = None
otto.host.transfer.base.validate_filename_lengths(files: list[Path], limit: int, host_name: str) → Result

Reject files whose basename exceeds the host’s filesystem cap.

Shared by UnixFileTransfer (Unix) and EmbeddedFileTransfer (embedded) so every backend surfaces the same self-explaining error. Without this guard the failure modes are:

  • Unix SCP/SFTP/FTP: server returns File name too long (errno 36), mid-transfer, after the local file is already read.

  • Embedded FAT (8.3, no LFN) or LittleFS over NAME_MAX: device fails fs_open with -ENOENT, giving no hint that the name was the problem.

Returns an ok Result when every basename fits, or a failing one whose msg names the offending file.

otto.host.transfer.base.MAX_FILE_MODE = 4095

Highest permission value mode accepts.

Twelve bits: setuid, setgid, sticky, then the three rwx triads — the same range chmod(1) accepts as an octal argument.

otto.host.transfer.base.parse_file_mode(value: int | str | None) → Result

Normalize a transfer permission mode, or explain why it is not one.

Strings are always interpreted base-8, with or without a 0o/0 prefix, because that is the only reading a permission mode can sensibly have — --mode 755 read as decimal would silently mean 0o1363. Integers are taken as-is: a Python caller writing mode=0o755 has already expressed the value, and re-reading it base-8 would corrupt it.

Returns an ok Result whose value is the int mode (or None when no mode was requested), or a failing one whose msg names the offending input. Mirrors validate_filename_lengths() so the two fold identically in BaseFileTransfer.put_files().

otto.host.transfer.base.chmod_command(mode: int, paths: list[Path]) → str

Build one batched chmod command covering every path in paths.

A single invocation for the whole batch, so a multi-file transfer costs one extra round trip rather than one per file. The mode is rendered as bare octal because that is what chmod(1) expects — a 0o prefix would be parsed as a filename on most implementations.

The -- terminator matters: shlex.quote leaves a leading-dash path like -rf unquoted (it contains no shell metacharacters), and a relative dest_dir collapses Path(".") / "-rf" to exactly that, so without -- chmod would read the destination as option flags.

otto.host.transfer.base.aggregate_transfer(per_file: dict[Path, Result]) → Result

Fold a per-file mapping into the aggregate transfer Result.

Aggregate status is the first non-ok entry’s status (Skipped counts as ok); aggregate msg joins each non-ok entry’s diagnostic. The mapping is carried through unchanged as value, keyed by the source paths exactly as passed.

otto.host.transfer.base.mark_skipped(per_file: dict[Path, Result], remaining: list[Path]) → None

Mark each not-yet-attempted source path Skipped after a sequential backend stops.

A sequential backend (ftp/console/nc) stops on the first failure; the files it never reached are recorded Status.Skipped (which is_ok treats as passing, so a trailing run of Skipped never fails the aggregate on its own). Keyed by the source path exactly as passed.

class otto.host.transfer.base.BaseFileTransfer(name: str, max_filename_len: int = 255)

Bases: ABC

Shared API + progress plumbing for any file-transfer backend.

The public put_files / get_files surface (filename-length validation, shared Rich progress acquisition) is owned by this base. Concrete backends (Unix’s UnixFileTransfer subclasses (ScpFileTransfer, SftpFileTransfer, FtpFileTransfer, NcFileTransfer), embedded’s EmbeddedFileTransfer subclasses (ConsoleFileTransfer, TftpFileTransfer), and any future ones) implement two abstract methods — _run_put and _run_get — both of which receive a TransferProgressFactory and are responsible for invoking it at least once per source file, terminating with bytes_done == bytes_total to mark completion.

The progress-bar capability is enforced at the type system level: abc.abstractmethod refuses to instantiate a subclass that omits either method, so a new backend cannot be defined without supplying a way to report progress. The runtime contract test (TestTransferProgressContract) verifies the factory is actually invoked, not just that the methods exist.

host_families : frozenset[str] = frozenset({})

Host-family selectors this backend serves — a subset of {'unix', 'embedded'}. Subclasses declare it; the spec field_validator rejects a backend on a host of the wrong family. A backend with an empty set can never validate and is rejected at registration.

supports_mode : bool = False

Whether this backend can apply a permission mode after a put.

Declarative, like host_families: put_files() reads it pre-flight and refuses a mode it could never honour before any bytes move — a 200 MB upload that ends in “this backend has no permission model” helps nobody. A backend setting this True must implement _apply_mode.

False for embedded backends (console, tftp): a Zephyr filesystem has no permission bits to set.

classmethod create(ctx: TransferContext) → BaseFileTransfer

Build a transfer backend from a TransferContext.

The uniform construction seam (WS#4). Concrete backends override this to run their exact construction against the ctx fields they need. Not an abstractmethod deliberately: only registered built-ins are ever constructed through create, and test doubles that subclass BaseFileTransfer only to exercise the progress contract must not be forced to implement it.

async put_files(src_files: list[Path], dest_dir: Path, show_progress: bool = True, mode: int | str | None = None) → Result

Upload src_files to dest_dir, validating filenames and driving progress display.

Rejects a bad or unhonourable mode and over-limit basenames up front — in that order, cheapest and most specific first — then acquires the process-wide shared Rich progress bar (if show_progress) and delegates to the concrete backend’s _run_put implementation. When mode is set, the files that landed are chmod-ed in one batch afterwards (see _apply_mode).

mode is the permission bits for the uploaded files: an int (0o755) from Python, or a string that is always read as octal ("755", "0755", "0o755"). None leaves whatever permissions the backend’s own defaults produce.

Returns the aggregate Result whose value maps each source path (exactly as passed) to its per-file Result.

async get_files(src_files: list[Path], dest_dir: Path, show_progress: bool = True) → Result

Download src_files into dest_dir, validating filenames and driving progress display.

Same validation and shared-progress contract as put_files(), but delegates to the concrete backend’s _run_get implementation. Returns the aggregate Result whose value maps each source path (exactly as passed) to its per-file Result.

otto.host.transfer.base.NcPortStrategy

Strategy for finding free ports on the remote host for netcat transfers.

Available strategies:

  • 'auto' (default) — try each built-in strategy in order (ss → netstat → python → proc) and cache the first one that succeeds.

  • 'ss' — parse ss -tln output to find unused ports.

  • 'netstat' — parse netstat -tln output (fallback for hosts without ss).

  • 'python' — bind a socket to port 0 via a python/python3 one-liner and let the OS assign a free port.

  • 'proc' — read /proc/net/tcp directly (Linux-only, always available as a last resort).

  • 'custom' — run the shell command specified in nc_port_cmd; the command must print a free port number to stdout.

alias of Literal[‘auto’, ‘ss’, ‘netstat’, ‘python’, ‘proc’, ‘custom’]

otto.host.transfer.base.NcListenerCheck

Strategy for checking if a remote nc listener is ready.

Available strategies:

  • 'auto' (default) — probe for ss, then netstat, falling back to proc. The first tool found is cached and reused for subsequent checks.

  • 'ss' — check for a LISTEN socket via ss -tln sport = :<port>.

  • 'netstat' — grep netstat -tln output for the port.

  • 'proc' — scan /proc/net/tcp for LISTEN state (0A) on the port (Linux-only, always available as a last resort).

  • 'custom' — run the shell command specified in nc_listener_cmd with a {port} placeholder. Must exit 0 when the port is listening.

alias of Literal[‘auto’, ‘ss’, ‘netstat’, ‘proc’, ‘custom’]