BGP

ANTA catalog for BGP tests

Deprecation Notice

As part of our ongoing effort to improve the ANTA catalog and align it with best practices, we are announcing the deprecation of certain BGP tests along with a specific decorator. These will be removed in a future major release of ANTA.

What is being deprecated?

• Tests: The following BGP tests in the ANTA catalog are marked for deprecation.

anta.tests.routing:
  bgp:
    - VerifyBGPIPv4UnicastState:
    - VerifyBGPIPv4UnicastCount:
    - VerifyBGPIPv6UnicastState:
    - VerifyBGPEVPNState:
    - VerifyBGPEVPNCount:
    - VerifyBGPRTCState:
    - VerifyBGPRTCCount:

• Decorator: The check_bgp_family_enable decorator is also being deprecated as it is no longer needed with the new refactored BGP tests.

What should you do?

We strongly recommend transitioning to the new set of BGP tests that have been introduced to replace the deprecated ones. Please refer to each test documentation on this page below.

anta.tests.routing:
  bgp:
    - VerifyBGPPeerCount:
    - VerifyBGPPeersHealth:
    - VerifyBGPSpecificPeers:

---

BGP test functions

VerifyBGPEVPNCount

Bases: 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.

Source code in anta/tests/routing/bgp.py

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

Input

Bases: Input

Source code in anta/tests/routing/bgp.py

class Input(AntaTest.Input):  # pylint: disable=missing-class-docstring
    number: int
    """The expected number of BGP EVPN neighbors in the default VRF"""

number instance-attribute

number: int

The expected number of BGP EVPN neighbors in the default VRF

VerifyBGPEVPNState

Bases: 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.

Source code in anta/tests/routing/bgp.py

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

VerifyBGPIPv4UnicastCount

Bases: 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.

Source code in anta/tests/routing/bgp.py

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

Input

Bases: Input

Source code in anta/tests/routing/bgp.py

class Input(AntaTest.Input):  # pylint: disable=missing-class-docstring
    vrfs: Dict[str, int]
    """VRFs associated with neighbors count to verify"""

vrfs instance-attribute

vrfs: Dict[str, int]

VRFs associated with neighbors count to verify

VerifyBGPIPv4UnicastState

Bases: 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.

Source code in anta/tests/routing/bgp.py

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

VerifyBGPIPv6UnicastState

Bases: 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.

Source code in anta/tests/routing/bgp.py

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

VerifyBGPPeerCount

Bases: AntaTest

This test verifies the count of BGP peers for a given address family.

It supports multiple types of address families (AFI) and subsequent service families (SAFI). Please refer to the Input class attributes below for details.

Expected Results

• success: If the count of BGP peers matches the expected count for each address family and VRF.
• failure: If the count of BGP peers does not match the expected count, or if BGP is not configured for an expected VRF or address family.

Source code in anta/tests/routing/bgp.py

