diff --git a/.github/workflows/continuous-integration.yml b/.github/workflows/continuous-integration.yml index 84b30c0..dd7eba6 100644 --- a/.github/workflows/continuous-integration.yml +++ b/.github/workflows/continuous-integration.yml @@ -20,12 +20,12 @@ jobs: fail-fast: false matrix: os: [ubuntu-latest] - python-version: [3.7, 3.8, 3.9] + python-version: ['3.10', '3.13', '3.14'] steps: - - uses: actions/checkout@master + - uses: actions/checkout@v4 - name: Set up Python ${{ matrix.python-version }} - uses: actions/setup-python@v1 + uses: actions/setup-python@v5 with: python-version: ${{ matrix.python-version }} - name: Install dependencies @@ -37,36 +37,34 @@ jobs: tox -e py$(echo ${{ matrix.python-version }} | tr -d .)-tests - name: Test cli run: | - tox -e py$(echo ${{ matrix.python-version }} | tr -d .)-cli + tox -e py$(echo ${{ matrix.python-version }} | tr -d .)-cli - name: Check style - if: ${{ matrix.python-version == 3.8 }} + if: ${{ matrix.python-version == '3.13' }} run: | - tox -e py$(echo ${{ matrix.python-version }} | tr -d .)-lint + tox -e py313-lint tox -e copying - name: Upload coverage to Codecov - if: ${{ matrix.python-version == 3.8 }} - uses: codecov/codecov-action@v1 + if: ${{ matrix.python-version == '3.13' }} + uses: codecov/codecov-action@v5 with: fail_ci_if_error: true - - name: Build a source tarball - if: matrix.python-version == 3.8 - run: python setup.py sdist check --strict --metadata + - name: Check package + if: ${{ matrix.python-version == '3.13' }} + run: tox -e py313-checksetup - name: Publish package to TestPyPI - uses: pypa/gh-action-pypi-publish@master + uses: pypa/gh-action-pypi-publish@release/v1 if: >- - matrix.python-version == 3.8 && + matrix.python-version == '3.13' && github.event_name == 'push' && startsWith(github.event.ref, 'refs/tags') with: - user: __token__ password: ${{ secrets.test_pypi_password }} - repository_url: https://test.pypi.org/legacy/ + repository-url: https://test.pypi.org/legacy/ - name: Publish package to PyPI - uses: pypa/gh-action-pypi-publish@master + uses: pypa/gh-action-pypi-publish@release/v1 if: >- - matrix.python-version == 3.8 && + matrix.python-version == '3.13' && github.event_name == 'push' && startsWith(github.event.ref, 'refs/tags') with: - user: __token__ password: ${{ secrets.pypi_password }} diff --git a/iotlabsshcli/__init__.py b/iotlabsshcli/__init__.py index 0ae8b8f..59c300c 100644 --- a/iotlabsshcli/__init__.py +++ b/iotlabsshcli/__init__.py @@ -20,4 +20,4 @@ # The fact that you are presently reading this means that you have had # knowledge of the CeCILL license and that you accept its terms. -__version__ = '1.1.0' +__version__ = "1.1.0" diff --git a/iotlabsshcli/_deprecated.py b/iotlabsshcli/_deprecated.py new file mode 100644 index 0000000..8f10b2c --- /dev/null +++ b/iotlabsshcli/_deprecated.py @@ -0,0 +1,29 @@ +# This file is a part of IoT-LAB ssh-cli-tools +# Copyright (C) 2015 INRIA (Contact: admin@iot-lab.info) +# Contributor(s) : see AUTHORS file +# +# This software is governed by the CeCILL license under French law +# and abiding by the rules of distribution of free software. You can use, +# modify and/ or redistribute the software under the terms of the CeCILL +# license as circulated by CEA, CNRS and INRIA at the following URL +# http://www.cecill.info. +# +# As a counterpart to the access to the source code and rights to copy, +# modify and redistribute granted by the license, users are provided only +# with a limited warranty and the software's author, the holder of the +# economic rights, and the successive licensors have only limited +# liability. +# +# The fact that you are presently reading this means that you have had +# knowledge of the CeCILL license and that you accept its terms. + +"""Deprecated entry points kept for backward compatibility.""" + +from iotlabcli.helpers import deprecate_cmd + +from iotlabsshcli.parser.open_linux_parser import main as _main + + +def open_a8_cli(): + """Entry point for the deprecated open-a8-cli command.""" + deprecate_cmd(_main, "open-a8-cli", "iotlab-ssh") diff --git a/iotlabsshcli/open_linux.py b/iotlabsshcli/open_linux.py index 6c39cb2..54f3c6b 100644 --- a/iotlabsshcli/open_linux.py +++ b/iotlabsshcli/open_linux.py @@ -22,6 +22,7 @@ import os.path from collections import OrderedDict + from iotlabsshcli.sshlib import OpenLinuxSsh @@ -29,20 +30,19 @@ def _nodes_grouped(nodes): """Group nodes per site from a list of nodes. >>> _nodes_grouped([]) OrderedDict() - >>> _nodes_grouped(['node-a8-1.grenoble.iot-lab.info', - ... 'node-a8-2.grenoble.iot-lab.info', - ... 'node-a8-2.saclay.iot-lab.info', - ... 'node-a8-2.lille.iot-lab.info']) + >>> list(_nodes_grouped(['node-a8-1.grenoble.iot-lab.info', + ... 'node-a8-2.grenoble.iot-lab.info', + ... 'node-a8-2.saclay.iot-lab.info', + ... 'node-a8-2.lille.iot-lab.info']).items()) ... # doctest: +NORMALIZE_WHITESPACE - OrderedDict([('grenoble.iot-lab.info', - ['node-a8-1.grenoble.iot-lab.info', - 'node-a8-2.grenoble.iot-lab.info']), - ('saclay.iot-lab.info', ['node-a8-2.saclay.iot-lab.info']), - ('lille.iot-lab.info', ['node-a8-2.lille.iot-lab.info'])]) + [('grenoble.iot-lab.info', + ['node-a8-1.grenoble.iot-lab.info', 'node-a8-2.grenoble.iot-lab.info']), + ('saclay.iot-lab.info', ['node-a8-2.saclay.iot-lab.info']), + ('lille.iot-lab.info', ['node-a8-2.lille.iot-lab.info'])] """ result = OrderedDict() for host in nodes: - site = host.split('.', 1)[1] + site = host.split(".", 1)[1] if site not in result: result.update({site: [host]}) else: @@ -50,17 +50,16 @@ def _nodes_grouped(nodes): return result -_FLASH_CMD = 'source /etc/profile && /usr/bin/iotlab_flash {fw} {bin}' -_RESET_CMD = 'source /etc/profile && /usr/bin/iotlab_reset' -_MAKE_EXECUTABLE_CMD = 'chmod +x {}' -_RUN_SCRIPT_CMD = 'source /etc/profile && ' -_RUN_SCRIPT_CMD += 'screen -S {screen} -dm bash -c \"{path}\"' -_QUIT_SCRIPT_CMD = 'screen -X -S {screen} quit' -_REMOTE_SHARED_DIR = 'shared/.iotlabsshcli' +_FLASH_CMD = "source /etc/profile && /usr/bin/iotlab_flash {fw} {bin}" +_RESET_CMD = "source /etc/profile && /usr/bin/iotlab_reset" +_MAKE_EXECUTABLE_CMD = "chmod +x {}" +_RUN_SCRIPT_CMD = 'source /etc/profile && screen -S {screen} -dm bash -c "{path}"' +_QUIT_SCRIPT_CMD = "screen -X -S {screen} quit" +_REMOTE_SHARED_DIR = "shared/.iotlabsshcli" def flash(config_ssh, nodes, firmware, verbose=False): - """Flash the firmware of co-microcontroller """ + """Flash the firmware of co-microcontroller""" failed_hosts = [] # configure ssh and remote firmware names. groups = _nodes_grouped(nodes) @@ -70,18 +69,18 @@ def flash(config_ssh, nodes, firmware, verbose=False): # scp create remote directory if it doesn't exist result = ssh.scp(firmware, remote_fw) # We delete failed hosts for the next command - if '1' in result: - failed_hosts = [ssh.groups.pop(res) for res in result['1']] + if "1" in result: + failed_hosts = [ssh.groups.pop(res) for res in result["1"]] # Run firmware update. bin_opt = "--bin" if remote_fw.endswith(".bin") else "" result = ssh.run(_FLASH_CMD.format(fw=remote_fw, bin=bin_opt)) for hosts in failed_hosts: - result.setdefault('1', []).extend(hosts) + result.setdefault("1", []).extend(hosts) return {"flash": result} def reset(config_ssh, nodes, verbose=False): - """Reset co-microcontroller """ + """Reset co-microcontroller""" # Configure ssh groups = _nodes_grouped(nodes) @@ -91,7 +90,7 @@ def reset(config_ssh, nodes, verbose=False): def wait_for_boot(config_ssh, nodes, max_wait=120, verbose=False): - """Wait for the open Linux nodes boot """ + """Wait for the open Linux nodes boot""" # Configure ssh. groups = _nodes_grouped(nodes) @@ -101,7 +100,7 @@ def wait_for_boot(config_ssh, nodes, max_wait=120, verbose=False): def run_cmd(config_ssh, nodes, cmd, run_on_frontend=False, verbose=False): - """Run a command on the Linux nodes or SSH frontend servers """ + """Run a command on the Linux nodes or SSH frontend servers""" # Configure ssh. groups = _nodes_grouped(nodes) @@ -110,7 +109,7 @@ def run_cmd(config_ssh, nodes, cmd, run_on_frontend=False, verbose=False): def copy_file(config_ssh, nodes, file_path, verbose=False): - """Copy a file to SSH frontend servers """ + """Copy a file to SSH frontend servers""" # Configure ssh. groups = _nodes_grouped(nodes) @@ -127,45 +126,41 @@ def _get_failed_result(groups, result, run_on_frontend): We delete failed hosts for the next commands in the groups """ failed = [] - if '1' in result and result['1']: + if "1" in result and result["1"]: if not run_on_frontend: # nodes list - for site in result['1']: + for site in result["1"]: failed.extend(groups[site]) del groups[site] else: # servers list - failed.extend(result['1']) - all(map(groups.pop, result['1'])) + failed.extend(result["1"]) + all(map(groups.pop, result["1"])) return failed -def run_script(config_ssh, nodes, script, run_on_frontend=False, - verbose=False): - """Run a script in background on Linux nodes or SSH frontend servers """ +def run_script(config_ssh, nodes, script, run_on_frontend=False, verbose=False): + """Run a script in background on Linux nodes or SSH frontend servers""" # Configure ssh. failed_hosts = [] groups = _nodes_grouped(nodes) ssh = OpenLinuxSsh(config_ssh, groups, verbose=verbose) - screen = f'{config_ssh["user"]}-{config_ssh["exp_id"]}' + screen = f"{config_ssh['user']}-{config_ssh['exp_id']}" remote_script = os.path.join(_REMOTE_SHARED_DIR, os.path.basename(script)) - script_data = {'screen': screen, 'path': remote_script} + script_data = {"screen": screen, "path": remote_script} # Copy script on SSH frontend servers scp_result = ssh.scp(script, remote_script) - failed_hosts.append(_get_failed_result(ssh.groups, scp_result, - run_on_frontend)) + failed_hosts.append(_get_failed_result(ssh.groups, scp_result, run_on_frontend)) # Make script executable - run_result = ssh.run(_MAKE_EXECUTABLE_CMD.format(remote_script), - with_proxy=False) - failed_hosts.append(_get_failed_result(ssh.groups, run_result, - run_on_frontend)) + run_result = ssh.run(_MAKE_EXECUTABLE_CMD.format(remote_script), with_proxy=False) + failed_hosts.append(_get_failed_result(ssh.groups, run_result, run_on_frontend)) # Try cleanup and kill any running script (don't check the result) - ssh.run(_QUIT_SCRIPT_CMD.format(**script_data), - with_proxy=not run_on_frontend) + ssh.run(_QUIT_SCRIPT_CMD.format(**script_data), with_proxy=not run_on_frontend) # Run script - result = ssh.run(_RUN_SCRIPT_CMD.format(**script_data), - with_proxy=not run_on_frontend, use_pty=False) + result = ssh.run( + _RUN_SCRIPT_CMD.format(**script_data), with_proxy=not run_on_frontend, use_pty=False + ) for hosts in filter(None, failed_hosts): - result.setdefault('1', []).extend(hosts) + result.setdefault("1", []).extend(hosts) return {"run-script": result} diff --git a/iotlabsshcli/parser/open_linux_parser.py b/iotlabsshcli/parser/open_linux_parser.py index 1fdb83f..21644ad 100644 --- a/iotlabsshcli/parser/open_linux_parser.py +++ b/iotlabsshcli/parser/open_linux_parser.py @@ -20,15 +20,14 @@ # The fact that you are presently reading this means that you have had # knowledge of the CeCILL license and that you accept its terms. - -import sys import argparse -from iotlabcli import auth -from iotlabcli import helpers -from iotlabcli import rest +import sys + +from iotlabcli import auth, helpers, rest +from iotlabcli.helpers import deprecate_warn_cmd from iotlabcli.parser import common from iotlabcli.parser.common import _get_experiment_nodes_list -from iotlabcli.helpers import deprecate_warn_cmd + import iotlabsshcli.open_linux @@ -36,9 +35,9 @@ def parse_options(): """Parse command line option.""" parent_parser = argparse.ArgumentParser(add_help=False) common.add_auth_arguments(parent_parser, False) - parent_parser.add_argument('-v', '--version', - action='version', - version=iotlabsshcli.__version__) + parent_parser.add_argument( + "-v", "--version", action="version", version=iotlabsshcli.__version__ + ) # We create top level parser parser = argparse.ArgumentParser( @@ -48,104 +47,95 @@ def parse_options(): common.add_expid_arg(parser) common.add_output_formatter(parser) - subparsers = parser.add_subparsers(dest='command') + subparsers = parser.add_subparsers(dest="command") subparsers.required = True # needed for python 3. # pylint: disable=super-with-arguments class DeprecateHelpFormatter(argparse.HelpFormatter): - """ Add drepecated help formatter """ + """Add drepecated help formatter""" def add_usage(self, usage, actions, groups, prefix=None): # self._prog = iotlab-ssh flash-m3 | reset-m3 old_cmd = self._prog.split()[-1] - new_cmd = old_cmd.split('-')[0] + new_cmd = old_cmd.split("-")[0] deprecate_warn_cmd(old_cmd, new_cmd, 20) - return super(DeprecateHelpFormatter, self).add_usage(usage, - actions, - groups, - prefix) - - update_parser = subparsers.add_parser('flash', - parents=[parent_parser], - help='Flash node\'s ' - 'co-microcontroller') - update_parser.add_argument('firmware', help='firmware path.') + return super(DeprecateHelpFormatter, self).add_usage(usage, actions, groups, prefix) + + update_parser = subparsers.add_parser( + "flash", parents=[parent_parser], help="Flash node's co-microcontroller" + ) + update_parser.add_argument("firmware", help="firmware path.") # nodes list or exclude list common.add_nodes_selection_list(update_parser) # reset parser - reset_parser = subparsers.add_parser('reset', - parents=[parent_parser], - help='Reset node\'s ' - 'co-microcontroller') + reset_parser = subparsers.add_parser( + "reset", parents=[parent_parser], help="Reset node's co-microcontroller" + ) # nodes list or exclude list common.add_nodes_selection_list(reset_parser) # wait-for-boot parser - boot_parser = subparsers.add_parser('wait-for-boot', - parents=[parent_parser], - help='Waits until Linux nodes ' - 'have boot') - boot_parser.add_argument('--max-wait', - type=int, - default=120, - help='Maximum waiting delay for nodes boot ' - '(in seconds)') + boot_parser = subparsers.add_parser( + "wait-for-boot", parents=[parent_parser], help="Waits until Linux nodes have boot" + ) + boot_parser.add_argument( + "--max-wait", + type=int, + default=120, + help="Maximum waiting delay for nodes boot (in seconds)", + ) # nodes list or exclude list common.add_nodes_selection_list(boot_parser) # run-script parser - run_script_parser = subparsers.add_parser('run-script', - parents=[parent_parser], - help='Run a script in ' - 'background on Linux ' - 'nodes') - run_script_parser.add_argument('script', help='script path.') - run_script_parser.add_argument('--frontend', action='store_true', - help='Execution on SSH frontend') + run_script_parser = subparsers.add_parser( + "run-script", parents=[parent_parser], help="Run a script in background on Linux nodes" + ) + run_script_parser.add_argument("script", help="script path.") + run_script_parser.add_argument( + "--frontend", action="store_true", help="Execution on SSH frontend" + ) # nodes list or exclude list common.add_nodes_selection_list(run_script_parser) # run-cmd parser - run_cmd_parser = subparsers.add_parser('run-cmd', - parents=[parent_parser], - help='Run a command on Linux nodes') - run_cmd_parser.add_argument('cmd', help='Command') - run_cmd_parser.add_argument('--frontend', action='store_true', - help='Execution on SSH frontend') + run_cmd_parser = subparsers.add_parser( + "run-cmd", parents=[parent_parser], help="Run a command on Linux nodes" + ) + run_cmd_parser.add_argument("cmd", help="Command") + run_cmd_parser.add_argument( + "--frontend", action="store_true", help="Execution on SSH frontend" + ) # nodes list or exclude list common.add_nodes_selection_list(run_cmd_parser) # copy-file parser - copy_file_parser = subparsers.add_parser('copy-file', - parents=[parent_parser], - help='Copy file on' - ' SSH frontend directory' - ' (~/shared/.iotlabsshcli)') - copy_file_parser.add_argument('file_path', help='File path') + copy_file_parser = subparsers.add_parser( + "copy-file", + parents=[parent_parser], + help="Copy file on SSH frontend directory (~/shared/.iotlabsshcli)", + ) + copy_file_parser.add_argument("file_path", help="File path") # nodes list or exclude list common.add_nodes_selection_list(copy_file_parser) - parser.add_argument('--verbose', - action='store_true', - help='Set verbose output') + parser.add_argument("--verbose", action="store_true", help="Set verbose output") # update-m3 parser - help_msg = 'DEPRECATED: use flash subcommand instead' - update_m3_parser = \ - subparsers.add_parser('flash-m3', help=help_msg, - parents=[parent_parser], - formatter_class=DeprecateHelpFormatter) - update_m3_parser.add_argument('firmware', help='firmware path.') + help_msg = "DEPRECATED: use flash subcommand instead" + update_m3_parser = subparsers.add_parser( + "flash-m3", help=help_msg, parents=[parent_parser], formatter_class=DeprecateHelpFormatter + ) + update_m3_parser.add_argument("firmware", help="firmware path.") # nodes list or exclude list common.add_nodes_selection_list(update_m3_parser) # reset-m3 parser - help_msg = 'DEPRECATED: use reset subcommand instead' - reset_m3_parser = \ - subparsers.add_parser('reset-m3', help=help_msg, - parents=[parent_parser], - formatter_class=DeprecateHelpFormatter) + help_msg = "DEPRECATED: use reset subcommand instead" + reset_m3_parser = subparsers.add_parser( + "reset-m3", help=help_msg, parents=[parent_parser], formatter_class=DeprecateHelpFormatter + ) # nodes list or exclude list common.add_nodes_selection_list(reset_m3_parser) @@ -158,62 +148,48 @@ def open_linux_parse_and_run(opts): api = rest.Api(user, passwd) exp_id = helpers.get_current_experiment(api, opts.experiment_id) - config_ssh = { - 'user': user, - 'exp_id': exp_id - } + config_ssh = {"user": user, "exp_id": exp_id} - nodes = common.list_nodes(api, exp_id, opts.nodes_list, - opts.exclude_nodes_list) + nodes = common.list_nodes(api, exp_id, opts.nodes_list, opts.exclude_nodes_list) # Only if nodes_list or exclude_nodes_list is not specify (nodes = []) if not nodes: nodes = _get_experiment_nodes_list(api, exp_id) # Only keep Linux nodes - nodes = [f'node-{node}' - for node in nodes - if node.startswith('a8') or node.startswith('rpi3')] + nodes = [f"node-{node}" for node in nodes if node.startswith("a8") or node.startswith("rpi3")] command = opts.command res = None - if command == 'reset': - res = iotlabsshcli.open_linux.reset(config_ssh, nodes, - verbose=opts.verbose) - elif command == 'flash': - res = iotlabsshcli.open_linux.flash(config_ssh, nodes, - opts.firmware, - verbose=opts.verbose) - elif command == 'wait-for-boot': - res = iotlabsshcli.open_linux.wait_for_boot(config_ssh, nodes, - max_wait=opts.max_wait, - verbose=opts.verbose) - elif command == 'run-script': - res = iotlabsshcli.open_linux.run_script(config_ssh, nodes, - opts.script, - opts.frontend, - verbose=opts.verbose) - elif command == 'run-cmd': - res = iotlabsshcli.open_linux.run_cmd(config_ssh, nodes, - opts.cmd, - opts.frontend, - verbose=opts.verbose) - elif command == 'copy-file': - res = iotlabsshcli.open_linux.copy_file(config_ssh, nodes, - opts.file_path, - verbose=opts.verbose) - if command == 'reset-m3': - deprecate_warn_cmd('reset-m3', 'reset', 7) - res = iotlabsshcli.open_linux.reset(config_ssh, nodes, - verbose=opts.verbose) - elif command == 'flash-m3': - deprecate_warn_cmd('flash-m3', 'flash', 7) - res = iotlabsshcli.open_linux.flash(config_ssh, nodes, - opts.firmware, - verbose=opts.verbose) + if command == "reset": + res = iotlabsshcli.open_linux.reset(config_ssh, nodes, verbose=opts.verbose) + elif command == "flash": + res = iotlabsshcli.open_linux.flash(config_ssh, nodes, opts.firmware, verbose=opts.verbose) + elif command == "wait-for-boot": + res = iotlabsshcli.open_linux.wait_for_boot( + config_ssh, nodes, max_wait=opts.max_wait, verbose=opts.verbose + ) + elif command == "run-script": + res = iotlabsshcli.open_linux.run_script( + config_ssh, nodes, opts.script, opts.frontend, verbose=opts.verbose + ) + elif command == "run-cmd": + res = iotlabsshcli.open_linux.run_cmd( + config_ssh, nodes, opts.cmd, opts.frontend, verbose=opts.verbose + ) + elif command == "copy-file": + res = iotlabsshcli.open_linux.copy_file( + config_ssh, nodes, opts.file_path, verbose=opts.verbose + ) + if command == "reset-m3": + deprecate_warn_cmd("reset-m3", "reset", 7) + res = iotlabsshcli.open_linux.reset(config_ssh, nodes, verbose=opts.verbose) + elif command == "flash-m3": + deprecate_warn_cmd("flash-m3", "flash", 7) + res = iotlabsshcli.open_linux.flash(config_ssh, nodes, opts.firmware, verbose=opts.verbose) if res is None: - raise ValueError(f'Unknown command {command}') + raise ValueError(f"Unknown command {command}") return res diff --git a/iotlabsshcli/sshlib/open_linux_ssh.py b/iotlabsshcli/sshlib/open_linux_ssh.py index 2c79946..f7722f7 100644 --- a/iotlabsshcli/sshlib/open_linux_ssh.py +++ b/iotlabsshcli/sshlib/open_linux_ssh.py @@ -22,12 +22,11 @@ import os import time + import pssh -from pssh.clients import SSHClient -from pssh.clients import ParallelSSHClient -from pssh.exceptions import SFTPError, UnknownHostError -from pssh.exceptions import AuthenticationError from pssh import utils +from pssh.clients import ParallelSSHClient, SSHClient +from pssh.exceptions import AuthenticationError, SFTPError, UnknownHostError def _cleanup_result(result): @@ -123,11 +122,10 @@ def _check_all_nodes_processed(result): # library uses SSH agent for authentication if no key is provided. # As we don't have SSH agent launched on the SSH frontend server by # default, we pass the key directly -SSH_KEY = ('~/.ssh/id_rsa' if os.getenv('IOT_LAB_FRONTEND_FQDN') - else None) +SSH_KEY = "~/.ssh/id_rsa" if os.getenv("IOT_LAB_FRONTEND_FQDN") else None -class OpenLinuxSsh(): +class OpenLinuxSsh: """Implement SSH API for Parallel SSH.""" def __init__(self, config_ssh, groups, verbose=False): @@ -144,12 +142,14 @@ def run(self, command, with_proxy=True, **kwargs): for site, hosts in self.groups.items(): proxy_host = site if with_proxy else None hosts = hosts if with_proxy else [site] - result_cmd = self.run_command(command, - hosts=hosts, - user=self.config_ssh['user'], - verbose=self.verbose, - proxy_host=proxy_host, - **kwargs) + result_cmd = self.run_command( + command, + hosts=hosts, + user=self.config_ssh["user"], + verbose=self.verbose, + proxy_host=proxy_host, + **kwargs, + ) result = _extend_result(result, result_cmd) return _cleanup_result(result) @@ -159,12 +159,15 @@ def scp(self, src, dst): sites = self.groups.keys() for site in sites: try: - client = SSHClient(site, user=self.config_ssh['user'], - pkey=SSH_KEY, timeout=10) + client = SSHClient(site, user=self.config_ssh["user"], pkey=SSH_KEY, timeout=10) client.copy_file(src, dst) result["0"].append(site) - except (SFTPError, UnknownHostError, AuthenticationError, - pssh.exceptions.ConnectionError): + except ( + SFTPError, + UnknownHostError, + AuthenticationError, + pssh.exceptions.ConnectionError, + ): result["1"].append(site) return _cleanup_result(result) @@ -173,36 +176,38 @@ def wait(self, max_wait): result = {"0": [], "1": []} start_time = time.time() groups = self.groups.copy() - while (start_time + max_wait > time.time() and - not _check_all_nodes_processed(groups)): + while start_time + max_wait > time.time() and not _check_all_nodes_processed(groups): for site, hosts in groups.copy().items(): - result_cmd = self.run_command("uptime", - hosts=hosts, - user=self.config_ssh['user'], - verbose=self.verbose, - proxy_host=site) + result_cmd = self.run_command( + "uptime", + hosts=hosts, + user=self.config_ssh["user"], + verbose=self.verbose, + proxy_host=site, + ) groups[site] = result_cmd["1"] groups = _cleanup_result(groups) result = _extend_result(result, result_cmd) return _cleanup_result(result) - # pylint: disable=too-many-arguments + # pylint: disable=too-many-arguments,too-many-positional-arguments @staticmethod - def run_command(command, hosts, user, verbose=False, proxy_host=None, - timeout=10, **kwargs): + def run_command(command, hosts, user, verbose=False, proxy_host=None, timeout=10, **kwargs): """Run ssh command using Parallel SSH.""" result = {"0": [], "1": []} if proxy_host: - client = ParallelSSHClient(hosts, user='root', pkey=SSH_KEY, - proxy_host=proxy_host, - proxy_user=user, - proxy_pkey=SSH_KEY, - timeout=timeout) + client = ParallelSSHClient( + hosts, + user="root", + pkey=SSH_KEY, + proxy_host=proxy_host, + proxy_user=user, + proxy_pkey=SSH_KEY, + timeout=timeout, + ) else: - client = ParallelSSHClient(hosts, user=user, pkey=SSH_KEY, - timeout=timeout) - output = client.run_command(command, stop_on_errors=False, - **kwargs) + client = ParallelSSHClient(hosts, user=user, pkey=SSH_KEY, timeout=timeout) + output = client.run_command(command, stop_on_errors=False, **kwargs) client.join(output) # output = pssh.output.HostOutput objects list for host in output: @@ -210,12 +215,12 @@ def run_command(command, hosts, user, verbose=False, proxy_host=None, if verbose and host.stdout: for line in host.stdout: print(line) - result['0'].append(host.host) + result["0"].append(host.host) elif host.host is not None: - result['1'].append(host.host) + result["1"].append(host.host) # find hosts that have raised Exception (Authentication, Connection) # host.exception = pssh.exceptions.* & host.host = None failed_hosts = list(set(hosts) - set(sum(result.values(), []))) if failed_hosts: - result['1'].extend(failed_hosts) + result["1"].extend(failed_hosts) return result diff --git a/iotlabsshcli/tests/iotlabsshcli_mock.py b/iotlabsshcli/tests/iotlabsshcli_mock.py index af96986..77c4a17 100644 --- a/iotlabsshcli/tests/iotlabsshcli_mock.py +++ b/iotlabsshcli/tests/iotlabsshcli_mock.py @@ -19,62 +19,66 @@ # The fact that you are presently reading this means that you have had # knowledge of the CeCILL license and that you accept its terms. -""" common TestCase class for testing commands """ +"""common TestCase class for testing commands""" import sys import unittest -from mock import patch, Mock +from unittest.mock import Mock, patch -from iotlabcli.rest import Api from iotlabcli.helpers import json_dumps +from iotlabcli.rest import Api API_RET = {"result": "test"} -class RequestRet(): # pylint:disable=too-few-public-methods - """ Mock of Request return value """ +class RequestRet: # pylint:disable=too-few-public-methods + """Mock of Request return value""" def __init__(self, status_code, content, headers=None): self.status_code = status_code - self.content = content.encode('utf-8') + self.content = content.encode("utf-8") self.headers = headers - self.text = self.content.decode('utf-8') + self.text = self.content.decode("utf-8") def api_mock(ret=None): - """ Return a mock of an api object + """Return a mock of an api object returned value for api methods will be 'ret' parameter or API_RET """ ret = ret or API_RET ret_val = RequestRet(content=json_dumps(ret), status_code=200) # HTTP OK - patch('requests.request', return_value=ret_val).start() - api_class = patch('iotlabcli.rest.Api').start() - api_class.return_value = Mock(wraps=Api('user', 'password')) + patch("requests.request", return_value=ret_val).start() + api_class = patch("iotlabcli.rest.Api").start() + api_class.return_value = Mock(wraps=Api("user", "password")) return api_class.return_value def api_mock_stop(): - """ Stop all patches started by api_mock. - Actually it stops everything but not a problem """ + """Stop all patches started by api_mock. + Actually it stops everything but not a problem""" patch.stopall() class MainMock(unittest.TestCase): - """ Common mock needed for testing main function of parsers """ + """Common mock needed for testing main function of parsers""" + def setUp(self): self.api = api_mock() - patch('sys.stderr', sys.stdout).start() - patch('iotlabcli.parser.common.sites_list', Mock( - return_value=['grenoble', 'strasbourg', - 'euratech', 'saclay'])).start() + patch("sys.stderr", sys.stdout).start() + patch( + "iotlabcli.parser.common.sites_list", + Mock(return_value=["grenoble", "strasbourg", "euratech", "saclay"]), + ).start() + + patch( + "iotlabcli.auth.get_user_credentials", Mock(return_value=("username", "password")) + ).start() - patch('iotlabcli.auth.get_user_credentials', - Mock(return_value=('username', 'password'))).start() + def get_exp(_, x, running_only=True): + return x if x is not None else (123 if running_only else 234) - get_exp = (lambda a, x, running_only=True: - x if x is not None else (123 if running_only else 234)) - patch('iotlabcli.helpers.get_current_experiment', get_exp).start() + patch("iotlabcli.helpers.get_current_experiment", get_exp).start() def tearDown(self): api_mock_stop() diff --git a/iotlabsshcli/tests/open_linux_parser_test.py b/iotlabsshcli/tests/open_linux_parser_test.py index 8b61d1f..736d25b 100644 --- a/iotlabsshcli/tests/open_linux_parser_test.py +++ b/iotlabsshcli/tests/open_linux_parser_test.py @@ -21,10 +21,14 @@ """Tests for iotlabsshcli.parser.open_linux package.""" +from unittest.mock import Mock, patch + import jmespath -from mock import patch, Mock +from iotlabsshcli._deprecated import _main as _deprecated_main +from iotlabsshcli._deprecated import open_a8_cli from iotlabsshcli.parser import open_linux_parser + from .iotlabsshcli_mock import MainMock # pylint: disable=too-many-public-methods @@ -34,256 +38,259 @@ class TestMainNodeParser(MainMock): """Test open-linux-cli main parser.""" - _nodes = [f'a8-{i}.saclay.iot-lab.info' for i in range(1, 6)] - _root_nodes = [f'node-{node}' for node in _nodes] + _nodes = [f"a8-{i}.saclay.iot-lab.info" for i in range(1, 6)] + _root_nodes = [f"node-{node}" for node in _nodes] - @patch('iotlabsshcli.open_linux.flash') - @patch('iotlabcli.parser.common.list_nodes') + @patch("iotlabsshcli.open_linux.flash") + @patch("iotlabcli.parser.common.list_nodes") def test_main_flash(self, list_nodes, flash): """Run the parser.node.main with update subparser function.""" - flash.return_value = {'result': 'test'} + flash.return_value = {"result": "test"} list_nodes.return_value = self._nodes - args = ['flash', 'firmware.elf', '-l', 'saclay,a8,1-5'] + args = ["flash", "firmware.elf", "-l", "saclay,a8,1-5"] open_linux_parser.main(args) list_nodes.assert_called_with(self.api, 123, [self._nodes], None) - flash.assert_called_with({'user': 'username', 'exp_id': 123}, - self._root_nodes, 'firmware.elf', - verbose=False) - - exp_info_res = {"items": [{"network_address": node} - for node in self._nodes]} - with patch.object(self.api, 'get_experiment_info', - Mock(return_value=exp_info_res)): + flash.assert_called_with( + {"user": "username", "exp_id": 123}, self._root_nodes, "firmware.elf", verbose=False + ) + + exp_info_res = {"items": [{"network_address": node} for node in self._nodes]} + with patch.object(self.api, "get_experiment_info", Mock(return_value=exp_info_res)): list_nodes.return_value = [] - args = ['flash', 'firmware.elf'] + args = ["flash", "firmware.elf"] open_linux_parser.main(args) list_nodes.assert_called_with(self.api, 123, None, None) - flash.assert_called_with({'user': 'username', 'exp_id': 123}, - self._root_nodes, 'firmware.elf', - verbose=False) - - @patch('iotlabsshcli.open_linux.reset') - @patch('iotlabcli.parser.common.list_nodes') + flash.assert_called_with( + {"user": "username", "exp_id": 123}, + self._root_nodes, + "firmware.elf", + verbose=False, + ) + + @patch("iotlabsshcli.open_linux.reset") + @patch("iotlabcli.parser.common.list_nodes") def test_main_reset(self, list_nodes, reset): """Run the parser.node.main with reset subparser function.""" - reset.return_value = {'result': 'test'} + reset.return_value = {"result": "test"} list_nodes.return_value = self._nodes - args = ['reset', '-l', 'saclay,a8,1-5'] + args = ["reset", "-l", "saclay,a8,1-5"] open_linux_parser.main(args) list_nodes.assert_called_with(self.api, 123, [self._nodes], None) - reset.assert_called_with({'user': 'username', 'exp_id': 123}, - self._root_nodes, - verbose=False) - - exp_info_res = {"items": [{"network_address": node} - for node in self._nodes]} - with patch.object(self.api, 'get_experiment_info', - Mock(return_value=exp_info_res)): + reset.assert_called_with( + {"user": "username", "exp_id": 123}, self._root_nodes, verbose=False + ) + + exp_info_res = {"items": [{"network_address": node} for node in self._nodes]} + with patch.object(self.api, "get_experiment_info", Mock(return_value=exp_info_res)): list_nodes.return_value = [] - args = ['reset'] + args = ["reset"] open_linux_parser.main(args) list_nodes.assert_called_with(self.api, 123, None, None) - reset.assert_called_with({'user': 'username', 'exp_id': 123}, - self._root_nodes, verbose=False) + reset.assert_called_with( + {"user": "username", "exp_id": 123}, self._root_nodes, verbose=False + ) - @patch('iotlabsshcli.open_linux.wait_for_boot') - @patch('iotlabcli.parser.common.list_nodes') + @patch("iotlabsshcli.open_linux.wait_for_boot") + @patch("iotlabcli.parser.common.list_nodes") def test_main_wait_for_boot(self, list_nodes, wait_for_boot): """Run the parser.node.main with wait-for-boot subparser function.""" - wait_for_boot.return_value = {'result': 'test'} + wait_for_boot.return_value = {"result": "test"} list_nodes.return_value = self._nodes - args = ['wait-for-boot', '-l', 'saclay,a8,1-5'] + args = ["wait-for-boot", "-l", "saclay,a8,1-5"] open_linux_parser.main(args) list_nodes.assert_called_with(self.api, 123, [self._nodes], None) - wait_for_boot.assert_called_with({'user': 'username', 'exp_id': 123}, - self._root_nodes, - max_wait=120, - verbose=False) + wait_for_boot.assert_called_with( + {"user": "username", "exp_id": 123}, self._root_nodes, max_wait=120, verbose=False + ) - args = ['wait-for-boot', "--max-wait", '10', '-l', 'saclay,a8,1-5'] + args = ["wait-for-boot", "--max-wait", "10", "-l", "saclay,a8,1-5"] open_linux_parser.main(args) list_nodes.assert_called_with(self.api, 123, [self._nodes], None) - wait_for_boot.assert_called_with({'user': 'username', 'exp_id': 123}, - self._root_nodes, - max_wait=10, - verbose=False) - - exp_info_res = {"items": [{"network_address": node} - for node in self._nodes]} - with patch.object(self.api, 'get_experiment_info', - Mock(return_value=exp_info_res)): + wait_for_boot.assert_called_with( + {"user": "username", "exp_id": 123}, self._root_nodes, max_wait=10, verbose=False + ) + + exp_info_res = {"items": [{"network_address": node} for node in self._nodes]} + with patch.object(self.api, "get_experiment_info", Mock(return_value=exp_info_res)): list_nodes.return_value = [] - args = ['wait-for-boot'] + args = ["wait-for-boot"] open_linux_parser.main(args) list_nodes.assert_called_with(self.api, 123, None, None) - wait_for_boot.assert_called_with({'user': 'username', - 'exp_id': 123}, - self._root_nodes, - max_wait=120, - verbose=False) - - @patch('iotlabsshcli.open_linux.run_script') - @patch('iotlabcli.parser.common.list_nodes') + wait_for_boot.assert_called_with( + {"user": "username", "exp_id": 123}, self._root_nodes, max_wait=120, verbose=False + ) + + @patch("iotlabsshcli.open_linux.run_script") + @patch("iotlabcli.parser.common.list_nodes") def test_main_run_script(self, list_nodes, run_script): """Run the parser.node.main with run-script subparser function.""" - run_script.return_value = {'result': 'test'} + run_script.return_value = {"result": "test"} list_nodes.return_value = self._nodes - args = ['run-script', 'script.sh', '-l', 'saclay,a8,1-5'] + args = ["run-script", "script.sh", "-l", "saclay,a8,1-5"] open_linux_parser.main(args) list_nodes.assert_called_with(self.api, 123, [self._nodes], None) - run_script.assert_called_with({'user': 'username', 'exp_id': 123}, - self._root_nodes, - 'script.sh', False, verbose=False) - - args = ['run-script', 'script.sh', '--frontend', '-l', - 'saclay,a8,1-5'] + run_script.assert_called_with( + {"user": "username", "exp_id": 123}, + self._root_nodes, + "script.sh", + False, + verbose=False, + ) + + args = ["run-script", "script.sh", "--frontend", "-l", "saclay,a8,1-5"] open_linux_parser.main(args) list_nodes.assert_called_with(self.api, 123, [self._nodes], None) - run_script.assert_called_with({'user': 'username', 'exp_id': 123}, - self._root_nodes, - 'script.sh', True, verbose=False) - - exp_info_res = {"items": [{"network_address": node} - for node in self._nodes]} - with patch.object(self.api, 'get_experiment_info', - Mock(return_value=exp_info_res)): + run_script.assert_called_with( + {"user": "username", "exp_id": 123}, self._root_nodes, "script.sh", True, verbose=False + ) + + exp_info_res = {"items": [{"network_address": node} for node in self._nodes]} + with patch.object(self.api, "get_experiment_info", Mock(return_value=exp_info_res)): list_nodes.return_value = [] - args = ['run-script', 'script.sh'] + args = ["run-script", "script.sh"] open_linux_parser.main(args) list_nodes.assert_called_with(self.api, 123, None, None) - run_script.assert_called_with({'user': 'username', 'exp_id': 123}, - self._root_nodes, - 'script.sh', False, verbose=False) - - @patch('iotlabsshcli.open_linux.run_cmd') - @patch('iotlabcli.parser.common.list_nodes') + run_script.assert_called_with( + {"user": "username", "exp_id": 123}, + self._root_nodes, + "script.sh", + False, + verbose=False, + ) + + @patch("iotlabsshcli.open_linux.run_cmd") + @patch("iotlabcli.parser.common.list_nodes") def test_main_run_cmd(self, list_nodes, run_cmd): """Run the parser.node.main with run-cmd subparser function.""" - run_cmd.return_value = {'result': 'test'} + run_cmd.return_value = {"result": "test"} list_nodes.return_value = self._nodes - args = ['run-cmd', 'uname -a', '-l', 'saclay,a8,1-5'] + args = ["run-cmd", "uname -a", "-l", "saclay,a8,1-5"] open_linux_parser.main(args) list_nodes.assert_called_with(self.api, 123, [self._nodes], None) - run_cmd.assert_called_with({'user': 'username', 'exp_id': 123}, - self._root_nodes, - 'uname -a', False, verbose=False) + run_cmd.assert_called_with( + {"user": "username", "exp_id": 123}, self._root_nodes, "uname -a", False, verbose=False + ) - args = ['run-cmd', 'uname -a', '--frontend', '-l', 'saclay,a8,1-5'] + args = ["run-cmd", "uname -a", "--frontend", "-l", "saclay,a8,1-5"] open_linux_parser.main(args) list_nodes.assert_called_with(self.api, 123, [self._nodes], None) - run_cmd.assert_called_with({'user': 'username', 'exp_id': 123}, - self._root_nodes, - 'uname -a', True, verbose=False) - - exp_info_res = {"items": [{"network_address": node} - for node in self._nodes]} - with patch.object(self.api, 'get_experiment_info', - Mock(return_value=exp_info_res)): + run_cmd.assert_called_with( + {"user": "username", "exp_id": 123}, self._root_nodes, "uname -a", True, verbose=False + ) + + exp_info_res = {"items": [{"network_address": node} for node in self._nodes]} + with patch.object(self.api, "get_experiment_info", Mock(return_value=exp_info_res)): list_nodes.return_value = [] - args = ['run-cmd', 'uname -a'] + args = ["run-cmd", "uname -a"] open_linux_parser.main(args) list_nodes.assert_called_with(self.api, 123, None, None) - run_cmd.assert_called_with({'user': 'username', 'exp_id': 123}, - self._root_nodes, - 'uname -a', False, verbose=False) - - @patch('iotlabsshcli.open_linux.copy_file') - @patch('iotlabcli.parser.common.list_nodes') + run_cmd.assert_called_with( + {"user": "username", "exp_id": 123}, + self._root_nodes, + "uname -a", + False, + verbose=False, + ) + + @patch("iotlabsshcli.open_linux.copy_file") + @patch("iotlabcli.parser.common.list_nodes") def test_main_copy_file(self, list_nodes, copy_file): """Run the parser.node.main with copy-file subparser function.""" - copy_file.return_value = {'result': 'test'} + copy_file.return_value = {"result": "test"} list_nodes.return_value = self._nodes - args = ['copy-file', 'script.sh', '-l', 'saclay,a8,1-5'] + args = ["copy-file", "script.sh", "-l", "saclay,a8,1-5"] open_linux_parser.main(args) list_nodes.assert_called_with(self.api, 123, [self._nodes], None) - copy_file.assert_called_with({'user': 'username', 'exp_id': 123}, - self._root_nodes, - 'script.sh', verbose=False) - - exp_info_res = {"items": [{"network_address": node} - for node in self._nodes]} - with patch.object(self.api, 'get_experiment_info', - Mock(return_value=exp_info_res)): + copy_file.assert_called_with( + {"user": "username", "exp_id": 123}, self._root_nodes, "script.sh", verbose=False + ) + + exp_info_res = {"items": [{"network_address": node} for node in self._nodes]} + with patch.object(self.api, "get_experiment_info", Mock(return_value=exp_info_res)): list_nodes.return_value = [] - args = ['copy-file', 'script.sh'] + args = ["copy-file", "script.sh"] open_linux_parser.main(args) list_nodes.assert_called_with(self.api, 123, None, None) - copy_file.assert_called_with({'user': 'username', 'exp_id': 123}, - self._root_nodes, - 'script.sh', verbose=False) + copy_file.assert_called_with( + {"user": "username", "exp_id": 123}, self._root_nodes, "script.sh", verbose=False + ) def test_main_unknown_function(self): """Run the parser.node.main with an unknown function.""" - args = ['unknown-cmd'] + args = ["unknown-cmd"] self.assertRaises(SystemExit, open_linux_parser.main, args) - @patch('iotlabcli.parser.common.list_nodes') + @patch("iotlabcli.parser.common.list_nodes") def test_run_unknown_function(self, list_nodes): # pylint:disable=unused-argument """Run the parser.node.main with an unknown function.""" parser = open_linux_parser.parse_options() - parser.command = 'unknown-cmd' - parser.username = 'username' - parser.password = 'password' - parser.experiment_id = 'experiment_id' + parser.command = "unknown-cmd" + parser.username = "username" + parser.password = "password" + parser.experiment_id = "experiment_id" parser.nodes_list = [] parser.exclude_nodes_list = [] - self.assertRaises(ValueError, - open_linux_parser.open_linux_parse_and_run, parser) + self.assertRaises(ValueError, open_linux_parser.open_linux_parse_and_run, parser) - @patch('iotlabsshcli.open_linux.reset') - @patch('iotlabcli.parser.common.list_nodes') - @patch('iotlabcli.parser.common.print_result') + @patch("iotlabsshcli.open_linux.reset") + @patch("iotlabcli.parser.common.list_nodes") + @patch("iotlabcli.parser.common.print_result") def test_reset_jmespath(self, print_result, list_nodes, reset): """Run reset subparser function with jmespath options.""" - reset.return_value = {'result': 'test'} + reset.return_value = {"result": "test"} list_nodes.return_value = self._nodes - args = ['--jmespath=\'test\'', '--fmt=\'int\'', 'reset', - '-l', 'saclay,a8,1-5'] + args = ["--jmespath='test'", "--fmt='int'", "reset", "-l", "saclay,a8,1-5"] open_linux_parser.main(args) print_result.assert_called_once() args, _ = print_result.call_args self.assertEqual(len(args), 3) - self.assertEqual(args[0], {'result': 'test'}) + self.assertEqual(args[0], {"result": "test"}) self.assertTrue(isinstance(args[1], jmespath.parser.ParsedResult)) - self.assertEqual(args[2], 'int') + self.assertEqual(args[2], "int") - @patch('iotlabsshcli.open_linux.reset') - @patch('iotlabsshcli.open_linux.flash') - @patch('iotlabcli.parser.common.list_nodes') - @patch('iotlabcli.parser.common.print_result') + @patch("iotlabsshcli.open_linux.reset") + @patch("iotlabsshcli.open_linux.flash") + @patch("iotlabcli.parser.common.list_nodes") + @patch("iotlabcli.parser.common.print_result") def test_deprecated_parser(self, print_result, list_nodes, flash, reset): """Run deprecated subparsers.""" - reset.return_value = flash.return_value = {'result': 'test'} + reset.return_value = flash.return_value = {"result": "test"} list_nodes.return_value = self._nodes # call deprecated subparser - args = ['reset-m3', '-h'] + args = ["reset-m3", "-h"] self.assertRaises(SystemExit, open_linux_parser.main, args) - args = ['flash-m3', 'firmware.elf', '-l', 'saclay,a8,1-5'] + args = ["flash-m3", "firmware.elf", "-l", "saclay,a8,1-5"] open_linux_parser.main(args) print_result.assert_called() args, _ = print_result.call_args self.assertEqual(len(args), 3) - self.assertEqual(args[0], {'result': 'test'}) + self.assertEqual(args[0], {"result": "test"}) self.assertEqual(args[1], None) self.assertEqual(args[2], None) - args = ['reset-m3', '-l', 'saclay,a8,1-5'] + args = ["reset-m3", "-l", "saclay,a8,1-5"] open_linux_parser.main(args) print_result.assert_called() args, _ = print_result.call_args self.assertEqual(len(args), 3) - self.assertEqual(args[0], {'result': 'test'}) + self.assertEqual(args[0], {"result": "test"}) self.assertEqual(args[1], None) self.assertEqual(args[2], None) + + @patch("iotlabsshcli._deprecated.deprecate_cmd") + def test_open_a8_cli_entry_point(self, mock_deprecate): + """open_a8_cli entry point delegates to deprecate_cmd.""" + open_a8_cli() + mock_deprecate.assert_called_once_with(_deprecated_main, "open-a8-cli", "iotlab-ssh") diff --git a/iotlabsshcli/tests/open_linux_ssh_test.py b/iotlabsshcli/tests/open_linux_ssh_test.py index 6d28b9d..e6eadee 100644 --- a/iotlabsshcli/tests/open_linux_ssh_test.py +++ b/iotlabsshcli/tests/open_linux_ssh_test.py @@ -21,13 +21,15 @@ """Tests for iotlabsshcli.open_linux package.""" -from pytest import mark +from unittest.mock import patch + from pssh.exceptions import SFTPError -from mock import patch +from pytest import mark from iotlabsshcli.open_linux import _nodes_grouped from iotlabsshcli.sshlib import OpenLinuxSsh -from .open_linux_test import _SACLAY_NODES, _GRENOBLE_NODES, _ROOT_NODES + +from .open_linux_test import _GRENOBLE_NODES, _ROOT_NODES, _SACLAY_NODES # pylint: disable=too-few-public-methods @@ -37,96 +39,94 @@ class HostOutput: ParallelSSH run_command returns a list of pssh.output.HostOutput objects since version 2.0.0. """ + def __init__(self, host=None, stdout=None, exit_code=None): self.host = host self.stdout = stdout self.exit_code = exit_code -@mark.parametrize('run_on_frontend', [False, True]) -@patch('pssh.clients.ParallelSSHClient.run_command') -@patch('pssh.clients.ParallelSSHClient.join') +@mark.parametrize("run_on_frontend", [False, True]) +@patch("pssh.clients.ParallelSSHClient.run_command") +@patch("pssh.clients.ParallelSSHClient.join") def test_run(join, run_command, run_on_frontend): # pylint: disable=unused-argument """Test running commands on ssh nodes.""" config_ssh = { - 'user': 'username', - 'exp_id': 123, + "user": "username", + "exp_id": 123, } - test_command = 'test' + test_command = "test" groups = _nodes_grouped(_ROOT_NODES) node_ssh = OpenLinuxSsh(config_ssh, groups, verbose=True) # Print output of run_command if run_on_frontend: - output = [HostOutput('saclay.iot-lab.info', 'test', 0), - HostOutput()] + output = [HostOutput("saclay.iot-lab.info", "test", 0), HostOutput()] else: - output = [HostOutput(host, 'test', 1) for host in _GRENOBLE_NODES] - output.extend([HostOutput(host, 'test', 0) for host in _SACLAY_NODES]) + output = [HostOutput(host, "test", 1) for host in _GRENOBLE_NODES] + output.extend([HostOutput(host, "test", 0) for host in _SACLAY_NODES]) run_command.return_value = output ret = node_ssh.run(test_command, with_proxy=not run_on_frontend) if run_on_frontend: - assert ret == {'0': ['saclay.iot-lab.info'], - '1': ['grenoble.iot-lab.info']} + assert ret == {"0": ["saclay.iot-lab.info"], "1": ["grenoble.iot-lab.info"]} else: - assert ret == {'0': _SACLAY_NODES, - '1': _GRENOBLE_NODES} + assert ret == {"0": _SACLAY_NODES, "1": _GRENOBLE_NODES} assert run_command.call_count == len(groups) run_command.assert_called_with(test_command, stop_on_errors=False) -@patch('pssh.clients.SSHClient._init') -@patch('pssh.clients.SSHClient.copy_file') +@patch("pssh.clients.SSHClient._init") +@patch("pssh.clients.SSHClient.copy_file") def test_scp(copy_file, init): # pylint: disable=unused-argument """Test wait for ssh nodes to be available.""" config_ssh = { - 'user': 'username', - 'exp_id': 123, + "user": "username", + "exp_id": 123, } - src = 'test_src' - dst = 'test_dst' + src = "test_src" + dst = "test_dst" groups = _nodes_grouped(_ROOT_NODES) node_ssh = OpenLinuxSsh(config_ssh, groups, verbose=True) ret = node_ssh.scp(src, dst) assert copy_file.call_count == 2 - assert ret == {'0': ['saclay.iot-lab.info', 'grenoble.iot-lab.info']} + assert ret == {"0": ["saclay.iot-lab.info", "grenoble.iot-lab.info"]} copy_file.side_effect = SFTPError() ret = node_ssh.scp(src, dst) - assert ret == {'1': ['saclay.iot-lab.info', 'grenoble.iot-lab.info']} + assert ret == {"1": ["saclay.iot-lab.info", "grenoble.iot-lab.info"]} -@patch('pssh.clients.ParallelSSHClient.run_command') -@patch('pssh.clients.ParallelSSHClient.join') +@patch("pssh.clients.ParallelSSHClient.run_command") +@patch("pssh.clients.ParallelSSHClient.join") def test_wait_all_boot(join, run_command): # pylint: disable=unused-argument """Test wait for ssh nodes to be available.""" config_ssh = { - 'user': 'username', - 'exp_id': 123, + "user": "username", + "exp_id": 123, } - test_command = 'test' + test_command = "test" groups = _nodes_grouped(_ROOT_NODES) # normal boot node_ssh = OpenLinuxSsh(config_ssh, groups, verbose=True) - output = [HostOutput(host, 'test', 0) for host in _ROOT_NODES] + output = [HostOutput(host, "test", 0) for host in _ROOT_NODES] run_command.return_value = output node_ssh.wait(120) assert run_command.call_count == 2 - run_command.assert_called_with('uptime', stop_on_errors=False) + run_command.assert_called_with("uptime", stop_on_errors=False) run_command.reset_mock() node_ssh.run(test_command) diff --git a/iotlabsshcli/tests/open_linux_test.py b/iotlabsshcli/tests/open_linux_test.py index 0e2a37a..3ea62cb 100644 --- a/iotlabsshcli/tests/open_linux_test.py +++ b/iotlabsshcli/tests/open_linux_test.py @@ -22,174 +22,167 @@ """Tests for iotlabsshcli.open_linux package.""" import os.path +from unittest.mock import patch + from pytest import mark -from mock import patch - -from iotlabsshcli.open_linux import reset, flash, wait_for_boot, run_script -from iotlabsshcli.open_linux import run_cmd, copy_file -from iotlabsshcli.open_linux import (_RESET_CMD, _FLASH_CMD, - _RUN_SCRIPT_CMD, - _QUIT_SCRIPT_CMD, _MAKE_EXECUTABLE_CMD) - -_SACLAY_NODES = [f'node-a8-{n}.saclay.iot-lab.info' - for n in range(1, 6)] -_GRENOBLE_NODES = [f'node-a8-{n}.grenoble.iot-lab.info' - for n in range(1, 4)] + +from iotlabsshcli.open_linux import ( + _FLASH_CMD, + _MAKE_EXECUTABLE_CMD, + _QUIT_SCRIPT_CMD, + _RESET_CMD, + _RUN_SCRIPT_CMD, + copy_file, + flash, + reset, + run_cmd, + run_script, + wait_for_boot, +) + +_SACLAY_NODES = [f"node-a8-{n}.saclay.iot-lab.info" for n in range(1, 6)] +_GRENOBLE_NODES = [f"node-a8-{n}.grenoble.iot-lab.info" for n in range(1, 4)] _ROOT_NODES = _SACLAY_NODES + _GRENOBLE_NODES -@patch('iotlabsshcli.sshlib.OpenLinuxSsh.run') -@patch('iotlabsshcli.sshlib.OpenLinuxSsh.scp') +@patch("iotlabsshcli.sshlib.OpenLinuxSsh.run") +@patch("iotlabsshcli.sshlib.OpenLinuxSsh.scp") def test_open_linux_flash(scp, run): """Test flashing a firmware.""" config_ssh = { - 'user': 'username', - 'exp_id': 123, + "user": "username", + "exp_id": 123, } - firmware = '/tmp/firmware.elf' - remote_fw = os.path.join('shared/.iotlabsshcli', - os.path.basename(firmware)) - scp.return_value = {'0': ['saclay.iot-lab.info'], - '1': ['grenoble.iot-lab.info']} + firmware = "/tmp/firmware.elf" + remote_fw = os.path.join("shared/.iotlabsshcli", os.path.basename(firmware)) + scp.return_value = {"0": ["saclay.iot-lab.info"], "1": ["grenoble.iot-lab.info"]} - return_value = {'0': _SACLAY_NODES} + return_value = {"0": _SACLAY_NODES} run.return_value = return_value - return_value.get('1', []).extend(_GRENOBLE_NODES) + return_value.get("1", []).extend(_GRENOBLE_NODES) ret = flash(config_ssh, _ROOT_NODES, firmware) - assert ret == {'flash': return_value} + assert ret == {"flash": return_value} scp.assert_called_once_with(firmware, remote_fw) assert run.call_count == 1 - run.mock_calls[0].assert_called_with( - _FLASH_CMD.format(fw=remote_fw, bin="") - ) + run.mock_calls[0].assert_called_with(_FLASH_CMD.format(fw=remote_fw, bin="")) # Cover the case where a .bin firmware is used - scp.call_count = 0 - run.call_count = 0 - firmware = '/tmp/firmware.bin' - remote_fw = os.path.join('shared/.iotlabsshcli', - os.path.basename(firmware)) + scp.reset_mock() + run.reset_mock() + firmware = "/tmp/firmware.bin" + remote_fw = os.path.join("shared/.iotlabsshcli", os.path.basename(firmware)) ret = flash(config_ssh, _ROOT_NODES, firmware) - assert ret == {'flash': return_value} + assert ret == {"flash": return_value} scp.assert_called_once_with(firmware, remote_fw) assert run.call_count == 1 - run.mock_calls[0].assert_called_with( - _FLASH_CMD.format(fw=remote_fw, bin="--bin") - ) + run.mock_calls[0].assert_called_with(_FLASH_CMD.format(fw=remote_fw, bin="--bin")) -@patch('iotlabsshcli.sshlib.OpenLinuxSsh.run') +@patch("iotlabsshcli.sshlib.OpenLinuxSsh.run") def test_open_linux_reset(run): """Test resetting co-microcontroller.""" config_ssh = { - 'user': 'username', - 'exp_id': 123, + "user": "username", + "exp_id": 123, } - return_value = {'0': 'test'} + return_value = {"0": "test"} run.return_value = return_value ret = reset(config_ssh, _ROOT_NODES) - assert ret == {'reset': return_value} + assert ret == {"reset": return_value} run.assert_called_once_with(_RESET_CMD) -@patch('iotlabsshcli.sshlib.OpenLinuxSsh.wait') +@patch("iotlabsshcli.sshlib.OpenLinuxSsh.wait") def test_open_linux_wait_for_boot(wait): """Test wait for Linux boot.""" config_ssh = { - 'user': 'username', - 'exp_id': 123, + "user": "username", + "exp_id": 123, } - return_value = {'0': 'test'} + return_value = {"0": "test"} wait.return_value = return_value ret = wait_for_boot(config_ssh, _ROOT_NODES) - assert ret == {'wait-for-boot': return_value} + assert ret == {"wait-for-boot": return_value} wait.assert_called_once_with(120) -@mark.parametrize('run_on_frontend', [False, True]) -@patch('iotlabsshcli.sshlib.OpenLinuxSsh.run') -@patch('iotlabsshcli.sshlib.OpenLinuxSsh.scp') +@mark.parametrize("run_on_frontend", [False, True]) +@patch("iotlabsshcli.sshlib.OpenLinuxSsh.run") +@patch("iotlabsshcli.sshlib.OpenLinuxSsh.scp") def test_open_linux_run_script(scp, run, run_on_frontend): """Test run script on Linux nodes.""" config_ssh = { - 'user': 'username', - 'exp_id': 123, + "user": "username", + "exp_id": 123, } - screen = '{user}-{exp_id}'.format(**config_ssh) - script = '/tmp/script.sh' - remote_script = os.path.join('shared/.iotlabsshcli', - os.path.basename(script)) - script_data = {'screen': screen, - 'path': remote_script} - scp.return_value = {'0': ['saclay.iot-lab.info'], - '1': ['grenoble.iot-lab.info']} + screen = "{user}-{exp_id}".format(**config_ssh) + script = "/tmp/script.sh" + remote_script = os.path.join("shared/.iotlabsshcli", os.path.basename(script)) + script_data = {"screen": screen, "path": remote_script} + scp.return_value = {"0": ["saclay.iot-lab.info"], "1": ["grenoble.iot-lab.info"]} if not run_on_frontend: - return_value = {'0': _SACLAY_NODES} + return_value = {"0": _SACLAY_NODES} else: - return_value = {'0': ['saclay.iot-lab.info']} + return_value = {"0": ["saclay.iot-lab.info"]} run.side_effect = [return_value for _n in range(3)] if not run_on_frontend: - return_value.get('1', []).extend(['grenoble.iot-lab.info']) + return_value.get("1", []).extend(["grenoble.iot-lab.info"]) else: - return_value.get('1', []).extend(_GRENOBLE_NODES) - ret = run_script(config_ssh, _ROOT_NODES, script, - run_on_frontend=run_on_frontend) + return_value.get("1", []).extend(_GRENOBLE_NODES) + ret = run_script(config_ssh, _ROOT_NODES, script, run_on_frontend=run_on_frontend) - assert ret == {'run-script': return_value} + assert ret == {"run-script": return_value} scp.assert_called_once_with(script, remote_script) assert run.call_count == 3 run.mock_calls[0].assert_called_with( - _MAKE_EXECUTABLE_CMD.format(os.path.dirname(remote_script)), - with_proxy=False) + _MAKE_EXECUTABLE_CMD.format(os.path.dirname(remote_script)), with_proxy=False + ) run.mock_calls[1].assert_called_with( - _QUIT_SCRIPT_CMD.format(**script_data), - with_proxy=not run_on_frontend) + _QUIT_SCRIPT_CMD.format(**script_data), with_proxy=not run_on_frontend + ) run.mock_calls[2].assert_called_with( - _RUN_SCRIPT_CMD.format(**script_data), - use_pty=False, - with_proxy=not run_on_frontend) + _RUN_SCRIPT_CMD.format(**script_data), use_pty=False, with_proxy=not run_on_frontend + ) -@patch('iotlabsshcli.sshlib.OpenLinuxSsh.scp') +@patch("iotlabsshcli.sshlib.OpenLinuxSsh.scp") def test_open_linux_copy_file(scp): """Test copy file on the SSH frontend.""" config_ssh = { - 'user': 'username', - 'exp_id': 123, + "user": "username", + "exp_id": 123, } - file_path = '/tmp/script.sh' - remote_file = os.path.join('shared/.iotlabsshcli', - os.path.basename(file_path)) - return_value = {'0': 'test'} + file_path = "/tmp/script.sh" + remote_file = os.path.join("shared/.iotlabsshcli", os.path.basename(file_path)) + return_value = {"0": "test"} scp.return_value = return_value ret = copy_file(config_ssh, _ROOT_NODES, file_path) - assert ret == {'copy-file': return_value} + assert ret == {"copy-file": return_value} scp.assert_called_once_with(file_path, remote_file) -@mark.parametrize('run_on_frontend', [False, True]) -@patch('iotlabsshcli.sshlib.OpenLinuxSsh.run') +@mark.parametrize("run_on_frontend", [False, True]) +@patch("iotlabsshcli.sshlib.OpenLinuxSsh.run") def test_open_linux_run_cmd(run, run_on_frontend): """Test run command on Linux nodes.""" config_ssh = { - 'user': 'username', - 'exp_id': 123, + "user": "username", + "exp_id": 123, } - cmd = 'uname -a' - return_value = {'0': 'test'} + cmd = "uname -a" + return_value = {"0": "test"} run.return_value = return_value - ret = run_cmd(config_ssh, _ROOT_NODES, cmd, - run_on_frontend=run_on_frontend) + ret = run_cmd(config_ssh, _ROOT_NODES, cmd, run_on_frontend=run_on_frontend) run.assert_called_once_with(cmd, with_proxy=not run_on_frontend) - assert ret == {'run-cmd': return_value} + assert ret == {"run-cmd": return_value} diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..74f5d12 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,72 @@ +[build-system] +requires = ["hatchling"] +build-backend = "hatchling.build" + +[project] +name = "iotlabsshcli" +dynamic = ["version"] +description = "IoT-LAB testbed ssh command-line client" +readme = "README.rst" +license = {text = "CeCILL v2.1"} +authors = [ + {name = "IoT-LAB Team", email = "admin@iot-lab.info"}, +] +classifiers = [ + "Development Status :: 5 - Production/Stable", + "Programming Language :: Python", + "Programming Language :: Python :: 3", + "Intended Audience :: End Users/Desktop", + "Environment :: Console", + "Topic :: Utilities", +] +requires-python = ">=3.10" +dependencies = [ + "iotlabcli>=2.0", + "parallel-ssh>=2.5.4", + "scp", + "gevent>=1.1", + "psutil==5.7.0", +] + +[project.urls] +Homepage = "http://www.iot-lab.info" +"Source Code" = "http://github.com/iot-lab/ssh-cli-tools/" + +[project.scripts] +iotlab-ssh = "iotlabsshcli.parser.open_linux_parser:main" +open-a8-cli = "iotlabsshcli._deprecated:open_a8_cli" + +[tool.hatch.version] +path = "iotlabsshcli/__init__.py" + +[tool.pytest.ini_options] +addopts = [ + "-v", + "--doctest-modules", + "--cov", "iotlabsshcli", + "--cov-report=term-missing", + "--cov-report=term", + "--cov-report=xml", +] +testpaths = ["iotlabsshcli"] + +[tool.pylint.messages_control] +disable = ["unspecified-encoding"] + +[tool.pylint.format] +msg-template = "{path}:{line}: [{msg_id}({symbol}), {obj}] {msg}" + +[tool.pylint.reports] +reports = false + +[tool.ruff] +line-length = 99 +target-version = "py310" +exclude = [".tox", "dist", "build", "*.egg"] + +[tool.ruff.lint] +select = ["E", "F", "W"] + +[tool.isort] +profile = "black" +known_first_party = ["iotlabsshcli"] diff --git a/setup.cfg b/setup.cfg deleted file mode 100644 index b676d98..0000000 --- a/setup.cfg +++ /dev/null @@ -1,23 +0,0 @@ -[tool:pytest] -addopts = - -v - --doctest-modules - --cov iotlabsshcli - --cov-report=term-missing --cov-report=term --cov-report=xml -testpaths = iotlabsshcli - -[pylint] -disable=unspecified-encoding -msg-template={path}:{line}: [{msg_id}({symbol}), {obj}] {msg} -reports=no - -[pycodestyle] -exclude = *.egg,.tox - -[flake8] -ignore = C901 -exclude = .tox,dist,doc,build,*.egg -max-complexity = 6 - -[metadata] -description-file = README.rst diff --git a/setup.py b/setup.py deleted file mode 100644 index 4bf409e..0000000 --- a/setup.py +++ /dev/null @@ -1,83 +0,0 @@ -#! /usr/bin/env python -# -*- coding:utf-8 -*- - -# This file is a part of IoT-LAB ssh-cli-tools -# Copyright (C) 2015 INRIA (Contact: admin@iot-lab.info) -# Contributor(s) : see AUTHORS file -# -# This software is governed by the CeCILL license under French law -# and abiding by the rules of distribution of free software. You can use, -# modify and/ or redistribute the software under the terms of the CeCILL -# license as circulated by CEA, CNRS and INRIA at the following URL -# http://www.cecill.info. -# -# As a counterpart to the access to the source code and rights to copy, -# modify and redistribute granted by the license, users are provided only -# with a limited warranty and the software's author, the holder of the -# economic rights, and the successive licensors have only limited -# liability. -# -# The fact that you are presently reading this means that you have had -# knowledge of the CeCILL license and that you accept its terms. - -"""SSH-CLI-Tools setuptools script.""" - -import os -from setuptools import setup, find_packages - -PACKAGE = 'iotlabsshcli' -# GPL compatible http://www.gnu.org/licenses/license-list.html#CeCILL -LICENSE = 'CeCILL v2.1' - - -def readme(fname): - """Utility function to read the README. Used for long description.""" - return open(os.path.join(os.path.dirname(__file__), fname)).read() - - -def get_version(package): - """ Extract package version without importing file - Importing cause issues with coverage, - (modules can be removed from sys.modules to prevent this) - Importing __init__.py triggers importing rest and then requests too - - Inspired from pep8 setup.py - """ - version = '-1' - with open(os.path.join(package, '__init__.py')) as init_fd: - for line in init_fd: - if line.startswith('__version__'): - version = eval(line.split('=')[-1]) # pylint:disable=eval-used - break - return version - - -SCRIPTS = ['iotlab-ssh'] -DEPRECATED_SCRIPTS = ['open-a8-cli'] - -SCRIPTS += DEPRECATED_SCRIPTS - -INSTALL_REQUIRES = ['argparse', 'iotlabcli>=2.0', 'parallel-ssh>=2.5.4', - 'scp', 'gevent>=1.1', 'psutil'] - -setup( - name=PACKAGE, - version=get_version(PACKAGE), - description='IoT-LAB testbed ssh command-line client', - long_description=readme('README.rst'), - long_description_content_type='text/x-rst', - author='IoT-LAB Team', - author_email='admin@iot-lab.info', - url='http://www.iot-lab.info', - license=LICENSE, - download_url='http://github.com/iot-lab/ssh-cli-tools/', - packages=find_packages(), - scripts=SCRIPTS, - classifiers=['Development Status :: 5 - Production/Stable', - 'Programming Language :: Python', - 'Programming Language :: Python :: 3', - 'Intended Audience :: End Users/Desktop', - 'Environment :: Console', - 'Topic :: Utilities', ], - install_requires=INSTALL_REQUIRES, -) diff --git a/tests_utils/check_license.sh b/tests_utils/check_license.sh index 4d61172..961b05a 100644 --- a/tests_utils/check_license.sh +++ b/tests_utils/check_license.sh @@ -9,7 +9,6 @@ files_list=$(git ls-tree -r HEAD --full-tree --name-only) files_list=$(echo "${files_list}" | grep -v \ -e 'tests_utils/' \ -e '.gitignore' \ - -e 'setup.cfg' \ -e 'tox.ini' \ -e '.md$' \ -e '.rst$' \ @@ -19,6 +18,7 @@ files_list=$(echo "${files_list}" | grep -v \ -e 'codecov.yml' \ -e '.github/workflows/continuous-integration.yml' \ -e 'utils/iotlabsshcli-bash-completion.sh' \ + -e 'pyproject.toml' \ ) # Verify that 'AUTHORS' and 'COPYING' files exist diff --git a/tests_utils/test-requirements.txt b/tests_utils/test-requirements.txt index 7715f79..9cd0a7f 100644 --- a/tests_utils/test-requirements.txt +++ b/tests_utils/test-requirements.txt @@ -1,9 +1,7 @@ pytest pytest-cov -mock pylint -pycodestyle -# issues with pep8 >= 1.6 -flake8==2.3.0 -codecov>=1.4.0 +ruff +isort twine +build diff --git a/tox.ini b/tox.ini index 95c747e..8696762 100644 --- a/tox.ini +++ b/tox.ini @@ -1,19 +1,20 @@ [tox] -envlist = copying,{py37,py38,py39}-{lint,tests,cli,checksetup} +envlist = copying,{py310,py313,py314}-{lint,tests,cli,checksetup} skip_missing_interpreters = True [testenv] -whitelist_externals = - cli: {[testenv:cli]whitelist_externals} +allowlist_externals = + /bin/bash + /usr/bin/bash deps= -rtests_utils/test-requirements.txt checksetup: {[testenv:checksetup]deps} commands= - tests: {[testenv:tests]commands} - lint: {[testenv:lint]commands} - cli: {[testenv:cli]commands} + tests: {[testenv:tests]commands} + lint: {[testenv:lint]commands} + cli: {[testenv:cli]commands} checksetup: {[testenv:checksetup]commands} - coverage: {[testenv:coverage]commands} + coverage: {[testenv:coverage]commands} [testenv:tests] commands= @@ -21,20 +22,20 @@ commands= [testenv:lint] commands= - pycodestyle iotlabsshcli setup.py - pylint --rcfile=setup.cfg iotlabsshcli setup.py - flake8 - twine check {distdir}/* + ruff check iotlabsshcli + ruff format --check iotlabsshcli + isort --check-only iotlabsshcli + pylint iotlabsshcli [testenv:copying] -whitelist_externals = +allowlist_externals = /bin/bash /usr/bin/bash commands= bash tests_utils/check_license.sh [testenv:cli] -whitelist_externals = +allowlist_externals = /bin/bash /usr/bin/bash commands= @@ -43,12 +44,13 @@ commands= [testenv:checksetup] deps = - docutils - readme-renderer + build + twine skip_install = True usedevelop = False commands = - python setup.py check --strict --metadata --restructuredtext + python -m build --outdir {envtmpdir}/dist + twine check {envtmpdir}/dist/* [testenv:coverage] passenv = CI TRAVIS TRAVIS_*