Merge branch 'main' into remove-aiocache

anta/custom_types.py

@@ -154,16 +154,38 @@ def validate_regex(value: str) -> str:
 ErrDisableReasons = Literal[
     "acl",
     "arp-inspection",
+    "bgp-session-tracking",
     "bpduguard",
+    "dot1x",
+    "dot1x-coa",
     "dot1x-session-replace",
+    "evpn-sa-mh",
+    "fabric-link-failure",
+    "fabric-link-flap",
     "hitless-reload-down",
+    "lacp-no-portid",
     "lacp-rate-limit",
+    "license-enforce",
     "link-flap",
+    "mlagasu",
+    "mlagdualprimary",
+    "mlagissu",
+    "mlagmaintdown",
     "no-internal-vlan",
+    "out-of-voqs",
     "portchannelguard",
+    "portgroup-disabled",
     "portsec",
+    "speed-misconfigured",
+    "storm-control",
+    "stp-no-portid",
+    "stuck-queue",
     "tapagg",
     "uplink-failure-detection",
+    "xcvr-misconfigured",
+    "xcvr-overheat",
+    "xcvr-power-unsupported",
+    "xcvr-unsupported",
 ]
 ErrDisableInterval = Annotated[int, Field(ge=30, le=86400)]
 Percent = Annotated[float, Field(ge=0.0, le=100.0)]

anta/input_models/routing/bgp.py

@@ -142,6 +142,8 @@ class BgpPeer(BaseModel):
     """IPv4 address of the BGP peer."""
     vrf: str = "default"
     """Optional VRF for the BGP peer. Defaults to `default`."""
+    peer_group: str | None = None
+    """Peer group of the BGP peer. Required field in the `VerifyBGPPeerGroup` test."""
     advertised_routes: list[IPv4Network] | None = None
     """List of advertised routes in CIDR format. Required field in the `VerifyBGPExchangedRoutes` test."""
     received_routes: list[IPv4Network] | None = None

anta/input_models/services.py

@@ -6,9 +6,13 @@
 from __future__ import annotations
 
 from ipaddress import IPv4Address, IPv6Address
+from typing import Any, Literal
+from warnings import warn
 
 from pydantic import BaseModel, ConfigDict, Field
 
+from anta.custom_types import ErrDisableReasons
+
 
 class DnsServer(BaseModel):
     """Model for a DNS server configuration."""
@@ -29,3 +33,42 @@ def __str__(self) -> str:
         Server 10.0.0.1 (VRF: default, Priority: 1)
         """
         return f"Server {self.server_address} (VRF: {self.vrf}, Priority: {self.priority})"
+
+
+class ErrdisableRecovery(BaseModel):
+    """Model for the error disable recovery functionality."""
+
+    model_config = ConfigDict(extra="forbid")
+    reason: ErrDisableReasons
+    """Name of the error disable reason."""
+    status: Literal["Enabled", "Disabled"] = "Enabled"
+    """Operational status of the reason. Defaults to 'Enabled'."""
+    interval: int = Field(ge=30, le=86400)
+    """Timer interval of the reason in seconds."""
+
+    def __str__(self) -> str:
+        """Return a human-readable string representation of the ErrdisableRecovery for reporting.
+
+        Examples
+        --------
+        Reason: acl Status: Enabled Interval: 300
+        """
+        return f"Reason: {self.reason} Status: {self.status} Interval: {self.interval}"
+
+
+class ErrDisableReason(ErrdisableRecovery):  # pragma: no cover
+    """Alias for the ErrdisableRecovery model to maintain backward compatibility.
+
+    When initialised, it will emit a deprecation warning and call the ErrdisableRecovery model.
+
+    TODO: Remove this class in ANTA v2.0.0.
+    """
+
+    def __init__(self, **data: Any) -> None:  # noqa: ANN401
+        """Initialize the ErrdisableRecovery class, emitting a depreciation warning."""
+        warn(
+            message="ErrDisableReason model is deprecated and will be removed in ANTA v2.0.0. Use the ErrdisableRecovery model instead.",
+            category=DeprecationWarning,
+            stacklevel=2,
+        )
+        super().__init__(**data)

anta/tests/bfd.py

@@ -8,9 +8,9 @@
 from __future__ import annotations
 
 from datetime import datetime, timezone
-from typing import TYPE_CHECKING, ClassVar
+from typing import TYPE_CHECKING, ClassVar, TypeVar
 
-from pydantic import Field
+from pydantic import Field, field_validator
 
 from anta.input_models.bfd import BFDPeer
 from anta.models import AntaCommand, AntaTest
@@ -19,6 +19,9 @@
 if TYPE_CHECKING:
     from anta.models import AntaTemplate
 
+# Using a TypeVar for the BFDPeer model since mypy thinks it's a ClassVar and not a valid type when used in field validators
+T = TypeVar("T", bound=BFDPeer)
+
 
 class VerifyBFDSpecificPeers(AntaTest):
     """Verifies the state of IPv4 BFD peer sessions.
@@ -143,6 +146,23 @@ class Input(AntaTest.Input):
         BFDPeer: ClassVar[type[BFDPeer]] = BFDPeer
         """To maintain backward compatibility"""
 
+        @field_validator("bfd_peers")
+        @classmethod
+        def validate_bfd_peers(cls, bfd_peers: list[T]) -> list[T]:
+            """Validate that 'tx_interval', 'rx_interval' and 'multiplier' fields are provided in each BFD peer."""
+            for peer in bfd_peers:
+                missing_fileds = []
+                if peer.tx_interval is None:
+                    missing_fileds.append("tx_interval")
+                if peer.rx_interval is None:
+                    missing_fileds.append("rx_interval")
+                if peer.multiplier is None:
+                    missing_fileds.append("multiplier")
+                if missing_fileds:
+                    msg = f"{peer} {', '.join(missing_fileds)} field(s) are missing in the input."
+                    raise ValueError(msg)
+            return bfd_peers
+
     @AntaTest.anta_test
     def test(self) -> None:
         """Main test function for VerifyBFDPeersIntervals."""
@@ -308,6 +328,16 @@ class Input(AntaTest.Input):
         BFDPeer: ClassVar[type[BFDPeer]] = BFDPeer
         """To maintain backward compatibility"""
 
+        @field_validator("bfd_peers")
+        @classmethod
+        def validate_bfd_peers(cls, bfd_peers: list[T]) -> list[T]:
+            """Validate that 'protocols' field is provided in each BFD peer."""
+            for peer in bfd_peers:
+                if peer.protocols is None:
+                    msg = f"{peer} 'protocols' field missing in the input."
+                    raise ValueError(msg)
+            return bfd_peers
+
     @AntaTest.anta_test
     def test(self) -> None:
         """Main test function for VerifyBFDPeersRegProtocols."""