class VerifyBGPPeerCount(AntaTest):
    """
    This test verifies the count of BGP peers for a given address family.

    It supports multiple types of address families (AFI) and subsequent service families (SAFI).
    Please refer to the Input class attributes below for details.

    Expected Results:
        * success: If the count of BGP peers matches the expected count for each address family and VRF.
        * failure: If the count of BGP peers does not match the expected count, or if BGP is not configured for an expected VRF or address family.
    """

    name = "VerifyBGPPeerCount"
    description = "Verifies the count of BGP peers."
    categories = ["routing", "bgp"]
    commands = [
        AntaTemplate(template="show bgp {afi} {safi} summary vrf {vrf}"),
        AntaTemplate(template="show bgp {afi} summary"),
    ]

    class Input(AntaTest.Input):  # pylint: disable=missing-class-docstring
        address_families: List[BgpAfi]
        """
        List of BGP address families (BgpAfi)
        """

        class BgpAfi(BaseModel):  # pylint: disable=missing-class-docstring
            afi: Afi
            """BGP address family (AFI)"""
            safi: Optional[Safi] = None
            """Optional BGP subsequent service family (SAFI).

            If the input `afi` is `ipv4` or `ipv6`, a valid `safi` must be provided.
            """
            vrf: str = "default"
            """
            Optional VRF for IPv4 and IPv6. If not provided, it defaults to `default`.

            If the input `afi` is not `ipv4` or `ipv6`, e.g. `evpn`, `vrf` must be `default`.
            """
            num_peers: PositiveInt
            """Number of expected BGP peer(s)"""

            @model_validator(mode="after")
            def validate_inputs(self: BaseModel) -> BaseModel:
                """
                Validate the inputs provided to the BgpAfi class.

                If afi is either ipv4 or ipv6, safi must be provided.

                If afi is not ipv4 or ipv6, safi must not be provided and vrf must be default.
                """
                if self.afi in ["ipv4", "ipv6"]:
                    if self.safi is None:
                        raise ValueError("'safi' must be provided when afi is ipv4 or ipv6")
                elif self.safi is not None:
                    raise ValueError("'safi' must not be provided when afi is not ipv4 or ipv6")
                elif self.vrf != "default":
                    raise ValueError("'vrf' must be default when afi is not ipv4 or ipv6")
                return self

    def render(self, template: AntaTemplate) -> list[AntaCommand]:
        commands = []
        for afi in self.inputs.address_families:
            if template == VerifyBGPPeerCount.commands[0] and afi.afi in ["ipv4", "ipv6"]:
                commands.append(template.render(afi=afi.afi, safi=afi.safi, vrf=afi.vrf, num_peers=afi.num_peers))
            elif template == VerifyBGPPeerCount.commands[1] and afi.afi not in ["ipv4", "ipv6"]:
                commands.append(template.render(afi=afi.afi, vrf=afi.vrf, num_peers=afi.num_peers))
        return commands

    @AntaTest.anta_test
    def test(self) -> None:
        self.result.is_success()

        failures: dict[tuple[str, Any], dict[str, Any]] = {}

        for command in self.instance_commands:
            peer_count = 0
            command_output = command.json_output

            afi = cast(Afi, command.params.get("afi"))
            safi = cast(Optional[Safi], command.params.get("safi"))
            afi_vrf = cast(str, command.params.get("vrf"))
            num_peers = cast(PositiveInt, command.params.get("num_peers"))

            if not (vrfs := command_output.get("vrfs")):
                _add_bgp_failures(failures=failures, afi=afi, safi=safi, vrf=afi_vrf, issue="Not Configured")
                continue

            if afi_vrf == "all":
                for vrf_data in vrfs.values():
                    peer_count += len(vrf_data["peers"])
            else:
                peer_count += len(command_output["vrfs"][afi_vrf]["peers"])

            if peer_count != num_peers:
                _add_bgp_failures(failures=failures, afi=afi, safi=safi, vrf=afi_vrf, issue=f"Expected: {num_peers}, Actual: {peer_count}")

        if failures:
            self.result.is_failure(f"Failures: {list(failures.values())}")

Input

Bases: Input

Source code in anta/tests/routing/bgp.py

class Input(AntaTest.Input):  # pylint: disable=missing-class-docstring
    address_families: List[BgpAfi]
    """
    List of BGP address families (BgpAfi)
    """

    class BgpAfi(BaseModel):  # pylint: disable=missing-class-docstring
        afi: Afi
        """BGP address family (AFI)"""
        safi: Optional[Safi] = None
        """Optional BGP subsequent service family (SAFI).

        If the input `afi` is `ipv4` or `ipv6`, a valid `safi` must be provided.
        """
        vrf: str = "default"
        """
        Optional VRF for IPv4 and IPv6. If not provided, it defaults to `default`.

        If the input `afi` is not `ipv4` or `ipv6`, e.g. `evpn`, `vrf` must be `default`.
        """
        num_peers: PositiveInt
        """Number of expected BGP peer(s)"""

        @model_validator(mode="after")
        def validate_inputs(self: BaseModel) -> BaseModel:
            """
            Validate the inputs provided to the BgpAfi class.

            If afi is either ipv4 or ipv6, safi must be provided.

            If afi is not ipv4 or ipv6, safi must not be provided and vrf must be default.
            """
            if self.afi in ["ipv4", "ipv6"]:
                if self.safi is None:
                    raise ValueError("'safi' must be provided when afi is ipv4 or ipv6")
            elif self.safi is not None:
                raise ValueError("'safi' must not be provided when afi is not ipv4 or ipv6")
            elif self.vrf != "default":
                raise ValueError("'vrf' must be default when afi is not ipv4 or ipv6")
            return self

