fix(anta)!: Make multiple tags filtering work properly (and fall pytest cleaning)

.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

@@ -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 ''}"

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,

anta/models.py

@@ -18,7 +18,7 @@
 from anta import GITHUB_SUGGESTION
 from anta.custom_types import REGEXP_EOS_BLACKLIST_CMDS, Revision
 from anta.logger import anta_log_exception, exc_to_str
-from anta.result_manager.models import TestResult
+from anta.result_manager.models import AntaTestStatus, TestResult
 
 if TYPE_CHECKING:
     from collections.abc import Coroutine
@@ -71,7 +71,6 @@ def __init__(
         *,
         use_cache: bool = True,
     ) -> None:
-        # pylint: disable=too-many-arguments
         self.template = template
         self.version = version
         self.revision = revision
@@ -430,7 +429,7 @@ def __init__(
             description=self.description,
         )
         self._init_inputs(inputs)
-        if self.result.result == "unset":
+        if self.result.result == AntaTestStatus.UNSET:
             self._init_commands(eos_data)
 
     def _init_inputs(self, inputs: dict[str, Any] | AntaTest.Input | None) -> None:
@@ -481,7 +480,7 @@ def _init_commands(self, eos_data: list[dict[Any, Any] | str] | None) -> None:
                     except NotImplementedError as e:
                         self.result.is_error(message=e.args[0])
                         return
-                    except Exception as e:  # pylint: disable=broad-exception-caught
+                    except Exception as e:  # noqa: BLE001
                         # render() is user-defined code.
                         # We need to catch everything if we want the AntaTest object
                         # to live until the reporting
@@ -559,7 +558,7 @@ async def collect(self) -> None:
         try:
             if self.blocked is False:
                 await self.device.collect_commands(self.instance_commands, collection_id=self.name)
-        except Exception as e:  # pylint: disable=broad-exception-caught
+        except Exception as e:  # noqa: BLE001
             # device._collect() is user-defined code.
             # We need to catch everything if we want the AntaTest object
             # to live until the reporting
@@ -631,7 +630,7 @@ async def wrapper(
 
             try:
                 function(self, **kwargs)
-            except Exception as e:  # pylint: disable=broad-exception-caught
+            except Exception as e:  # noqa: BLE001
                 # test() is user-defined code.
                 # We need to catch everything if we want the AntaTest object
                 # to live until the reporting