conftest.py

# pylint: disable=W0613,W0621

"""
Central pytest definitions.

See https://docs.pytest.org/en/stable/fixture.html#conftest-py-sharing-fixture-functions
"""  # noqa: E501

import random
import re
import os
import subprocess
import sys
import time

from collections.abc import Iterable

import pytest
from riotctrl.ctrl import RIOTCtrl

import testutils.github
import testutils.pytest
from testutils.iotlab import IoTLABExperiment, DEFAULT_SITE

from testutils.pytest import get_required_envvar
from testutils import ttn


IOTLAB_EXPERIMENT_DURATION = 120
RIOTBASE = os.environ.get('RIOTBASE')
RUNNING_CTRLS = []
RUNNING_EXPERIMENTS = []
DEFAULT_PAN_ID = str(random.randint(0, 0xFFFD))


def pytest_addoption(parser):
    # pylint: disable=C0301
    """
    register argparse-style options and ini-style config values, called once at
    the beginning of a test run.

    See https://docs.pytest.org/en/stable/reference.html#_pytest.hookspec.pytest_addoption
    """  # noqa: E501
    parser.addoption(
        "--local",
        action="store_true",
        default=False,
        help="use local boards",
    )
    parser.addoption(
        "--hide-output",
        action="store_true",
        default=False,
        help="Don't log output of nodes",
    )
    parser.addoption(
        "--boards",
        type=testutils.pytest.list_from_string,
        help="list of BOARD's or IOTLAB_NODEs for the test",
    )
    parser.addoption(
        "--non-RC",
        action="store_true",
        default=False,
        help="Runs test even if RIOT version under test is not an RC",
    )
    parser.addoption(
        "--self-test",
        action="store_true",
        default=False,
        help="Tests the testutils rather than running the release tests",
    )
    parser.addoption(
        "--log-file-fmt",
        nargs="?",
        default=None,
        type=testutils.pytest.log_file_fmt,
        help="Format for the log file name. The available variables are: "
        "`module`: The module (=specXX) of the test, "
        "`function`: The function (=taskXX) of the test, "
        "`node`: Name of the node (on IoT-LAB the URL of the node, "
        "locally board name + port), "
        "`time`: UNIX timestamp at creation time. "
        "If the provided argument is an empty string the format will be "
        "'{module}-{function}-{node}-{time}.log' and stored in the "
        "current work directory",
    )


def pytest_ignore_collect(path, config):
    # pylint: disable=C0301
    """
    return True to prevent considering this path for collection.

    See: https://docs.pytest.org/en/stable/reference.html#_pytest.hookspec.pytest_ignore_collect
    """  # noqa: E501
    # This is about the --self-test option so I don't agree with pylint here
    # pylint: disable=R1705
    if config.getoption("--self-test"):
        return "testutils" not in str(path)
    else:
        return "testutils" in str(path)


def pytest_collection_modifyitems(config, items):
    # pylint: disable=C0301
    """
    called after collection has been performed, may filter or re-order the
    items in-place.

    See: https://docs.pytest.org/en/stable/reference.html#_pytest.hookspec.pytest_collection_modifyitems
    """  # noqa: E501
    # --local given by CLI
    run_local = config.getoption("--local")
    sudo_only_mark = testutils.pytest.check_sudo()
    local_only_mark = testutils.pytest.check_local(run_local)
    iotlab_creds_mark = testutils.pytest.check_credentials(run_local)
    rc_only_mark = testutils.pytest.check_rc(not config.getoption("--non-RC"))

    for item in items:
        if local_only_mark and "local_only" in item.keywords:
            item.add_marker(local_only_mark)
        if sudo_only_mark and "sudo_only" in item.keywords:
            item.add_marker(sudo_only_mark)
        if iotlab_creds_mark and "iotlab_creds" in item.keywords:
            item.add_marker(iotlab_creds_mark)
        if rc_only_mark and "rc_only" in item.keywords:
            item.add_marker(rc_only_mark)


def pytest_configure(config):
    plugin = GithubCommentReportPlugin(config)
    config.pluginmanager.register(plugin, 'github_comment_report_plugin')