address_families instance-attribute

address_families: List[BgpAfi]

List of BGP address families (BgpAfi)

BgpAfi

Bases: BaseModel

Source code in anta/tests/routing/bgp.py

class BgpAfi(BaseModel):  # pylint: disable=missing-class-docstring
    afi: Afi
    """BGP address family (AFI)"""
    safi: Optional[Safi] = None
    """Optional BGP subsequent service family (SAFI).

    If the input `afi` is `ipv4` or `ipv6`, a valid `safi` must be provided.
    """
    vrf: str = "default"
    """
    Optional VRF for IPv4 and IPv6. If not provided, it defaults to `default`.

    If the input `afi` is not `ipv4` or `ipv6`, e.g. `evpn`, `vrf` must be `default`.
    """
    num_peers: PositiveInt
    """Number of expected BGP peer(s)"""

    @model_validator(mode="after")
    def validate_inputs(self: BaseModel) -> BaseModel:
        """
        Validate the inputs provided to the BgpAfi class.

        If afi is either ipv4 or ipv6, safi must be provided.

        If afi is not ipv4 or ipv6, safi must not be provided and vrf must be default.
        """
        if self.afi in ["ipv4", "ipv6"]:
            if self.safi is None:
                raise ValueError("'safi' must be provided when afi is ipv4 or ipv6")
        elif self.safi is not None:
            raise ValueError("'safi' must not be provided when afi is not ipv4 or ipv6")
        elif self.vrf != "default":
            raise ValueError("'vrf' must be default when afi is not ipv4 or ipv6")
        return self

afi instance-attribute

afi: Afi

BGP address family (AFI)

num_peers instance-attribute

num_peers: PositiveInt

Number of expected BGP peer(s)

safi class-attribute instance-attribute

safi: Optional[Safi] = None

Optional BGP subsequent service family (SAFI).

If the input afi is ipv4 or ipv6, a valid safi must be provided.

vrf class-attribute instance-attribute

vrf: str = 'default'

Optional VRF for IPv4 and IPv6. If not provided, it defaults to default.

If the input afi is not ipv4 or ipv6, e.g. evpn, vrf must be default.

validate_inputs

validate_inputs() -> BaseModel

Validate the inputs provided to the BgpAfi class.

If afi is either ipv4 or ipv6, safi must be provided.

If afi is not ipv4 or ipv6, safi must not be provided and vrf must be default.

Source code in anta/tests/routing/bgp.py

@model_validator(mode="after")
def validate_inputs(self: BaseModel) -> BaseModel:
    """
    Validate the inputs provided to the BgpAfi class.

    If afi is either ipv4 or ipv6, safi must be provided.

    If afi is not ipv4 or ipv6, safi must not be provided and vrf must be default.
    """
    if self.afi in ["ipv4", "ipv6"]:
        if self.safi is None:
            raise ValueError("'safi' must be provided when afi is ipv4 or ipv6")
    elif self.safi is not None:
        raise ValueError("'safi' must not be provided when afi is not ipv4 or ipv6")
    elif self.vrf != "default":
        raise ValueError("'vrf' must be default when afi is not ipv4 or ipv6")
    return self

