Merge branch 'main' into issue_810

.github/generate_release.py

@@ -30,7 +30,7 @@ class SafeDumper(yaml.SafeDumper):
     https://github.com/yaml/pyyaml/issues/234#issuecomment-765894586.
     """
 
-    # pylint: disable=R0901,W0613,W1113
+    # pylint: disable=R0901
 
     def increase_indent(self, flow=False, *args, **kwargs):
         return super().increase_indent(flow=flow, indentless=False)

.gitignore

@@ -1,8 +1,10 @@
 __pycache__
 *.pyc
 .pages
-.coverage
 .pytest_cache
+.mypy_cache
+.ruff_cache
+.cache
 build
 dist
 *.egg-info
@@ -46,14 +48,13 @@ htmlcov/
 .tox/
 .nox/
 .coverage
+coverage_html_report
 .coverage.*
-.cache
 nosetests.xml
 coverage.xml
 *.cover
 *.py,cover
 .hypothesis/
-.pytest_cache/
 cover/
 report.html
 
@@ -97,17 +98,4 @@ venv.bak/
 /site
 
 # VScode settings
-.vscode
-test.env
-tech-support/
-tech-support/*
-2*
-
-**/report.html
-.*report.html
-
-# direnv file
-.envrc
-
-clab-atd-anta/*
-clab-atd-anta/
+.vscode

.pre-commit-config.yaml

@@ -43,7 +43,7 @@ repos:
           - '<!--|  ~|  -->'
 
   - repo: https://github.com/astral-sh/ruff-pre-commit
-    rev: v0.6.4
+    rev: v0.6.5
     hooks:
     - id: ruff
       name: Run Ruff linter
@@ -69,6 +69,7 @@ repos:
           - types-pyOpenSSL
           - pylint_pydantic
           - pytest
+          - respx
 
   - repo: https://github.com/codespell-project/codespell
     rev: v2.3.0

.vscode/settings.json

@@ -3,28 +3,13 @@
   "ruff.configuration": "pyproject.toml",
   "python.testing.unittestEnabled": false,
   "python.testing.pytestEnabled": true,
-  "pylint.importStrategy": "fromEnvironment",
-  "pylint.severity": {
-    "refactor": "Warning"
-  },
-  "pylint.args": [
-    "--load-plugins",
-    "pylint_pydantic",
-    "--rcfile=pyproject.toml"
-  ],
   "python.testing.pytestArgs": [
     "tests"
   ],
-  "autoDocstring.docstringFormat": "numpy",
-  "autoDocstring.includeName": false,
-  "autoDocstring.includeExtendedSummary": true,
-  "autoDocstring.startOnNewLine": true,
-  "autoDocstring.guessTypes": true,
   "python.languageServer": "Pylance",
   "githubIssues.issueBranchTitle": "issues/${issueNumber}-${issueTitle}",
   "editor.formatOnPaste": true,
   "files.trimTrailingWhitespace": true,
-  "mypy.configFile": "pyproject.toml",
   "workbench.remoteIndicator.showExtensionRecommendations": true,
 
 }

anta/catalog.py

@@ -173,7 +173,7 @@ def flatten_modules(data: dict[str, Any], package: str | None = None) -> dict[Mo
                 module_name = f".{module_name}"  # noqa: PLW2901
             try:
                 module: ModuleType = importlib.import_module(name=module_name, package=package)
-            except Exception as e:  # pylint: disable=broad-exception-caught
+            except Exception as e:
                 # A test module is potentially user-defined code.
                 # We need to catch everything if we want to have meaningful logs
                 module_str = f"{module_name[1:] if module_name.startswith('.') else module_name}{f' from package {package}' if package else ''}"
@@ -328,6 +328,10 @@ def parse(filename: str | Path, file_format: Literal["yaml", "json"] = "yaml") -
         file_format
             Format of the file, either 'yaml' or 'json'.
 
+        Returns
+        -------
+        AntaCatalog
+            An AntaCatalog populated with the file content.
         """
         if file_format not in ["yaml", "json"]:
             message = f"'{file_format}' is not a valid format for an AntaCatalog file. Only 'yaml' and 'json' are supported."
@@ -356,8 +360,13 @@ def from_dict(data: RawCatalogInput, filename: str | Path | None = None) -> Anta
         ----------
         data
             Python dictionary used to instantiate the AntaCatalog instance.
-            filename: value to be set as AntaCatalog instance attribute
+        filename
+            value to be set as AntaCatalog instance attribute
 
+        Returns
+        -------
+        AntaCatalog
+            An AntaCatalog populated with the 'data' dictionary content.
         """
         tests: list[AntaTestDefinition] = []
         if data is None:
@@ -392,6 +401,10 @@ def from_list(data: ListAntaTestTuples) -> AntaCatalog:
         data
             Python list used to instantiate the AntaCatalog instance.
 
+        Returns
+        -------
+        AntaCatalog
+            An AntaCatalog populated with the 'data' list content.
         """
         tests: list[AntaTestDefinition] = []
         try:

anta/cli/_main.py

@@ -61,7 +61,7 @@ def cli() -> None:
     """Entrypoint for pyproject.toml."""
     try:
         anta(obj={}, auto_envvar_prefix="ANTA")
-    except Exception as exc:  # pylint: disable=broad-exception-caught
+    except Exception as exc:  # noqa: BLE001
         anta_log_exception(
             exc,
             f"Uncaught Exception when running ANTA CLI\n{GITHUB_SUGGESTION}",

anta/cli/debug/commands.py

@@ -35,7 +35,6 @@ def run_cmd(
     version: Literal["1", "latest"],
     revision: int,
 ) -> None:
-    # pylint: disable=too-many-arguments
     """Run arbitrary command to an ANTA device."""
     console.print(f"Run command [green]{command}[/green] on [red]{device.name}[/red]")
     # I do not assume the following line, but click make me do it
@@ -71,7 +70,6 @@ def run_template(
     version: Literal["1", "latest"],
     revision: int,
 ) -> None:
-    # pylint: disable=too-many-arguments
     # Using \b for click
     # ruff: noqa: D301
     """Run arbitrary templated command to an ANTA device.

anta/cli/debug/utils.py

@@ -48,7 +48,6 @@ def wrapper(
         **kwargs: Any,
     ) -> Any:
         # TODO: @gmuloc - tags come from context https://github.com/aristanetworks/anta/issues/584
-        # pylint: disable=unused-argument
         # ruff: noqa: ARG001
         if (d := inventory.get(device)) is None:
             logger.error("Device '%s' does not exist in Inventory", device)

anta/cli/exec/__init__.py

@@ -9,7 +9,7 @@
 
 
 @click.group("exec")
-def _exec() -> None:  # pylint: disable=redefined-builtin
+def _exec() -> None:
     """Commands to execute various scripts on EOS devices."""
 
 

anta/cli/exec/utils.py

@@ -10,16 +10,15 @@
 import itertools
 import json
 import logging
-import re
 from pathlib import Path
 from typing import TYPE_CHECKING, Literal
 
 from click.exceptions import UsageError
 from httpx import ConnectError, HTTPError
 
-from anta.custom_types import REGEXP_PATH_MARKERS
 from anta.device import AntaDevice, AsyncEOSDevice
 from anta.models import AntaCommand
+from anta.tools import safe_command
 from asynceapi import EapiCommandError
 
 if TYPE_CHECKING:
@@ -52,7 +51,7 @@ async def clear(dev: AntaDevice) -> None:
 
 async def collect_commands(
     inv: AntaInventory,
-    commands: dict[str, str],
+    commands: dict[str, list[str]],
     root_dir: Path,
     tags: set[str] | None = None,
 ) -> None:
@@ -61,17 +60,16 @@ async def collect_commands(
     async def collect(dev: AntaDevice, command: str, outformat: Literal["json", "text"]) -> None:
         outdir = Path() / root_dir / dev.name / outformat
         outdir.mkdir(parents=True, exist_ok=True)
-        safe_command = re.sub(rf"{REGEXP_PATH_MARKERS}", "_", command)
         c = AntaCommand(command=command, ofmt=outformat)
         await dev.collect(c)
         if not c.collected:
             logger.error("Could not collect commands on device %s: %s", dev.name, c.errors)
             return
         if c.ofmt == "json":
-            outfile = outdir / f"{safe_command}.json"
+            outfile = outdir / f"{safe_command(command)}.json"
             content = json.dumps(c.json_output, indent=2)
         elif c.ofmt == "text":
-            outfile = outdir / f"{safe_command}.log"
+            outfile = outdir / f"{safe_command(command)}.log"
             content = c.text_output
         else:
             logger.error("Command outformat is not in ['json', 'text'] for command '%s'", command)
@@ -83,6 +81,9 @@ async def collect(dev: AntaDevice, command: str, outformat: Literal["json", "tex
     logger.info("Connecting to devices...")
     await inv.connect_inventory()
     devices = inv.get_inventory(established_only=True, tags=tags).devices
+    if not devices:
+        logger.info("No online device found. Exiting")
+        return
     logger.info("Collecting commands from remote devices")
     coros = []
     if "json_format" in commands:
@@ -134,8 +135,8 @@ async def collect(device: AntaDevice) -> None:
                 if not isinstance(device, AsyncEOSDevice):
                     msg = "anta exec collect-tech-support is only supported with AsyncEOSDevice for now."
                     raise UsageError(msg)
-                if device.enable and device._enable_password is not None:  # pylint: disable=protected-access
-                    commands.append({"cmd": "enable", "input": device._enable_password})  # pylint: disable=protected-access
+                if device.enable and device._enable_password is not None:
+                    commands.append({"cmd": "enable", "input": device._enable_password})
                 elif device.enable:
                     commands.append({"cmd": "enable"})
                 commands.extend(
@@ -146,7 +147,7 @@ async def collect(device: AntaDevice) -> None:
                 )
                 logger.warning("Configuring 'aaa authorization exec default local' on device %s", device.name)
                 command = AntaCommand(command="show running-config | include aaa authorization exec default local", ofmt="text")
-                await device._session.cli(commands=commands)  # pylint: disable=protected-access
+                await device._session.cli(commands=commands)
                 logger.info("Configured 'aaa authorization exec default local' on device %s", device.name)
 
             logger.debug("'aaa authorization exec default local' is already configured on device %s", device.name)

anta/cli/get/commands.py

@@ -45,7 +45,6 @@
     default=False,
 )
 def from_cvp(ctx: click.Context, output: Path, host: str, username: str, password: str, container: str | None, *, ignore_cert: bool) -> None:
-    # pylint: disable=too-many-arguments
     """Build ANTA inventory from CloudVision.
 
     NOTE: Only username/password authentication is supported for on-premises CloudVision instances.
@@ -127,7 +126,6 @@ def inventory(inventory: AntaInventory, tags: set[str] | None, *, connected: boo
 @click.command
 @inventory_options
 def tags(inventory: AntaInventory, **kwargs: Any) -> None:
-    # pylint: disable=unused-argument
     """Get list of configured tags in user inventory."""
     tags: set[str] = set()
     for device in inventory.values():

anta/cli/nrfu/__init__.py

@@ -103,7 +103,6 @@ def parse_args(self, ctx: click.Context, args: list[str]) -> list[str]:
     is_flag=True,
     default=False,
 )
-# pylint: disable=too-many-arguments
 def nrfu(
     ctx: click.Context,
     inventory: AntaInventory,

anta/cli/utils.py

@@ -40,7 +40,6 @@ class ExitCode(enum.IntEnum):
 
 
 def parse_tags(ctx: click.Context, param: Option, value: str | None) -> set[str] | None:
-    # pylint: disable=unused-argument
     # ruff: noqa: ARG001
     """Click option callback to parse an ANTA inventory tags."""
     if value is not None:
@@ -207,7 +206,6 @@ def wrapper(
         disable_cache: bool,
         **kwargs: dict[str, Any],
     ) -> Any:
-        # pylint: disable=too-many-arguments
         # If help is invoke somewhere, do not parse inventory
         if ctx.obj.get("_anta_help"):
             return f(*args, inventory=None, **kwargs)
@@ -272,7 +270,6 @@ def wrapper(
         tags: set[str] | None,
         **kwargs: dict[str, Any],
     ) -> Any:
-        # pylint: disable=too-many-arguments
         # If help is invoke somewhere, do not parse inventory
         if ctx.obj.get("_anta_help"):
             return f(*args, tags=tags, **kwargs)

anta/device.py

@@ -106,7 +106,7 @@ def _init_cache(self) -> None:
 
     @property
     def cache_statistics(self) -> dict[str, Any] | None:
-        """Returns the device cache statistics for logging purposes."""
+        """Return the device cache statistics for logging purposes."""
         # Need to ignore pylint no-member as Cache is a proxy class and pylint is not smart enough
         # https://github.com/pylint-dev/pylint/issues/7258
         if self.cache is not None:
@@ -126,6 +126,17 @@ def __rich_repr__(self) -> Iterator[tuple[str, Any]]:
         yield "established", self.established
         yield "disable_cache", self.cache is None
 
+    def __repr__(self) -> str:
+        """Return a printable representation of an AntaDevice."""
+        return (
+            f"AntaDevice({self.name!r}, "
+            f"tags={self.tags!r}, "
+            f"hw_model={self.hw_model!r}, "
+            f"is_online={self.is_online!r}, "
+            f"established={self.established!r}, "
+            f"disable_cache={self.cache is None!r})"
+        )
+
     @abstractmethod
     async def _collect(self, command: AntaCommand, *, collection_id: str | None = None) -> None:
         """Collect device command output.
@@ -244,7 +255,6 @@ class AsyncEOSDevice(AntaDevice):
 
     """
 
-    # pylint: disable=R0913
     def __init__(
         self,
         host: str,
@@ -338,6 +348,22 @@ def __rich_repr__(self) -> Iterator[tuple[str, Any]]:
             yield ("_session", vars(self._session))
             yield ("_ssh_opts", _ssh_opts)
 
+    def __repr__(self) -> str:
+        """Return a printable representation of an AsyncEOSDevice."""
+        return (
+            f"AsyncEOSDevice({self.name!r}, "
+            f"tags={self.tags!r}, "
+            f"hw_model={self.hw_model!r}, "
+            f"is_online={self.is_online!r}, "
+            f"established={self.established!r}, "
+            f"disable_cache={self.cache is None!r}, "
+            f"host={self._session.host!r}, "
+            f"eapi_port={self._session.port!r}, "
+            f"username={self._ssh_opts.username!r}, "
+            f"enable={self.enable!r}, "
+            f"insecure={self._ssh_opts.known_hosts is None!r})"
+        )
+
     @property
     def _keys(self) -> tuple[Any, ...]:
         """Two AsyncEOSDevice objects are equal if the hostname and the port are the same.
@@ -346,7 +372,7 @@ def _keys(self) -> tuple[Any, ...]:
         """
         return (self._session.host, self._session.port)
 
-    async def _collect(self, command: AntaCommand, *, collection_id: str | None = None) -> None:  # noqa: C901  function is too complex - because of many required except blocks #pylint: disable=line-too-long
+    async def _collect(self, command: AntaCommand, *, collection_id: str | None = None) -> None:  # noqa: C901  function is too complex - because of many required except blocks
         """Collect device command output from EOS using aio-eapi.
 
         Supports outformat `json` and `text` as output structure.

anta/inventory/__init__.py

@@ -171,7 +171,6 @@ def _parse_ranges(
             anta_log_exception(e, message, logger)
             raise InventoryIncorrectSchemaError(message) from e
 
-    # pylint: disable=too-many-arguments
     @staticmethod
     def parse(
         filename: str | Path,