From 4d0956d36c22c4a0b3bbb04a57b5d76225639418 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Matthieu=20T=C3=A2che?= Date: Wed, 15 May 2024 15:11:50 +0200 Subject: [PATCH] refactor: use imported aioeapi module --- aioeapi/config_session.py | 8 ++++---- aioeapi/device.py | 11 +++++++---- anta/cli/exec/utils.py | 2 +- anta/device.py | 9 +++++---- tests/lib/fixture.py | 2 +- tests/units/test_device.py | 8 ++++---- 6 files changed, 22 insertions(+), 18 deletions(-) diff --git a/aioeapi/config_session.py b/aioeapi/config_session.py index 2327d762e..2a654c781 100644 --- a/aioeapi/config_session.py +++ b/aioeapi/config_session.py @@ -189,7 +189,7 @@ async def push(self, content: list[str] | str, *, replace: bool = False) -> None # prepare the initial set of command to enter the config session and # rollback clean if the `replace` argument is True. - commands = [self._cli_config_session] + commands: list[str | dict[str, Any]] = [self._cli_config_session] if replace: commands.append(self.CLI_CFG_FACTORY_RESET) @@ -244,7 +244,7 @@ async def diff(self) -> str: ---------- * https://www.gnu.org/software/diffutils/manual/diffutils.txt """ - return await self._cli(f"show session-config named {self.name} diffs", ofmt="text")# type: ignore[return-value] # text outformat returns str + return await self._cli(f"show session-config named {self.name} diffs", ofmt="text") # type: ignore[return-value] # text outformat returns str async def load_file(self, filename: str, *, replace: bool = False) -> None: """ @@ -268,12 +268,12 @@ async def load_file(self, filename: str, *, replace: bool = False) -> None: If there are any issues with loading the configuration file then a RuntimeError is raised with the error messages content. """ - commands = [self._cli_config_session] + commands: list[str | dict[str, Any]] = [self._cli_config_session] if replace: commands.append(self.CLI_CFG_FACTORY_RESET) commands.append(f"copy {filename} session-config") - res: list[dict[str, Any]] = await self._cli(commands=commands) # type: ignore[assignment] # JSON outformat of multiple commands returns list[dict[str, Any]] + res: list[dict[str, Any]] = await self._cli(commands=commands) # type: ignore[assignment] # JSON outformat of multiple commands returns list[dict[str, Any]] checks_re = re.compile(r"error|abort|invalid", flags=re.I) messages = res[-1]["messages"] diff --git a/aioeapi/device.py b/aioeapi/device.py index 0cea1799d..4da52ea51 100644 --- a/aioeapi/device.py +++ b/aioeapi/device.py @@ -6,7 +6,7 @@ from __future__ import annotations from socket import getservbyname -from typing import Any +from typing import TYPE_CHECKING, Any # ----------------------------------------------------------------------------- # Public Imports @@ -20,6 +20,9 @@ from .config_session import SessionConfig from .errors import EapiCommandError +if TYPE_CHECKING: + from collections.abc import Sequence + # ----------------------------------------------------------------------------- # Exports # ----------------------------------------------------------------------------- @@ -110,8 +113,8 @@ async def check_connection(self) -> bool: async def cli( # noqa: PLR0913 self, - command: str | None = None, - commands: list[str] | None = None, + command: str | dict[str, Any] | None = None, + commands: Sequence[str | dict[str, Any]] | None = None, ofmt: str | None = None, version: int | str | None = "latest", *, @@ -184,7 +187,7 @@ async def cli( # noqa: PLR0913 def _jsonrpc_command( # noqa: PLR0913 self, - commands: list[str] | None = None, + commands: Sequence[str | dict[str, Any]] | None = None, ofmt: str | None = None, version: int | str | None = "latest", *, diff --git a/anta/cli/exec/utils.py b/anta/cli/exec/utils.py index 5a28912e1..2c9ffd3b6 100644 --- a/anta/cli/exec/utils.py +++ b/anta/cli/exec/utils.py @@ -14,10 +14,10 @@ from pathlib import Path from typing import TYPE_CHECKING, Literal -from aioeapi import EapiCommandError from click.exceptions import UsageError from httpx import ConnectError, HTTPError +from aioeapi import EapiCommandError from anta.device import AntaDevice, AsyncEOSDevice from anta.models import AntaCommand diff --git a/anta/device.py b/anta/device.py index d517b8fb0..8c16a020b 100644 --- a/anta/device.py +++ b/anta/device.py @@ -18,7 +18,8 @@ from asyncssh import SSHClientConnection, SSHClientConnectionOptions from httpx import ConnectError, HTTPError, TimeoutException -from anta import __DEBUG__, aioeapi +import aioeapi +from anta import __DEBUG__ from anta.logger import anta_log_exception, exc_to_str from anta.models import AntaCommand @@ -316,7 +317,7 @@ async def _collect(self, command: AntaCommand) -> None: # noqa: C901 function ---- command: the AntaCommand to collect. """ - commands: list[dict[str, Any]] = [] + commands: list[dict[str, str | int]] = [] if self.enable and self._enable_password is not None: commands.append( { @@ -329,11 +330,11 @@ async def _collect(self, command: AntaCommand) -> None: # noqa: C901 function commands.append({"cmd": "enable"}) commands += [{"cmd": command.command, "revision": command.revision}] if command.revision else [{"cmd": command.command}] try: - response: list[dict[str, Any]] = await self._session.cli( + response: list[dict[str, Any] | str] = await self._session.cli( commands=commands, ofmt=command.ofmt, version=command.version, - ) + ) # type: ignore[assignment] # multiple commands returns a list # Do not keep response of 'enable' command command.output = response[-1] except aioeapi.EapiCommandError as e: diff --git a/tests/lib/fixture.py b/tests/lib/fixture.py index 43fb60a84..f0f25b838 100644 --- a/tests/lib/fixture.py +++ b/tests/lib/fixture.py @@ -13,7 +13,7 @@ import pytest from click.testing import CliRunner, Result -from anta import aioeapi +import aioeapi from anta.cli.console import console from anta.device import AntaDevice, AsyncEOSDevice from anta.inventory import AntaInventory diff --git a/tests/units/test_device.py b/tests/units/test_device.py index c901a3dd8..cb5d69928 100644 --- a/tests/units/test_device.py +++ b/tests/units/test_device.py @@ -15,7 +15,7 @@ from asyncssh import SSHClientConnection, SSHClientConnectionOptions from rich import print as rprint -from anta import aioeapi +import aioeapi from anta.device import AntaDevice, AsyncEOSDevice from anta.models import AntaCommand from tests.lib.fixture import COMMAND_OUTPUT @@ -705,9 +705,9 @@ async def test_refresh(self, async_device: AsyncEOSDevice, patch_kwargs: list[di """Test AsyncEOSDevice.refresh().""" with patch.object(async_device._session, "check_connection", **patch_kwargs[0]), patch.object(async_device._session, "cli", **patch_kwargs[1]): await async_device.refresh() - async_device._session.check_connection.assert_called_once() + async_device._session.check_connection.assert_called_once() # type: ignore[attr-defined] # aioeapi.Device.check_connection is patched if expected["is_online"]: - async_device._session.cli.assert_called_once() + async_device._session.cli.assert_called_once() # type: ignore[attr-defined] # aioeapi.Device.cli is patched assert async_device.is_online == expected["is_online"] assert async_device.established == expected["established"] assert async_device.hw_model == expected["hw_model"] @@ -740,7 +740,7 @@ async def test__collect(self, async_device: AsyncEOSDevice, command: dict[str, A commands.append({"cmd": cmd.command, "revision": cmd.revision}) else: commands.append({"cmd": cmd.command}) - async_device._session.cli.assert_called_once_with(commands=commands, ofmt=cmd.ofmt, version=cmd.version) + async_device._session.cli.assert_called_once_with(commands=commands, ofmt=cmd.ofmt, version=cmd.version) # type: ignore[attr-defined] # aioeapi.Device.cli is patched # pylint: disable=line-too-long assert cmd.output == expected["output"] assert cmd.errors == expected["errors"]