From 460bf2eb1ab729a28bf09155e5407088b2fb224e Mon Sep 17 00:00:00 2001 From: Francisco Molina Date: Tue, 30 Oct 2018 15:35:22 +0100 Subject: [PATCH] testutils: initial import Co-Authored-By: Jose Alamos Co-Authored-By: Martine S. Lenders --- .gitignore | 137 ++++++++++++++++++- conftest.py | 238 +++++++++++++++++++++++++++++++++ setup.cfg | 27 ++++ testutils/__init__.py | 0 testutils/asyncio.py | 32 +++++ testutils/iotlab.py | 178 ++++++++++++++++++++++++ testutils/native.py | 140 +++++++++++++++++++ testutils/pytest.py | 89 ++++++++++++ testutils/shell.py | 171 +++++++++++++++++++++++ testutils/tests/__init__.py | 3 + testutils/tests/test_iotlab.py | 188 ++++++++++++++++++++++++++ testutils/tests/test_native.py | 136 +++++++++++++++++++ testutils/tests/test_pytest.py | 65 +++++++++ testutils/tests/test_shell.py | 179 +++++++++++++++++++++++++ tox.ini | 57 ++++++++ 15 files changed, 1639 insertions(+), 1 deletion(-) create mode 100644 conftest.py create mode 100644 setup.cfg create mode 100644 testutils/__init__.py create mode 100644 testutils/asyncio.py create mode 100644 testutils/iotlab.py create mode 100644 testutils/native.py create mode 100644 testutils/pytest.py create mode 100644 testutils/shell.py create mode 100644 testutils/tests/__init__.py create mode 100644 testutils/tests/test_iotlab.py create mode 100644 testutils/tests/test_native.py create mode 100644 testutils/tests/test_pytest.py create mode 100644 testutils/tests/test_shell.py create mode 100644 tox.ini diff --git a/.gitignore b/.gitignore index b45f5596..7a78a20f 100644 --- a/.gitignore +++ b/.gitignore @@ -1,2 +1,137 @@ -results/ +test-report.xml + +#### joe made this: http://goel.io/joe + +#### python #### +# Byte-compiled / optimized / DLL files __pycache__/ +*.py[cod] +*$py.class + +# C extensions +*.so + +# Distribution / packaging +.Python +build/ +develop-eggs/ +dist/ +downloads/ +eggs/ +.eggs/ +lib/ +lib64/ +parts/ +sdist/ +var/ +wheels/ +*.egg-info/ +.installed.cfg +*.egg +MANIFEST + +# PyInstaller +# Usually these files are written by a python script from a template +# before PyInstaller builds the exe, so as to inject date/other infos into it. +*.manifest +*.spec + +# Installer logs +pip-log.txt +pip-delete-this-directory.txt + +# Unit test / coverage reports +htmlcov/ +.tox/ +.coverage +.coverage.* +.cache +nosetests.xml +coverage.xml +*.cover +.hypothesis/ +.pytest_cache/ + +# Translations +*.mo +*.pot + +# Django stuff: +*.log +local_settings.py +db.sqlite3 + +# Flask stuff: +instance/ +.webassets-cache + +# Scrapy stuff: +.scrapy + +# Sphinx documentation +docs/_build/ + +# PyBuilder +target/ + +# Jupyter Notebook +.ipynb_checkpoints + +# pyenv +.python-version + +# celery beat schedule file +celerybeat-schedule + +# SageMath parsed files +*.sage.py + +# Environments +.env +.venv +env/ +venv/ +ENV/ +env.bak/ +venv.bak/ + +# Spyder project settings +.spyderproject +.spyproject + +# Rope project settings +.ropeproject + +# mkdocs documentation +/site + +# mypy +.mypy_cache/ + + +#### vim #### +# Swap +[._]*.s[a-v][a-z] +[._]*.sw[a-p] +[._]s[a-rt-v][a-z] +[._]ss[a-gi-z] +[._]sw[a-p] + +# Session +Session.vim + +# Temporary +.netrwhist +*~ +# Auto-generated tag files +tags +# Persistent undo +[._]*.un~ + + +#### visualstudiocode #### +.vscode/* +!.vscode/settings.json +!.vscode/tasks.json +!.vscode/launch.json +!.vscode/extensions.json diff --git a/conftest.py b/conftest.py new file mode 100644 index 00000000..f4f83484 --- /dev/null +++ b/conftest.py @@ -0,0 +1,238 @@ +# pylint: disable=W0613,W0621 + +""" +Central pytest definitions. + +See https://docs.pytest.org/en/stable/fixture.html#conftest-py-sharing-fixture-functions +""" # noqa: E501 + +import os +import sys +from collections.abc import Iterable + +import pytest +from riotctrl.ctrl import RIOTCtrl + +import testutils.pytest +from testutils.iotlab import IoTLABExperiment, DEFAULT_SITE + + +IOTLAB_EXPERIMENT_DURATION = 120 +RIOTBASE = os.environ.get('RIOTBASE') +DEVNULL = open(os.devnull, 'w') +RUNNING_CTRLS = [] +RUNNING_EXPERIMENTS = [] + + +def pytest_addoption(parser): + """ + 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", + ) + + +def pytest_ignore_collect(path, config): + """ + 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_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 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 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") + + +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): + """ + 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': '{}'.format(board)} + else: + env = { + 'BOARD': IoTLABExperiment.board_from_iotlab_node(board), + 'IOTLAB_NODE': '{}'.format(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( + name="RIOT-release-test{module}{function}".format(**name_fmt), + ctrls=ctrls, + site=os.environ.get("IOTLAB_SITE", DEFAULT_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): + 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 cflags is not None: + node.env['CFLAGS'] = cflags + if port is not None: + node.env['PORT'] = port + + +@pytest.fixture +def riot_ctrl(log_nodes, nodes, riotbase): + """ + Factory to create RIOTCtrl objects from list nodes provided by nodes + fixture + """ + factory_ctrls = [] + + # pylint: disable=R0913 + def ctrl(nodes_idx, application_dir, shell_interaction_cls, + board_type=None, modules=None, cflags=None, port=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) + # need to access private member here isn't possible otherwise sadly :( + # pylint: disable=W0212 + node._application_directory = os.path.join(riotbase, application_dir) + node.make_run(['flash'], stdout=DEVNULL, stderr=DEVNULL) + 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 diff --git a/setup.cfg b/setup.cfg new file mode 100644 index 00000000..401de397 --- /dev/null +++ b/setup.cfg @@ -0,0 +1,27 @@ +[tool:pytest] +addopts = -rs -v --junit-xml=test-report.xml +markers = + rc_only: mark tests that should be skipped if the version under test is not an RC + local_only: marks tests as local_only (deselect with '-m "not local"') + sudo_only: marks tests as sudo_only (deselect with '-m "not sudo"') + iotlab_creds: marks tests to require IoT-LAB access if not run locally (deselect with '-m "not iotlab_creds"') + self_test: marks tests that are testing the testutils rather than the release +junit_logging = all +junit_family = xunit2 + +[flake8] +exclude = .tox,dist,doc,build,*.egg,09-coap/task*.py,09-coap/server.py +max-complexity = 10 + +[pylint.master] +ignore = server.py,task03.py,task04.py,task05.py + +[pylint.messages control] +disable= + duplicate-code, + fixme, + invalid-name, + logging-format-interpolation, + missing-module-docstring, + missing-class-docstring, + missing-function-docstring diff --git a/testutils/__init__.py b/testutils/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/testutils/asyncio.py b/testutils/asyncio.py new file mode 100644 index 00000000..f44b3809 --- /dev/null +++ b/testutils/asyncio.py @@ -0,0 +1,32 @@ +""" +Helpers for asyncio +""" + +import asyncio + + +def wait_for_futures(futures): + loop = asyncio.get_event_loop() + loop.run_until_complete(asyncio.gather( + *futures + )) + + +def timeout_futures(futures, timeout): + gather = None + + async def wait_for_timeout(): + await asyncio.sleep(timeout) + if gather: + return gather.cancel() + return False + + gather = asyncio.gather( + wait_for_timeout(), *futures + ) + + loop = asyncio.get_event_loop() + try: + loop.run_until_complete(gather) + except asyncio.CancelledError: + pass diff --git a/testutils/iotlab.py b/testutils/iotlab.py new file mode 100644 index 00000000..c762e944 --- /dev/null +++ b/testutils/iotlab.py @@ -0,0 +1,178 @@ +import logging +import re + +from iotlabcli.auth import get_user_credentials +from iotlabcli.rest import Api +from iotlabcli.experiment import (submit_experiment, wait_experiment, + stop_experiment, get_experiment, + exp_resources, AliasNodes) + + +DEFAULT_SITE = 'saclay' +IOTLAB_DOMAIN = 'iot-lab.info' + + +class IoTLABExperiment(): + """Utility for running iotlab-experiments based on a list of RIOTCtrls + expects BOARD or IOTLAB_NODE variable to be set for received nodes""" + BOARD_ARCHI_MAP = { + 'arduino-zero': {'name': 'arduino-zero', 'radio': 'xbee'}, + 'b-l072z-lrwan1': {'name': 'st-lrwan1', 'radio': 'sx1276'}, + 'b-l475e-iot01a': {'name': 'st-iotnode', 'radio': 'multi'}, + 'firefly': {'name': 'firefly', 'radio': 'multi'}, + 'frdm-kw41z': {'name': 'frdm-kw41z', 'radio': 'multi'}, + 'iotlab-a8-m3': {'name': 'a8', 'radio': 'at86rf231'}, + 'iotlab-m3': {'name': 'm3', 'radio': 'at86rf231'}, + 'microbit': {'name': 'microbit', 'radio': 'ble'}, + 'nrf51dk': {'name': 'nrf51dk', 'radio': 'ble'}, + 'nrf52dk': {'name': 'nrf52dk', 'radio': 'ble'}, + 'nrf52832-mdk': {'name': 'nrf52832mdk', 'radio': 'ble'}, + 'nrf52840dk': {'name': 'nrf52840dk', 'radio': 'multi'}, + 'nrf52840-mdk': {'name': 'nrf52840mdk', 'radio': 'multi'}, + 'pba-d-01-kw2x': {'name': 'phynode', 'radio': 'kw2xrf'}, + 'samr21-xpro': {'name': 'samr21', 'radio': 'at86rf233'}, + 'samr30-xpro': {'name': 'samr30', 'radio': 'at86rf212b'}, + } + + SITES = ['grenoble', 'lille', 'saclay'] + + def __init__(self, name, ctrls, site=DEFAULT_SITE): + IoTLABExperiment._check_site(site) + self.site = site + IoTLABExperiment._check_ctrls(site, ctrls) + self.ctrls = ctrls + self.name = name + self.exp_id = None + + @staticmethod + def board_from_iotlab_node(iotlab_node): + """Return BOARD matching iotlab_node""" + reg = r'([0-9a-zA-Z\-]+)-\d+\.[a-z]+\.iot-lab\.info' + match = re.search(reg, iotlab_node) + if match is None: + raise ValueError("Unable to parse {} as IoT-LAB node name of " + "format ..iot-lab.info" + .format(iotlab_node)) + iotlab_node_name = match.group(1) + dict_values = IoTLABExperiment.BOARD_ARCHI_MAP.values() + dict_names = [value['name'] for value in dict_values] + dict_keys = list(IoTLABExperiment.BOARD_ARCHI_MAP.keys()) + return dict_keys[dict_names.index(iotlab_node_name)] + + @staticmethod + def valid_board(board): + return board in IoTLABExperiment.BOARD_ARCHI_MAP + + @staticmethod + def valid_iotlab_node(iotlab_node, site, board=None): + if site not in iotlab_node: + raise ValueError("All nodes must be on the same site") + if board is not None: + if IoTLABExperiment.board_from_iotlab_node(iotlab_node) != board: + raise ValueError("IOTLAB_NODE doesn't match BOARD") + + @classmethod + def check_user_credentials(cls): + res = cls.user_credentials() + return res != (None, None) + + @staticmethod + def user_credentials(): + return get_user_credentials() + + @staticmethod + def _archi_from_board(board): + """Return iotlab 'archi' format for BOARD""" + return '{}:{}'.format(IoTLABExperiment.BOARD_ARCHI_MAP[board]['name'], + IoTLABExperiment.BOARD_ARCHI_MAP[board]['radio']) + + @staticmethod + def _check_site(site): + if site not in IoTLABExperiment.SITES: + raise ValueError("iotlab site must be one of {}" + .format(IoTLABExperiment.SITES)) + + @staticmethod + def _valid_addr(ctrl, addr): + """Check id addr matches a specific RIOTCtrl BOARD""" + return addr.startswith( + IoTLABExperiment.BOARD_ARCHI_MAP[ctrl.board()]['name']) + + @staticmethod + def _check_ctrls(site, ctrls): + """Takes a list of RIOTCtrls and validates BOARD or IOTLAB_NODE""" + for ctrl in ctrls: + # If BOARD is set it must be supported in iotlab + if ctrl.board() is not None: + if not IoTLABExperiment.valid_board(ctrl.board()): + raise ValueError("{} BOARD unsupported in iotlab" + .format(ctrl)) + if ctrl.env.get('IOTLAB_NODE') is not None: + IoTLABExperiment.valid_iotlab_node(ctrl.env['IOTLAB_NODE'], + site, + ctrl.board()) + elif ctrl.env.get('IOTLAB_NODE') is not None: + IoTLABExperiment.valid_iotlab_node(ctrl.env['IOTLAB_NODE'], + site) + board = IoTLABExperiment.board_from_iotlab_node( + ctrl.env["IOTLAB_NODE"] + ) + ctrl.env['BOARD'] = board + else: + raise ValueError("BOARD or IOTLAB_NODE must be set") + + def stop(self): + """If running stop the experiment""" + ret = None + if self.exp_id is not None: + ret = stop_experiment(Api(*self.user_credentials()), self.exp_id) + self.exp_id = None + return ret + + def start(self, duration=60): + """Submit an experiment, wait for it to be ready and map assigned + nodes""" + logging.info("Submitting experiment") + self.exp_id = self._submit(site=self.site, duration=duration) + logging.info("Waiting for experiment {} to go to state \"Running\"" + .format(self.exp_id)) + self._wait() + self._map_iotlab_nodes_to_riot_ctrl(self._get_nodes()) + + def _wait(self): + """Wait for the experiment to finish launching""" + ret = wait_experiment(Api(*self.user_credentials()), self.exp_id) + return ret + + def _submit(self, site, duration): + """Submit an experiment with required nodes""" + api = Api(*self.user_credentials()) + resources = [] + for ctrl in self.ctrls: + if ctrl.env.get('IOTLAB_NODE') is not None: + resources.append(exp_resources([ctrl.env.get('IOTLAB_NODE')])) + elif ctrl.board() is not None: + board = IoTLABExperiment._archi_from_board(ctrl.board()) + alias = AliasNodes(1, site, board) + resources.append(exp_resources(alias)) + else: + raise ValueError("neither BOARD or IOTLAB_NODE are set") + return submit_experiment(api, self.name, duration, resources)['id'] + + def _map_iotlab_nodes_to_riot_ctrl(self, iotlab_nodes): + """Fetch reserved nodes and map each one to an RIOTCtrl""" + for ctrl in self.ctrls: + if ctrl.env.get('IOTLAB_NODE') in iotlab_nodes: + iotlab_nodes.remove(ctrl.env['IOTLAB_NODE']) + else: + for iotlab_node in iotlab_nodes: + if IoTLABExperiment._valid_addr(ctrl, iotlab_node): + iotlab_nodes.remove(iotlab_node) + ctrl.env['IOTLAB_NODE'] = str(iotlab_node) + break + ctrl.env['IOTLAB_EXP_ID'] = str(self.exp_id) + + def _get_nodes(self): + """Return all nodes reserved by the experiment""" + ret = get_experiment(Api(*self.user_credentials()), self.exp_id) + return ret['nodes'] diff --git a/testutils/native.py b/testutils/native.py new file mode 100644 index 00000000..de1712aa --- /dev/null +++ b/testutils/native.py @@ -0,0 +1,140 @@ +""" +Helpers for native +""" + +import re +import subprocess + + +TAP_MASTER_C = re.compile(r"^\d+:\s+(?P[^:]+):.+" + r"master\s+(?P\S+)?") +TAP_LINK_LOCAL_C = re.compile(r"inet6\s+(?Pfe80:[0-9a-f:]+)/\d+\s+" + r"scope\s+link") + + +def _run_check(cmd, shell=False): + try: + subprocess.check_call(cmd, shell=shell, stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL) + except subprocess.CalledProcessError: + return False + else: + return True + + +def command_exists(cmd): + # command usually is a shell build-in so run it in shell + return _run_check(' '.join(["command", "-v", cmd]), shell=True) + + +def _check_bridged(ip_link_output, taps): + """ + Checks "ip link" output if a given list of TAP interfaces are in the same + bridge + + >>> _check_bridged("", ["tap0", "tap1"]) + False + >>> _check_bridged( + ... "49: tap0: <...> mtu 1500 ... master tapbr0 state ...\\n" + ... " link/ether e2:bc:7d:cb:f5:4f brd ff:ff:ff:ff:ff:ff\\n" + ... "50: tap1: <...> mtu 1500 ... master tapbr0 state ...\\n" + ... " link/ether da:27:1d:a8:64:23 brd ff:ff:ff:ff:ff:ff\\n", + ... ["tap0", "tap1"]) + True + >>> _check_bridged( + ... "50: tap1: <...> mtu 1500 ... master tapbr0 state ...\\n" + ... " link/ether da:27:1d:a8:64:23 brd ff:ff:ff:ff:ff:ff\\n", + ... ["tap0", "tap1"]) + False + >>> _check_bridged( + ... "50: tap1: <...> mtu 1500 ... master tapbr0 state ...\\n" + ... " link/ether da:27:1d:a8:64:23 brd ff:ff:ff:ff:ff:ff\\n" + ... "60: tap0: <...> mtu 1500 qdisc fq_codel state ...\\n" + ... " link/ether e2:bc:7d:cb:f5:4f brd ff:ff:ff:ff:ff:ff\\n", + ... ["tap0", "tap1"]) + False + """ + taps_in_bridges = set() + tap_bridges = set() + for line in ip_link_output.splitlines(): + m = TAP_MASTER_C.match(line) + if m is not None and m.group("tap") in taps: + taps_in_bridges.add(m.group("tap")) + tap_bridges.add(m.group("master")) + return set(taps) == taps_in_bridges and len(tap_bridges) == 1 + + +def ip_addr_add(iface, addr): + subprocess.check_call(["ip", "addr", "add", addr, "dev", iface]) + + +def ip_addr_del(iface, addr): + subprocess.run(["ip", "addr", "del", addr, "dev", iface], + stderr=subprocess.DEVNULL, check=False) + + +def ip_route_add(iface, route, via=None): + cmd = ["ip", "route", "add", route] + if via: + cmd += ["via", via] + cmd += ["dev", iface] + subprocess.check_call(cmd) + + +def ip_route_del(iface, route, via=None): + cmd = ["ip", "route", "del", route] + if via: + cmd += ["via", via] + cmd += ["dev", iface] + subprocess.run(cmd, stderr=subprocess.DEVNULL, check=False) + + +def ip_link(iface=None): + cmd = ["ip", "link", "show"] + if iface is not None: + cmd.append(iface) + return subprocess.check_output(cmd).decode() + + +def bridged(taps): + """ + Checks if a list of TAP interface `taps` are all in the same bridge + (and exist) + """ + return _check_bridged(ip_link(), taps) + + +def interface_exists(iface): + return _run_check(["ip", "link", "show", iface]) + + +def get_link_local(iface): + out = subprocess.check_output(["ip", "a", "s", "dev", iface]).decode() + for line in out.splitlines(): + m = TAP_LINK_LOCAL_C.search(line) + if m is not None: + return m.group("link_local") + return None + + +def bridge(tap): + """ + Get the bridge a TAP interface is assigned to. If it is not assigned to a + bridge, the TAP interface itself is returned. + """ + out = ip_link(tap) + for line in out.splitlines(): + m = TAP_MASTER_C.match(line) + if m is not None and m.group("tap") == tap: + return m.group("master") + return tap + + +def get_ping_cmd(): + if command_exists("ping6"): + ping_cmd = "ping6" + elif command_exists("ping"): + ping_cmd = "ping -6" + else: + raise FileNotFoundError("No ping command found on host machine") + return ping_cmd diff --git a/testutils/pytest.py b/testutils/pytest.py new file mode 100644 index 00000000..dae2a286 --- /dev/null +++ b/testutils/pytest.py @@ -0,0 +1,89 @@ +""" +Helpers for pytest +""" + +import os +import re +import subprocess + +import pexpect.replwrap +import pytest + +from .iotlab import IoTLABExperiment, DEFAULT_SITE, IOTLAB_DOMAIN + + +def list_from_string(list_str=None): + """Get list of items from `list_str` + + >>> list_from_string(None) + [] + >>> list_from_string("") + [] + >>> list_from_string(" ") + [] + >>> list_from_string("a") + ['a'] + >>> list_from_string("a ") + ['a'] + >>> list_from_string("a b c") + ['a', 'b', 'c'] + """ + value = (list_str or '').split(' ') + return [v for v in value if v] + + +def check_ssh(): + user, _ = IoTLABExperiment.user_credentials() + if user is None: + return False + spawn = pexpect.spawnu("ssh {}@{}.{} /bin/bash".format(user, DEFAULT_SITE, + IOTLAB_DOMAIN)) + spawn.sendline("echo $USER") + return bool(spawn.expect([pexpect.TIMEOUT, "{}".format(user)], + timeout=5)) + + +def check_sudo(): + sudo_only_mark = None + if os.geteuid() != 0: + sudo_only_mark = pytest.mark.skip(reason="Test needs sudo to run") + return sudo_only_mark + + +def check_local(run_local): + local_only_mark = None + if not run_local: + local_only_mark = pytest.mark.skip(reason="Test can't run on IoT-LAB") + return local_only_mark + + +def check_credentials(run_local): + iotlab_creds_mark = None + if not run_local and not IoTLABExperiment.check_user_credentials(): + iotlab_creds_mark = pytest.mark.skip( + reason="Test requires IoT-LAB credentials in {}. " + "Use `iotlab-auth` to create".format( + os.path.join(os.environ["HOME"], ".iotlabrc")) + ) + elif not run_local and not check_ssh(): + iotlab_creds_mark = pytest.mark.skip( + reason="Can't access IoT-LAB front-end {}.{} via SSH. " + "Use key without password or `ssh-agent`".format( + DEFAULT_SITE, IOTLAB_DOMAIN) + ) + return iotlab_creds_mark + + +def check_rc(only_rc_allowed): + rc_only_mark = None + output = subprocess.check_output([ + "git", "-C", os.environ["RIOTBASE"], "log", "-1", "--oneline", + "--decorate" + ]).decode() + is_rc = re.search(r"tag:\s\d{4}.\d{2}-RC\d+", output) is not None + + if only_rc_allowed and not is_rc: + rc_only_mark = pytest.mark.skip( + reason="RIOT version under test is not a release candidate" + ) + return rc_only_mark diff --git a/testutils/shell.py b/testutils/shell.py new file mode 100644 index 00000000..8094e55b --- /dev/null +++ b/testutils/shell.py @@ -0,0 +1,171 @@ +""" +Extra shell interactions and parsers not covered by `riotctrl_shell` and +convenience functions for the use of ShellInteractions and +ShellInteractionParsers +""" + +import math +import re + +import pexpect +from riotctrl.shell import ShellInteraction, ShellInteractionParser + +from riotctrl_shell.gnrc import GNRCICMPv6EchoParser, GNRCPktbufStatsParser +from riotctrl_shell.netif import IfconfigListParser + + +PARSERS = { + "ping6": GNRCICMPv6EchoParser(), + "pktbuf": GNRCPktbufStatsParser(), + "ifconfig": IfconfigListParser(), +} + + +# pylint: disable=R0903 +class GNRCUDPClientSendParser(ShellInteractionParser): + """ + Shell interaction parser for + + - $RIOTBASE/examples/gnrc_networking + - $RIOTBASE/tests/gnrc_udp. + + As the `udp` shell command is application specific, a central + ShellInteraction in `riotctrl_shell` does not make much sense + """ + def __init__(self): + self.success_c = re.compile(r"Success:\s+sen[td]\s+" + r"(?P\d+)\s+" + r"byte(\(s\))?\s+to\s+" + r"\[(?P[0-9a-f:]+(%\S+)?)\]:" + r"(?P\d+)$") + + def parse(self, cmd_output): + """ + Parses output of GNRCUDP::udp_client_send() + + >>> parser = GNRCUDPClientSendParser() + >>> res = parser.parse("Success: send 12 byte to [fe80::1%2]:1337 \\n" + ... "Success: sent 5 byte(s) to [abcd::2]:52\\n") + >>> len(res) + 2 + >>> sorted(res[0]) + ['dport', 'dst', 'payload_len'] + >>> sorted(res[1]) + ['dport', 'dst', 'payload_len'] + >>> res[0]["payload_len"], res[1]["payload_len"] + (12, 5) + >>> res[0]["dst"], res[1]["dst"] + ('fe80::1%2', 'abcd::2') + >>> res[0]["dport"], res[1]["dport"] + (1337, 52) + """ + res = [] + for line in cmd_output.splitlines(): + m = self.success_c.search(line.strip()) + if m is not None: + msg = m.groupdict() + msg["payload_len"] = int(msg["payload_len"]) + msg["dport"] = int(msg["dport"]) + res.append(msg) + return res + + +class GNRCUDP(ShellInteraction): + """ + Shell interaction for + + - $RIOTBASE/examples/gnrc_networking + - $RIOTBASE/tests/gnrc_udp. + + As the `udp` shell command is application specific, a central + ShellInteraction in `riotctrl_shell` does not make much sense + """ + @ShellInteraction.check_term + def udp_server_start(self, port, timeout=-1, async_=False): + res = self.cmd("udp server start {}".format(port), timeout=timeout, + async_=async_) + if "Success:" not in res: + raise RuntimeError(res) + return res + + @ShellInteraction.check_term + def udp_server_stop(self, timeout=-1, async_=False): + return self.cmd("udp server stop", timeout=timeout, async_=async_) + + def udp_server_check_output(self, count, delay_ms): + packets_lost = 0 + if delay_ms > 0: + timeout = (delay_ms / 1000) * 10 + else: + timeout = 1 + for _ in range(count): + exp = self.riotctrl.term.expect([ + pexpect.TIMEOUT, + r"Packets received:\s+\d", + r"PKTDUMP: data received:" + ], timeout=timeout) + if not exp: # expect timed out + packets_lost += 1 + if exp < 2: + continue + try: + self.riotctrl.term.expect( + r"~~ SNIP 0 - size:\s+\d+ byte, type: NETTYPE_UNDEF " + r"\(\d+\)" + ) + self.riotctrl.term.expect( + r"~~ SNIP 1 - size:\s+\d+ byte, type: NETTYPE_UDP \(\d+\)" + ) + self.riotctrl.term.expect( + r"~~ SNIP 2 - size:\s+40 byte, type: NETTYPE_IPV6 \(\d+\)" + ) + self.riotctrl.term.expect( + r"~~ SNIP 3 - size:\s+\d+ byte, type: NETTYPE_NETIF " + r"\(-1\)" + ) + self.riotctrl.term.expect( + r"~~ PKT\s+-\s+4 snips, total size:\s+\d+ byte" + ) + except pexpect.TIMEOUT: + packets_lost += 1 + return (packets_lost / count) * 100 + + # pylint: disable=R0913 + @ShellInteraction.check_term + def udp_client_send(self, dest_addr, port, payload, + count=1, delay_ms=1000, async_=False): + if delay_ms: + # wait .5 sec more per message + timeout = math.ceil((delay_ms * count) / 1000) + (5 * (count / 10)) + else: + # wait 1 sec per message + timeout = count * 1 + res = self.cmd( + "udp send {dest_addr} {port} {payload} {count} {delay_us}" + .format(dest_addr=dest_addr, port=port, payload=payload, + count=count, delay_us=int(delay_ms * 1000)), + timeout=timeout, async_=async_ + ) + if "Error:" in res: + raise RuntimeError(res) + return res + + +def ping6(pinger, hostname, count, packet_size, interval): + out = pinger.ping6(hostname, count=count, packet_size=packet_size, + interval=interval) + return PARSERS["ping6"].parse(out) + + +def pktbuf(node): + out = node.pktbuf_stats() + res = PARSERS["pktbuf"].parse(out) + return res + + +def lladdr(ifconfig_out): + netifs = PARSERS["ifconfig"].parse(ifconfig_out) + key = next(iter(netifs)) + netif = netifs[key] + return key, [addr["addr"] for addr in netif["ipv6_addrs"] if + addr["scope"] == "link"][0] diff --git a/testutils/tests/__init__.py b/testutils/tests/__init__.py new file mode 100644 index 00000000..b84fc35e --- /dev/null +++ b/testutils/tests/__init__.py @@ -0,0 +1,3 @@ +import pytest + +pytestmark = pytest.mark.self_test diff --git a/testutils/tests/test_iotlab.py b/testutils/tests/test_iotlab.py new file mode 100644 index 00000000..2b12b2e8 --- /dev/null +++ b/testutils/tests/test_iotlab.py @@ -0,0 +1,188 @@ +import pytest + +import testutils.iotlab + + +# pylint: disable=R0903 +class MockRIOTCtrl(): + def __init__(self, env): + self.env = env + + def board(self): + return self.env.get("BOARD") + + +@pytest.mark.parametrize( + "iotlab_node,expected", + [("arduino-zero-1.saclay.iot-lab.info", "arduino-zero"), + ("st-lrwan1-2.saclay.iot-lab.info", "b-l072z-lrwan1"), + ("st-iotnode-3.saclay.iot-lab.info", "b-l475e-iot01a"), + ("firefly-10.lille.iot-lab.info", "firefly"), + ("frdm-kw41z-4.saclay.iot-lab.info", "frdm-kw41z"), + ("a8-125.grenoble.iot-lab.info", "iotlab-a8-m3"), + ("m3-23.lyon.iot-lab.info", "iotlab-m3"), + ("microbit-5.saclay.iot-lab.info", "microbit"), + ("nrf51dk-2.saclay.iot-lab.info", "nrf51dk"), + ("nrf52dk-6.saclay.iot-lab.info", "nrf52dk"), + ("nrf52832mdk-1.saclay.iot-lab.info", "nrf52832-mdk"), + ("nrf52840dk-7.saclay.iot-lab.info", "nrf52840dk"), + ("nrf52840mdk-1.saclay.iot-lab.info", "nrf52840-mdk"), + ("phynode-1.saclay.iot-lab.info", "pba-d-01-kw2x"), + ("samr21-19.saclay.iot-lab.info", "samr21-xpro"), + ("samr30-3.saclay.iot-lab.info", "samr30-xpro")] +) +def test_board_from_iotlab_node(iotlab_node, expected): + assert testutils.iotlab.IoTLABExperiment.board_from_iotlab_node( + iotlab_node + ) == expected + + +def test_board_from_iotlab_node_invalid(): + with pytest.raises(ValueError): + testutils.iotlab.IoTLABExperiment.board_from_iotlab_node("foobar") + + +def test_valid_board(): + assert testutils.iotlab.IoTLABExperiment.valid_board( + next(iter(testutils.iotlab.IoTLABExperiment.BOARD_ARCHI_MAP)) + ) + + +def test_invalid_board(): + assert not testutils.iotlab.IoTLABExperiment.valid_board("ghgsoqwczoe") + + +@pytest.mark.parametrize( + "iotlab_node,site,board", + [("m3-84.grenoble.iot-lab.info", "grenoble", None), + ("a8-11.lyon.iot-lab.info", "lyon", None), + ("m3-84.lille.iot-lab.info", "lille", "iotlab-m3"), + ("a8-84.saclay.iot-lab.info", "saclay", "iotlab-a8-m3")] +) +def test_valid_iotlab_node(iotlab_node, site, board): + testutils.iotlab.IoTLABExperiment.valid_iotlab_node(iotlab_node, site, + board) + + +@pytest.mark.parametrize( + "iotlab_node,site,board", + [("m3-84.grenoble.iot-lab.info", "lyon", None), + ("wuadngum", "gvcedudng", None), + ("m3-84.lille.iot-lab.info", "lille", "samr21-xpro"), + ("a8-84.saclay.iot-lab.info", "saclay", "eauneguä")] +) +def test_invalid_iotlab_node(iotlab_node, site, board): + with pytest.raises(ValueError): + testutils.iotlab.IoTLABExperiment.valid_iotlab_node(iotlab_node, site, + board) + + +def test_user_credentials(monkeypatch): + creds = ("user", "password") + monkeypatch.setattr(testutils.iotlab, "get_user_credentials", + lambda: creds) + assert testutils.iotlab.IoTLABExperiment.user_credentials() == creds + + +def test_check_user_credentials(monkeypatch): + monkeypatch.setattr(testutils.iotlab, "get_user_credentials", + lambda: ("user", "password")) + assert testutils.iotlab.IoTLABExperiment.check_user_credentials() + + +def test_check_user_credentials_unset(monkeypatch): + monkeypatch.setattr(testutils.iotlab, "get_user_credentials", + lambda: (None, None)) + assert not testutils.iotlab.IoTLABExperiment.check_user_credentials() + + +@pytest.mark.parametrize( + "ctrl_envs,args,exp_boards", + [([{"IOTLAB_NODE": "m3-23.saclay.iot-lab.info", "BOARD": "iotlab-m3"}], + (), ["iotlab-m3"]), + ([{"IOTLAB_NODE": "m3-23.saclay.iot-lab.info"}], (), ["iotlab-m3"]), + ([{"BOARD": "iotlab-m3"}], (), ["iotlab-m3"]), + ([{"IOTLAB_NODE": "m3-23.saclay.iot-lab.info", "BOARD": "iotlab-m3"}], + ("saclay",), ["iotlab-m3"]), + ([{"IOTLAB_NODE": "m3-23.grenoble.iot-lab.info"}], ("grenoble",), + ["iotlab-m3"]), + ([{"BOARD": "iotlab-m3"}], ("lille",), ["iotlab-m3"])] +) +def test_init(ctrl_envs, args, exp_boards): + assert testutils.iotlab.DEFAULT_SITE == "saclay" + ctrls = [MockRIOTCtrl(env) for env in ctrl_envs] + exp = testutils.iotlab.IoTLABExperiment("test", ctrls, *args) + if args: + assert exp.site == args[0] + else: + assert exp.site == "saclay" + assert exp.ctrls == ctrls + for ctrl, exp_board in zip(exp.ctrls, exp_boards): + assert ctrl.board() == exp_board + assert exp.name == "test" + assert exp.exp_id is None + + +@pytest.mark.parametrize( + "ctrl_envs,args", + [([{"IOTLAB_NODE": "m3-23.saclay.iot-lab.info", "BOARD": "iotlab-m3"}], + ("khseaip",)), + ([{"BOARD": "uaek-eaqfgnic"}], ()), + ([{"IOTLAB_NODE": "go5wxbp-124.saclay.iot-lab.info"}], ()), + ([{}], ())] +) +def test_init_value_error(ctrl_envs, args): + ctrls = [MockRIOTCtrl(env) for env in ctrl_envs] + with pytest.raises(ValueError): + testutils.iotlab.IoTLABExperiment("test", ctrls, *args) + + +@pytest.mark.parametrize( + "exp_id,expected", + [(None, None), (1234, "This is a test")] +) +def test_stop(monkeypatch, exp_id, expected): + monkeypatch.setattr(testutils.iotlab.IoTLABExperiment, "user_credentials", + lambda cls: ("user", "password")) + monkeypatch.setattr(testutils.iotlab, "Api", lambda user, password: None) + monkeypatch.setattr(testutils.iotlab, "stop_experiment", + lambda api, exp_id: expected) + ctrls = [MockRIOTCtrl({"BOARD": "iotlab-m3"})] + exp = testutils.iotlab.IoTLABExperiment("test", ctrls) + exp.exp_id = exp_id + assert exp.stop() == expected + + +@pytest.mark.parametrize( + "ctrl_envs, exp_nodes", + [([{"BOARD": "nrf52dk"}], ["nrf52dk-5.saclay.iot-lab.info"]), + ([{"IOTLAB_NODE": "samr21-21.saclay.iot-lab.info"}], + ["samr21-21.saclay.iot-lab.info"])] +) +def test_start(monkeypatch, ctrl_envs, exp_nodes): + monkeypatch.setattr(testutils.iotlab.IoTLABExperiment, "user_credentials", + lambda cls: ("user", "password")) + monkeypatch.setattr(testutils.iotlab, "Api", lambda user, password: None) + monkeypatch.setattr(testutils.iotlab, "exp_resources", + lambda arg: arg) + monkeypatch.setattr(testutils.iotlab, "submit_experiment", + lambda api, name, duration, resources: {"id": 12345}) + monkeypatch.setattr(testutils.iotlab, "get_experiment", + lambda api, exp_id: {"nodes": exp_nodes}) + monkeypatch.setattr(testutils.iotlab, "wait_experiment", + lambda api, exp_id: {}) + ctrls = [MockRIOTCtrl(env) for env in ctrl_envs] + exp = testutils.iotlab.IoTLABExperiment("test", ctrls) + exp.start() + assert exp.exp_id == 12345 + + +def test_start_error(monkeypatch): + monkeypatch.setattr(testutils.iotlab.IoTLABExperiment, "user_credentials", + lambda cls: ("user", "password")) + monkeypatch.setattr(testutils.iotlab, "Api", lambda user, password: None) + ctrls = [MockRIOTCtrl({'BOARD': 'iotlab-m3'})] + exp = testutils.iotlab.IoTLABExperiment("test", ctrls) + ctrls[0].env.pop("BOARD") + with pytest.raises(ValueError): + exp.start() diff --git a/testutils/tests/test_native.py b/testutils/tests/test_native.py new file mode 100644 index 00000000..260eb20c --- /dev/null +++ b/testutils/tests/test_native.py @@ -0,0 +1,136 @@ +import pytest + +import testutils.native + + +def test_command_exists(monkeypatch): + monkeypatch.setattr(testutils.native.subprocess, "check_call", + lambda *args, **kwargs: None) + assert testutils.native.command_exists("test") + + +def test_command_not_exists(monkeypatch): + def func(*args, **kwargs): + raise testutils.native.subprocess.CalledProcessError(1, "test") + monkeypatch.setattr(testutils.native.subprocess, "check_call", func) + assert not testutils.native.command_exists("test") + + +@pytest.mark.parametrize( + "args", + ((), ("test",)) +) +def test_ip_link(monkeypatch, args): + monkeypatch.setattr(testutils.native.subprocess, "check_output", + lambda *args, **kwargs: b"This is a test") + assert testutils.native.ip_link(*args) == "This is a test" + + +def test_ip_addr_add(monkeypatch): + monkeypatch.setattr(testutils.native.subprocess, "check_call", + lambda *args, **kwargs: None) + testutils.native.ip_addr_add("test", "2001:db8::1") + + +def test_ip_addr_del(monkeypatch): + monkeypatch.setattr(testutils.native.subprocess, "run", + lambda *args, **kwargs: None) + testutils.native.ip_addr_del("test", "2001:db8::1") + + +@pytest.mark.parametrize( + "args", + ((), ("fe80::1",)) +) +def test_ip_route_add(monkeypatch, args): + monkeypatch.setattr(testutils.native.subprocess, "check_call", + lambda *args, **kwargs: None) + testutils.native.ip_route_add("test", "2001:db8::/64", *args) + + +@pytest.mark.parametrize( + "args", + ((), ("fe80::1",)) +) +def test_ip_route_del(monkeypatch, args): + monkeypatch.setattr(testutils.native.subprocess, "run", + lambda *args, **kwargs: None) + testutils.native.ip_route_del("test", "2001:db8::/64", *args) + + +@pytest.mark.parametrize("expected", [True, False]) +def test_bridged(monkeypatch, expected): + monkeypatch.setattr(testutils.native, "_check_bridged", + lambda *args, **kwargs: expected) + assert testutils.native.bridged(["tap0", "tap1"]) == expected + + +def test_interface_exists(monkeypatch): + monkeypatch.setattr(testutils.native.subprocess, "check_call", + lambda *args, **kwargs: None) + assert testutils.native.interface_exists("test") + + +def test_interface_not_exists(monkeypatch): + def func(*args, **kwargs): + raise testutils.native.subprocess.CalledProcessError(1, "test") + monkeypatch.setattr(testutils.native.subprocess, "check_call", func) + assert not testutils.native.interface_exists("test") + + +def test_get_link_local(monkeypatch): + output = """ +95: tap0: ... + link/ether e2:bc:7d:cb:f5:4f brd ff:ff:ff:ff:ff:ff + inet6 fe80::e0bc:7dff:fecb:f54f/64 scope link + valid_lft forever preferred_lft forever""" + monkeypatch.setattr(testutils.native.subprocess, "check_output", + lambda *args, **kwargs: output.encode()) + assert testutils.native.get_link_local("tap0") == \ + "fe80::e0bc:7dff:fecb:f54f" + + +def test_get_link_local_not_none(monkeypatch): + output = """ +95: tap0: ... + link/ether e2:bc:7d:cb:f5:4f brd ff:ff:ff:ff:ff:ff""" + monkeypatch.setattr(testutils.native.subprocess, "check_output", + lambda *args, **kwargs: output.encode()) + assert testutils.native.get_link_local("tap0") is None + + +def test_bridge_bridged(monkeypatch): + monkeypatch.setattr(testutils.native, "ip_link", + lambda *args, **kwargs: """ +49: tap0: <...> mtu 1500 ... master tapbr0 state ... + link/ether e2:bc:7d:cb:f5:4f brd ff:ff:ff:ff:ff:ff +50: tap1: <...> mtu 1500 ... master tapbr0 state ... + link/ether da:27:1d:a8:64:23 brd ff:ff:ff:ff:ff:ff +""") + assert testutils.native.bridge("tap0") == "tapbr0" + + +def test_bridge_unbridged(monkeypatch): + monkeypatch.setattr(testutils.native, "ip_link", + lambda *args, **kwargs: """ +60: tap0: <...> mtu 1500 qdisc fq_codel state ... + link/ether e2:bc:7d:cb:f5:4f brd ff:ff:ff:ff:ff:ff +""") + assert testutils.native.bridge("tap0") == "tap0" + + +@pytest.mark.parametrize( + "ping_cmd,expected", + [("ping6", "ping6"), ("ping", "ping -6")] +) +def test_get_ping_cmd(monkeypatch, ping_cmd, expected): + monkeypatch.setattr(testutils.native, "command_exists", + lambda cmd: cmd == ping_cmd) + assert testutils.native.get_ping_cmd() == expected + + +def test_get_ping_cmd_no_ping(monkeypatch): + monkeypatch.setattr(testutils.native, "command_exists", + lambda cmd: False) + with pytest.raises(FileNotFoundError): + testutils.native.get_ping_cmd() diff --git a/testutils/tests/test_pytest.py b/testutils/tests/test_pytest.py new file mode 100644 index 00000000..595b3cae --- /dev/null +++ b/testutils/tests/test_pytest.py @@ -0,0 +1,65 @@ +import pytest + +import testutils.pytest + + +class MockSpawn(): + expect_ret = None + + # pylint: disable=W0613 + def __init__(self, cmd, *args, **kwargs): + self.cmd = cmd + + # pylint: disable=W0613 + def sendline(self, *args, **kwargs): + pass + + # pylint: disable=W0613 + def expect(self, *args, **kwargs): + return self.expect_ret + + +@pytest.mark.parametrize( + "iotlab_creds,expect_ret,expect_func", + [(("test", None), 1, lambda x: x), + (("test", None), 0, lambda x: not x), + ((None, None), 0, lambda x: not x), + ((None, None), 1, lambda x: not x)] +) +def test_check_ssh_creds(monkeypatch, iotlab_creds, expect_ret, expect_func): + monkeypatch.setattr(testutils.pytest.IoTLABExperiment, "user_credentials", + lambda: iotlab_creds) + monkeypatch.setattr(testutils.pytest.pexpect, "spawnu", MockSpawn) + MockSpawn.expect_ret = expect_ret + assert expect_func(testutils.pytest.check_ssh()) + + +@pytest.mark.parametrize( + "iotlab_creds,expect_ret,run_local,expect_func", + [(("test", None), 1, False, lambda x: not x), + (("test", None), 0, False, lambda x: x), + ((None, None), 0, False, lambda x: x), + ((None, None), 1, False, lambda x: x), + ((None, None), 0, True, lambda x: not x), + (("test", None), 1, True, lambda x: not x)] +) +def test_check_credentials(monkeypatch, iotlab_creds, expect_ret, expect_func, + run_local): + monkeypatch.setattr(testutils.pytest.IoTLABExperiment, "user_credentials", + lambda: iotlab_creds) + monkeypatch.setattr(testutils.pytest.pexpect, "spawnu", MockSpawn) + MockSpawn.expect_ret = expect_ret + assert expect_func(testutils.pytest.check_credentials(run_local)) + + +@pytest.mark.parametrize( + "output,only_rc_allowed,expect_func", + [(b"foobar", False, lambda x: not x), + (b"foobar", True, lambda x: x), + (b"tag: 2042.06-RC13", True, lambda x: not x), + (b"tag: 2042.06-RC13", False, lambda x: not x)] +) +def test_check_rc(monkeypatch, output, only_rc_allowed, expect_func): + monkeypatch.setattr(testutils.pytest.subprocess, "check_output", + lambda *args, **kwargs: output) + assert expect_func(testutils.pytest.check_rc(only_rc_allowed)) diff --git a/testutils/tests/test_shell.py b/testutils/tests/test_shell.py new file mode 100644 index 00000000..e101316f --- /dev/null +++ b/testutils/tests/test_shell.py @@ -0,0 +1,179 @@ +#! /usr/bin/env python +# -*- coding: utf-8 -*- +# vim:fenc=utf-8 +# +# Copyright © 2020 Martine Lenders +# +# Distributed under terms of the MIT license. + +import pexpect +import pytest + +import riotctrl_shell.gnrc +import riotctrl_shell.tests.common +from riotctrl_shell.tests.common import init_ctrl + +import testutils.shell + + +class ExpectMockSpawn(riotctrl_shell.tests.common.MockSpawn): + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + self._expect_sequence = None + + @property + def expect_sequence(self): + return self._expect_sequence + + @expect_sequence.setter + def expect_sequence(self, expect_sequence): + self._expect_sequence = expect_sequence + self._count = 0 + + # pylint: disable=W0613 + def expect(self, *args, **kwargs): + if self._expect_sequence is None: + raise RuntimeError("No expect_sequence given") + if self._count < len(self._expect_sequence): + res = self._expect_sequence[self._count] + if res is pexpect.TIMEOUT: + raise pexpect.TIMEOUT("") + self._count += 1 + return res + raise RuntimeError("State error") + + +class ExpectMockRIOTCtrl(riotctrl_shell.tests.common.MockRIOTCtrl): + def start_term(self, **spawnkwargs): + self.term = ExpectMockSpawn(ctrl=self) + + +def test_udp_server_start(): + ctrl = init_ctrl(output="Success: server was started") + shell = testutils.shell.GNRCUDP(ctrl) + res = shell.udp_server_start(1337) + assert res.startswith("Success:") + assert ctrl.term.last_command == "udp server start 1337" + + +def test_udp_server_start_error(): + ctrl = init_ctrl() + shell = testutils.shell.GNRCUDP(ctrl) + with pytest.raises(RuntimeError): + shell.udp_server_start(1337) + assert ctrl.term.last_command == "udp server start 1337" + + +def test_udp_server_stop(): + ctrl = init_ctrl() + shell = testutils.shell.GNRCUDP(ctrl) + res = shell.udp_server_stop() + # mock just returns last input + assert res == "udp server stop" + + +@pytest.mark.parametrize( + "expect_sequence,count,delay,expected", + [ + ([0, 0, 0, 0, 0], 5, 10, 100.0), + ([1, 1, 1, 1, 1], 5, 10, 0.0), + ([1, 1, 1, 1, 0], 5, 0, 20.0), + ([1] + ([0] * 9), 10, 10, 90.0), + ([2] + ([0] * 5), 1, 10, 0.0), + ([2] + ([0] * 5) + [2, 0, 0, pexpect.TIMEOUT], 2, 10, 50.0), + ([2] + ([0] * 5) + [2, pexpect.TIMEOUT], 2, 10, 50.0), + ] +) +def test_udp_server_check_output(expect_sequence, count, delay, expected): + ctrl = ExpectMockRIOTCtrl("foobar", env={"BOARD": "native"}) + shell = testutils.shell.GNRCUDP(ctrl) + ctrl.start_term() + ctrl.term.expect_sequence = expect_sequence + assert shell.udp_server_check_output(count, delay) == expected + ctrl.stop_term() + + +@pytest.mark.parametrize( + "dest_addr,port,payload,count,delay_ms,expected", + [ + ("ff02::1", 1337, '"Hallo World"', 1000, 1000, + 'udp send ff02::1 1337 "Hallo World" 1000 1000000'), + ("affe::1", 61616, 15, 1000, 0, 'udp send affe::1 61616 15 1000 0'), + ("fe80::1", 52, 1000, 1000, 0.01, 'udp send fe80::1 52 1000 1000 10'), + ] +) +# pylint: disable=R0913 +def test_udp_client_send(dest_addr, port, payload, count, delay_ms, expected): + ctrl = init_ctrl() + shell = testutils.shell.GNRCUDP(ctrl) + res = shell.udp_client_send(dest_addr, port, payload, count, delay_ms) + # mock just returns last input + assert res == expected + + +def test_udp_client_send_error(): + ctrl = init_ctrl(output="Error: we don't want to send") + shell = testutils.shell.GNRCUDP(ctrl) + with pytest.raises(RuntimeError): + shell.udp_client_send("ff02::1", 1337, 10, 1000, 1000) + + +def test_ping6(): + ctrl = init_ctrl(output=""" +12 bytes from ::1: icmp_seq=0 ttl=64 +12 bytes from ::1: icmp_seq=1 ttl=64 +12 bytes from ::1: icmp_seq=2 ttl=64 + +--- ::1 PING statistics --- +3 packets transmitted, 3 packets received, 0% packet loss""") + shell = riotctrl_shell.gnrc.GNRCICMPv6Echo(ctrl) + ping_res = testutils.shell.ping6(shell, "ff02::1", 3, 4, 1000) + assert ping_res + assert len(ping_res["replies"]) == 3 + assert ping_res["stats"]["packet_loss"] == 0 + + +def test_pktbuf_empty(): + ctrl = init_ctrl(output=""" +packet buffer: first byte: 0x5660dce0, last byte: 0x5660fce0 (size: 8192) + position of last byte used: 1792 +~ unused: 0x5660dce0 (next: (nil), size: 8192) ~""") + shell = riotctrl_shell.gnrc.GNRCPktbufStats(ctrl) + pktbuf_res = testutils.shell.pktbuf(shell) + assert pktbuf_res + assert pktbuf_res.is_empty() + + +def test_pktbuf_not_empty(): + ctrl = init_ctrl(output=""" +packet buffer: first byte: 0x5660dce0, last byte: 0x5660fce0 (size: 8192) + position of last byte used: 1792 +~ unused: 0x5660de00 (next: (nil), size: 7904) ~""") + shell = riotctrl_shell.gnrc.GNRCPktbufStats(ctrl) + pktbuf_res = testutils.shell.pktbuf(shell) + assert pktbuf_res + assert not pktbuf_res.is_empty() + + +def test_lladdr(): + netif, lladdr = testutils.shell.lladdr(""" +Iface 6 HWaddr: 6A:2E:4F:3D:DF:CB + L2-PDU:1500 MTU:1500 HL:64 RTR + RTR_ADV + Source address length: 6 + Link type: wired + inet6 addr: fe80::682e:4fff:fe3d:dfcb scope: link VAL + inet6 group: ff02::2 + inet6 group: ff02::1 + inet6 group: ff02::1:ff3d:dfcb + """) + assert netif == "6" + assert lladdr == "fe80::682e:4fff:fe3d:dfcb" + + +def test_lladdr_no_lladdr(): + with pytest.raises(KeyError): + testutils.shell.lladdr(""" +Iface 4 HWaddr: 6A:2E:4F:3D:DF:CB + L2-PDU:1500 Source address length: 6 + """) diff --git a/tox.ini b/tox.ini new file mode 100644 index 00000000..5e1f2119 --- /dev/null +++ b/tox.ini @@ -0,0 +1,57 @@ +[tox] +envlist = test +skipsdist = True + +[testenv] +basepython = python3 +commands = + test: {[testenv:test]commands} + flake8: {[testenv:flake8]commands} + pylint: {[testenv:pylint]commands} + +[testenv:test] +passenv = + BUILD_IN_DOCKER + HOME + IOTLAB_SITE + RIOTBASE + SSH_AUTH_SOCK + SSH_AGENT_PID +setenv = + PYTHONPATH = {env:RIOTBASE}/dist/pythonlibs:{env:PYTHONPATH:} +deps = + aiocoap + iotlabcli + pytest + riotctrl + scapy +commands = + pytest {posargs} + +[testenv:flake8] +deps = flake8 +commands = + flake8 + +[testenv:pylint] +deps = + aiocoap + iotlabcli + pylint + pytest + riotctrl + scapy +setenv = + PYTHONPATH = {env:RIOTBASE}/dist/pythonlibs:{env:PYTHONPATH:} +commands = + pylint \ + conftest.py \ + testutils/ \ + 03-single-hop-ipv6-icmp/ \ + 04-single-hop-6lowpan-icmp/ \ + 05-single-hop-route/ \ + 06-single-hop-udp/ \ + 07-multi-hop/ \ + 08-interop/ \ + 09-coap/ \ + 10-icmpv6-error/