class GithubCommentReportPlugin:  # pylint: disable=R0903
    def __init__(self, config):
        self.config = config

    def pytest_runtest_logreport(self, report):
        # pylint: disable=W0212
        basetemp = self.config._tmp_path_factory.getbasetemp()
        testutils.github.update_issue(report, basetemp)


def pytest_keyboard_interrupt(excinfo):
    # pylint: disable=C0301
    """
    Called on KeyInterrupt

    See: https://docs.pytest.org/en/stable/reference.html?highlight=hooks#_pytest.hookspec.pytest_keyboard_interrupt
    """  # noqa: E501
    for child in RUNNING_CTRLS:
        child.stop_term()
    for exp in RUNNING_EXPERIMENTS:
        exp.stop()


@pytest.fixture
def dev_id(request):
    if request.param == "otaa":
        return ttn.DEVICE_ID

    return ttn.DEVICE_ID_ABP


@pytest.fixture
def appkey():
    return get_required_envvar("APPKEY")


@pytest.fixture
def appeui():
    return get_required_envvar("APPEUI")


@pytest.fixture
def deveui():
    return get_required_envvar("DEVEUI")


@pytest.fixture
def devaddr():
    return get_required_envvar("DEVADDR")


@pytest.fixture
def nwkskey():
    return get_required_envvar("NWKSKEY")


@pytest.fixture
def appskey():
    return get_required_envvar("APPSKEY")


@pytest.fixture
def ttn_client():
    with ttn.TTNClient() as client:
        yield client


@pytest.fixture
def log_nodes(request):
    """
    Show output of nodes

    :return: True if output of nodes should be shown, False otherwise
    """
    # use reverse, since from outside we most of the time _want_ to log
    return not request.config.getoption("--hide-output")


@pytest.fixture
def log_file_fmt(request):
    """
    Show output of nodes

    :return: True if output of nodes should be shown, False otherwise
    """
    # use reverse, since from outside we most of the time _want_ to log
    return request.config.getoption("--log-file-fmt")


@pytest.fixture
def local(request):
    """
    Use local boards

    :return: True if local boards should be used, False otherwise
    """
    return request.config.getoption("--local")


@pytest.fixture
def riotbase(request):
    """
    RIOT directory to test. Taken from the variable `RIOTBASE`
    """
    return os.path.abspath(RIOTBASE)


@pytest.fixture
def boards(request):
    """
    String list of boards to use for the test.
    Values are used for the RIOT environment variables `IOTLAB_NODE` or `BOARD`
    """
    return request.config.getoption("--boards")


@pytest.fixture
def iotlab_site(request):
    """
    IoT-LAB site where the nodes are reserved.
    If not specified by a test, it's fetched from the IOTLAB_SITE environment
    variable or falls back to DEFAULT_SITE.
    """

    return getattr(request, "param", os.environ.get("IOTLAB_SITE", DEFAULT_SITE))


def get_namefmt(request):
    name_fmt = {}
    if request.module:
        name_fmt["module"] = request.module.__name__.replace("test_", "")
    if request.function:
        name_fmt["function"] = request.function.__name__.replace("test_", "")
    return name_fmt


@pytest.fixture
def nodes(local, request, boards, iotlab_site):
    """
    Provides the nodes for a test as a list of RIOTCtrl objects
    """
    ctrls = []
    if boards is None:
        boards = request.param
    only_native = all(b.startswith("native") for b in boards)
    for board in boards:
        if local or only_native or IoTLABExperiment.valid_board(board):
            env = {'BOARD': f'{board}'}
            if only_native:
                # XXX this does not work for a mix of native and non-native boards,
                # but we do not have these in the release tests at the moment.
                env["RIOT_TERMINAL"] = "native"
        else:
            env = {
                'BOARD': IoTLABExperiment.board_from_iotlab_node(board),
                'IOTLAB_NODE': f'{board}',
            }
        ctrls.append(RIOTCtrl(env=env))
    if local or only_native:
        yield ctrls
    else:
        name_fmt = get_namefmt(request)
        # Start IoT-LAB experiment if requested
        exp = IoTLABExperiment(
            # pylint: disable=C0209
            name="RIOT-release-test-{module}-{function}".format(**name_fmt),
            ctrls=ctrls,
            site=iotlab_site,
        )
        RUNNING_EXPERIMENTS.append(exp)
        exp.start(duration=IOTLAB_EXPERIMENT_DURATION)
        yield ctrls
        exp.stop()
        RUNNING_EXPERIMENTS.remove(exp)


