Merge branch 'main' into issue_820

.github/workflows/code-testing.yml

@@ -133,3 +133,20 @@ jobs:
         run: pip install .[doc]
       - name: "Build mkdocs documentation offline"
         run: mkdocs build
+  benchmarks:
+    name: Benchmark ANTA for Python 3.12
+    runs-on: ubuntu-latest
+    needs: [test-python]
+    steps:
+      - uses: actions/checkout@v4
+      - name: Setup Python
+        uses: actions/setup-python@v5
+        with:
+          python-version: "3.12"
+      - name: Install dependencies
+        run: pip install .[dev]
+      - name: Run benchmarks
+        uses: CodSpeedHQ/action@v3
+        with:
+          token: ${{ secrets.CODSPEED_TOKEN }}
+          run: pytest --codspeed --no-cov --log-cli-level INFO tests/benchmark

.github/workflows/codspeed.yml

@@ -0,0 +1,22 @@
+---
+name: Run benchmarks manually
+on:
+  workflow_dispatch:
+
+jobs:
+  benchmarks:
+    name: Benchmark ANTA for Python 3.12
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - name: Setup Python
+        uses: actions/setup-python@v5
+        with:
+          python-version: "3.12"
+      - name: Install dependencies
+        run: pip install .[dev]
+      - name: Run benchmarks
+        uses: CodSpeedHQ/action@v3
+        with:
+          token: ${{ secrets.CODSPEED_TOKEN }}
+          run: pytest --codspeed --no-cov --log-cli-level INFO tests/benchmark

.pre-commit-config.yaml

@@ -43,7 +43,7 @@ repos:
           - '<!--|  ~|  -->'
 
   - repo: https://github.com/astral-sh/ruff-pre-commit
-    rev: v0.6.5
+    rev: v0.6.8
     hooks:
     - id: ruff
       name: Run Ruff linter
@@ -52,7 +52,7 @@ repos:
       name: Run Ruff formatter
 
   - repo: https://github.com/pycqa/pylint
-    rev: "v3.2.7"
+    rev: "v3.3.1"
     hooks:
       - id: pylint
         name: Check code style with pylint
@@ -69,6 +69,7 @@ repos:
           - types-pyOpenSSL
           - pylint_pydantic
           - pytest
+          - pytest-codspeed
           - respx
 
   - repo: https://github.com/codespell-project/codespell

anta/catalog.py

@@ -25,8 +25,14 @@
 from anta.models import AntaTest
 
 if TYPE_CHECKING:
+    import sys
     from types import ModuleType
 
+    if sys.version_info >= (3, 11):
+        from typing import Self
+    else:
+        from typing_extensions import Self
+
 logger = logging.getLogger(__name__)
 
 # { <module_name> : [ { <test_class_name>: <input_as_dict_or_None> }, ... ] }
