Source code for tbot_contrib.uboot._testpy

import hashlib
import os
import select
import time
from typing import List, Optional, Tuple, TypeVar

import tbot
from tbot import machine
from tbot.machine import channel, linux

BH = TypeVar("BH", bound=linux.LinuxShell)

    # Hook which test/py uses to access the console:
    "u-boot-test-console": """\
#!/usr/bin/env bash

stty raw -echo

exec 10<&0
exec 11>&1

cat <&10 >{fifo_console_send} &
cat <{fifo_console_recv} >&11 &

sleep infinity
    # Script which tbot uses to 'emulate' the console:
    "tbot-console": """\
#!/usr/bin/env bash

stty raw -echo

exec 10<&0
exec 11>&1

# Open fifos for reading AND writing to make sure that we can
# never receive EOF (which would happen when no reader fds are open)
cat <&10 1<>{fifo_console_recv} &
cat 0<>{fifo_console_send} >&11 &

sleep infinity
    # Hook which test/py uses to reset the board
    "u-boot-test-reset": """\
#!/usr/bin/env bash

echo "RESET" >{fifo_commands}
    # Hook which test/py calls to flash U-Boot (unused here)
    "u-boot-test-flash": """\
#!/usr/bin/env bash
    # Script which tbot uses to listen to incoming commands
    # (only for triggering resets at the moment)
    "tbot-commands": """\
#!/usr/bin/env bash

# Open for reading and writing so the fifo can never receive EOF
cat <>{fifo_commands}
# If something goes wrong, send an invalid command to abort
echo "FAIL"

def setup_testhooks(
    bh: BH, m_console: BH, m_command: BH
) -> Tuple[channel.Channel, channel.Channel]:
    Setup u-boot-test-* hook scripts for tbot interaction.

    The scripts use 3 FIFOs for sending/receiving the console data to/from tbot
    which in turn communicates with the board.  The third FIFO is used for
    commands (currently only RESET).

    This testcase tries to be smart and only rewrite the hooks when something
    changed.  For this purpose a hash-value is stored on the build-host and
    checked agains the locally computed one.

    # m_console and m_command must not be the same machine
    assert id(m_console) != id(m_command), "testhook channels must be separate"

    hookdir = bh.workdir / "uboot-testpy-tbot"
    if not hookdir.is_dir():
        bh.exec0("mkdir", hookdir)

    # FIFOs --- {{{
    tbot.log.message("Creating FIFOs ...")

    # This dict is used for resolving the paths in the scripts later on
    fifos = {}
    for fifoname in ["fifo_console_send", "fifo_console_recv", "fifo_commands"]:
        fifo = hookdir / fifoname
        bh.exec0("rm", "-rf", fifo)
        bh.exec0("mkfifo", fifo)
        fifos[fifoname] = fifo.at_host(bh)
    # }}}

    # Hook Scripts --- {{{

    # Generate a hash for the version of the control files
    script_hasher = hashlib.sha256()
    for script in sorted(HOOK_SCRIPTS.values()):
    script_hash = script_hasher.hexdigest()

    hashfile = hookdir / "tbot-scripts.sha256"
        up_to_date = script_hash == hashfile.read_text().strip()
    except Exception:
        up_to_date = False

    if up_to_date:
        tbot.log.message("Hooks are up to date, skipping deployment ...")
        tbot.log.message("Updating hook scripts ...")

        for scriptname, script in HOOK_SCRIPTS.items():
            rendered = script.format(**fifos)
            (hookdir / scriptname).write_text(rendered)
            bh.exec0("chmod", "+x", hookdir / scriptname)

        # Write checksum so we don't re-deploy next time
    # }}}

    tbot.log.message("Adding hooks to $PATH ...")
    oldpath = bh.env("PATH")
    bh.env("PATH", f"{hookdir.at_host(bh)}:{oldpath}")

    tbot.log.message("Open console & command channels ...")
    chan_console = m_console.open_channel(hookdir / "tbot-console")
    chan_command = m_command.open_channel(hookdir / "tbot-commands")

    return (chan_console, chan_command)

[docs]@tbot.named_testcase("uboot_testpy") def testpy( uboot_sources: linux.Path[BH], *, board: Optional[tbot.role.Board] = None, uboot: Optional[tbot.role.BoardUBoot] = None, boardenv: Optional[str] = None, testpy_args: Optional[List[str]] = None, ) -> None: """ Run U-Boot's test/py test-framework against a tbot-machine. .. note:: This function supersedes the old :py:func:`` testcase. Please read the docs below carefully on how to use it. This function is meant to be integrated into a custom testcase for test/py which sets up the environment as needed. A simple example could look like this: .. code-block:: python from tbot_contrib import uboot BOARDENV = r\"""# Boardenv for xyz board. env__net_dhcp_server = True \""" @tbot.testcase def run_testpy() -> None: with tbot.ctx.request(tbot.role.BuildHost) as h: # location of the U-Boot sources - these must have been # pre-configured elsewhere (or configured here manually) build_dir = h.workdir / "u-boot-mainline" # subshell for the build environment with h.subshell(): # setup toolchain h.env("CROSS_COMPILE", "arm-linux-") # if needed, you could configure sources here # h.exec0("make", "xyz_defconfig") uboot.testpy( build_dir, boardenv=BOARDENV, testpy_args=["--maxfail", "6"], ) As shown above, the testcase will attempt to instanciate the default board machine from :py:class:`tbot.role.Board` and :py:class:`tbot.role.BoardUBoot`. If a different board machine should be used, it can be passed in explicitly like this: .. code-block:: python @tbot.testcase def run_testpy() -> None: with tbot.ctx() as cx: h = cx.request(tbot.role.BuildHost) build_dir = ... ... # for demonstration - this is exactly what uboot.testpy() # would do on its own when board and uboot are not passed. b = cx.request(tbot.role.Board) ub = cx.request(tbot.role.BoardUBoot, exclusive=True) uboot.testpy( build_dir, board=b, uboot=ub, boardenv=BOARDENV, ) It is advisable to request the U-Boot machine with ``exclusive=True`` so no other code will attempt interacting with it while testpy is running. This is not a hard requirement, though. :param tbot.machine.linux.Path uboot_sources: Path to the U-Boot sources on the host where test/py should run. This could, for example, be your build-host. Importantly, before calling ``testpy()``, you must ensure that these U-Boot sources have been configured appropriately for your board. ``testpy()`` will make no attempt to do this itself. :param tbot.role.Board board: Optional board machine if requesting :py:class:`tbot.role.Board` is not enough. ``uboot`` must also be passed if ``board`` is passed. :param tbot.role.BoardUBoot uboot: Optional U-Boot machine if requesting :py:class:`tbot.role.BoardUBoot` is not enough. ``board`` must also be passed if ``uboot`` is passed. :param str boardenv: Optional "boardenv" file contents to configure test/py. This can, for example, be used to configure where mmc tests can perform test reads. The individual test implementations in the U-Boot sources document available options. :param list(str) testpy_args: Optional additional args to be passed to the test/py invocation. For example, you can use ``["-k", "mmc"]`` to filter for mmc tests only. Or ``["-v"]`` to show the names of all testcases as they are executed (or skipped). .. versionadded:: 0.9.5 """ if board is not None: assert uboot is not None, "when passing `board`, `uboot` is also required!" with tbot.ctx() as cx: bh = with tbot.ctx.request(tbot.role.LabHost) as lh: if id(bh) == id(lh): tbot.log.warning( "The host we're using for test/py might" + " be needed elsewhere during the test run." + " Creating a clone..." ) bh: BH = cx.enter_context(bh.clone()) # type: ignore # Spawn a subshell to not mess up the parent shell's environment and PWD cx.enter_context(bh.subshell()) m_console: BH = cx.enter_context(bh.clone()) # type: ignore m_command: BH = cx.enter_context(bh.clone()) # type: ignore chan_console, chan_command = setup_testhooks(bh, m_console, m_command) assert ( uboot_sources / ".config" ).exists(), "u-boot does not seem configured (.config is missing)!" assert ( uboot_sources / "include" / "" ).exists(), "include/ is missing!" if board is not None: b = board else: b = cx.request(tbot.role.Board, reset=True) assert isinstance( b, machine.board.PowerControl ), "board does not support power-control!" if board is not None: assert uboot is not None # for type checking ub = uboot else: ub = cx.request(tbot.role.BoardUBoot, exclusive=True) chan_uboot = boardtype = "unknown" if boardenv is not None: boardtype = f"tbot-{}" boardtype_filename = boardtype.replace("-", "_") boardenv_file = ( uboot_sources / "test" / "py" / f"u_boot_boardenv_{boardtype_filename}.py" ) boardenv_file.write_text(boardenv) bh.exec0("cd", uboot_sources) chan_testpy = cx.enter_context( "./test/py/", "--build-dir", ".", "--board-type", boardtype, *(testpy_args or []), ) ) # We have to deal with incoming data on any of the following channels. # The comments denote what needs to be done for each channel: readfds = [ chan_console, # Send data to U-Boot chan_uboot, # Send data to chan_console (test/py) chan_command, # Powercycle the board chan_testpy, # Read data so the log-event picks it up ] try: while True: r, _, _ =, [], []) if chan_console in r: # Send data to U-Boot data =, 4096) os.write(chan_uboot.fileno(), data) if chan_uboot in r: # Send data to chan_console (test/py) data =, 4096) os.write(chan_console.fileno(), data) if chan_command in r: # Powercycle the board msg = if msg[:2] == b"RE": b.poweroff() if b.powercycle_delay > 0: time.sleep(b.powercycle_delay) b.poweron() else: raise Exception(f"Got unknown command {msg!r}!") if chan_testpy in r: # Read data so the log-event picks it up. If a # DeathStringException occurs here, test/py finished and we # need to properly terminate the context. try: except linux.CommandEndedException: chan_testpy.terminate0() break except KeyboardInterrupt: # on keyboard interrupt, try to abort test/py chan_testpy.sendcontrol("C") chan_testpy.terminate() raise