def update_env(node, modules=None, cflags=None, port=None, termflags=None, extras=None):
    # pylint: disable=too-many-arguments
    node.env['QUIETER'] = '1'
    if not isinstance(modules, str) and isinstance(modules, Iterable):
        node.env['USEMODULE'] = ' '.join(str(m) for m in modules)
    elif modules is not None:
        node.env['USEMODULE'] = modules
    if "USEMODULE" in node.env and os.environ.get('BUILD_IN_DOCKER', 0) == '1':
        # workaround to inject USEMODULE into docker container
        # see: https://github.com/RIOT-OS/RIOT/issues/14504
        node.env['DOCKER_ENVIRONMENT_CMDLINE'] = (
            node.env.get('DOCKER_ENVIRONMENT_CMDLINE', '')
            + f" -e 'USEMODULE={node.env['USEMODULE']}'"
        )
    node.env['DEFAULT_PAN_ID'] = DEFAULT_PAN_ID
    if os.environ.get('BUILD_IN_DOCKER', 0) == '1':
        node.env['DOCKER_ENVIRONMENT_CMDLINE'] = (
            node.env.get('DOCKER_ENVIRONMENT_CMDLINE', '')
            + f" -e 'DEFAULT_PAN_ID={node.env['DEFAULT_PAN_ID']}'"
        )
    if cflags is not None:
        node.env['CFLAGS'] = cflags
    if port is not None:
        node.env['PORT'] = port
    if termflags is not None:
        node.env['TERMFLAGS'] = termflags
    if extras is not None:
        node.env.update(extras)


@pytest.fixture
def riot_ctrl(log_nodes, log_file_fmt, nodes, riotbase, request):
    """
    Factory to create RIOTCtrl objects from list nodes provided by nodes
    fixture
    """
    factory_ctrls = []

    # pylint: disable=R0913
    # flake8: noqa: C901
    def ctrl(
        nodes_idx,
        application_dir,
        shell_interaction_cls,
        board_type=None,
        modules=None,
        cflags=None,
        port=None,
        termflags=None,
        extras=None,
    ):
        if board_type is not None:
            node = next(n for n in nodes if n.board() == board_type)
        else:
            node = nodes[nodes_idx]
        update_env(node, modules, cflags, port, termflags, extras)
        # if the nodes are not logged, there is no sense in logging to a file
        # so check if nodes are logged as well as if they should be logged to a
        # file
        if log_nodes and log_file_fmt:
            if node.env.get("IOTLAB_NODE"):
                node_name = node.env["IOTLAB_NODE"]
            else:
                node_port = re.sub(r'\W+', '-', node.env["PORT"])
                node_name = f"{node.board()}-{node_port}"
            node.env["TERMLOG"] = os.path.join(
                os.getcwd(),
                log_file_fmt.format(
                    node=node_name, time=int(time.time()), **get_namefmt(request)
                ),
            )
        # need to access private member here isn't possible otherwise sadly :(
        # pylint: disable=W0212
        node._application_directory = os.path.join(riotbase, application_dir)
        flash_cmd = "flash"
        if "BINFILE" in node.env:
            flash_cmd = "flash-only"
        node.make_run(
            [flash_cmd],
            check=True,
            stdout=None if log_nodes else subprocess.DEVNULL,
            stderr=None if log_nodes else subprocess.DEVNULL,
        )
        if node.env.get("IOTLAB_NODE"):
            # reset to prevent at86rf2xx `ifconfig` issue
            time.sleep(1)
            node.make_run(
                ['reset'],
                check=True,
                stdout=None if log_nodes else subprocess.DEVNULL,
                stderr=None if log_nodes else subprocess.DEVNULL,
            )
            time.sleep(1)
        termargs = {}
        if log_nodes:
            termargs["logfile"] = sys.stdout
        RUNNING_CTRLS.append(node)
        node.start_term(**termargs)
        factory_ctrls.append(node)
        return shell_interaction_cls(node)

    yield ctrl

    for node in factory_ctrls:
        try:
            node.stop_term()
            RUNNING_CTRLS.remove(node)
        except RuntimeError:
            # Process had to be forced kill, happens with native
            pass