ci: add codspeed to benchmark ANTA

.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

.pre-commit-config.yaml

@@ -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`.

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

anta/tests/security.py

@@ -9,14 +9,22 @@
 # mypy: disable-error-code=attr-defined
 from datetime import datetime, timezone
 from ipaddress import IPv4Address
-from typing import ClassVar
+from typing import TYPE_CHECKING, ClassVar, get_args
 
 from pydantic import BaseModel, Field, model_validator
 
 from anta.custom_types import EcdsaKeySize, EncryptionAlgorithm, PositiveInteger, RsaKeySize
 from anta.models import AntaCommand, AntaTemplate, AntaTest
 from anta.tools import get_failed_logs, 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
+
 
 class VerifySSHStatus(AntaTest):
     """Verifies if the SSHD agent is disabled in the default VRF.
@@ -47,7 +55,7 @@ def test(self) -> None:
         try:
             line = next(line for line in command_output.split("\n") if line.startswith("SSHD status"))
         except StopIteration:
-            self.result.is_error("Could not find SSH status in returned output.")
+            self.result.is_failure("Could not find SSH status in returned output.")
             return
         status = line.split("is ")[1]
 
@@ -416,19 +424,19 @@ class APISSLCertificate(BaseModel):
             """The encryption algorithm key size of the certificate."""
 
             @model_validator(mode="after")
-            def validate_inputs(self: BaseModel) -> BaseModel:
+            def validate_inputs(self) -> Self:
                 """Validate the key size provided to the APISSLCertificates class.
 
                 If encryption_algorithm is RSA then key_size should be in {2048, 3072, 4096}.
 
                 If encryption_algorithm is ECDSA then key_size should be in {256, 384, 521}.
                 """
-                if self.encryption_algorithm == "RSA" and self.key_size not in RsaKeySize.__args__:
-                    msg = f"`{self.certificate_name}` key size {self.key_size} is invalid for RSA encryption. Allowed sizes are {RsaKeySize.__args__}."
+                if self.encryption_algorithm == "RSA" and self.key_size not in get_args(RsaKeySize):
+                    msg = f"`{self.certificate_name}` key size {self.key_size} is invalid for RSA encryption. Allowed sizes are {get_args(RsaKeySize)}."
                     raise ValueError(msg)
 
-                if self.encryption_algorithm == "ECDSA" and self.key_size not in EcdsaKeySize.__args__:
-                    msg = f"`{self.certificate_name}` key size {self.key_size} is invalid for ECDSA encryption. Allowed sizes are {EcdsaKeySize.__args__}."
+                if self.encryption_algorithm == "ECDSA" and self.key_size not in get_args(EcdsaKeySize):
+                    msg = f"`{self.certificate_name}` key size {self.key_size} is invalid for ECDSA encryption. Allowed sizes are {get_args(EcdsaKeySize)}."
                     raise ValueError(msg)
 
                 return self

anta/tests/system.py

@@ -89,9 +89,6 @@ class VerifyReloadCause(AntaTest):
     def test(self) -> None:
         """Main test function for VerifyReloadCause."""
         command_output = self.instance_commands[0].json_output
-        if "resetCauses" not in command_output:
-            self.result.is_error(message="No reload causes available")
-            return
         if len(command_output["resetCauses"]) == 0:
             # No reload causes
             self.result.is_success()

pyproject.toml

@@ -68,6 +68,8 @@ dev = [
   "pytest-asyncio>=0.21.1",
   "pytest-cov>=4.1.0",
   "pytest-dependency",
+  "pytest-codspeed>=2.2.0",
+  "respx",
   "pytest-html>=3.2.0",
   "pytest-httpx>=0.30.0",
   "pytest-metadata>=3.0.0",
@@ -171,6 +173,7 @@ render_collapsed = true
 testpaths = ["tests"]
 asyncio_mode = "auto"
 asyncio_default_fixture_loop_scope = "function"
+norecursedirs = ["tests/benchmark"] # Do not run performance testing outside of Codspeed
 filterwarnings = [
   # cvprac is raising the next warning
   "default:pkg_resources is deprecated:DeprecationWarning",
@@ -450,13 +453,17 @@ disable = [ # Any rule listed here can be disabled: https://github.com/astral-sh
     "keyword-arg-before-vararg",
     "protected-access",
     "too-many-arguments",
-    "too-many-positional-arguments",  # New in pylint 3.3.0
+    "too-many-positional-arguments",
     "wrong-import-position",
     "pointless-statement",
     "broad-exception-caught",
     "line-too-long",
     "unused-variable",
     "redefined-builtin",
+    "global-statement",
+    "reimported",
+    "wrong-import-order",
+    "wrong-import-position",
     "abstract-class-instantiated", # Overlap with https://mypy.readthedocs.io/en/stable/error_code_list.html#check-instantiation-of-abstract-classes-abstract
     "unexpected-keyword-arg", # Overlap with https://mypy.readthedocs.io/en/stable/error_code_list.html#check-arguments-in-calls-call-arg and other rules
     "no-value-for-parameter" # Overlap with https://mypy.readthedocs.io/en/stable/error_code_list.html#check-arguments-in-calls-call-arg

tests/benchmark/__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.
+"""Benchmark tests for ANTA."""

tests/benchmark/conftest.py

@@ -0,0 +1,40 @@
+# 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.
+"""Fixtures for benchmarking ANTA."""
+
+import logging
+
+import pytest
+import respx
+from _pytest.terminal import TerminalReporter
+
+from anta.catalog import AntaCatalog
+
+from .utils import AntaMockEnvironment
+
+logger = logging.getLogger(__name__)
+
+TEST_CASE_COUNT = None
+
+
+@pytest.fixture(name="anta_mock_env", scope="session")  # We want this fixture to have a scope set to session to avoid reparsing all the unit tests data.
+def anta_mock_env_fixture() -> AntaMockEnvironment:
+    """Return an AntaMockEnvironment for this test session. Also configure respx to mock eAPI responses."""
+    global TEST_CASE_COUNT  # noqa: PLW0603
+    eapi_route = respx.post(path="/command-api", headers={"Content-Type": "application/json-rpc"})
+    env = AntaMockEnvironment()
+    TEST_CASE_COUNT = env.tests_count
+    eapi_route.side_effect = env.eapi_response
+    return env
+
+
+@pytest.fixture  # This fixture should have a scope set to function as the indexing result is stored in this object
+def catalog(anta_mock_env: AntaMockEnvironment) -> AntaCatalog:
+    """Fixture that return an ANTA catalog from the AntaMockEnvironment of this test session."""
+    return anta_mock_env.catalog
+
+
+def pytest_terminal_summary(terminalreporter: TerminalReporter) -> None:
+    """Display the total number of ANTA unit test cases used to benchmark."""
+    terminalreporter.write_sep("=", f"{TEST_CASE_COUNT} ANTA test cases")