riotctrl 0.5.0

RIOT Ctrl - A RIOT node python abstraction

This provides python object abstraction of a RIOT device. The first goal is to be the starting point for the serial abstraction and build on top of that to provide higher level abstraction like over the shell.

It could provide an RPC interface to a device in Python over the serial port and maybe also over network.

The goal is here to be test environment agnostic and be usable in any test framework and also without it.

Testing

Run tox to run the whole test suite:

tox
...
________________________________ summary ________________________________
  test: commands succeeded
  lint: commands succeeded
  flake8: commands succeeded
  congratulations :)

Usage

RIOTCtrl provides a python object abstraction of a RIOT device. It’s meant as a starting point for any serial abstraction on which higher level abstractions (like a shell) can be built.

from riotctrl.ctrl import RIOTCtrl

env = {'BOARD': 'native'}
# if not running from the application directory the a path must be provided
ctrl = RIOTCtrl(env=env, application_directory='.')
# flash the application
ctrl.make_run(['flash'])
# run the terminal through a contextmanager
with ctrl.run_term():
    ctrl.term.expect('>')       # wait for shell to start
    ctrl.term.sendline("help")  # send the help command
    ctrl.term.expect('>')       # wait for the command result to finnish
    print(ctrl.term.before)     # print the command result
# run without a contextmanager
ctrl.start_term()               # start a serial terminal
ctrl.term.sendline("help")      # send the help command
ctrl.term.expect('>')           # wait for the command result to finnish
print(ctrl.term.before)         # print the command result
ctrl.stop_term()                # close the terminal

Creating a RIOTCtrl object is done via environments. If empty then all configuration will come from the target application makefile. But any Make environment variable can be overridden, for example setting BOARD to a target BOARD which is not the default for that application.

Any make target used on RIOT devices can be used on the abstraction like: make flash => ctrl.make_run(['flash']).

ctrl.start_term() (make term’s alter ego) by default spawns a pexpect child application. From there interactions with the application under use can be atomized. In the example below the output of the "help" command is captured:

ShellInteractions

RIOTCtrl provides a minimal extensions by using: pexpect replwrap “[A] Generic wrapper for read-eval-print-loops, a.k.a. interactive shells”. This implements a nice wrapper for RIOT shell commands since it will wait for a command to finish before returning its output.

RIOT already provides a ShellInteraction for the "help" command as well as many others. To make importing them as from riotctrl_shell.sys import Help possible RIOT’s pythonlibs needs to be part of the PYTHONPATH, this can be done by setting in the environment PYTHONPATH=$PYTHONPATH:${RIOTBASE}/dist/pythonlibs or doing so in the script sys.path.append('/path/to/RIOTBASE/dist/pythonlibs')

The previous example can be re-written using ShellInteraction:

from riotctrl.ctrl import RIOTCtrl
from riotctrl.shell import ShellInteraction

env = {'BOARD': 'native'}
# if not running from the application directory the a path must be provided
ctrl = RIOTCtrl(env=env, application_directory='.')
# flash the application
ctrl.flash()                     # alias for ctrl.make_run(['flash'])
# shell interaction instance
shell = ShellInteraction(ctrl)
shell.start_term()               # start a serial terminal
print(shell.cmd("help"))         # print the command result
shell.stop_term()                # close the terminal

or using the already provided Help ShellInteraction:

from riotctrl.ctrl import RIOTCtrl
from riotctrl_shell.sys import Help

env = {'BOARD': 'native'}
# if not running from the application directory the a path must be provided
ctrl = RIOTCtrl(env=env, application_directory='.')
# flash the application
ctrl.flash()                     # alias for ctrl.make_run(['flash'])
# shell interaction instance, Help uses the @ShellInteraction.check_term
# decorator, it will start the terminal if its not yet running, and close
# it after the command ends
shell = Help(ctrl)              # create ShellInteraction
print(shell.help())             # print the command result

Writing ShellInteraction

Lets use this simple C shell application as an example:

#include <stdio.h>
#include <stdlib.h>
#include "shell.h"

static unsigned int counter = 0;