VerifyBGPPeersHealth

Bases: AntaTest

This test verifies the health of BGP peers.

It will validate that all BGP sessions are established and all message queues for these BGP sessions are empty for a given address family.

It supports multiple types of address families (AFI) and subsequent service families (SAFI). Please refer to the Input class attributes below for details.

Expected Results

• success: If all BGP sessions are established and all messages queues are empty for each address family and VRF.
• failure: If there are issues with any of the BGP sessions, or if BGP is not configured for an expected VRF or address family.

Source code in anta/tests/routing/bgp.py

class VerifyBGPPeersHealth(AntaTest):
    """
    This test verifies the health of BGP peers.

    It will validate that all BGP sessions are established and all message queues for these BGP sessions are empty for a given address family.

    It supports multiple types of address families (AFI) and subsequent service families (SAFI).
    Please refer to the Input class attributes below for details.

    Expected Results:
        * success: If all BGP sessions are established and all messages queues are empty for each address family and VRF.
        * failure: If there are issues with any of the BGP sessions, or if BGP is not configured for an expected VRF or address family.
    """

    name = "VerifyBGPPeersHealth"
    description = "Verifies the health of BGP peers"
    categories = ["routing", "bgp"]
    commands = [
        AntaTemplate(template="show bgp {afi} {safi} summary vrf {vrf}"),
        AntaTemplate(template="show bgp {afi} summary"),
    ]

    class Input(AntaTest.Input):  # pylint: disable=missing-class-docstring
        address_families: List[BgpAfi]
        """
        List of BGP address families (BgpAfi)
        """

        class BgpAfi(BaseModel):  # pylint: disable=missing-class-docstring
            afi: Afi
            """BGP address family (AFI)"""
            safi: Optional[Safi] = None
            """Optional BGP subsequent service family (SAFI).

            If the input `afi` is `ipv4` or `ipv6`, a valid `safi` must be provided.
            """
            vrf: str = "default"
            """
            Optional VRF for IPv4 and IPv6. If not provided, it defaults to `default`.

            If the input `afi` is not `ipv4` or `ipv6`, e.g. `evpn`, `vrf` must be `default`.
            """

            @model_validator(mode="after")
            def validate_inputs(self: BaseModel) -> BaseModel:
                """
                Validate the inputs provided to the BgpAfi class.

                If afi is either ipv4 or ipv6, safi must be provided.

                If afi is not ipv4 or ipv6, safi must not be provided and vrf must be default.
                """
                if self.afi in ["ipv4", "ipv6"]:
                    if self.safi is None:
                        raise ValueError("'safi' must be provided when afi is ipv4 or ipv6")
                elif self.safi is not None:
                    raise ValueError("'safi' must not be provided when afi is not ipv4 or ipv6")
                elif self.vrf != "default":
                    raise ValueError("'vrf' must be default when afi is not ipv4 or ipv6")
                return self

    def render(self, template: AntaTemplate) -> list[AntaCommand]:
        commands = []
        for afi in self.inputs.address_families:
            if template == VerifyBGPPeersHealth.commands[0] and afi.afi in ["ipv4", "ipv6"]:
                commands.append(template.render(afi=afi.afi, safi=afi.safi, vrf=afi.vrf))
            elif template == VerifyBGPPeersHealth.commands[1] and afi.afi not in ["ipv4", "ipv6"]:
                commands.append(template.render(afi=afi.afi, vrf=afi.vrf))
        return commands

    @AntaTest.anta_test
    def test(self) -> None:
        self.result.is_success()

        failures: dict[tuple[str, Any], dict[str, Any]] = {}

        for command in self.instance_commands:
            command_output = command.json_output

            afi = cast(Afi, command.params.get("afi"))
            safi = cast(Optional[Safi], command.params.get("safi"))
            afi_vrf = cast(str, command.params.get("vrf"))

            if not (vrfs := command_output.get("vrfs")):
                _add_bgp_failures(failures=failures, afi=afi, safi=safi, vrf=afi_vrf, issue="Not Configured")
                continue

            for vrf, vrf_data in vrfs.items():
                if not (peers := vrf_data.get("peers")):
                    _add_bgp_failures(failures=failures, afi=afi, safi=safi, vrf=afi_vrf, issue="No Peers")
                    continue

                peer_issues = {}
                for peer, peer_data in peers.items():
                    issues = _check_peer_issues(peer_data)

                    if issues:
                        peer_issues[peer] = issues

                if peer_issues:
                    _add_bgp_failures(failures=failures, afi=afi, safi=safi, vrf=vrf, issue=peer_issues)

        if failures:
            self.result.is_failure(f"Failures: {list(failures.values())}")

