linting

anta/device.py

@@ -10,15 +10,14 @@
 import logging
 from abc import ABC, abstractmethod
 from collections import defaultdict
-from collections.abc import Iterator
 from pathlib import Path
 from typing import Any, AnyStr, Dict, Iterator, List, Literal, Optional, Union
 
 import aioeapi
 import asyncssh
 from aiocache import Cache
 from aiocache.plugins import HitMissRatioPlugin
-from aioeapi import Device, EapiCommandError
+from aioeapi import Device
 from asyncssh import SSHClientConnection, SSHClientConnectionOptions
 from httpx import ConnectError, HTTPError
 
@@ -41,16 +40,17 @@ class EapiCommandError(RuntimeError):
     not_exec: List[str] - a list of commands that were not executed
     """
 
-    def __init__(self, failed: str, errors: list[str], errmsg: str, passed, not_exec):
+    # pylint: disable=too-many-arguments
+    def __init__(self, failed: str, errors: list[str], errmsg: str, passed: list[Union[str, dict[str, Any]]], not_exec: list[dict[str, Any]]):
         """Initializer for the EapiCommandError exception"""
         self.failed = failed
         self.errors = errors
         self.errmsg = errmsg
         self.passed = passed
         self.not_exec = not_exec
-        super(EapiCommandError, self).__init__()
+        super().__init__()
 
-    def __str__(self):
+    def __str__(self) -> str:
         """returns the error message associated with the exception"""
         return self.errmsg
 
@@ -59,7 +59,7 @@ def __str__(self):
 
 
 # patching aioeapi.Device.jsonrpc_exec to test some aio-eapi enhancement
-async def jsonrpc_exec(self, jsonrpc: dict) -> List[Union[Dict, AnyStr]]:
+async def jsonrpc_exec(self, jsonrpc: dict) -> List[Union[Dict, AnyStr]]:  # type: ignore
     """
     Execute the JSON-RPC dictionary object.
 
@@ -105,10 +105,10 @@ async def jsonrpc_exec(self, jsonrpc: dict) -> List[Union[Dict, AnyStr]]:
 
     raise EapiCommandError(
         passed=[get_output(cmd_data[cmd_i]) for cmd_i, cmd in enumerate(commands[:err_at])],
-        failed=commands[err_at],
+        failed=commands[err_at]["cmd"],
         errors=cmd_data[err_at]["errors"],
         errmsg=err_msg,
-        not_exec=commands[err_at + 1 :],
+        not_exec=commands[err_at + 1 :],  # noqa: E203
     )
 
 
@@ -379,8 +379,8 @@ def _keys(self) -> tuple[Any, ...]:
     def is_supported(command: AntaCommand) -> bool:
         """Returns True if the command is supported on the device hardware platform, False otherwise.
         Can only be called if the command has failed."""
-        return (
-            not isinstance(command.failed, EapiCommandError)
+        return not (
+            isinstance(command.failed, EapiCommandError)
             and command.failed.errors
             and any("not supported on this hardware platform" in e for e in command.failed.errors)
         )

tests/units/test_device.py

@@ -11,9 +11,8 @@
 from unittest.mock import patch
 
 import pytest
-from aioeapi import EapiCommandError
 
-from anta.device import AntaDevice, AsyncEOSDevice
+from anta.device import AntaDevice, AsyncEOSDevice, EapiCommandError
 from anta.models import AntaCommand
 from tests.lib.utils import generate_test_ids_list
 
@@ -104,6 +103,9 @@ def test__init__(self, device_data: dict[str, Any]) -> None:
             assert device.cache_locks is not None
 
     def test_is_supported(self) -> None:
+        """
+        Test if the is_supported() static method parses correctly the aioeapi.EapiCommandError exception
+        """
         DATA: dict[str, Any] = {
             "host": "42.42.42.42",
             "username": "anta",
@@ -126,7 +128,7 @@ def test_is_supported(self) -> None:
             command="show hardware counter drop",
             failed=EapiCommandError(
                 passed=[],
-                failed={"cmd": "show hardware counter drop"},
+                failed="show hardware counter drop",
                 errors=["Unavailable command (not supported on this hardware platform) (at token 2: 'counter')"],
                 errmsg="CLI command 1 of 1 'show hardware counter drop' failed: invalid command",
                 not_exec=[],