diff --git a/iotlabsshcli/_deprecated.py b/iotlabsshcli/_deprecated.py index 8f10b2c..e4b1b29 100644 --- a/iotlabsshcli/_deprecated.py +++ b/iotlabsshcli/_deprecated.py @@ -24,6 +24,6 @@ from iotlabsshcli.parser.open_linux_parser import main as _main -def open_a8_cli(): +def open_a8_cli() -> None: """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 54f3c6b..c1c7c12 100644 --- a/iotlabsshcli/open_linux.py +++ b/iotlabsshcli/open_linux.py @@ -22,11 +22,12 @@ import os.path from collections import OrderedDict +from typing import Any from iotlabsshcli.sshlib import OpenLinuxSsh -def _nodes_grouped(nodes): +def _nodes_grouped(nodes: list[str]) -> OrderedDict[str, list[str]]: """Group nodes per site from a list of nodes. >>> _nodes_grouped([]) OrderedDict() @@ -58,7 +59,9 @@ def _nodes_grouped(nodes): _REMOTE_SHARED_DIR = "shared/.iotlabsshcli" -def flash(config_ssh, nodes, firmware, verbose=False): +def flash( + config_ssh: dict[str, Any], nodes: list[str], firmware: str, verbose: bool = False +) -> dict[str, Any]: """Flash the firmware of co-microcontroller""" failed_hosts = [] # configure ssh and remote firmware names. @@ -79,7 +82,7 @@ def flash(config_ssh, nodes, firmware, verbose=False): return {"flash": result} -def reset(config_ssh, nodes, verbose=False): +def reset(config_ssh: dict[str, Any], nodes: list[str], verbose: bool = False) -> dict[str, Any]: """Reset co-microcontroller""" # Configure ssh @@ -89,7 +92,9 @@ def reset(config_ssh, nodes, verbose=False): return {"reset": ssh.run(_RESET_CMD)} -def wait_for_boot(config_ssh, nodes, max_wait=120, verbose=False): +def wait_for_boot( + config_ssh: dict[str, Any], nodes: list[str], max_wait: int = 120, verbose: bool = False +) -> dict[str, Any]: """Wait for the open Linux nodes boot""" # Configure ssh. @@ -99,7 +104,13 @@ def wait_for_boot(config_ssh, nodes, max_wait=120, verbose=False): return {"wait-for-boot": ssh.wait(max_wait)} -def run_cmd(config_ssh, nodes, cmd, run_on_frontend=False, verbose=False): +def run_cmd( + config_ssh: dict[str, Any], + nodes: list[str], + cmd: str, + run_on_frontend: bool = False, + verbose: bool = False, +) -> dict[str, Any]: """Run a command on the Linux nodes or SSH frontend servers""" # Configure ssh. @@ -108,7 +119,9 @@ def run_cmd(config_ssh, nodes, cmd, run_on_frontend=False, verbose=False): return {"run-cmd": ssh.run(cmd, with_proxy=not run_on_frontend)} -def copy_file(config_ssh, nodes, file_path, verbose=False): +def copy_file( + config_ssh: dict[str, Any], nodes: list[str], file_path: str, verbose: bool = False +) -> dict[str, Any]: """Copy a file to SSH frontend servers""" # Configure ssh. @@ -120,7 +133,9 @@ def copy_file(config_ssh, nodes, file_path, verbose=False): return {"copy-file": result} -def _get_failed_result(groups, result, run_on_frontend): +def _get_failed_result( + groups: OrderedDict[str, list[str]], result: dict[str, list[str]], run_on_frontend: bool +) -> list[str]: """Returns failed nodes or SSH frontend servers list. We delete failed hosts for the next commands in the groups @@ -139,7 +154,13 @@ def _get_failed_result(groups, result, run_on_frontend): return failed -def run_script(config_ssh, nodes, script, run_on_frontend=False, verbose=False): +def run_script( + config_ssh: dict[str, Any], + nodes: list[str], + script: str, + run_on_frontend: bool = False, + verbose: bool = False, +) -> dict[str, Any]: """Run a script in background on Linux nodes or SSH frontend servers""" # Configure ssh. diff --git a/iotlabsshcli/parser/open_linux_parser.py b/iotlabsshcli/parser/open_linux_parser.py index 21644ad..2fc2924 100644 --- a/iotlabsshcli/parser/open_linux_parser.py +++ b/iotlabsshcli/parser/open_linux_parser.py @@ -22,6 +22,7 @@ import argparse import sys +from typing import Any from iotlabcli import auth, helpers, rest from iotlabcli.helpers import deprecate_warn_cmd @@ -31,7 +32,7 @@ import iotlabsshcli.open_linux -def parse_options(): +def parse_options() -> argparse.ArgumentParser: """Parse command line option.""" parent_parser = argparse.ArgumentParser(add_help=False) common.add_auth_arguments(parent_parser, False) @@ -54,7 +55,7 @@ def parse_options(): class DeprecateHelpFormatter(argparse.HelpFormatter): """Add drepecated help formatter""" - def add_usage(self, usage, actions, groups, prefix=None): + def add_usage(self, usage, actions, groups, prefix=None) -> None: # type: ignore[override] # self._prog = iotlab-ssh flash-m3 | reset-m3 old_cmd = self._prog.split()[-1] new_cmd = old_cmd.split("-")[0] @@ -142,7 +143,7 @@ def add_usage(self, usage, actions, groups, prefix=None): return parser -def open_linux_parse_and_run(opts): +def open_linux_parse_and_run(opts: argparse.Namespace) -> dict[str, Any]: """Parse namespace 'opts' object.""" user, passwd = auth.get_user_credentials(opts.username, opts.password) api = rest.Api(user, passwd) @@ -194,7 +195,7 @@ def open_linux_parse_and_run(opts): return res -def main(args=None): +def main(args: list[str] | None = None) -> None: """Open Linux SSH cli parser.""" args = args or sys.argv[1:] # required for easy testing. parser = parse_options() diff --git a/iotlabsshcli/sshlib/open_linux_ssh.py b/iotlabsshcli/sshlib/open_linux_ssh.py index 46b72ba..d0821b7 100644 --- a/iotlabsshcli/sshlib/open_linux_ssh.py +++ b/iotlabsshcli/sshlib/open_linux_ssh.py @@ -23,11 +23,12 @@ import asyncio import os import time +from typing import Any import asyncssh -def _cleanup_result(result): +def _cleanup_result(result: dict[str, list[str]]) -> dict[str, list[str]]: """Remove empty list from result. >>> _cleanup_result({ '0': [], '1': []}) @@ -49,7 +50,9 @@ def _cleanup_result(result): return result -def _extend_result(result, new_result): +def _extend_result( + result: dict[str, list[str]], new_result: dict[str, list[str]] +) -> dict[str, list[str]]: """Extend result dictionnary values with new result dictionnary values @@ -100,7 +103,7 @@ def _extend_result(result, new_result): return result -def _check_all_nodes_processed(result): +def _check_all_nodes_processed(result: dict[str, list[str]]) -> bool: """Verify all nodes are successful or failed. >>> _check_all_nodes_processed({ 'saclay': [], 'grenoble': []}) @@ -126,12 +129,14 @@ def _check_all_nodes_processed(result): class OpenLinuxSsh: """Implement SSH API using asyncssh.""" - def __init__(self, config_ssh, groups, verbose=False): + def __init__( + self, config_ssh: dict[str, Any], groups: dict[str, list[str]], verbose: bool = False + ) -> None: self.config_ssh = config_ssh self.groups = groups self.verbose = verbose - def run(self, command, with_proxy=True, **kwargs): + def run(self, command: str, with_proxy: bool = True, **kwargs: Any) -> dict[str, list[str]]: """Run ssh command on nodes, optionally through a proxy.""" result = {"0": [], "1": []} for site, hosts in self.groups.items(): @@ -143,7 +148,7 @@ def run(self, command, with_proxy=True, **kwargs): result = _extend_result(result, result_cmd) return _cleanup_result(result) - def scp(self, src, dst): + def scp(self, src: str, dst: str) -> dict[str, list[str]]: """Copy file to SSH frontend via SFTP.""" result = {"0": [], "1": []} for site in self.groups: @@ -154,7 +159,7 @@ def scp(self, src, dst): result["1"].append(site) return _cleanup_result(result) - def wait(self, max_wait): + def wait(self, max_wait: int) -> dict[str, list[str]]: """Wait for requested Linux nodes until they boot.""" result = {"0": [], "1": []} start_time = time.time() @@ -167,13 +172,20 @@ def wait(self, max_wait): result = _extend_result(result, result_cmd) return _cleanup_result(result) - def _connect_kwargs(self, timeout=10): + def _connect_kwargs(self, timeout: int = 10) -> dict[str, Any]: kwargs = {"known_hosts": None, "connect_timeout": timeout} if SSH_KEY: kwargs["client_keys"] = [os.path.expanduser(SSH_KEY)] return kwargs - async def _run_command(self, command, hosts, proxy_host=None, timeout=10, **kwargs): + async def _run_command( + self, + command: str, + hosts: list[str], + proxy_host: str | None = None, + timeout: int = 10, + **kwargs: Any, + ) -> dict[str, list[str]]: tasks = [ self._run_on_host(host, command, proxy_host=proxy_host, timeout=timeout, **kwargs) for host in hosts @@ -190,7 +202,14 @@ async def _run_command(self, command, hosts, proxy_host=None, timeout=10, **kwar result["0"].append(host) return result - async def _run_on_host(self, host, command, proxy_host=None, timeout=10, **kwargs): + async def _run_on_host( + self, + host: str, + command: str, + proxy_host: str | None = None, + timeout: int = 10, + **kwargs: Any, + ) -> tuple[int, str]: ck = self._connect_kwargs(timeout) if proxy_host: async with asyncssh.connect( @@ -204,7 +223,7 @@ async def _run_on_host(self, host, command, proxy_host=None, timeout=10, **kwarg result = await conn.run(command, **kwargs) return result.exit_status, result.stdout or "" - async def _copy_file(self, site, src, dst): + async def _copy_file(self, site: str, src: str, dst: str) -> None: ck = self._connect_kwargs() async with asyncssh.connect(site, username=self.config_ssh["user"], **ck) as conn: async with conn.start_sftp_client() as sftp: diff --git a/iotlabsshcli/tests/iotlabsshcli_mock.py b/iotlabsshcli/tests/iotlabsshcli_mock.py index 77c4a17..0d99ef1 100644 --- a/iotlabsshcli/tests/iotlabsshcli_mock.py +++ b/iotlabsshcli/tests/iotlabsshcli_mock.py @@ -23,6 +23,7 @@ import sys import unittest +from typing import Any from unittest.mock import Mock, patch from iotlabcli.helpers import json_dumps @@ -34,14 +35,14 @@ class RequestRet: # pylint:disable=too-few-public-methods """Mock of Request return value""" - def __init__(self, status_code, content, headers=None): + def __init__(self, status_code: int, content: str, headers: Any = None) -> None: self.status_code = status_code self.content = content.encode("utf-8") self.headers = headers self.text = self.content.decode("utf-8") -def api_mock(ret=None): +def api_mock(ret: dict[str, Any] | None = None) -> Mock: """Return a mock of an api object returned value for api methods will be 'ret' parameter or API_RET """ @@ -53,7 +54,7 @@ def api_mock(ret=None): return api_class.return_value -def api_mock_stop(): +def api_mock_stop() -> None: """Stop all patches started by api_mock. Actually it stops everything but not a problem""" patch.stopall() @@ -62,7 +63,7 @@ def api_mock_stop(): class MainMock(unittest.TestCase): """Common mock needed for testing main function of parsers""" - def setUp(self): + def setUp(self) -> None: self.api = api_mock() patch("sys.stderr", sys.stdout).start() @@ -75,11 +76,11 @@ def setUp(self): "iotlabcli.auth.get_user_credentials", Mock(return_value=("username", "password")) ).start() - def get_exp(_, x, running_only=True): + def get_exp(_: Any, x: int | None, running_only: bool = True) -> int: return x if x is not None else (123 if running_only else 234) patch("iotlabcli.helpers.get_current_experiment", get_exp).start() - def tearDown(self): + def tearDown(self) -> None: api_mock_stop() patch.stopall() diff --git a/iotlabsshcli/tests/open_linux_parser_test.py b/iotlabsshcli/tests/open_linux_parser_test.py index 736d25b..585f66c 100644 --- a/iotlabsshcli/tests/open_linux_parser_test.py +++ b/iotlabsshcli/tests/open_linux_parser_test.py @@ -21,6 +21,7 @@ """Tests for iotlabsshcli.parser.open_linux package.""" +from typing import Any from unittest.mock import Mock, patch import jmespath @@ -43,7 +44,7 @@ class TestMainNodeParser(MainMock): @patch("iotlabsshcli.open_linux.flash") @patch("iotlabcli.parser.common.list_nodes") - def test_main_flash(self, list_nodes, flash): + def test_main_flash(self, list_nodes: Any, flash: Any) -> None: """Run the parser.node.main with update subparser function.""" flash.return_value = {"result": "test"} @@ -71,7 +72,7 @@ def test_main_flash(self, list_nodes, flash): @patch("iotlabsshcli.open_linux.reset") @patch("iotlabcli.parser.common.list_nodes") - def test_main_reset(self, list_nodes, reset): + def test_main_reset(self, list_nodes: Any, reset: Any) -> None: """Run the parser.node.main with reset subparser function.""" reset.return_value = {"result": "test"} list_nodes.return_value = self._nodes @@ -95,7 +96,7 @@ def test_main_reset(self, list_nodes, reset): @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): + def test_main_wait_for_boot(self, list_nodes: Any, wait_for_boot: Any) -> None: """Run the parser.node.main with wait-for-boot subparser function.""" wait_for_boot.return_value = {"result": "test"} list_nodes.return_value = self._nodes @@ -126,7 +127,7 @@ def test_main_wait_for_boot(self, list_nodes, wait_for_boot): @patch("iotlabsshcli.open_linux.run_script") @patch("iotlabcli.parser.common.list_nodes") - def test_main_run_script(self, list_nodes, run_script): + def test_main_run_script(self, list_nodes: Any, run_script: Any) -> None: """Run the parser.node.main with run-script subparser function.""" run_script.return_value = {"result": "test"} list_nodes.return_value = self._nodes @@ -165,7 +166,7 @@ def test_main_run_script(self, list_nodes, run_script): @patch("iotlabsshcli.open_linux.run_cmd") @patch("iotlabcli.parser.common.list_nodes") - def test_main_run_cmd(self, list_nodes, run_cmd): + def test_main_run_cmd(self, list_nodes: Any, run_cmd: Any) -> None: """Run the parser.node.main with run-cmd subparser function.""" run_cmd.return_value = {"result": "test"} list_nodes.return_value = self._nodes @@ -200,7 +201,7 @@ def test_main_run_cmd(self, list_nodes, run_cmd): @patch("iotlabsshcli.open_linux.copy_file") @patch("iotlabcli.parser.common.list_nodes") - def test_main_copy_file(self, list_nodes, copy_file): + def test_main_copy_file(self, list_nodes: Any, copy_file: Any) -> None: """Run the parser.node.main with copy-file subparser function.""" copy_file.return_value = {"result": "test"} list_nodes.return_value = self._nodes @@ -222,13 +223,13 @@ def test_main_copy_file(self, list_nodes, copy_file): {"user": "username", "exp_id": 123}, self._root_nodes, "script.sh", verbose=False ) - def test_main_unknown_function(self): + def test_main_unknown_function(self) -> None: """Run the parser.node.main with an unknown function.""" args = ["unknown-cmd"] self.assertRaises(SystemExit, open_linux_parser.main, args) @patch("iotlabcli.parser.common.list_nodes") - def test_run_unknown_function(self, list_nodes): + def test_run_unknown_function(self, list_nodes: Any) -> None: # pylint:disable=unused-argument """Run the parser.node.main with an unknown function.""" parser = open_linux_parser.parse_options() @@ -243,7 +244,7 @@ def test_run_unknown_function(self, list_nodes): @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): + def test_reset_jmespath(self, print_result: Any, list_nodes: Any, reset: Any) -> None: """Run reset subparser function with jmespath options.""" reset.return_value = {"result": "test"} list_nodes.return_value = self._nodes @@ -262,7 +263,9 @@ def test_reset_jmespath(self, print_result, list_nodes, 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): + def test_deprecated_parser( + self, print_result: Any, list_nodes: Any, flash: Any, reset: Any + ) -> None: """Run deprecated subparsers.""" reset.return_value = flash.return_value = {"result": "test"} list_nodes.return_value = self._nodes @@ -290,7 +293,7 @@ def test_deprecated_parser(self, print_result, list_nodes, flash, reset): self.assertEqual(args[2], None) @patch("iotlabsshcli._deprecated.deprecate_cmd") - def test_open_a8_cli_entry_point(self, mock_deprecate): + def test_open_a8_cli_entry_point(self, mock_deprecate: Any) -> None: """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 573c63c..5c6c174 100644 --- a/iotlabsshcli/tests/open_linux_ssh_test.py +++ b/iotlabsshcli/tests/open_linux_ssh_test.py @@ -22,6 +22,7 @@ """Tests for iotlabsshcli.open_linux package.""" import os +from typing import Any from unittest.mock import AsyncMock, MagicMock, patch import asyncssh @@ -33,7 +34,7 @@ from .open_linux_test import _GRENOBLE_NODES, _ROOT_NODES, _SACLAY_NODES -def _make_conn(exit_status=0, stdout="test"): +def _make_conn(exit_status: int = 0, stdout: str = "test") -> MagicMock: """Return a mock asyncssh connection usable as an async context manager.""" run_result = MagicMock() run_result.exit_status = exit_status @@ -54,7 +55,7 @@ def _make_conn(exit_status=0, stdout="test"): @mark.parametrize("run_on_frontend", [False, True]) -def test_run(run_on_frontend): +def test_run(run_on_frontend: bool) -> None: """Test running commands on ssh nodes.""" config_ssh = {"user": "username", "exp_id": 123} groups = _nodes_grouped(_ROOT_NODES) @@ -63,7 +64,7 @@ def test_run(run_on_frontend): conn_ok = _make_conn(exit_status=0) conn_fail = _make_conn(exit_status=1) - def connect_side_effect(host, **_): + def connect_side_effect(host: str, **_: Any) -> MagicMock: return conn_ok if "saclay" in host else conn_fail with patch( @@ -78,7 +79,7 @@ def connect_side_effect(host, **_): assert ret == {"0": _SACLAY_NODES, "1": _GRENOBLE_NODES} -def test_scp(): +def test_scp() -> None: """Test copying a file to SSH frontend nodes.""" config_ssh = {"user": "username", "exp_id": 123} groups = _nodes_grouped(_ROOT_NODES) @@ -109,7 +110,7 @@ def test_scp(): assert ret.get("0", []) == [] -def test_run_with_ssh_key(): +def test_run_with_ssh_key() -> None: """Test that a configured SSH_KEY is forwarded as client_keys.""" config_ssh = {"user": "username", "exp_id": 123} groups = _nodes_grouped(_ROOT_NODES) @@ -129,7 +130,7 @@ def test_run_with_ssh_key(): assert call_kwargs["client_keys"] == [os.path.expanduser("~/.ssh/id_rsa")] -def test_wait_all_boot(): +def test_wait_all_boot() -> None: """Test waiting for ssh nodes to become available.""" config_ssh = {"user": "username", "exp_id": 123} groups = _nodes_grouped(_ROOT_NODES) diff --git a/iotlabsshcli/tests/open_linux_test.py b/iotlabsshcli/tests/open_linux_test.py index 3ea62cb..04ea384 100644 --- a/iotlabsshcli/tests/open_linux_test.py +++ b/iotlabsshcli/tests/open_linux_test.py @@ -22,6 +22,7 @@ """Tests for iotlabsshcli.open_linux package.""" import os.path +from typing import Any from unittest.mock import patch from pytest import mark @@ -47,7 +48,7 @@ @patch("iotlabsshcli.sshlib.OpenLinuxSsh.run") @patch("iotlabsshcli.sshlib.OpenLinuxSsh.scp") -def test_open_linux_flash(scp, run): +def test_open_linux_flash(scp: Any, run: Any) -> None: """Test flashing a firmware.""" config_ssh = { "user": "username", @@ -81,7 +82,7 @@ def test_open_linux_flash(scp, run): @patch("iotlabsshcli.sshlib.OpenLinuxSsh.run") -def test_open_linux_reset(run): +def test_open_linux_reset(run: Any) -> None: """Test resetting co-microcontroller.""" config_ssh = { "user": "username", @@ -97,7 +98,7 @@ def test_open_linux_reset(run): @patch("iotlabsshcli.sshlib.OpenLinuxSsh.wait") -def test_open_linux_wait_for_boot(wait): +def test_open_linux_wait_for_boot(wait: Any) -> None: """Test wait for Linux boot.""" config_ssh = { "user": "username", @@ -115,7 +116,7 @@ def test_open_linux_wait_for_boot(wait): @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): +def test_open_linux_run_script(scp: Any, run: Any, run_on_frontend: bool) -> None: """Test run script on Linux nodes.""" config_ssh = { "user": "username", @@ -154,7 +155,7 @@ def test_open_linux_run_script(scp, run, run_on_frontend): @patch("iotlabsshcli.sshlib.OpenLinuxSsh.scp") -def test_open_linux_copy_file(scp): +def test_open_linux_copy_file(scp: Any) -> None: """Test copy file on the SSH frontend.""" config_ssh = { "user": "username", @@ -173,7 +174,7 @@ def test_open_linux_copy_file(scp): @mark.parametrize("run_on_frontend", [False, True]) @patch("iotlabsshcli.sshlib.OpenLinuxSsh.run") -def test_open_linux_run_cmd(run, run_on_frontend): +def test_open_linux_run_cmd(run: Any, run_on_frontend: bool) -> None: """Test run command on Linux nodes.""" config_ssh = { "user": "username",