Input

Bases: Input

Source code in anta/tests/routing/bgp.py

class Input(AntaTest.Input):  # pylint: disable=missing-class-docstring
    address_families: List[BgpAfi]
    """
    List of BGP address families (BgpAfi)
    """

    class BgpAfi(BaseModel):  # pylint: disable=missing-class-docstring
        afi: Afi
        """BGP address family (AFI)"""
        safi: Optional[Safi] = None
        """Optional BGP subsequent service family (SAFI).

        If the input `afi` is `ipv4` or `ipv6`, a valid `safi` must be provided.
        """
        vrf: str = "default"
        """
        Optional VRF for IPv4 and IPv6. If not provided, it defaults to `default`.

        If the input `afi` is not `ipv4` or `ipv6`, e.g. `evpn`, `vrf` must be `default`.
        """

        @model_validator(mode="after")
        def validate_inputs(self: BaseModel) -> BaseModel:
            """
            Validate the inputs provided to the BgpAfi class.

            If afi is either ipv4 or ipv6, safi must be provided.

            If afi is not ipv4 or ipv6, safi must not be provided and vrf must be default.
            """
            if self.afi in ["ipv4", "ipv6"]:
                if self.safi is None:
                    raise ValueError("'safi' must be provided when afi is ipv4 or ipv6")
            elif self.safi is not None:
                raise ValueError("'safi' must not be provided when afi is not ipv4 or ipv6")
            elif self.vrf != "default":
                raise ValueError("'vrf' must be default when afi is not ipv4 or ipv6")
            return self

address_families instance-attribute

address_families: List[BgpAfi]

List of BGP address families (BgpAfi)

BgpAfi

Bases: BaseModel

Source code in anta/tests/routing/bgp.py

class BgpAfi(BaseModel):  # pylint: disable=missing-class-docstring
    afi: Afi
    """BGP address family (AFI)"""
    safi: Optional[Safi] = None
    """Optional BGP subsequent service family (SAFI).

    If the input `afi` is `ipv4` or `ipv6`, a valid `safi` must be provided.
    """
    vrf: str = "default"
    """
    Optional VRF for IPv4 and IPv6. If not provided, it defaults to `default`.

    If the input `afi` is not `ipv4` or `ipv6`, e.g. `evpn`, `vrf` must be `default`.
    """

    @model_validator(mode="after")
    def validate_inputs(self: BaseModel) -> BaseModel:
        """
        Validate the inputs provided to the BgpAfi class.

        If afi is either ipv4 or ipv6, safi must be provided.

        If afi is not ipv4 or ipv6, safi must not be provided and vrf must be default.
        """
        if self.afi in ["ipv4", "ipv6"]:
            if self.safi is None:
                raise ValueError("'safi' must be provided when afi is ipv4 or ipv6")
        elif self.safi is not None:
            raise ValueError("'safi' must not be provided when afi is not ipv4 or ipv6")
        elif self.vrf != "default":
            raise ValueError("'vrf' must be default when afi is not ipv4 or ipv6")
        return self

afi instance-attribute

afi: Afi

BGP address family (AFI)

safi class-attribute instance-attribute

safi: Optional[Safi] = None

Optional BGP subsequent service family (SAFI).

