Source code for

# tbot, Embedded Automation Tool
# Copyright (C) 2019  Harald Seiler
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# GNU General Public License for more details.
# You should have received a copy of the GNU General Public License
# along with this program.  If not, see <>.

import typing
import tbot
from tbot.machine import linux, connector
from tbot.machine.linux import auth

__all__ = ("copy",)

H1 = typing.TypeVar("H1", bound=linux.LinuxShell)
H2 = typing.TypeVar("H2", bound=linux.LinuxShell)

def _scp_copy(
    local_path: linux.Path[H1],
    remote_path: linux.Path[H2],
    copy_to_remote: bool,
    username: str,
    hostname: str,
    ignore_hostkey: bool,
    port: int,
    ssh_config: typing.List[str],
    authenticator: auth.Authenticator,
) -> None:
    local_host =

    hk_disable = ["-o", "StrictHostKeyChecking=no"] if ignore_hostkey else []

    scp_command = [
        *["-P", str(port)],
        *[arg for opt in ssh_config for arg in ["-o", opt]],

    if isinstance(authenticator, auth.NoneAuthenticator):
        scp_command += ["-o", "BatchMode=yes"]
    elif isinstance(authenticator, auth.PrivateKeyAuthenticator):
        scp_command += [
    elif isinstance(authenticator, auth.PasswordAuthenticator):
        scp_command = ["sshpass", "-p", authenticator.password] + scp_command
        if typing.TYPE_CHECKING:
        raise ValueError("Unknown authenticator {authenticator!r}")

    if copy_to_remote:

[docs]@tbot.testcase def copy(p1: linux.Path[H1], p2: linux.Path[H2]) -> None: """ Copy a file, possibly from one host to another. The following transfers are currently supported: * ``H`` 🢥 ``H`` (transfer without changing host) * **lab-host** 🢥 **ssh-machine** (:py:class:`~tbot.machine.connector.SSHConnector`, using ``scp``) * **ssh-machine** 🢥 **lab-host** (Using ``scp``) * **local-host** 🢥 **paramiko-host** (:py:class:`~tbot.machine.connector.ParamikoConnector`, using ``scp``) * **local-host** 🢥 **ssh-machine** (:py:class:`~tbot.machine.connector.SSHConnector`, using ``scp``) * **paramiko-host**/**ssh-machine** 🢥 **local-host** (Using ``scp``) The following transfers are **not** supported: * **ssh-machine** 🢥 **ssh-machine** (There is no guarantee that two remote hosts can connect to each other. If you need this, transfer to the lab-host first and then to the other remote) * **lab-host** 🢥 **board-machine** (Transfers over serial are not (yet) implemented. To 'upload' files, connect to your target via ssh or use a tftp download) :param linux.Path p1: Exisiting path to be copied :param linux.Path p2: Target where ``p1`` should be copied .. note:: You can combine this function with :py:func:`tbot_contrib.utils.hashcmp` to only copy a file when the hashsums of source and destination mismatch: .. code-block:: python from tbot_contrib import utils from import shell ... if not utils.hashcmp(path_a, path_b) shell.copy(path_a, path_b) """ if isinstance(, or isinstance(, # Both paths are on the same host p2_w1 = linux.Path(, p2)"cp", p1, p2_w1) return elif isinstance(, connector.SSHConnector) and is # Copy from an SSH machine _scp_copy( local_path=p2, remote_path=p1, copy_to_remote=False,,,,,,, ) elif isinstance(, connector.SSHConnector) and is # Copy to an SSH machine _scp_copy( local_path=p1, remote_path=p2, copy_to_remote=True,,,,,,, ) elif isinstance(, connector.SubprocessConnector) and ( isinstance(, connector.ParamikoConnector) or isinstance(, connector.SSHConnector) ): # Copy from local to ssh labhost _scp_copy( local_path=p1, remote_path=p2, copy_to_remote=True,,,,, ssh_config=getattr(, "ssh_config", []),, ) elif isinstance(, connector.SubprocessConnector) and ( isinstance(, connector.ParamikoConnector) or isinstance(, connector.SSHConnector) ): # Copy to local from ssh labhost _scp_copy( local_path=p2, remote_path=p1, copy_to_remote=False,,,,, ssh_config=getattr(, "ssh_config", []),, ) else: raise NotImplementedError(f"Can't copy from {} to {}!")
_TOOL_CACHE: typing.Dict[linux.LinuxShell, typing.Dict[str, bool]] = {}
[docs]def check_for_tool(host: linux.LinuxShell, tool: str) -> bool: """ Check whether a certain tool/program is installed on a host. Results from previous invocations are cached. **Example**: .. code-block:: python if shell.check_for_tool(lh, "wget"): lh.exec0("wget", download_url, "-O", lh.workdir / "download.tgz") elif shell.check_for_tool(lh, "curl"): lh.exec0("curl", download_url, "-o", lh.workdir / "download.tgz") else: raise Exception("Need either 'wget' or 'curl'!") :param linux.LinuxShell host: The host to ceck on. :param str tool: Name of the binary to check for. :rtype: bool :returns: ``True`` if the tool was found and ``False`` otherwise. """ if host not in _TOOL_CACHE: _TOOL_CACHE[host] = {} if tool not in _TOOL_CACHE[host]: with tbot.testcase("check_for_tool"): has_tool = host.test("which", tool) _TOOL_CACHE[host][tool] = has_tool if has_tool: tbot.log.message( f"Host '{tbot.log.c(}' has " + f"'{tbot.log.c(tool).bold}' installed." ) else: tbot.log.message( f"Host '{tbot.log.c(}' does " + f"{tbot.log.c('not').red} have '{tbot.log.c(tool).bold}' installed." ) return _TOOL_CACHE[host][tool]