@@ -123,7 +129,7 @@ def instantiate_inputs(
         raise ValueError(msg)
 
     @model_validator(mode="after")
-    def check_inputs(self) -> AntaTestDefinition:
+    def check_inputs(self) -> Self:
         """Check the `inputs` field typing.
 
         The `inputs` class attribute needs to be an instance of the AntaTest.Input subclass defined in the class `test`.
@@ -290,11 +296,16 @@ def __init__(
             else:
                 self._filename = Path(filename)
 
-        # Default indexes for faster access
-        self.tag_to_tests: defaultdict[str | None, set[AntaTestDefinition]] = defaultdict(set)
-        self.tests_without_tags: set[AntaTestDefinition] = set()
-        self.indexes_built: bool = False
-        self.final_tests_count: int = 0
+        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
     def filename(self) -> Path | None:
@@ -479,7 +490,7 @@ def build_indexes(self, filtered_tests: set[str] | None = None) -> None:
 
         - 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.
+        - _tests_without_tags: A set of tests that do not have any tags.
 
         Once the indexes are built, the `indexes_built` attribute is set to True.
         """
@@ -493,11 +504,15 @@ 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._tests_without_tags.add(test)
 
-        self.tag_to_tests[None] = self.tests_without_tags
+        self.tag_to_tests[None] = self._tests_without_tags
         self.indexes_built = True
 
+    def clear_indexes(self) -> None:
+        """Clear this AntaCatalog instance indexes."""
+        self._init_indexes()
+
     def get_tests_by_tags(self, tags: set[str], *, strict: bool = False) -> set[AntaTestDefinition]:
         """Return all tests that match a given set of tags, according to the specified strictness.
 

anta/runner.py

@@ -147,6 +147,7 @@ def prepare_tests(
     device_to_tests: defaultdict[AntaDevice, set[AntaTestDefinition]] = defaultdict(set)
 
     # Create AntaTestRunner tuples from the tags
+    final_tests_count = 0
     for device in inventory.devices:
         if tags:
             if not any(tag in device.tags for tag in tags):
@@ -159,9 +160,9 @@ def prepare_tests(
         # Add the tests with matching tags from device tags
         device_to_tests[device].update(catalog.get_tests_by_tags(device.tags))
 
-        catalog.final_tests_count += len(device_to_tests[device])
+        final_tests_count += len(device_to_tests[device])
 
-    if catalog.final_tests_count == 0:
+    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."
         )
@@ -171,13 +172,15 @@ def prepare_tests(
     return device_to_tests
 
 
-def get_coroutines(selected_tests: defaultdict[AntaDevice, set[AntaTestDefinition]]) -> list[Coroutine[Any, Any, TestResult]]:
+def get_coroutines(selected_tests: defaultdict[AntaDevice, set[AntaTestDefinition]], manager: ResultManager) -> list[Coroutine[Any, Any, TestResult]]:
     """Get the coroutines for the ANTA run.
 
     Parameters
     ----------
     selected_tests
         A mapping of devices to the tests to run. The selected tests are generated by the `prepare_tests` function.
+    manager
+        A ResultManager
 
     Returns
     -------
@@ -189,6 +192,7 @@ def get_coroutines(selected_tests: defaultdict[AntaDevice, set[AntaTestDefinitio
         for test in test_definitions:
             try:
                 test_instance = test.test(device=device, inputs=test.inputs)
+                manager.add(test_instance.result)
                 coros.append(test_instance.test())
             except Exception as e:  # noqa: PERF203, BLE001
                 # An AntaTest instance is potentially user-defined code.
@@ -256,25 +260,26 @@ async def main(  # noqa: PLR0913
             selected_tests = prepare_tests(selected_inventory, catalog, tests, tags)
             if selected_tests is None:
                 return
+            final_tests_count = sum(len(tests) for tests in selected_tests.values())
 
         run_info = (
             "--- ANTA NRFU Run Information ---\n"
             f"Number of devices: {len(inventory)} ({len(selected_inventory)} established)\n"
-            f"Total number of selected tests: {catalog.final_tests_count}\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"
             "---------------------------------"
         )
 
         logger.info(run_info)
 
-        if catalog.final_tests_count > limits[0]:
+        if final_tests_count > limits[0]:
             logger.warning(
                 "The number of concurrent tests is higher than the open file descriptors limit for this ANTA process.\n"
                 "Errors may occur while running the tests.\n"
                 "Please consult the ANTA FAQ."
             )
 
-        coroutines = get_coroutines(selected_tests)
+        coroutines = get_coroutines(selected_tests, manager)
 
     if dry_run:
         logger.info("Dry-run mode, exiting before running the tests.")
@@ -286,8 +291,6 @@ async def main(  # noqa: PLR0913
         AntaTest.nrfu_task = AntaTest.progress.add_task("Running NRFU Tests...", total=len(coroutines))
 
     with Catchtime(logger=logger, message="Running ANTA tests"):
-        test_results = await asyncio.gather(*coroutines)
-        for r in test_results:
-            manager.add(r)
+        await asyncio.gather(*coroutines)
 
     log_cache_statistics(selected_inventory.devices)

anta/tests/field_notices.py

@@ -196,4 +196,4 @@ def test(self) -> None:
                     self.result.is_success("FN72 is mitigated")
                 return
         # We should never hit this point
-        self.result.is_error("Error in running test - FixedSystemvrm1 not found")
+        self.result.is_failure("Error in running test - Component FixedSystemvrm1 not found in 'show version'")

anta/tests/interfaces.py

@@ -71,7 +71,7 @@ def test(self) -> None:
             if ((duplex := (interface := interfaces["interfaces"][intf]).get("duplex", None)) is not None and duplex != duplex_full) or (
                 (members := interface.get("memberInterfaces", None)) is not None and any(stats["duplex"] != duplex_full for stats in members.values())
             ):
-                self.result.is_error(f"Interface {intf} or one of its member interfaces is not Full-Duplex. VerifyInterfaceUtilization has not been implemented.")
+                self.result.is_failure(f"Interface {intf} or one of its member interfaces is not Full-Duplex. VerifyInterfaceUtilization has not been implemented.")
                 return
 
             if (bandwidth := interfaces["interfaces"][intf]["bandwidth"]) == 0:
@@ -705,7 +705,7 @@ def test(self) -> None:
                     input_interface_detail = interface
                     break
             else:
-                self.result.is_error(f"Could not find `{intf}` in the input interfaces. {GITHUB_SUGGESTION}")
+                self.result.is_failure(f"Could not find `{intf}` in the input interfaces. {GITHUB_SUGGESTION}")
                 continue
 
             input_primary_ip = str(input_interface_detail.primary_ip)

anta/tests/mlag.py

@@ -123,10 +123,7 @@ class VerifyMlagConfigSanity(AntaTest):
     def test(self) -> None:
         """Main test function for VerifyMlagConfigSanity."""
         command_output = self.instance_commands[0].json_output
-        if (mlag_status := get_value(command_output, "mlagActive")) is None:
-            self.result.is_error(message="Incorrect JSON response - 'mlagActive' state was not found")
-            return
-        if mlag_status is False:
+        if command_output["mlagActive"] is False:
             self.result.is_skipped("MLAG is disabled")
             return
         keys_to_verify = ["globalConfiguration", "interfaceConfiguration"]

anta/tests/routing/bgp.py

@@ -8,7 +8,7 @@
 from __future__ import annotations
 
 from ipaddress import IPv4Address, IPv4Network, IPv6Address
-from typing import Any, ClassVar
+from typing import TYPE_CHECKING, Any, ClassVar
 
 from pydantic import BaseModel, Field, PositiveInt, model_validator
 from pydantic.v1.utils import deep_update
@@ -18,6 +18,14 @@
 from anta.models import AntaCommand, AntaTemplate, AntaTest
 from anta.tools import get_item, get_value
 
+if TYPE_CHECKING:
+    import sys
+
+    if sys.version_info >= (3, 11):
+        from typing import Self
+    else:
+        from typing_extensions import Self
+
 
 def _add_bgp_failures(failures: dict[tuple[str, str | None], dict[str, Any]], afi: Afi, safi: Safi | None, vrf: str, issue: str | dict[str, Any]) -> None:
     """Add a BGP failure entry to the given `failures` dictionary.
@@ -235,7 +243,7 @@ class BgpAfi(BaseModel):
             """Number of expected BGP peer(s)."""
 
             @model_validator(mode="after")
-            def validate_inputs(self: BaseModel) -> BaseModel:
+            def validate_inputs(self) -> Self:
                 """Validate the inputs provided to the BgpAfi class.
 
                 If afi is either ipv4 or ipv6, safi must be provided.
@@ -375,7 +383,7 @@ class BgpAfi(BaseModel):
             """
 
             @model_validator(mode="after")
-            def validate_inputs(self: BaseModel) -> BaseModel:
+            def validate_inputs(self) -> Self:
                 """Validate the inputs provided to the BgpAfi class.
 
                 If afi is either ipv4 or ipv6, safi must be provided.
@@ -522,7 +530,7 @@ class BgpAfi(BaseModel):
             """List of BGP IPv4 or IPv6 peer."""
 
             @model_validator(mode="after")
-            def validate_inputs(self: BaseModel) -> BaseModel:
+            def validate_inputs(self) -> Self:
                 """Validate the inputs provided to the BgpAfi class.
 
                 If afi is either ipv4 or ipv6, safi must be provided and vrf must NOT be all.
@@ -1485,7 +1493,7 @@ class BgpPeer(BaseModel):
             """Outbound route map applied, defaults to None."""
 
             @model_validator(mode="after")
-            def validate_inputs(self: BaseModel) -> BaseModel:
+            def validate_inputs(self) -> Self:
                 """Validate the inputs provided to the BgpPeer class.
 
                 At least one of 'inbound' or 'outbound' route-map must be provided.

anta/tests/routing/generic.py

@@ -9,12 +9,21 @@
 
 from functools import cache
 from ipaddress import IPv4Address, IPv4Interface
-from typing import ClassVar, Literal
+from typing import TYPE_CHECKING, ClassVar, Literal
 
 from pydantic import model_validator
 
+from anta.custom_types import PositiveInteger
 from anta.models import AntaCommand, AntaTemplate, AntaTest
 
+if TYPE_CHECKING:
+    import sys
+
+    if sys.version_info >= (3, 11):
+        from typing import Self
+    else:
+        from typing_extensions import Self
+
 
 class VerifyRoutingProtocolModel(AntaTest):
     """Verifies the configured routing protocol model is the one we expect.
@@ -84,13 +93,13 @@ class VerifyRoutingTableSize(AntaTest):
     class Input(AntaTest.Input):
         """Input model for the VerifyRoutingTableSize test."""
 
-        minimum: int
+        minimum: PositiveInteger
         """Expected minimum routing table size."""
-        maximum: int
+        maximum: PositiveInteger
         """Expected maximum routing table size."""
 
-        @model_validator(mode="after")  # type: ignore[misc]
-        def check_min_max(self) -> AntaTest.Input:
+        @model_validator(mode="after")
+        def check_min_max(self) -> Self:
             """Validate that maximum is greater than minimum."""
             if self.minimum > self.maximum:
                 msg = f"Minimum {self.minimum} is greater than maximum {self.maximum}"