If the input afi is ipv4 or ipv6, a valid safi must be provided.

vrf class-attribute instance-attribute

vrf: str = 'default'

Optional VRF for IPv4 and IPv6. If not provided, it defaults to default.

If the input afi is not ipv4 or ipv6, e.g. evpn, vrf must be default.

validate_inputs

validate_inputs() -> BaseModel

Validate the inputs provided to the BgpAfi class.

If afi is either ipv4 or ipv6, safi must be provided.

If afi is not ipv4 or ipv6, safi must not be provided and vrf must be default.

Source code in anta/tests/routing/bgp.py

@model_validator(mode="after")
def validate_inputs(self: BaseModel) -> BaseModel:
    """
    Validate the inputs provided to the BgpAfi class.

    If afi is either ipv4 or ipv6, safi must be provided.

    If afi is not ipv4 or ipv6, safi must not be provided and vrf must be default.
    """
    if self.afi in ["ipv4", "ipv6"]:
        if self.safi is None:
            raise ValueError("'safi' must be provided when afi is ipv4 or ipv6")
    elif self.safi is not None:
        raise ValueError("'safi' must not be provided when afi is not ipv4 or ipv6")
    elif self.vrf != "default":
        raise ValueError("'vrf' must be default when afi is not ipv4 or ipv6")
    return self

VerifyBGPRTCCount

Bases: 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.

Source code in anta/tests/routing/bgp.py

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

Input

Bases: Input

Source code in anta/tests/routing/bgp.py

class Input(AntaTest.Input):  # pylint: disable=missing-class-docstring
    number: int
    """The expected number of BGP RTC neighbors in the default VRF"""

number instance-attribute

number: int

The expected number of BGP RTC neighbors in the default VRF

VerifyBGPRTCState

Bases: 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.

Source code in anta/tests/routing/bgp.py

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

VerifyBGPSpecificPeers

Bases: AntaTest

This test verifies the health of specific BGP peer(s).

It will validate that the BGP session is established and all message queues for this BGP session are empty for the given peer(s).

It supports multiple types of address families (AFI) and subsequent service families (SAFI). Please refer to the Input class attributes below for details.

Expected Results

• success: If the BGP session is established and all messages queues are empty for each given peer.
• failure: If the BGP session has issues or is not configured, or if BGP is not configured for an expected VRF or address family.

Source code in anta/tests/routing/bgp.py

