Merge branch 'main' into mcsClientMounts

.github/workflows/code-testing.yml

@@ -46,7 +46,7 @@ jobs:
     runs-on: ubuntu-20.04
     strategy:
       matrix:
-        python-version: ["3.9", "3.10", "3.11", "3.12"]
+        python-version: ["3.9", "3.10", "3.11", "3.12", "3.13"]
     needs: file-changes
     steps:
       - uses: actions/checkout@v4
@@ -108,7 +108,7 @@ jobs:
     needs: [lint-python, type-python]
     strategy:
       matrix:
-        python: ["3.9", "3.10", "3.11", "3.12"]
+        python: ["3.9", "3.10", "3.11", "3.12", "3.13"]
     steps:
       - uses: actions/checkout@v4
       - name: Setup Python
@@ -149,4 +149,4 @@ jobs:
         uses: CodSpeedHQ/action@v3
         with:
           token: ${{ secrets.CODSPEED_TOKEN }}
-          run: pytest --codspeed --no-cov --log-cli-level INFO tests/benchmark
+          run: pytest --codspeed --no-cov --log-cli-level INFO tests/benchmark

.pre-commit-config.yaml

@@ -46,7 +46,7 @@ repos:
           - '<!--|  ~|  -->'
 
   - repo: https://github.com/astral-sh/ruff-pre-commit
-    rev: v0.7.0
+    rev: v0.7.2
     hooks:
     - id: ruff
       name: Run Ruff linter
@@ -85,7 +85,7 @@ repos:
         types: [text]
 
   - repo: https://github.com/pre-commit/mirrors-mypy
-    rev: v1.12.1
+    rev: v1.13.0
     hooks:
       - id: mypy
         name: Check typing with mypy

anta/cli/exec/commands.py

