Merge branch 'main' into feat/AVT

.github/workflows/code-testing.yml

@@ -119,6 +119,23 @@ jobs:
         run: pip install tox tox-gh-actions
       - name: "Run pytest via tox for ${{ matrix.python }}"
         run: tox
+  test-python-windows:
+    name: Pytest on 3.12 for windows
+    runs-on: windows-2022
+    needs: [lint-python, type-python]
+    env:
+      # Required to prevent asyncssh to fail.
+      USERNAME: WindowsUser
+    steps:
+      - uses: actions/checkout@v4
+      - name: Setup Python
+        uses: actions/setup-python@v5
+        with:
+          python-version: 3.12
+      - name: Install dependencies
+        run: pip install tox tox-gh-actions
+      - name: Run pytest via tox for 3.12 on Windows
+        run: tox
   test-documentation:
     name: Build offline documentation for testing
     runs-on: ubuntu-20.04

.pre-commit-config.yaml

@@ -46,7 +46,7 @@ repos:
           - '<!--|  ~|  -->'
 
   - repo: https://github.com/astral-sh/ruff-pre-commit
-    rev: v0.8.2
+    rev: v0.8.3
     hooks:
     - id: ruff
       name: Run Ruff linter

anta/constants.py

@@ -17,3 +17,12 @@
     - [Summary Totals Per Category](#summary-totals-per-category)
   - [Test Results](#test-results)"""
 """Table of Contents for the Markdown report."""
+
+KNOWN_EOS_ERRORS = [
+    r"BGP inactive",
+    r"VRF '.*' is not active",
+    r".* does not support IP",
+    r"IS-IS (.*) is disabled because: .*",
+    r"No source interface .*",
+]
+"""List of known EOS errors that should set a test status to 'failure' with the error message."""

anta/device.py

@@ -255,7 +255,7 @@ class AsyncEOSDevice(AntaDevice):
 
     """
 
-    def __init__(
+    def __init__(  # noqa: PLR0913
         self,
         host: str,
         username: str,
@@ -372,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
+    async def _collect(self, command: AntaCommand, *, collection_id: str | None = None) -> None:
         """Collect device command output from EOS using aio-eapi.
 
         Supports outformat `json` and `text` as output structure.
@@ -409,15 +409,7 @@ async def _collect(self, command: AntaCommand, *, collection_id: str | None = No
             command.output = response[-1]
         except asynceapi.EapiCommandError as e:
             # This block catches exceptions related to EOS issuing an error.
-            command.errors = e.errors
-            if command.requires_privileges:
-                logger.error(
-                    "Command '%s' requires privileged mode on %s. Verify user permissions and if the `enable` option is required.", command.command, self.name
-                )
-            if command.supported:
-                logger.error("Command '%s' failed on %s: %s", command.command, self.name, e.errors[0] if len(e.errors) == 1 else e.errors)
-            else:
-                logger.debug("Command '%s' is not supported on '%s' (%s)", command.command, self.name, self.hw_model)
+            self._log_eapi_command_error(command, e)
         except TimeoutException as e:
             # This block catches Timeout exceptions.
             command.errors = [exc_to_str(e)]
@@ -446,6 +438,18 @@ async def _collect(self, command: AntaCommand, *, collection_id: str | None = No
             anta_log_exception(e, f"An error occurred while issuing an eAPI request to {self.name}", logger)
         logger.debug("%s: %s", self.name, command)
 
+    def _log_eapi_command_error(self, command: AntaCommand, e: asynceapi.EapiCommandError) -> None:
+        """Appropriately log the eapi command error."""
+        command.errors = e.errors
+        if command.requires_privileges:
+            logger.error("Command '%s' requires privileged mode on %s. Verify user permissions and if the `enable` option is required.", command.command, self.name)
+        if not command.supported:
+            logger.debug("Command '%s' is not supported on '%s' (%s)", command.command, self.name, self.hw_model)
+        elif command.returned_known_eos_error:
+            logger.debug("Command '%s' returned a known error '%s': %s", command.command, self.name, command.errors)
+        else:
+            logger.error("Command '%s' failed on %s: %s", command.command, self.name, e.errors[0] if len(e.errors) == 1 else e.errors)
+
     async def refresh(self) -> None:
         """Update attributes of an AsyncEOSDevice instance.
 

anta/models.py

@@ -16,6 +16,7 @@
 from pydantic import BaseModel, ConfigDict, ValidationError, create_model
 
 from anta import GITHUB_SUGGESTION
+from anta.constants import KNOWN_EOS_ERRORS
 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 AntaTestStatus, TestResult
@@ -240,18 +241,37 @@ def requires_privileges(self) -> bool:
 
     @property
     def supported(self) -> bool:
-        """Return True if the command is supported on the device hardware platform, False otherwise.
+        """Indicates if the command is supported on the device.
+
+        Returns
+        -------
+        bool
+            True if the command is supported on the device hardware platform, False otherwise.
 
         Raises
         ------
+        RuntimeError
+            If the command has not been collected and has not returned an error.
+            AntaDevice.collect() must be called before this property.
+        """
+        if not self.collected and not self.error:
+            msg = f"Command '{self.command}' has not been collected and has not returned an error. Call AntaDevice.collect()."
+
+            raise RuntimeError(msg)
+        return all("not supported on this hardware platform" not in e for e in self.errors)
+
+    @property
+    def returned_known_eos_error(self) -> bool:
+        """Return True if the command returned a known_eos_error on the device, False otherwise.
+
         RuntimeError
             If the command has not been collected and has not returned an error.
             AntaDevice.collect() must be called before this property.
         """
         if not self.collected and not self.error:
             msg = f"Command '{self.command}' has not been collected and has not returned an error. Call AntaDevice.collect()."
             raise RuntimeError(msg)
-        return not any("not supported on this hardware platform" in e for e in self.errors)
+        return any(any(re.match(pattern, e) for e in self.errors) for pattern in KNOWN_EOS_ERRORS)
 
 
 class AntaTemplateRenderError(RuntimeError):
@@ -630,14 +650,9 @@ async def wrapper(
                     AntaTest.update_progress()
                     return self.result
 
-                if cmds := self.failed_commands:
-                    unsupported_commands = [f"'{c.command}' is not supported on {self.device.hw_model}" for c in cmds if not c.supported]
-                    if unsupported_commands:
-                        msg = f"Test {self.name} has been skipped because it is not supported on {self.device.hw_model}: {GITHUB_SUGGESTION}"
-                        self.logger.warning(msg)
-                        self.result.is_skipped("\n".join(unsupported_commands))
-                    else:
-                        self.result.is_error(message="\n".join([f"{c.command} has failed: {', '.join(c.errors)}" for c in cmds]))
+                if self.failed_commands:
+                    self._handle_failed_commands()
+
                     AntaTest.update_progress()
                     return self.result
 
@@ -657,6 +672,28 @@ async def wrapper(
 
         return wrapper
 
+    def _handle_failed_commands(self) -> None:
+        """Handle failed commands inside a test.
+
+        There can be 3 types:
+        * unsupported on hardware commands which set the test status to 'skipped'
+        * known EOS error which set the test status to 'failure'
+        * unknown failure which set the test status to 'error'
+        """
+        cmds = self.failed_commands
+        unsupported_commands = [f"'{c.command}' is not supported on {self.device.hw_model}" for c in cmds if not c.supported]
+        if unsupported_commands:
+            msg = f"Test {self.name} has been skipped because it is not supported on {self.device.hw_model}: {GITHUB_SUGGESTION}"
+            self.logger.warning(msg)
+            self.result.is_skipped("\n".join(unsupported_commands))
+            return
+        returned_known_eos_error = [f"'{c.command}' failed on {self.device.name}: {', '.join(c.errors)}" for c in cmds if c.returned_known_eos_error]
+        if returned_known_eos_error:
+            self.result.is_failure("\n".join(returned_known_eos_error))
+            return
+
+        self.result.is_error(message="\n".join([f"{c.command} has failed: {', '.join(c.errors)}" for c in cmds]))
+
     @classmethod
     def update_progress(cls: type[AntaTest]) -> None:
         """Update progress bar for all AntaTest objects if it exists."""

anta/reporter/csv_reporter.py

@@ -107,7 +107,7 @@ def generate(cls, results: ResultManager, csv_filename: pathlib.Path) -> None:
         ]
 
         try:
-            with csv_filename.open(mode="w", encoding="utf-8") as csvfile:
+            with csv_filename.open(mode="w", encoding="utf-8", newline="") as csvfile:
                 csvwriter = csv.writer(
                     csvfile,
                     delimiter=",",

anta/result_manager/__init__.py

@@ -9,6 +9,7 @@
 from collections import defaultdict
 from functools import cached_property
 from itertools import chain
+from typing import Any
 
 from anta.result_manager.models import AntaTestStatus, TestResult
 
@@ -89,6 +90,10 @@ def __init__(self) -> None:
         If the status of the added test is error, the status is untouched and the
         error_status is set to True.
         """
+        self.reset()
+
+    def reset(self) -> None:
+        """Create or reset the attributes of the ResultManager instance."""
         self._result_entries: list[TestResult] = []
         self.status: AntaTestStatus = AntaTestStatus.UNSET
         self.error_status = False
@@ -122,10 +127,15 @@ def results(self, value: list[TestResult]) -> None:
         for result in value:
             self.add(result)
 
+    @property
+    def dump(self) -> list[dict[str, Any]]:
+        """Get a list of dictionary of the results."""
+        return [result.model_dump() for result in self._result_entries]
+
     @property
     def json(self) -> str:
         """Get a JSON representation of the results."""
-        return json.dumps([result.model_dump() for result in self._result_entries], indent=4)
+        return json.dumps(self.dump, indent=4)
 
     @property
     def sorted_category_stats(self) -> dict[str, CategoryStats]:

anta/runner.py

@@ -8,7 +8,7 @@
 import asyncio
 import logging
 import os
-import resource
+import sys
 from collections import defaultdict
 from typing import TYPE_CHECKING, Any
 
@@ -26,35 +26,38 @@
     from anta.result_manager import ResultManager
     from anta.result_manager.models import TestResult
 
-logger = logging.getLogger(__name__)
+if os.name == "posix":
+    import resource
 
-DEFAULT_NOFILE = 16384
+    DEFAULT_NOFILE = 16384
 
+    def adjust_rlimit_nofile() -> tuple[int, int]:
+        """Adjust the maximum number of open file descriptors for the ANTA process.
 
-def adjust_rlimit_nofile() -> tuple[int, int]:
-    """Adjust the maximum number of open file descriptors for the ANTA process.
+        The limit is set to the lower of the current hard limit and the value of the ANTA_NOFILE environment variable.
 
-    The limit is set to the lower of the current hard limit and the value of the ANTA_NOFILE environment variable.
+        If the `ANTA_NOFILE` environment variable is not set or is invalid, `DEFAULT_NOFILE` is used.
 
-    If the `ANTA_NOFILE` environment variable is not set or is invalid, `DEFAULT_NOFILE` is used.
+        Returns
+        -------
+        tuple[int, int]
+            The new soft and hard limits for open file descriptors.
+        """
+        try:
+            nofile = int(os.environ.get("ANTA_NOFILE", DEFAULT_NOFILE))
+        except ValueError as exception:
+            logger.warning("The ANTA_NOFILE environment variable value is invalid: %s\nDefault to %s.", exc_to_str(exception), DEFAULT_NOFILE)
+            nofile = DEFAULT_NOFILE
+
+        limits = resource.getrlimit(resource.RLIMIT_NOFILE)
+        logger.debug("Initial limit numbers for open file descriptors for the current ANTA process: Soft Limit: %s | Hard Limit: %s", limits[0], limits[1])
+        nofile = min(limits[1], nofile)
+        logger.debug("Setting soft limit for open file descriptors for the current ANTA process to %s", nofile)
+        resource.setrlimit(resource.RLIMIT_NOFILE, (nofile, limits[1]))
+        return resource.getrlimit(resource.RLIMIT_NOFILE)
 
-    Returns
-    -------
-    tuple[int, int]
-        The new soft and hard limits for open file descriptors.
-    """
-    try:
-        nofile = int(os.environ.get("ANTA_NOFILE", DEFAULT_NOFILE))
-    except ValueError as exception:
-        logger.warning("The ANTA_NOFILE environment variable value is invalid: %s\nDefault to %s.", exc_to_str(exception), DEFAULT_NOFILE)
-        nofile = DEFAULT_NOFILE
 
-    limits = resource.getrlimit(resource.RLIMIT_NOFILE)
-    logger.debug("Initial limit numbers for open file descriptors for the current ANTA process: Soft Limit: %s | Hard Limit: %s", limits[0], limits[1])
-    nofile = min(limits[1], nofile)
-    logger.debug("Setting soft limit for open file descriptors for the current ANTA process to %s", nofile)
-    resource.setrlimit(resource.RLIMIT_NOFILE, (nofile, limits[1]))
-    return resource.getrlimit(resource.RLIMIT_NOFILE)
+logger = logging.getLogger(__name__)
 
 
 def log_cache_statistics(devices: list[AntaDevice]) -> None:
@@ -167,7 +170,8 @@ def prepare_tests(
 
     if total_test_count == 0:
         msg = (
-            f"There are no tests{f' matching the tags {tags} ' if tags else ' '}to run in the current test catalog and device inventory, please verify your inputs."
+            f"There are no tests{f' matching the tags {tags} ' if tags else ' '}to run in the current "
+            "test catalog and device inventory, please verify your inputs."
         )
         logger.warning(msg)
         return None
@@ -211,7 +215,7 @@ def get_coroutines(selected_tests: defaultdict[AntaDevice, set[AntaTestDefinitio
 
 
 @cprofile()
-async def main(  # noqa: PLR0913
+async def main(
     manager: ResultManager,
     inventory: AntaInventory,
     catalog: AntaCatalog,
@@ -246,9 +250,6 @@ async def main(  # noqa: PLR0913
     dry_run
         Build the list of coroutine to run and stop before test execution.
     """
-    # Adjust the maximum number of open file descriptors for the ANTA process
-    limits = adjust_rlimit_nofile()
-
     if not catalog.tests:
         logger.info("The list of tests is empty, exiting")
         return
@@ -269,10 +270,19 @@ async def main(  # noqa: PLR0913
             "--- ANTA NRFU Run Information ---\n"
             f"Number of devices: {len(inventory)} ({len(selected_inventory)} established)\n"
             f"Total number of selected tests: {final_tests_count}\n"
-            f"Maximum number of open file descriptors for the current ANTA process: {limits[0]}\n"
-            "---------------------------------"
         )
 
+        if os.name == "posix":
+            # Adjust the maximum number of open file descriptors for the ANTA process
+            limits = adjust_rlimit_nofile()
+            run_info += f"Maximum number of open file descriptors for the current ANTA process: {limits[0]}\n"
+        else:
+            # Running on non-Posix system, cannot manage the resource.
+            limits = (sys.maxsize, sys.maxsize)
+            run_info += "Running on a non-POSIX system, cannot adjust the maximum number of file descriptors.\n"
+
+        run_info += "---------------------------------"
+
         logger.info(run_info)
 
         if final_tests_count > limits[0]:

anta/tests/connectivity.py

@@ -56,13 +56,12 @@ class Input(AntaTest.Input):
 
     def render(self, template: AntaTemplate) -> list[AntaCommand]:
         """Render the template for each host in the input list."""
-        commands = []
-        for host in self.inputs.hosts:
-            # df_bit includes leading space when enabled, empty string when disabled
-            df_bit = " df-bit" if host.df_bit else ""
-            command = template.render(destination=host.destination, source=host.source, vrf=host.vrf, repeat=host.repeat, size=host.size, df_bit=df_bit)
-            commands.append(command)
-        return commands
+        return [
+            template.render(
+                destination=host.destination, source=host.source, vrf=host.vrf, repeat=host.repeat, size=host.size, df_bit=" df-bit" if host.df_bit else ""
+            )
+            for host in self.inputs.hosts
+        ]
 
     @AntaTest.anta_test
     def test(self) -> None: