refactor(anta.tests): Remove decorator and deprecated tests

anta/decorators.py

@@ -7,8 +7,7 @@
 from functools import wraps
 from typing import TYPE_CHECKING, Any, Callable, Optional, TypeVar, cast
 
-from anta.models import AntaCommand, AntaTest, logger
-from anta.tools.misc import exc_to_str
+from anta.models import AntaTest, logger
 
 if TYPE_CHECKING:
     from anta.result_manager.models import TestResult
@@ -103,80 +102,3 @@ async def wrapper(*args: Any, **kwargs: Any) -> TestResult:
         return cast(F, wrapper)
 
     return decorator
-
-
-def check_bgp_family_enable(family: str) -> Callable[[F], F]:
-    """
-    Return a decorator to conditionally skip a test based on BGP address family availability.
-
-    This is a decorator factory that generates a decorator that will skip the test
-    if there's no BGP configuration or peer for the given address family.
-
-    !!! warning
-        This decorator is deprecated and will eventually be removed in a future major release of ANTA.
-        New BGP tests have been created to address this. For more details, please refer to the BGP tests documentation.
-
-    Args:
-        family (str): BGP address family to check. Accepted values are 'ipv4', 'ipv6', 'evpn', 'rtc'.
-
-    Returns:
-        Callable[[F], F]: A decorator that can be used to wrap test functions.
-    """
-
-    def decorator(function: F) -> F:
-        """
-        Actual decorator that either runs the test or skips it based on BGP address family state.
-
-        Args:
-            function (F): The test function to be decorated.
-
-        Returns:
-            F: The decorated function.
-        """
-
-        @wraps(function)
-        async def wrapper(*args: Any, **kwargs: Any) -> TestResult:  # pylint: disable=too-many-branches
-            """
-            Check BGP address family and conditionally run or skip the test.
-
-            This wrapper checks the BGP address family state on the device.
-            If the required BGP configuration or peer is not found, the test is either skipped.
-            """
-            anta_test = args[0]
-
-            if anta_test.result.result != "unset":
-                AntaTest.update_progress()
-                return anta_test.result
-
-            if family == "ipv4":
-                command = AntaCommand(command="show bgp ipv4 unicast summary vrf all")
-            elif family == "ipv6":
-                command = AntaCommand(command="show bgp ipv6 unicast summary vrf all")
-            elif family == "evpn":
-                command = AntaCommand(command="show bgp evpn summary")
-            elif family == "rtc":
-                command = AntaCommand(command="show bgp rt-membership summary")
-            else:
-                anta_test.result.is_error(message=f"Wrong address family for BGP decorator: {family}")
-                return anta_test.result
-
-            await anta_test.device.collect(command=command)
-
-            if not command.collected and command.failed is not None:
-                anta_test.result.is_error(message=f"{command.command}: {exc_to_str(command.failed)}")
-                return anta_test.result
-            if "vrfs" not in command.json_output:
-                anta_test.result.is_skipped(f"No BGP configuration for address family {family} on this device")
-                AntaTest.update_progress()
-                return anta_test.result
-            if len(bgp_vrfs := command.json_output["vrfs"]) == 0 or len(bgp_vrfs["default"]["peers"]) == 0:
-                # No VRF
-                anta_test.result.is_skipped(f"No BGP peer for address family {family} on this device")
-                AntaTest.update_progress()
-                return anta_test.result
-
-            return await function(*args, **kwargs)
-
-        return cast(F, wrapper)
-
-    return decorator

anta/models.py

@@ -171,27 +171,6 @@ def collected(self) -> bool:
         return self.output is not None and self.failed is None
 
 
-class AntaTestFilter(ABC):
-    """Class to define a test Filter"""
-
-    # pylint: disable=too-few-public-methods
-
-    @abstractmethod
-    def should_skip(
-        self,
-        device: AntaDevice,
-        result: TestResult,
-        *args: list[Any],
-        **kwagrs: dict[str, Any],
-    ) -> bool:
-        """
-        Sets the TestResult status to skip with the appropriate skip message
-
-        Returns:
-            bool: True if the test should be skipped, False otherwise
-        """
-
-
 class AntaTemplateRenderError(RuntimeError):
     """
     Raised when an AntaTemplate object could not be rendered
@@ -262,8 +241,6 @@ def test(self) -> None:
     description: ClassVar[str]
     categories: ClassVar[list[str]]
     commands: ClassVar[list[Union[AntaTemplate, AntaCommand]]]
-    # Optional class attributes
-    test_filters: ClassVar[list[AntaTestFilter]]
     # Class attributes to handle the progress bar of ANTA CLI
     progress: Optional[Progress] = None
     nrfu_task: Optional[TaskID] = None

anta/tests/routing/bgp.py

@@ -9,17 +9,16 @@
 from __future__ import annotations
 
 from ipaddress import IPv4Address, IPv6Address
-
-# Need to keep Dict and List for pydantic in python 3.8
-from typing import Any, Dict, List, Optional, Union, cast
+from typing import Any, List, Optional, Union, cast
 
 from pydantic import BaseModel, PositiveInt, model_validator
 
 from anta.custom_types import Afi, Safi
-from anta.decorators import check_bgp_family_enable, deprecated_test
 from anta.models import AntaCommand, AntaTemplate, AntaTest
 from anta.tools.get_value import get_value
 
+# Need to keep List for pydantic in python 3.8
+
 
 def _check_bgp_vrfs(bgp_vrfs: dict[str, Any]) -> dict[str, Any]:
     """
@@ -133,239 +132,6 @@ def _check_peer_issues(peer_data: Optional[dict[str, Any]]) -> dict[str, Any]:
     return {}
 
 
-class VerifyBGPIPv4UnicastState(AntaTest):
-    """
-    Verifies all IPv4 unicast BGP sessions are established (for all VRF)
-    and all BGP messages queues for these sessions are empty (for all VRF).
-
-    * self.result = "skipped" if no BGP vrf are returned by the device
-    * self.result = "success" if all IPv4 unicast BGP sessions are established (for all VRF)
-                         and all BGP messages queues for these sessions are empty (for all VRF).
-    * self.result = "failure" otherwise.
-    """
-
-    name = "VerifyBGPIPv4UnicastState"
-    description = "Verifies all IPv4 unicast BGP sessions are established (for all VRF) and all BGP messages queues for these sessions are empty (for all VRF)."
-    categories = ["routing", "bgp"]
-    commands = [AntaCommand(command="show bgp ipv4 unicast summary vrf all")]
-
-    @deprecated_test(new_tests=["VerifyBGPPeersHealth"])
-    @check_bgp_family_enable("ipv4")
-    @AntaTest.anta_test
-    def test(self) -> None:
-        command_output = self.instance_commands[0].json_output
-        state_issue = _check_bgp_vrfs(command_output["vrfs"])
-        if not state_issue:
-            self.result.is_success()
-        else:
-            self.result.is_failure(f"Some IPv4 Unicast BGP Peer are not up: {state_issue}")
-
-
-class VerifyBGPIPv4UnicastCount(AntaTest):
-    """
-    Verifies all IPv4 unicast BGP sessions are established
-    and all BGP messages queues for these sessions are empty
-    and the actual number of BGP IPv4 unicast neighbors is the one we expect
-    in all VRFs specified as input.
-
-    * self.result = "success" if all IPv4 unicast BGP sessions are established
-                         and if all BGP messages queues for these sessions are empty
-                         and if the actual number of BGP IPv4 unicast neighbors is equal to `number
-                         in all VRFs specified as input.
-    * self.result = "failure" otherwise.
-    """
-
-    name = "VerifyBGPIPv4UnicastCount"
-    description = (
-        "Verifies all IPv4 unicast BGP sessions are established and all their BGP messages queues are empty and "
-        " the actual number of BGP IPv4 unicast neighbors is the one we expect."
-    )
-    categories = ["routing", "bgp"]
-    commands = [AntaTemplate(template="show bgp ipv4 unicast summary vrf {vrf}")]
-
-    class Input(AntaTest.Input):  # pylint: disable=missing-class-docstring
-        vrfs: Dict[str, int]
-        """VRFs associated with neighbors count to verify"""
-
-    def render(self, template: AntaTemplate) -> list[AntaCommand]:
-        return [template.render(vrf=vrf) for vrf in self.inputs.vrfs]
-
-    @deprecated_test(new_tests=["VerifyBGPPeerCount", "VerifyBGPPeersHealth"])
-    @check_bgp_family_enable("ipv4")
-    @AntaTest.anta_test
-    def test(self) -> None:
-        self.result.is_success()
-        for command in self.instance_commands:
-            if "vrf" in command.params:
-                vrf = command.params["vrf"]
-                count = self.inputs.vrfs[vrf]
-                if vrf not in command.json_output["vrfs"]:
-                    self.result.is_failure(f"VRF {vrf} is not configured")
-                    return
-                peers = command.json_output["vrfs"][vrf]["peers"]
-                state_issue = _check_bgp_vrfs(command.json_output["vrfs"])
-                if len(peers) != count:
-                    self.result.is_failure(f"Expecting {count} BGP peer(s) in vrf {vrf} but got {len(peers)} peer(s)")
-                if state_issue:
-                    self.result.is_failure(f"The following IPv4 peer(s) are not established: {state_issue}")
-
-
-class VerifyBGPIPv6UnicastState(AntaTest):
-    """
-    Verifies all IPv6 unicast BGP sessions are established (for all VRF)
-    and all BGP messages queues for these sessions are empty (for all VRF).
-
-    * self.result = "skipped" if no BGP vrf are returned by the device
-    * self.result = "success" if all IPv6 unicast BGP sessions are established (for all VRF)
-                         and all BGP messages queues for these sessions are empty (for all VRF).
-    * self.result = "failure" otherwise.
-    """
-
-    name = "VerifyBGPIPv6UnicastState"
-    description = "Verifies all IPv6 unicast BGP sessions are established (for all VRF) and all BGP messages queues for these sessions are empty (for all VRF)."
-    categories = ["routing", "bgp"]
-    commands = [AntaCommand(command="show bgp ipv6 unicast summary vrf all")]
-
-    @deprecated_test(new_tests=["VerifyBGPPeersHealth"])
-    @check_bgp_family_enable("ipv6")
-    @AntaTest.anta_test
-    def test(self) -> None:
-        command_output = self.instance_commands[0].json_output
-        state_issue = _check_bgp_vrfs(command_output["vrfs"])
-        if not state_issue:
-            self.result.is_success()
-        else:
-            self.result.is_failure(f"Some IPv4 Unicast BGP Peer are not up: {state_issue}")
-
-
-class VerifyBGPEVPNState(AntaTest):
-    """
-    Verifies all EVPN BGP sessions are established (default VRF).
-
-    * self.result = "skipped" if no BGP EVPN peers are returned by the device
-    * self.result = "success" if all EVPN BGP sessions are established.
-    * self.result = "failure" otherwise.
-    """
-
-    name = "VerifyBGPEVPNState"
-    description = "Verifies all EVPN BGP sessions are established (default VRF)."
-    categories = ["routing", "bgp"]
-    commands = [AntaCommand(command="show bgp evpn summary")]
-
-    @deprecated_test(new_tests=["VerifyBGPPeersHealth"])
-    @check_bgp_family_enable("evpn")
-    @AntaTest.anta_test
-    def test(self) -> None:
-        command_output = self.instance_commands[0].json_output
-        bgp_vrfs = command_output["vrfs"]
-        peers = bgp_vrfs["default"]["peers"]
-        non_established_peers = [peer for peer, peer_dict in peers.items() if peer_dict["peerState"] != "Established"]
-        if not non_established_peers:
-            self.result.is_success()
-        else:
-            self.result.is_failure(f"The following EVPN peers are not established: {non_established_peers}")
-
-
-class VerifyBGPEVPNCount(AntaTest):
-    """
-    Verifies all EVPN BGP sessions are established (default VRF)
-    and the actual number of BGP EVPN neighbors is the one we expect (default VRF).
-
-    * self.result = "success" if all EVPN BGP sessions are Established and if the actual
-                         number of BGP EVPN neighbors is the one we expect.
-    * self.result = "failure" otherwise.
-    """
-
-    name = "VerifyBGPEVPNCount"
-    description = "Verifies all EVPN BGP sessions are established (default VRF) and the actual number of BGP EVPN neighbors is the one we expect (default VRF)."
-    categories = ["routing", "bgp"]
-    commands = [AntaCommand(command="show bgp evpn summary")]
-
-    class Input(AntaTest.Input):  # pylint: disable=missing-class-docstring
-        number: int
-        """The expected number of BGP EVPN neighbors in the default VRF"""
-
-    @deprecated_test(new_tests=["VerifyBGPPeerCount", "VerifyBGPPeersHealth"])
-    @check_bgp_family_enable("evpn")
-    @AntaTest.anta_test
-    def test(self) -> None:
-        command_output = self.instance_commands[0].json_output
-        peers = command_output["vrfs"]["default"]["peers"]
-        non_established_peers = [peer for peer, peer_dict in peers.items() if peer_dict["peerState"] != "Established"]
-        if not non_established_peers and len(peers) == self.inputs.number:
-            self.result.is_success()
-        else:
-            self.result.is_failure()
-            if len(peers) != self.inputs.number:
-                self.result.is_failure(f"Expecting {self.inputs.number} BGP EVPN peers and got {len(peers)}")
-            if non_established_peers:
-                self.result.is_failure(f"The following EVPN peers are not established: {non_established_peers}")
-
-
-class VerifyBGPRTCState(AntaTest):
-    """
-    Verifies all RTC BGP sessions are established (default VRF).
-
-    * self.result = "skipped" if no BGP RTC peers are returned by the device
-    * self.result = "success" if all RTC BGP sessions are established.
-    * self.result = "failure" otherwise.
-    """
-
-    name = "VerifyBGPRTCState"
-    description = "Verifies all RTC BGP sessions are established (default VRF)."
-    categories = ["routing", "bgp"]
-    commands = [AntaCommand(command="show bgp rt-membership summary")]
-
-    @deprecated_test(new_tests=["VerifyBGPPeersHealth"])
-    @check_bgp_family_enable("rtc")
-    @AntaTest.anta_test
-    def test(self) -> None:
-        command_output = self.instance_commands[0].json_output
-        bgp_vrfs = command_output["vrfs"]
-        peers = bgp_vrfs["default"]["peers"]
-        non_established_peers = [peer for peer, peer_dict in peers.items() if peer_dict["peerState"] != "Established"]
-        if not non_established_peers:
-            self.result.is_success()
-        else:
-            self.result.is_failure(f"The following RTC peers are not established: {non_established_peers}")
-
-
-class VerifyBGPRTCCount(AntaTest):
-    """
-    Verifies all RTC BGP sessions are established (default VRF)
-    and the actual number of BGP RTC neighbors is the one we expect (default VRF).
-
-    * self.result = "success" if all RTC BGP sessions are Established and if the actual
-                         number of BGP RTC neighbors is the one we expect.
-    * self.result = "failure" otherwise.
-    """
-
-    name = "VerifyBGPRTCCount"
-    description = "Verifies all RTC BGP sessions are established (default VRF) and the actual number of BGP RTC neighbors is the one we expect (default VRF)."
-    categories = ["routing", "bgp"]
-    commands = [AntaCommand(command="show bgp rt-membership summary")]
-
-    class Input(AntaTest.Input):  # pylint: disable=missing-class-docstring
-        number: int
-        """The expected number of BGP RTC neighbors in the default VRF"""
-
-    @deprecated_test(new_tests=["VerifyBGPPeerCount", "VerifyBGPPeersHealth"])
-    @check_bgp_family_enable("rtc")
-    @AntaTest.anta_test
-    def test(self) -> None:
-        command_output = self.instance_commands[0].json_output
-        peers = command_output["vrfs"]["default"]["peers"]
-        non_established_peers = [peer for peer, peer_dict in peers.items() if peer_dict["peerState"] != "Established"]
-        if not non_established_peers and len(peers) == self.inputs.number:
-            self.result.is_success()
-        else:
-            self.result.is_failure()
-            if len(peers) != self.inputs.number:
-                self.result.is_failure(f"Expecting {self.inputs.number} BGP RTC peers and got {len(peers)}")
-            if non_established_peers:
-                self.result.is_failure(f"The following RTC peers are not established: {non_established_peers}")
-
-
 class VerifyBGPPeerCount(AntaTest):
     """
     This test verifies the count of BGP peers for a given address family.