class VerifyBGPSpecificPeers(AntaTest):
    """
    This test verifies the health of specific BGP peer(s).

    It will validate that the BGP session is established and all message queues for this BGP session are empty for the given peer(s).

    It supports multiple types of address families (AFI) and subsequent service families (SAFI).
    Please refer to the Input class attributes below for details.

    Expected Results:
        * success: If the BGP session is established and all messages queues are empty for each given peer.
        * failure: If the BGP session has issues or is not configured, or if BGP is not configured for an expected VRF or address family.
    """

    name = "VerifyBGPSpecificPeers"
    description = "Verifies the health of specific BGP peer(s)."
    categories = ["routing", "bgp"]
    commands = [
        AntaTemplate(template="show bgp {afi} {safi} summary vrf {vrf}"),
        AntaTemplate(template="show bgp {afi} summary"),
    ]

    class Input(AntaTest.Input):  # pylint: disable=missing-class-docstring
        address_families: List[BgpAfi]
        """
        List of BGP address families (BgpAfi)
        """

        class BgpAfi(BaseModel):  # pylint: disable=missing-class-docstring
            afi: Afi
            """BGP address family (AFI)"""
            safi: Optional[Safi] = None
            """Optional BGP subsequent service family (SAFI).

            If the input `afi` is `ipv4` or `ipv6`, a valid `safi` must be provided.
            """
            vrf: str = "default"
            """
            Optional VRF for IPv4 and IPv6. If not provided, it defaults to `default`.

            `all` is NOT supported.

            If the input `afi` is not `ipv4` or `ipv6`, e.g. `evpn`, `vrf` must be `default`.
            """
            peers: List[Union[IPv4Address, IPv6Address]]
            """List of BGP IPv4 or IPv6 peer"""

            @model_validator(mode="after")
            def validate_inputs(self: BaseModel) -> BaseModel:
                """
                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.

                If afi is not ipv4 or ipv6, safi must not be provided and vrf must be default.
                """
                if self.afi in ["ipv4", "ipv6"]:
                    if self.safi is None:
                        raise ValueError("'safi' must be provided when afi is ipv4 or ipv6")
                    if self.vrf == "all":
                        raise ValueError("'all' is not supported in this test. Use VerifyBGPPeersHealth test instead.")
                elif self.safi is not None:
                    raise ValueError("'safi' must not be provided when afi is not ipv4 or ipv6")
                elif self.vrf != "default":
                    raise ValueError("'vrf' must be default when afi is not ipv4 or ipv6")
                return self

    def render(self, template: AntaTemplate) -> list[AntaCommand]:
        commands = []
        for afi in self.inputs.address_families:
            if template == VerifyBGPSpecificPeers.commands[0] and afi.afi in ["ipv4", "ipv6"]:
                commands.append(template.render(afi=afi.afi, safi=afi.safi, vrf=afi.vrf, peers=afi.peers))
            elif template == VerifyBGPSpecificPeers.commands[1] and afi.afi not in ["ipv4", "ipv6"]:
                commands.append(template.render(afi=afi.afi, vrf=afi.vrf, peers=afi.peers))
        return commands

    @AntaTest.anta_test
    def test(self) -> None:
        self.result.is_success()

        failures: dict[tuple[str, Any], dict[str, Any]] = {}

        for command in self.instance_commands:
            command_output = command.json_output

            afi = cast(Afi, command.params.get("afi"))
            safi = cast(Optional[Safi], command.params.get("safi"))
            afi_vrf = cast(str, command.params.get("vrf"))
            afi_peers = cast(List[Union[IPv4Address, IPv6Address]], command.params.get("peers", []))

            if not (vrfs := command_output.get("vrfs")):
                _add_bgp_failures(failures=failures, afi=afi, safi=safi, vrf=afi_vrf, issue="Not Configured")
                continue

            peer_issues = {}
            for peer in afi_peers:
                peer_ip = str(peer)
                peer_data = get_value(dictionary=vrfs, key=f"{afi_vrf}_peers_{peer_ip}", separator="_")
                issues = _check_peer_issues(peer_data)
                if issues:
                    peer_issues[peer_ip] = issues

            if peer_issues:
                _add_bgp_failures(failures=failures, afi=afi, safi=safi, vrf=afi_vrf, issue=peer_issues)

        if failures:
            self.result.is_failure(f"Failures: {list(failures.values())}")

Input

Bases: Input

Source code in anta/tests/routing/bgp.py

class Input(AntaTest.Input):  # pylint: disable=missing-class-docstring
    address_families: List[BgpAfi]
    """
    List of BGP address families (BgpAfi)
    """

    class BgpAfi(BaseModel):  # pylint: disable=missing-class-docstring
        afi: Afi
        """BGP address family (AFI)"""
        safi: Optional[Safi] = None
        """Optional BGP subsequent service family (SAFI).

        If the input `afi` is `ipv4` or `ipv6`, a valid `safi` must be provided.
        """
        vrf: str = "default"
        """
        Optional VRF for IPv4 and IPv6. If not provided, it defaults to `default`.

        `all` is NOT supported.

        If the input `afi` is not `ipv4` or `ipv6`, e.g. `evpn`, `vrf` must be `default`.
        """
        peers: List[Union[IPv4Address, IPv6Address]]
        """List of BGP IPv4 or IPv6 peer"""

        @model_validator(mode="after")
        def validate_inputs(self: BaseModel) -> BaseModel:
            """
            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.

            If afi is not ipv4 or ipv6, safi must not be provided and vrf must be default.
            """
            if self.afi in ["ipv4", "ipv6"]:
                if self.safi is None:
                    raise ValueError("'safi' must be provided when afi is ipv4 or ipv6")
                if self.vrf == "all":
                    raise ValueError("'all' is not supported in this test. Use VerifyBGPPeersHealth test instead.")
            elif self.safi is not None:
                raise ValueError("'safi' must not be provided when afi is not ipv4 or ipv6")
            elif self.vrf != "default":
                raise ValueError("'vrf' must be default when afi is not ipv4 or ipv6")
            return self