@@ -84,7 +84,10 @@ def snapshot(inventory: AntaInventory, tags: set[str] | None, commands_list: Pat
 )
 @click.option(
     "--configure",
-    help="Ensure devices have 'aaa authorization exec default local' configured (required for SCP on EOS). THIS WILL CHANGE THE CONFIGURATION OF YOUR NETWORK.",
+    help=(
+        "[DEPRECATED] Ensure devices have 'aaa authorization exec default local' configured (required for SCP on EOS). "
+        "THIS WILL CHANGE THE CONFIGURATION OF YOUR NETWORK."
+    ),
     default=False,
     is_flag=True,
     show_default=True,

anta/cli/exec/utils.py

@@ -128,6 +128,13 @@ async def collect(device: AntaDevice) -> None:
                     logger.error("Unable to collect tech-support on %s: configuration 'aaa authorization exec default local' is not present", device.name)
                     return
 
+                # TODO: ANTA 2.0.0
+                msg = (
+                    "[DEPRECATED] Using '--configure' for collecting show-techs is deprecated and will be removed in ANTA 2.0.0. "
+                    "Please add the required configuration on your devices before running this command from ANTA."
+                )
+                logger.warning(msg)
+
                 commands = []
                 # TODO: @mtache - add `config` field to `AntaCommand` object to handle this use case.
                 # Otherwise mypy complains about enable as it is only implemented for AsyncEOSDevice

anta/cli/nrfu/utils.py

@@ -116,8 +116,12 @@ def print_text(ctx: click.Context) -> None:
     """Print results as simple text."""
     console.print()
     for test in _get_result_manager(ctx).results:
-        message = f" ({test.messages[0]!s})" if len(test.messages) > 0 else ""
-        console.print(f"{test.name} :: {test.test} :: [{test.result}]{test.result.upper()}[/{test.result}]{message}", highlight=False)
+        if len(test.messages) <= 1:
+            message = test.messages[0] if len(test.messages) == 1 else ""
+            console.print(f"{test.name} :: {test.test} :: [{test.result}]{test.result.upper()}[/{test.result}]({message})", highlight=False)
+        else:  # len(test.messages) > 1
+            console.print(f"{test.name} :: {test.test} :: [{test.result}]{test.result.upper()}[/{test.result}]", highlight=False)
+            console.print("\n".join(f"    {message}" for message in test.messages), highlight=False)
 
 
 def print_jinja(results: ResultManager, template: pathlib.Path, output: pathlib.Path | None = None) -> None:

anta/input_models/__init__.py

@@ -0,0 +1,4 @@
+# Copyright (c) 2023-2024 Arista Networks, Inc.
+# Use of this source code is governed by the Apache License 2.0
+# that can be found in the LICENSE file.
+"""Package related to all ANTA tests input models."""

anta/input_models/connectivity.py

@@ -0,0 +1,41 @@
+# Copyright (c) 2023-2024 Arista Networks, Inc.
+# Use of this source code is governed by the Apache License 2.0
+# that can be found in the LICENSE file.
+"""Module containing input models for connectivity tests."""
+
+from __future__ import annotations
+
+from ipaddress import IPv4Address
+
+from pydantic import BaseModel, ConfigDict
+
+from anta.custom_types import Interface
+
+
+class Host(BaseModel):
+    """Model for a remote host to ping."""
+
+    model_config = ConfigDict(extra="forbid")
+    destination: IPv4Address
+    """IPv4 address to ping."""
+    source: IPv4Address | Interface
+    """IPv4 address source IP or egress interface to use."""
+    vrf: str = "default"
+    """VRF context. Defaults to `default`."""
+    repeat: int = 2
+    """Number of ping repetition. Defaults to 2."""
+    size: int = 100
+    """Specify datagram size. Defaults to 100."""
+    df_bit: bool = False
+    """Enable do not fragment bit in IP header. Defaults to False."""
+
+    def __str__(self) -> str:
+        """Return a human-readable string representation of the Host for reporting.
+
+        Examples
+        --------
+        Host 10.1.1.1 (src: 10.2.2.2, vrf: mgmt, size: 100B, repeat: 2)
+
+        """
+        df_status = ", df-bit: enabled" if self.df_bit else ""
+        return f"Host {self.destination} (src: {self.source}, vrf: {self.vrf}, size: {self.size}B, repeat: {self.repeat}{df_status})"

anta/input_models/interfaces.py

@@ -0,0 +1,23 @@
+# Copyright (c) 2023-2024 Arista Networks, Inc.
+# Use of this source code is governed by the Apache License 2.0
+# that can be found in the LICENSE file.
+"""Module containing input models for interface tests."""
+
+from __future__ import annotations
+
+from typing import Literal
+
+from pydantic import BaseModel
+
+from anta.custom_types import Interface
+
+
+class InterfaceState(BaseModel):
+    """Model for an interface state."""
+
+    name: Interface
+    """Interface to validate."""
+    status: Literal["up", "down", "adminDown"]
+    """Expected status of the interface."""
+    line_protocol_status: Literal["up", "down", "testing", "unknown", "dormant", "notPresent", "lowerLayerDown"] | None = None
+    """Expected line protocol status of the interface."""

anta/input_models/system.py

@@ -0,0 +1,31 @@
+# Copyright (c) 2023-2024 Arista Networks, Inc.
+# Use of this source code is governed by the Apache License 2.0
+# that can be found in the LICENSE file.
+"""Module containing input models for system tests."""
+
+from __future__ import annotations
+
+from ipaddress import IPv4Address
+
+from pydantic import BaseModel, ConfigDict, Field
+
+from anta.custom_types import Hostname
+
+
+class NTPServer(BaseModel):
+    """Model for a NTP server."""
+
+    model_config = ConfigDict(extra="forbid")
+    server_address: Hostname | IPv4Address
+    """The NTP server address as an IPv4 address or hostname. The NTP server name defined in the running configuration
+    of the device may change during DNS resolution, which is not handled in ANTA. Please provide the DNS-resolved server name.
+    For example, 'ntp.example.com' in the configuration might resolve to 'ntp3.example.com' in the device output."""
+    preferred: bool = False
+    """Optional preferred for NTP server. If not provided, it defaults to `False`."""
+    stratum: int = Field(ge=0, le=16)
+    """NTP stratum level (0 to 15) where 0 is the reference clock and 16 indicates unsynchronized.
+    Values should be between 0 and 15 for valid synchronization and 16 represents an out-of-sync state."""
+
+    def __str__(self) -> str:
+        """Representation of the NTPServer model."""
+        return f"{self.server_address} (Preferred: {self.preferred}, Stratum: {self.stratum})"

anta/models.py

@@ -284,8 +284,7 @@ class AntaTest(ABC):
     The following is an example of an AntaTest subclass implementation:
         ```python
             class VerifyReachability(AntaTest):
-                name = "VerifyReachability"
-                description = "Test the network reachability to one or many destination IP(s)."
+                '''Test the network reachability to one or many destination IP(s).'''
                 categories = ["connectivity"]
                 commands = [AntaTemplate(template="ping vrf {vrf} {dst} source {src} repeat 2")]
 
@@ -326,12 +325,17 @@ def test(self) -> None:
         Python logger for this test instance.
     """
 
-    # Mandatory class attributes
-    # TODO: find a way to tell mypy these are mandatory for child classes - maybe Protocol
+    # Optional class attributes
     name: ClassVar[str]
     description: ClassVar[str]
+
+    # Mandatory class attributes
+    # TODO: find a way to tell mypy these are mandatory for child classes
+    #       follow this https://discuss.python.org/t/provide-a-canonical-way-to-declare-an-abstract-class-variable/69416
+    #       for now only enforced at runtime with __init_subclass__
     categories: ClassVar[list[str]]
     commands: ClassVar[list[AntaTemplate | AntaCommand]]
+
     # Class attributes to handle the progress bar of ANTA CLI
     progress: Progress | None = None
     nrfu_task: TaskID | None = None
@@ -505,12 +509,19 @@ def save_commands_data(self, eos_data: list[dict[str, Any] | str]) -> None:
             self.instance_commands[index].output = data
 
     def __init_subclass__(cls) -> None:
-        """Verify that the mandatory class attributes are defined."""
-        mandatory_attributes = ["name", "description", "categories", "commands"]
-        for attr in mandatory_attributes:
-            if not hasattr(cls, attr):
-                msg = f"Class {cls.__module__}.{cls.__name__} is missing required class attribute {attr}"
-                raise NotImplementedError(msg)
+        """Verify that the mandatory class attributes are defined and set name and description if not set."""
+        mandatory_attributes = ["categories", "commands"]
+        if missing_attrs := [attr for attr in mandatory_attributes if not hasattr(cls, attr)]:
+            msg = f"Class {cls.__module__}.{cls.__name__} is missing required class attribute(s): {', '.join(missing_attrs)}"
+            raise AttributeError(msg)
+
+        cls.name = getattr(cls, "name", cls.__name__)
+        if not hasattr(cls, "description"):
+            if not cls.__doc__ or cls.__doc__.strip() == "":
+                # No doctsring or empty doctsring - raise
+                msg = f"Cannot set the description for class {cls.name}, either set it in the class definition or add a docstring to the class."
+                raise AttributeError(msg)
+            cls.description = cls.__doc__.split(sep="\n", maxsplit=1)[0]
 
     @property
     def module(self) -> str:

anta/reporter/csv_reporter.py

@@ -58,8 +58,7 @@ def split_list_to_txt_list(cls, usr_list: list[str], delimiter: str = " - ") ->
 
     @classmethod
     def convert_to_list(cls, result: TestResult) -> list[str]:
-        """
-        Convert a TestResult into a list of string for creating file content.
+        """Convert a TestResult into a list of string for creating file content.
 
         Parameters
         ----------

anta/runner.py

@@ -146,20 +146,26 @@ def prepare_tests(
     # Using a set to avoid inserting duplicate tests
     device_to_tests: defaultdict[AntaDevice, set[AntaTestDefinition]] = defaultdict(set)
 
+    total_test_count = 0
+
     # Create the device to tests mapping from the tags
     for device in inventory.devices:
         if tags:
-            if not any(tag in device.tags for tag in tags):
+            # If there are CLI tags, execute tests with matching tags for this device
+            if not (matching_tags := tags.intersection(device.tags)):
                 # The device does not have any selected tag, skipping
                 continue
+            device_to_tests[device].update(catalog.get_tests_by_tags(matching_tags))
         else:
             # If there is no CLI tags, execute all tests that do not have any tags
             device_to_tests[device].update(catalog.tag_to_tests[None])
 
-        # Add the tests with matching tags from device tags
-        device_to_tests[device].update(catalog.get_tests_by_tags(device.tags))
+            # Then add the tests with matching tags from device tags
+            device_to_tests[device].update(catalog.get_tests_by_tags(device.tags))
+
+        total_test_count += len(device_to_tests[device])
 
-    if len(device_to_tests.values()) == 0:
+    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."
         )

anta/tests/aaa.py

@@ -35,8 +35,6 @@ class VerifyTacacsSourceIntf(AntaTest):
     ```
     """
 
-    name = "VerifyTacacsSourceIntf"
-    description = "Verifies TACACS source-interface for a specified VRF."
     categories: ClassVar[list[str]] = ["aaa"]
     commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show tacacs", revision=1)]
 
@@ -81,8 +79,6 @@ class VerifyTacacsServers(AntaTest):
     ```
     """
 
-    name = "VerifyTacacsServers"
-    description = "Verifies TACACS servers are configured for a specified VRF."
     categories: ClassVar[list[str]] = ["aaa"]
     commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show tacacs", revision=1)]
 
@@ -134,8 +130,6 @@ class VerifyTacacsServerGroups(AntaTest):
     ```
     """
 
-    name = "VerifyTacacsServerGroups"
-    description = "Verifies if the provided TACACS server group(s) are configured."
     categories: ClassVar[list[str]] = ["aaa"]
     commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show tacacs", revision=1)]
 
@@ -184,8 +178,6 @@ class VerifyAuthenMethods(AntaTest):
     ```
     """
 
-    name = "VerifyAuthenMethods"
-    description = "Verifies the AAA authentication method lists for different authentication types (login, enable, dot1x)."
     categories: ClassVar[list[str]] = ["aaa"]
     commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show aaa methods authentication", revision=1)]
 
@@ -245,8 +237,6 @@ class VerifyAuthzMethods(AntaTest):
     ```
     """
 
-    name = "VerifyAuthzMethods"
-    description = "Verifies the AAA authorization method lists for different authorization types (commands, exec)."
     categories: ClassVar[list[str]] = ["aaa"]
     commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show aaa methods authorization", revision=1)]
 
@@ -301,8 +291,6 @@ class VerifyAcctDefaultMethods(AntaTest):
     ```
     """
 
-    name = "VerifyAcctDefaultMethods"
-    description = "Verifies the AAA accounting default method lists for different accounting types (system, exec, commands, dot1x)."
     categories: ClassVar[list[str]] = ["aaa"]
     commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show aaa methods accounting", revision=1)]
 
@@ -364,8 +352,6 @@ class VerifyAcctConsoleMethods(AntaTest):
     ```
     """
 
-    name = "VerifyAcctConsoleMethods"
-    description = "Verifies the AAA accounting console method lists for different accounting types (system, exec, commands, dot1x)."
     categories: ClassVar[list[str]] = ["aaa"]
     commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show aaa methods accounting", revision=1)]