static int _cmd_counter(int argc, char **argv)
{
    if (argc == 1) {
        printf("counter: %d\n", counter);
    }
    else if (argc == 2) {
        counter += atoi(argv[1]);
    }
    else {
        puts("Usage: counter [value]");
        return -1;
    }
    return 0;
}

static const shell_command_t shell_commands[] = {
    { "counter", "prints current counter or adds input", _cmd_counter },
    { NULL, NULL, NULL }
};

int main(void)
{
    char line_buf[SHELL_DEFAULT_BUFSIZE];

    shell_run(shell_commands, line_buf, SHELL_DEFAULT_BUFSIZE);

    return 0;
}

This simple command allows to return the current counter value or modifying by adding a value to it.

main(): This is RIOT! (Version: 2021.10-devel-645-g2c3266-pr_kconfig_mtd)
> boardinfo
board: native
cpu: native
> counter 5
> counter -3
> counter
counter: 2

A ShellInteraction for this could look as follows:

from riotctrl.shell import ShellInteraction


class CounterCmdShell(ShellInteraction):
    @ShellInteraction.check_term
    def counter_cmd(self, args=None, timeout=-1, async_=False):
        cmd = "counter"
        if args is not None:
            cmd += " {args}".format(args=" ".join(str(a) for a in args))
        return self.cmd(cmd, timeout=timeout, async_=False)

Parsing Interaction Results

Parsers can be written for the result of ShellInteraction commands, these can then be returned in any format, for this a base class ShellInteractionParser is provided where the parse() method needs to be implemented.

An examples for the counter command

import re
from riotctrl.shell import ShellInteractionParser


class CounterCmdShellParser(ShellInteractionParser):
    pattern = re.compile(r"counter: (?P<counter>\d+)$")

    def parse(self, cmd_output):
        devices = None
        for line in cmd_output.splitlines():
            m = self.pattern.search(line)
            if m is not None:
                return m.group["counter"]

env = {'BOARD': 'native'}
# if not running from the application directory the a path must be provided
ctrl = RIOTCtrl(env=env, application_directory='.')
# flash the application
ctrl.flash()                     # alias for ctrl.make_run(['flash'])
# shell interaction instance
shell = CounterCmdShell(ctrl)
 with ctrl.run_term():
    parser = CounterCmdShellParser()
    counter = parse.parse(shell_counter_cmd())
    shell.counter_cmd(4)
    assert counter + 4 = parse.parse(shell_counter_cmd())

Interacting with multiple RIOT devices

RIOTCtrl only wrap’s a single RIOT device, handling multiple devices is not yet handled in RIOTCtrl, but through different environments multiple RIOT devices can be created and controlled.

Users of RIOT and FIT IoT-LAB may have already ran experiments on multiple ctrls of the same type (e.g: iotlab-m3) using the IOTLAB_NODE make environment variable. With this one can easily control which device it is targeting.

But if running this locally, with e.g.: multiple samr21-xpro connected the serial or DEBUG_ADAPTER_ID must be used to flash the correct device, and for some BOARDs also the serial port PORT. These variables can be appended to the environment of the spawned object, e.g:

• FIT IoT-LAB:

# first device using dwm1001-1 on the saclay site
env1 = {'BOARD': 'dwm10001', 'IOTLAB_NODE': 'dwm1001-1.saclay.iot-lab.info'}
ctrl1 = RIOTCtrl(env=env1, application_directory='.')
# second device using dwm1001-2 on the saclay site
env2 = {'BOARD': 'dwm10001', 'IOTLAB_NODE': 'dwm1001-2.saclay.iot-lab.info'}
ctrl2 = RIOTCtrl(env=env2, application_directory='.')

• locally:

# first samr21-xpro
env1 = {'BOARD': 'samr21-xpro', 'DEBUG_ADAPTER_ID': 'ATML2127031800004957'}
ctrl1 = RIOTCtrl(env=env1, application_directory='.')
# second samr21-xpro
env2 = {'BOARD': 'samr21-xpro', 'DEBUG_ADAPTER_ID': 'ATML2127031800011458'}
ctrl2 = RIOTCtrl(env=env2, application_directory='.')

For the advanced user one could also do as suggested in multiple-boards-udev and use an easy to remember variable to identify BOARDs (which would allow also running the same python code on different setups), if following the above guide:

# first samr21-xpro
env1 = {'BOARD': 'samr21-xpro', 'BOARD_NUM': 0}
ctrl1 = RIOTCtrl(env=env1, application_directory='.')
# second samr21-xpro
env2 = {'BOARD': 'samr21-xpro', 'BOARD_NUM': 1}
ctrl2 = RIOTCtrl(env=env2, application_directory='.')

Factories

The same tasks are done multiple times creating the object flashing it, starting the terminal and making sure its clean up. Once experiments grow and take over multiple ctrls this can become tedious, using a Factory together with a context manager can help with this.

Going back to our example lets write a factory inheriting from RIOTCtrlBoardFactoryBase (or directly from RIOTCtrlFactoryBase base class).

from contextlib import ContextDecorator
from riotctrl.ctrl import RIOTCtrl, RIOTCtrlBoardFactory
from riotctrl_ctrl import native

class RIOTCtrlAppFactory(RIOTCtrlBoardFactory, ContextDecorator):

    def __init__(self):
        super().__init__(board_cls={
            'native': native.NativeRIOTCtrl,
        })
        self.ctrl_list = list()

    def __enter__(self):
        return self

    def __exit__(self, *exc):
        for ctrl in self.ctrl_list:
            ctrl.stop_term()

    def get_ctrl(self, application_directory='.', env=None):
        # retrieve a RIOTCtrl Object
        ctrl = super().get_ctrl(
            env=env,
            application_directory=application_directory
        )
        # append ctrl to list
        self.ctrl_list.append(ctrl)
        # flash and start terminal
        ctrl.flash()
        ctrl.start_term()
        # return ctrl with started terminal
        return ctrl

And the script itself can be re-written as:

with RIOTCtrlAppFactory() as factory:
    env = {'BOARD': 'native'}
    ctrl = factory.get_ctrl(env=env)
    shell = SaulShell(ctrl)
    parser = SaulShellCmdParser()
    print(parser.parse(shell.saul_cmd()))

GNRC Networking example native

Lets put all the above into practice and script an experiment verifying connectivity between two ctrls, here multiple native instance will be used.

First create two tap interfaces connected through a bridge interface, e.g. on linux:

ip link add name tapbr0 type bridge
ip link set tapbr0 up
ip tuntap add dev tap0 mode tap user $USER
ip tuntap add dev tap1 mode tap user $USER
ip link set dev tap0 master tapbr0
ip link set dev tap1 master tapbr0
ip link set dev tap0 up
ip link set dev tap1 up

Then we can ping and parse the results asserting than packet loss is under a threshold or that an mount of responses was received..

from riotctrl_shell.gnrc import GNRCICMPv6Echo, GNRCICMPv6EchoParser
from riotctrl_shell.netif import Ifconfig


class Shell(ifconfig, GNRCICMPv6Echo):
  pass


with RIOTCtrlAppFactory() as factory:
    # Create two native instances, specifying the tap interface
    native_0 = factory.get_ctrl(env={'BOARD':'native', 'PORT':'tap0'})
    native_1 = factory.get_ctrl(env={'BOARD':'native', 'PORT':'tap1'})
    # `NativeRIOTCtrl` allows for `make reset` with `native`
    native_0.reset()
    native_1.reset()
    # Perform a multicast ping and parse results
    pinger = Shell(native_0)
    parser = GNRCICMPv6EchoParser()
    result = parser.parse(pinger.ping6("ff02::1
    # assert packetloss is under 10%"))
    assert result['stats']['packet_loss'] < 10
    # assert at least one responder
    assert result['stats']['rx'] > 0

A more complex example can be seen in the Release Tests: 04-single-hop-6lowpan-icmp

Examples

• pytest: ReleaseSpecs

• unittests:

  tests/turo, tests/congure_test

Discussion

RIOTCtrl base class is not tied into having a serial based interaction, its the most common usage so far but a new interface or Interaction could use different different transports (e.g. COAP), and does not need to provide a CLI type interface.

Test applications could also use Structured Output, like RIOT’s turo, and in this case parsing CBOR/JSON/XML output could be close to a NOP.