address_families instance-attribute

address_families: List[BgpAfi]

List of BGP address families (BgpAfi)

BgpAfi

Bases: BaseModel

Source code in anta/tests/routing/bgp.py

class BgpAfi(BaseModel):  # pylint: disable=missing-class-docstring
    afi: Afi
    """BGP address family (AFI)"""
    safi: Optional[Safi] = None
    """Optional BGP subsequent service family (SAFI).

    If the input `afi` is `ipv4` or `ipv6`, a valid `safi` must be provided.
    """
    vrf: str = "default"
    """
    Optional VRF for IPv4 and IPv6. If not provided, it defaults to `default`.

    `all` is NOT supported.

    If the input `afi` is not `ipv4` or `ipv6`, e.g. `evpn`, `vrf` must be `default`.
    """
    peers: List[Union[IPv4Address, IPv6Address]]
    """List of BGP IPv4 or IPv6 peer"""

    @model_validator(mode="after")
    def validate_inputs(self: BaseModel) -> BaseModel:
        """
        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.

        If afi is not ipv4 or ipv6, safi must not be provided and vrf must be default.
        """
        if self.afi in ["ipv4", "ipv6"]:
            if self.safi is None:
                raise ValueError("'safi' must be provided when afi is ipv4 or ipv6")
            if self.vrf == "all":
                raise ValueError("'all' is not supported in this test. Use VerifyBGPPeersHealth test instead.")
        elif self.safi is not None:
            raise ValueError("'safi' must not be provided when afi is not ipv4 or ipv6")
        elif self.vrf != "default":
            raise ValueError("'vrf' must be default when afi is not ipv4 or ipv6")
        return self

afi instance-attribute

afi: Afi

BGP address family (AFI)

peers instance-attribute

peers: List[Union[IPv4Address, IPv6Address]]

List of BGP IPv4 or IPv6 peer

safi class-attribute instance-attribute

safi: Optional[Safi] = None

Optional BGP subsequent service family (SAFI).

If the input afi is ipv4 or ipv6, a valid safi must be provided.

vrf class-attribute instance-attribute

vrf: str = 'default'

Optional VRF for IPv4 and IPv6. If not provided, it defaults to default.

all is NOT supported.

If the input afi is not ipv4 or ipv6, e.g. evpn, vrf must be default.

validate_inputs

validate_inputs() -> BaseModel

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.

If afi is not ipv4 or ipv6, safi must not be provided and vrf must be default.

Source code in anta/tests/routing/bgp.py

@model_validator(mode="after")
def validate_inputs(self: BaseModel) -> BaseModel:
    """
    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.

    If afi is not ipv4 or ipv6, safi must not be provided and vrf must be default.
    """
    if self.afi in ["ipv4", "ipv6"]:
        if self.safi is None:
            raise ValueError("'safi' must be provided when afi is ipv4 or ipv6")
        if self.vrf == "all":
            raise ValueError("'all' is not supported in this test. Use VerifyBGPPeersHealth test instead.")
    elif self.safi is not None:
        raise ValueError("'safi' must not be provided when afi is not ipv4 or ipv6")
    elif self.vrf != "default":
        raise ValueError("'vrf' must be default when afi is not ipv4 or ipv6")
    return self

---

Last update: September 13, 2023