Merge branch 'aristanetworks:main' into check-connection-with-head

anta/catalog.py

@@ -298,13 +298,11 @@ def __init__(
 
         self.indexes_built: bool
         self.tag_to_tests: defaultdict[str | None, set[AntaTestDefinition]]
-        self._tests_without_tags: set[AntaTestDefinition]
         self._init_indexes()
 
     def _init_indexes(self) -> None:
         """Init indexes related variables."""
         self.tag_to_tests = defaultdict(set)
-        self._tests_without_tags = set()
         self.indexes_built = False
 
     @property
@@ -486,11 +484,7 @@ def build_indexes(self, filtered_tests: set[str] | None = None) -> None:
 
         If a `filtered_tests` set is provided, only the tests in this set will be indexed.
 
-        This method populates two attributes:
-
-        - tag_to_tests: A dictionary mapping each tag to a set of tests that contain it.
-
-        - _tests_without_tags: A set of tests that do not have any tags.
+        This method populates the tag_to_tests attribute, which is a dictionary mapping tags to sets of tests.
 
         Once the indexes are built, the `indexes_built` attribute is set to True.
         """
@@ -504,9 +498,8 @@ def build_indexes(self, filtered_tests: set[str] | None = None) -> None:
                 for tag in test_tags:
                     self.tag_to_tests[tag].add(test)
             else:
-                self._tests_without_tags.add(test)
+                self.tag_to_tests[None].add(test)
 
-        self.tag_to_tests[None] = self._tests_without_tags
         self.indexes_built = True
 
     def clear_indexes(self) -> None:

anta/cli/_main.py

@@ -25,7 +25,8 @@
 
 @click.group(cls=AliasedGroup)
 @click.pass_context
-@click.version_option(__version__)
+@click.help_option(allow_from_autoenv=False)
+@click.version_option(__version__, allow_from_autoenv=False)
 @click.option(
     "--log-file",
     help="Send the logs to a file. If logging level is DEBUG, only INFO or higher will be sent to stdout.",

anta/custom_types.py

@@ -204,3 +204,7 @@ def validate_regex(value: str) -> str:
 ]
 BgpUpdateError = Literal["inUpdErrWithdraw", "inUpdErrIgnore", "inUpdErrDisableAfiSafi", "disabledAfiSafi", "lastUpdErrTime"]
 BfdProtocol = Literal["bgp", "isis", "lag", "ospf", "ospfv3", "pim", "route-input", "static-bfd", "static-route", "vrrp", "vxlan"]
+SnmpPdu = Literal["inGetPdus", "inGetNextPdus", "inSetPdus", "outGetResponsePdus", "outTrapPdus"]
+SnmpErrorCounter = Literal[
+    "inVersionErrs", "inBadCommunityNames", "inBadCommunityUses", "inParseErrs", "outTooBigErrs", "outNoSuchNameErrs", "outBadValueErrs", "outGeneralErrs"
+]

anta/reporter/__init__.py

@@ -105,7 +105,7 @@ def _color_result(self, status: AntaTestStatus) -> str:
     def report_all(self, manager: ResultManager, title: str = "All tests results") -> Table:
         """Create a table report with all tests for one or all devices.
 
-        Create table with full output: Host / Test / Status / Message
+        Create table with full output: Device | Test Name | Test Status | Message(s) | Test description | Test category
 
         Parameters
         ----------
@@ -141,7 +141,8 @@ def report_summary_tests(
     ) -> Table:
         """Create a table report with result aggregated per test.
 
-        Create table with full output: Test | Number of success | Number of failure | Number of error | List of nodes in error or failure
+        Create table with full output:
+        Test Name | # of success | # of skipped | # of failure | # of errors | List of failed or error nodes
 
         Parameters
         ----------
@@ -187,7 +188,7 @@ def report_summary_devices(
     ) -> Table:
         """Create a table report with result aggregated per device.
 
-        Create table with full output: Host | Number of success | Number of failure | Number of error | List of nodes in error or failure
+        Create table with full output: Device | # of success | # of skipped | # of failure | # of errors | List of failed or error test cases
 
         Parameters
         ----------

anta/reporter/md_reporter.py

@@ -112,8 +112,8 @@ def generate_heading_name(self) -> str:
 
         Example
         -------
-        - `ANTAReport` will become ANTA Report.
-        - `TestResultsSummary` will become Test Results Summary.
+        - `ANTAReport` will become `ANTA Report`.
+        - `TestResultsSummary` will become `Test Results Summary`.
         """
         class_name = self.__class__.__name__
 
@@ -153,7 +153,7 @@ def write_heading(self, heading_level: int) -> None:
 
         Example
         -------
-        ## Test Results Summary
+        `## Test Results Summary`
         """
         # Ensure the heading level is within the valid range of 1 to 6
         heading_level = max(1, min(heading_level, 6))

anta/runner.py

@@ -146,8 +146,7 @@ def prepare_tests(
     # Using a set to avoid inserting duplicate tests
     device_to_tests: defaultdict[AntaDevice, set[AntaTestDefinition]] = defaultdict(set)
 
-    # Create AntaTestRunner tuples from the tags
-    final_tests_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):
@@ -160,8 +159,6 @@ def prepare_tests(
         # Add the tests with matching tags from device tags
         device_to_tests[device].update(catalog.get_tests_by_tags(device.tags))
 
-        final_tests_count += len(device_to_tests[device])
-
     if len(device_to_tests.values()) == 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/flow_tracking.py

@@ -11,6 +11,7 @@
 
 from pydantic import BaseModel
 
+from anta.decorators import skip_on_platforms
 from anta.models import AntaCommand, AntaTemplate, AntaTest
 from anta.tools import get_failed_logs
 
@@ -151,6 +152,7 @@ def render(self, template: AntaTemplate) -> list[AntaCommand]:
         """Render the template for each hardware tracker."""
         return [template.render(name=tracker.name) for tracker in self.inputs.trackers]
 
+    @skip_on_platforms(["cEOSLab", "vEOS-lab"])
     @AntaTest.anta_test
     def test(self) -> None:
         """Main test function for VerifyHardwareFlowTrackerStatus."""

anta/tests/security.py

@@ -57,7 +57,7 @@ def test(self) -> None:
         except StopIteration:
             self.result.is_failure("Could not find SSH status in returned output.")
             return
-        status = line.split("is ")[1]
+        status = line.split()[-1]
 
         if status == "disabled":
             self.result.is_success()

anta/tests/snmp.py

@@ -7,9 +7,9 @@
 # mypy: disable-error-code=attr-defined
 from __future__ import annotations
 
-from typing import TYPE_CHECKING, ClassVar
+from typing import TYPE_CHECKING, ClassVar, get_args
 
-from anta.custom_types import PositiveInteger
+from anta.custom_types import PositiveInteger, SnmpErrorCounter, SnmpPdu
 from anta.models import AntaCommand, AntaTest
 from anta.tools import get_value
 
@@ -237,3 +237,116 @@ def test(self) -> None:
             self.result.is_failure(f"Expected `{self.inputs.contact}` as the contact, but found `{contact}` instead.")
         else:
             self.result.is_success()
+
+
+class VerifySnmpPDUCounters(AntaTest):
+    """Verifies the SNMP PDU counters.
+
+    By default, all SNMP PDU counters will be checked for any non-zero values.
+    An optional list of specific SNMP PDU(s) can be provided for granular testing.
+
+    Expected Results
+    ----------------
+    * Success: The test will pass if the SNMP PDU counter(s) are non-zero/greater than zero.
+    * Failure: The test will fail if the SNMP PDU counter(s) are zero/None/Not Found.
+
+    Examples
+    --------
+    ```yaml
+    anta.tests.snmp:
+      - VerifySnmpPDUCounters:
+          pdus:
+            - outTrapPdus
+            - inGetNextPdus
+    ```
+    """
+
+    name = "VerifySnmpPDUCounters"
+    description = "Verifies the SNMP PDU counters."
+    categories: ClassVar[list[str]] = ["snmp"]
+    commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show snmp", revision=1)]
+
+    class Input(AntaTest.Input):
+        """Input model for the VerifySnmpPDUCounters test."""
+
+        pdus: list[SnmpPdu] | None = None
+        """Optional list of SNMP PDU counters to be verified. If not provided, test will verifies all PDU counters."""
+
+    @AntaTest.anta_test
+    def test(self) -> None:
+        """Main test function for VerifySnmpPDUCounters."""
+        snmp_pdus = self.inputs.pdus
+        command_output = self.instance_commands[0].json_output
+
+        # Verify SNMP PDU counters.
+        if not (pdu_counters := get_value(command_output, "counters")):
+            self.result.is_failure("SNMP counters not found.")
+            return
+
+        # In case SNMP PDUs not provided, It will check all the update error counters.
+        if not snmp_pdus:
+            snmp_pdus = list(get_args(SnmpPdu))
+
+        failures = {pdu: value for pdu in snmp_pdus if (value := pdu_counters.get(pdu, "Not Found")) == "Not Found" or value == 0}
+
+        # Check if any failures
+        if not failures:
+            self.result.is_success()
+        else:
+            self.result.is_failure(f"The following SNMP PDU counters are not found or have zero PDU counters:\n{failures}")
+
+
+class VerifySnmpErrorCounters(AntaTest):
+    """Verifies the SNMP error counters.
+
+    By default, all  error counters will be checked for any non-zero values.
+    An optional list of specific error counters can be provided for granular testing.
+
+    Expected Results
+    ----------------
+    * Success: The test will pass if the SNMP error counter(s) are zero/None.
+    * Failure: The test will fail if the SNMP error counter(s) are non-zero/not None/Not Found or is not configured.
+
+    Examples
+    --------
+    ```yaml
+    anta.tests.snmp:
+      - VerifySnmpErrorCounters:
+          error_counters:
+            - inVersionErrs
+            - inBadCommunityNames
+    """
+
+    name = "VerifySnmpErrorCounters"
+    description = "Verifies the SNMP error counters."
+    categories: ClassVar[list[str]] = ["snmp"]
+    commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show snmp", revision=1)]
+
+    class Input(AntaTest.Input):
+        """Input model for the VerifySnmpErrorCounters test."""
+
+        error_counters: list[SnmpErrorCounter] | None = None
+        """Optional list of SNMP error counters to be verified. If not provided, test will verifies all error counters."""
+
+    @AntaTest.anta_test
+    def test(self) -> None:
+        """Main test function for VerifySnmpErrorCounters."""
+        error_counters = self.inputs.error_counters
+        command_output = self.instance_commands[0].json_output
+
+        # Verify SNMP PDU counters.
+        if not (snmp_counters := get_value(command_output, "counters")):
+            self.result.is_failure("SNMP counters not found.")
+            return
+
+        # In case SNMP error counters not provided, It will check all the error counters.
+        if not error_counters:
+            error_counters = list(get_args(SnmpErrorCounter))
+
+        error_counters_not_ok = {counter: value for counter in error_counters if (value := snmp_counters.get(counter))}
+
+        # Check if any failures
+        if not error_counters_not_ok:
+            self.result.is_success()
+        else:
+            self.result.is_failure(f"The following SNMP error counters are not found or have non-zero error counters:\n{error_counters_not_ok}")

docs/advanced_usages/as-python-lib.md

@@ -44,98 +44,14 @@ The [AntaInventory](../api/inventory.md#anta.inventory.AntaInventory) class is a
 ### Parse an ANTA inventory file
 
 ```python
-"""
-This script parses an ANTA inventory file, connects to devices and print their status.
-"""
-import asyncio
-
-from anta.inventory import AntaInventory
-
-
-async def main(inv: AntaInventory) -> None:
-    """
-    Take an AntaInventory and:
-    1. try to connect to every device in the inventory
-    2. print a message for every device connection status
-    """
-    await inv.connect_inventory()
-
-    for device in inv.values():
-        if device.established:
-            print(f"Device {device.name} is online")
-        else:
-            print(f"Could not connect to device {device.name}")
-
-if __name__ == "__main__":
-    # Create the AntaInventory instance
-    inventory = AntaInventory.parse(
-        filename="inv.yml",
-        username="arista",
-        password="@rista123",
-    )
-
-    # Run the main coroutine
-    res = asyncio.run(main(inventory))
+--8<-- "parse_anta_inventory_file.py"
 ```
 
-??? note "How to create your inventory file"
+!!! note "How to create your inventory file"
     Please visit this [dedicated section](../usage-inventory-catalog.md) for how to use inventory and catalog files.
 
 ### Run EOS commands
 
 ```python
-"""
-This script runs a list of EOS commands on reachable devices.
-"""
-# This is needed to run the script for python < 3.10 for typing annotations
-from __future__ import annotations
-
-import asyncio
-from pprint import pprint
-
-from anta.inventory import AntaInventory
-from anta.models import AntaCommand
-
-
-async def main(inv: AntaInventory, commands: list[str]) -> dict[str, list[AntaCommand]]:
-    """
-    Take an AntaInventory and a list of commands as string and:
-    1. try to connect to every device in the inventory
-    2. collect the results of the commands from each device
-
-    Returns:
-      a dictionary where key is the device name and the value is the list of AntaCommand ran towards the device
-    """
-    await inv.connect_inventory()
-
-    # Make a list of coroutine to run commands towards each connected device
-    coros = []
-    # dict to keep track of the commands per device
-    result_dict = {}
-    for name, device in inv.get_inventory(established_only=True).items():
-        anta_commands = [AntaCommand(command=command, ofmt="json") for command in commands]
-        result_dict[name] = anta_commands
-        coros.append(device.collect_commands(anta_commands))
-
-    # Run the coroutines
-    await asyncio.gather(*coros)
-
-    return result_dict
-
-
-if __name__ == "__main__":
-    # Create the AntaInventory instance
-    inventory = AntaInventory.parse(
-        filename="inv.yml",
-        username="arista",
-        password="@rista123",
-    )
-
-    # Create a list of commands with json output
-    commands = ["show version", "show ip bgp summary"]
-
-    # Run the main asyncio  entry point
-    res = asyncio.run(main(inventory, commands))
-
-    pprint(res)
+--8<-- "run_eos_commands.py"
 ```

docs/advanced_usages/custom-tests.md

@@ -71,9 +71,6 @@ Full AntaTest API documentation is available in the [API documentation section](
 
 ### Instance Attributes
 
-!!! info
-    You can access an instance attribute in your code using the `self` reference. E.g. you can access the test input values using `self.inputs`.
-
 ::: anta.models.AntaTest
     options:
         show_docstring_attributes: true