ANTA catalog for BGP tests

BGP Test Compatibility Note

ANTA BGP tests are designed for the multi-agent routing protocol model. Starting from EOS 4.30.1F, service routing protocols models is set to multi-agent by default, and from EOS 4.32.0F it becomes the only supported model.

The following tests are available for devices using the legacy ribd model on earlier EOS versions:

VerifyBGPPeerSessionRibd
VerifyBGPPeersHealthRibd

Tests¶

Module related to BGP tests.

VerifyBGPAdvCommunities ¶

Verifies that advertised communities are standard, extended and large for BGP IPv4 peer(s).

This test performs the following checks for each specified peer:

Verifies that the peer is found in its VRF in the BGP configuration.
Validates that all required community types are advertised:
- Standard communities
- Extended communities
- Large communities

Expected Results

Success: If all of the following conditions are met:
- All specified peers are found in the BGP configuration.
- Each peer advertises standard, extended and large communities.
Failure: If any of the following occur:
- A specified peer is not found in the BGP configuration.
- A peer does not advertise standard, extended or large communities.

Examples

anta.tests.routing:
  bgp:
    - VerifyBGPAdvCommunities:
        bgp_peers:
          - peer_address: 172.30.11.17
            vrf: default
          - peer_address: 172.30.11.21
            vrf: default

Inputs¶

Name	Type	Description	Default
`bgp_peers`	`list[BgpPeer]`	List of BGP IPv4 peers.	-

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

class VerifyBGPAdvCommunities(AntaTest):
    """Verifies that advertised communities are standard, extended and large for BGP IPv4 peer(s).

    This test performs the following checks for each specified peer:

      1. Verifies that the peer is found in its VRF in the BGP configuration.
      2. Validates that all required community types are advertised:
        - Standard communities
        - Extended communities
        - Large communities

    Expected Results
    ----------------
    * Success: If all of the following conditions are met:
        - All specified peers are found in the BGP configuration.
        - Each peer advertises standard, extended and large communities.
    * Failure: If any of the following occur:
        - A specified peer is not found in the BGP configuration.
        - A peer does not advertise standard, extended or large communities.

    Examples
    --------
    ```yaml
    anta.tests.routing:
      bgp:
        - VerifyBGPAdvCommunities:
            bgp_peers:
              - peer_address: 172.30.11.17
                vrf: default
              - peer_address: 172.30.11.21
                vrf: default
    ```
    """

    categories: ClassVar[list[str]] = ["bgp"]
    commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show bgp neighbors vrf all", revision=3)]

    class Input(AntaTest.Input):
        """Input model for the VerifyBGPAdvCommunities test."""

        bgp_peers: list[BgpPeer]
        """List of BGP IPv4 peers."""
        BgpPeer: ClassVar[type[BgpPeer]] = BgpPeer

    @AntaTest.anta_test
    def test(self) -> None:
        """Main test function for VerifyBGPAdvCommunities."""
        self.result.is_success()

        output = self.instance_commands[0].json_output

        for peer in self.inputs.bgp_peers:
            peer_ip = str(peer.peer_address)
            peer_list = get_value(output, f"vrfs.{peer.vrf}.peerList", default=[])

            # Check if the peer is found
            if (peer_data := get_item(peer_list, "peerAddress", peer_ip)) is None:
                self.result.is_failure(f"{peer} - Not found")
                continue

            # Check BGP peer advertised communities
            if not all(get_value(peer_data, f"advertisedCommunities.{community}") is True for community in ["standard", "extended", "large"]):
                self.result.is_failure(f"{peer} - {format_data(peer_data['advertisedCommunities'])}")

VerifyBGPExchangedRoutes ¶

Verifies the advertised and received routes of BGP IPv4 peer(s).

This test performs the following checks for each specified peer:

For each advertised and received route: - Confirms that the route exists in the BGP route table. - Verifies that the route is in an ‘active’ and ‘valid’ state.

Expected Results

Success: If all of the following conditions are met:
- All specified advertised/received routes are found in the BGP route table.
- All routes are in both ‘active’ and ‘valid’ states.
Failure: If any of the following occur:
- An advertised/received route is not found in the BGP route table.
- Any route is not in an ‘active’ or ‘valid’ state.

Examples

anta.tests.routing:
  bgp:
    - VerifyBGPExchangedRoutes:
        bgp_peers:
          - peer_address: 172.30.255.5
            vrf: default
            advertised_routes:
              - 192.0.254.5/32
            received_routes:
              - 192.0.255.4/32
          - peer_address: 172.30.255.1
            vrf: default
            advertised_routes:
              - 192.0.255.1/32
              - 192.0.254.5/32

Inputs¶

Name	Type	Description	Default
`bgp_peers`	`list[BgpPeer]`	List of BGP IPv4 peers.	-

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

class VerifyBGPExchangedRoutes(AntaTest):
    """Verifies the advertised and received routes of BGP IPv4 peer(s).

    This test performs the following checks for each specified peer:

      For each advertised and received route:
        - Confirms that the route exists in the BGP route table.
        - Verifies that the route is in an 'active' and 'valid' state.

    Expected Results
    ----------------
    * Success: If all of the following conditions are met:
        - All specified advertised/received routes are found in the BGP route table.
        - All routes are in both 'active' and 'valid' states.
    * Failure: If any of the following occur:
        - An advertised/received route is not found in the BGP route table.
        - Any route is not in an 'active' or 'valid' state.

    Examples
    --------
    ```yaml
    anta.tests.routing:
      bgp:
        - VerifyBGPExchangedRoutes:
            bgp_peers:
              - peer_address: 172.30.255.5
                vrf: default
                advertised_routes:
                  - 192.0.254.5/32
                received_routes:
                  - 192.0.255.4/32
              - peer_address: 172.30.255.1
                vrf: default
                advertised_routes:
                  - 192.0.255.1/32
                  - 192.0.254.5/32
    ```
    """

    categories: ClassVar[list[str]] = ["bgp"]
    commands: ClassVar[list[AntaCommand | AntaTemplate]] = [
        AntaTemplate(template="show bgp neighbors {peer} advertised-routes vrf {vrf}", revision=3),
        AntaTemplate(template="show bgp neighbors {peer} routes vrf {vrf}", revision=3),
    ]

    class Input(AntaTest.Input):
        """Input model for the VerifyBGPExchangedRoutes test."""

        bgp_peers: list[BgpPeer]
        """List of BGP IPv4 peers."""
        BgpNeighbor: ClassVar[type[BgpNeighbor]] = BgpNeighbor

        @field_validator("bgp_peers")
        @classmethod
        def validate_bgp_peers(cls, bgp_peers: list[BgpPeer]) -> list[BgpPeer]:
            """Validate that 'advertised_routes' or 'received_routes' field is provided in each BGP peer."""
            for peer in bgp_peers:
                if peer.advertised_routes is None and peer.received_routes is None:
                    msg = f"{peer} 'advertised_routes' or 'received_routes' field missing in the input"
                    raise ValueError(msg)
            return bgp_peers

    def render(self, template: AntaTemplate) -> list[AntaCommand]:
        """Render the template for each BGP peer in the input list."""
        return [template.render(peer=str(bgp_peer.peer_address), vrf=bgp_peer.vrf) for bgp_peer in self.inputs.bgp_peers]

    def _validate_bgp_route_paths(self, peer: str, route_type: str, route: str, entries: dict[str, Any]) -> str | None:
        """Validate the BGP route paths."""
        # Check if the route is found
        if route in entries:
            # Check if the route is active and valid
            route_paths = entries[route]["bgpRoutePaths"][0]["routeType"]
            is_active = route_paths["active"]
            is_valid = route_paths["valid"]
            if not is_active or not is_valid:
                return f"{peer} {route_type} route: {route} - Valid: {is_valid} Active: {is_active}"
            return None

        return f"{peer} {route_type} route: {route} - Not found"

    @AntaTest.anta_test
    def test(self) -> None:
        """Main test function for VerifyBGPExchangedRoutes."""
        self.result.is_success()

        num_peers = len(self.inputs.bgp_peers)

        # Process each peer and its corresponding command pair
        for peer_idx, peer in enumerate(self.inputs.bgp_peers):
            # For n peers, advertised routes are at indices 0 to n-1, and received routes are at indices n to 2n-1
            advertised_routes_cmd = self.instance_commands[peer_idx]
            received_routes_cmd = self.instance_commands[peer_idx + num_peers]

            # Get the BGP route entries of each command
            command_output = {
                "Advertised": get_value(advertised_routes_cmd.json_output, f"vrfs.{peer.vrf}.bgpRouteEntries", default={}),
                "Received": get_value(received_routes_cmd.json_output, f"vrfs.{peer.vrf}.bgpRouteEntries", default={}),
            }

            # Validate both advertised and received routes
            for route_type, routes in zip(["Advertised", "Received"], [peer.advertised_routes, peer.received_routes]):
                # Skipping the validation for routes if user input is None
                if not routes:
                    continue

                entries = command_output[route_type]
                for route in routes:
                    # Check if the route is found. If yes then checks the route is active and valid
                    failure_msg = self._validate_bgp_route_paths(str(peer), route_type, str(route), entries)
                    if failure_msg:
                        self.result.is_failure(failure_msg)

VerifyBGPNlriAcceptance ¶

Verifies that all received NLRI are accepted for all AFI/SAFI configured for BGP IPv4 peer(s).

This test performs the following checks for each specified peer:

Verifies that the peer is found in its VRF in the BGP configuration.
Verifies that all received NLRI were accepted by comparing nlrisReceived with nlrisAccepted.

Expected Results

Success: If nlrisReceived equals nlrisAccepted, indicating all NLRI were accepted.
Failure: If any of the following occur:
- The specified VRF is not configured.
- nlrisReceived does not equal nlrisAccepted, indicating some NLRI were rejected or filtered.

Examples

anta.tests.routing:
  bgp:
    - VerifyBGPNlriAcceptance:
        bgp_peers:
          - peer_address: 10.100.0.128
            vrf: default
            capabilities:
              - ipv4Unicast

Inputs¶

Name	Type	Description	Default
`bgp_peers`	`list[BgpPeer]`	List of BGP IPv4 peers.	-

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

class VerifyBGPNlriAcceptance(AntaTest):
    """Verifies that all received NLRI are accepted for all AFI/SAFI configured for BGP IPv4 peer(s).

    This test performs the following checks for each specified peer:

      1. Verifies that the peer is found in its VRF in the BGP configuration.
      2. Verifies that all received NLRI were accepted by comparing `nlrisReceived` with `nlrisAccepted`.

    Expected Results
    ----------------
    * Success: If `nlrisReceived` equals `nlrisAccepted`, indicating all NLRI were accepted.
    * Failure: If any of the following occur:
        - The specified VRF is not configured.
        - `nlrisReceived` does not equal `nlrisAccepted`, indicating some NLRI were rejected or filtered.

    Examples
    --------
    ```yaml
    anta.tests.routing:
      bgp:
        - VerifyBGPNlriAcceptance:
            bgp_peers:
              - peer_address: 10.100.0.128
                vrf: default
                capabilities:
                  - ipv4Unicast
    ```
    """

    categories: ClassVar[list[str]] = ["bgp"]
    commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show bgp summary vrf all", revision=1)]

    class Input(AntaTest.Input):
        """Input model for the VerifyBGPNlriAcceptance test."""

        bgp_peers: list[BgpPeer]
        """List of BGP IPv4 peers."""

        @field_validator("bgp_peers")
        @classmethod
        def validate_bgp_peers(cls, bgp_peers: list[T]) -> list[T]:
            """Validate that 'capabilities' field is provided in each BGP peer."""
            for peer in bgp_peers:
                if peer.capabilities is None:
                    msg = f"{peer} 'capabilities' field missing in the input"
                    raise ValueError(msg)
            return bgp_peers

    @AntaTest.anta_test
    def test(self) -> None:
        """Main test function for VerifyBGPNlriAcceptance."""
        self.result.is_success()

        output = self.instance_commands[0].json_output

        for peer in self.inputs.bgp_peers:
            # Check if the peer is found
            if not (peer_data := get_value(output, f"vrfs..{peer.vrf}..peers..{peer.peer_address}", separator="..")):
                self.result.is_failure(f"{peer} - Not found")
                continue

            # Fetching the multiprotocol capabilities
            for capability in peer.capabilities:
                # Check if the capability is found
                if (capability_status := get_value(peer_data, capability)) is None:
                    self.result.is_failure(f"{peer} - {capability} not found")
                    continue

                if capability_status["afiSafiState"] != "negotiated":
                    self.result.is_failure(f"{peer} - {capability} not negotiated")

                if (received := capability_status.get("nlrisReceived")) != (accepted := capability_status.get("nlrisAccepted")):
                    self.result.is_failure(f"{peer} AFI/SAFI: {capability} - Some NLRI were filtered or rejected - Accepted: {accepted} Received: {received}")

VerifyBGPPeerASNCap ¶

Verifies the four octet ASN capability of BGP IPv4 peer(s).

This test performs the following checks for each specified peer:

Verifies that the peer is found in its VRF in the BGP configuration.
Validates that the capability is present in the peer configuration.
Confirms that the capability is advertised, received, and enabled.

Expected Results

Success: If all of the following conditions are met:
- All specified peers are found in the BGP configuration.
- The four octet ASN capability is present in each peer configuration.
- The capability is properly negotiated (advertised, received, and enabled) for all peers.
Failure: If any of the following occur:
- A specified peer is not found in the BGP configuration.
- The four octet ASN capability is not present for a peer.
- The capability is not properly negotiated (not advertised, received, or enabled) for any peer.

Examples

anta.tests.routing:
  bgp:
    - VerifyBGPPeerASNCap:
        bgp_peers:
          - peer_address: 172.30.11.1
            vrf: default

Inputs¶

Name	Type	Description	Default
`bgp_peers`	`list[BgpPeer]`	List of BGP IPv4 peers.	-

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

class VerifyBGPPeerASNCap(AntaTest):
    """Verifies the four octet ASN capability of BGP IPv4 peer(s).

    This test performs the following checks for each specified peer:

      1. Verifies that the peer is found in its VRF in the BGP configuration.
      2. Validates that the capability is present in the peer configuration.
      3. Confirms that the capability is advertised, received, and enabled.

    Expected Results
    ----------------
    * Success: If all of the following conditions are met:
        - All specified peers are found in the BGP configuration.
        - The four octet ASN capability is present in each peer configuration.
        - The capability is properly negotiated (advertised, received, and enabled) for all peers.
    * Failure: If any of the following occur:
        - A specified peer is not found in the BGP configuration.
        - The four octet ASN capability is not present for a peer.
        - The capability is not properly negotiated (not advertised, received, or enabled) for any peer.

    Examples
    --------
    ```yaml
    anta.tests.routing:
      bgp:
        - VerifyBGPPeerASNCap:
            bgp_peers:
              - peer_address: 172.30.11.1
                vrf: default
    ```
    """

    categories: ClassVar[list[str]] = ["bgp"]
    commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show bgp neighbors vrf all", revision=3)]

    class Input(AntaTest.Input):
        """Input model for the VerifyBGPPeerASNCap test."""

        bgp_peers: list[BgpPeer]
        """List of BGP IPv4 peers."""
        BgpPeer: ClassVar[type[BgpPeer]] = BgpPeer

    @AntaTest.anta_test
    def test(self) -> None:
        """Main test function for VerifyBGPPeerASNCap."""
        self.result.is_success()

        output = self.instance_commands[0].json_output

        for peer in self.inputs.bgp_peers:
            peer_ip = str(peer.peer_address)
            peer_list = get_value(output, f"vrfs.{peer.vrf}.peerList", default=[])

            # Check if the peer is found
            if (peer_data := get_item(peer_list, "peerAddress", peer_ip)) is None:
                self.result.is_failure(f"{peer} - Not found")
                continue

            # Check if the 4-octet ASN capability is found
            if (capablity_status := get_value(peer_data, "neighborCapabilities.fourOctetAsnCap")) is None:
                self.result.is_failure(f"{peer} - 4-octet ASN capability not found")
                continue

            # Check if the 4-octet ASN capability is advertised, received, and enabled
            if not _check_bgp_neighbor_capability(capablity_status):
                self.result.is_failure(f"{peer} - 4-octet ASN capability not negotiated - {format_data(capablity_status)}")

VerifyBGPPeerCount ¶

Verifies the count of BGP peers for given address families.

This test performs the following checks for each specified address family:

Confirms that the specified VRF is configured.
Counts the number of peers that are:
- If check_peer_state is set to True, Counts the number of BGP peers that are in the Established state and have successfully negotiated the specified AFI/SAFI
- If check_peer_state is set to False, skips validation of the Established state and AFI/SAFI negotiation.

Expected Results

Success: If the count of BGP peers matches the expected count with check_peer_state enabled/disabled.
Failure: If any of the following occur:
- The specified VRF is not configured.
- The BGP peer count does not match expected value with check_peer_state enabled/disabled.”

Examples

anta.tests.routing:
  bgp:
    - VerifyBGPPeerCount:
        address_families:
          - afi: "evpn"
            num_peers: 2
          - afi: "ipv4"
            safi: "unicast"
            vrf: "PROD"
            num_peers: 2
          - afi: "ipv4"
            safi: "unicast"
            vrf: "default"
            num_peers: 3
          - afi: "ipv4"
            safi: "multicast"
            vrf: "DEV"
            num_peers: 3

Inputs¶

Name	Type	Description	Default
`address_families`	`list[BgpAddressFamily]`	List of BGP address families.	-

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

class VerifyBGPPeerCount(AntaTest):
    """Verifies the count of BGP peers for given address families.

    This test performs the following checks for each specified address family:

      1. Confirms that the specified VRF is configured.
      2. Counts the number of peers that are:
        - If `check_peer_state` is set to True, Counts the number of BGP peers that are in the `Established` state and
        have successfully negotiated the specified AFI/SAFI
        - If `check_peer_state` is set to False, skips validation of the `Established` state and AFI/SAFI negotiation.

    Expected Results
    ----------------
    * Success: If the count of BGP peers matches the expected count with `check_peer_state` enabled/disabled.
    * Failure: If any of the following occur:
        - The specified VRF is not configured.
        - The BGP peer count does not match expected value with `check_peer_state` enabled/disabled."

    Examples
    --------
    ```yaml
    anta.tests.routing:
      bgp:
        - VerifyBGPPeerCount:
            address_families:
              - afi: "evpn"
                num_peers: 2
              - afi: "ipv4"
                safi: "unicast"
                vrf: "PROD"
                num_peers: 2
              - afi: "ipv4"
                safi: "unicast"
                vrf: "default"
                num_peers: 3
              - afi: "ipv4"
                safi: "multicast"
                vrf: "DEV"
                num_peers: 3
    ```
    """

    categories: ClassVar[list[str]] = ["bgp"]
    commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show bgp summary vrf all", revision=1)]

    class Input(AntaTest.Input):
        """Input model for the VerifyBGPPeerCount test."""

        address_families: list[BgpAddressFamily]
        """List of BGP address families."""
        BgpAfi: ClassVar[type[BgpAfi]] = BgpAfi

        @field_validator("address_families")
        @classmethod
        def validate_address_families(cls, address_families: list[BgpAddressFamily]) -> list[BgpAddressFamily]:
            """Validate that 'num_peers' field is provided in each address family."""
            for af in address_families:
                if af.num_peers is None:
                    msg = f"{af} 'num_peers' field missing in the input"
                    raise ValueError(msg)
            return address_families

    @AntaTest.anta_test
    def test(self) -> None:
        """Main test function for VerifyBGPPeerCount."""
        self.result.is_success()

        output = self.instance_commands[0].json_output

        for address_family in self.inputs.address_families:
            # Check if the VRF is configured
            if (vrf_output := get_value(output, f"vrfs.{address_family.vrf}")) is None:
                self.result.is_failure(f"{address_family} - VRF not configured")
                continue

            peers_data = vrf_output.get("peers", {}).values()
            if not address_family.check_peer_state:
                # Count the number of peers without considering the state and negotiated AFI/SAFI check if the count matches the expected count
                peer_count = sum(1 for peer_data in peers_data if address_family.eos_key in peer_data)
            else:
                # Count the number of established peers with negotiated AFI/SAFI
                peer_count = sum(
                    1
                    for peer_data in peers_data
                    if peer_data.get("peerState") == "Established" and get_value(peer_data, f"{address_family.eos_key}.afiSafiState") == "negotiated"
                )

            # Check if the count matches the expected count
            if address_family.num_peers != peer_count:
                self.result.is_failure(f"{address_family} - Peer count mismatch - Expected: {address_family.num_peers} Actual: {peer_count}")

VerifyBGPPeerDropStats ¶

Verifies BGP NLRI drop statistics for the provided BGP IPv4 peer(s).

This test performs the following checks for each specified peer:

Verifies that the peer is found in its VRF in the BGP configuration.
Validates the BGP drop statistics:
- If specific drop statistics are provided, checks only those counters.
- If no specific drop statistics are provided, checks all available counters.
- Confirms that all checked counters have a value of zero.

Expected Results

Success: If all of the following conditions are met:
- All specified peers are found in the BGP configuration.
- All specified drop statistics counters (or all counters if none specified) are zero.
Failure: If any of the following occur:
- A specified peer is not found in the BGP configuration.
- Any checked drop statistics counter has a non-zero value.
- A specified drop statistics counter does not exist.

Examples

anta.tests.routing:
  bgp:
    - VerifyBGPPeerDropStats:
        bgp_peers:
          - peer_address: 172.30.11.1
            vrf: default
            drop_stats:
              - inDropAsloop
              - prefixEvpnDroppedUnsupportedRouteType

Inputs¶

Name	Type	Description	Default
`bgp_peers`	`list[BgpPeer]`	List of BGP IPv4 peers.	-

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

class VerifyBGPPeerDropStats(AntaTest):
    """Verifies BGP NLRI drop statistics for the provided BGP IPv4 peer(s).

    This test performs the following checks for each specified peer:

      1. Verifies that the peer is found in its VRF in the BGP configuration.
      2. Validates the BGP drop statistics:
        - If specific drop statistics are provided, checks only those counters.
        - If no specific drop statistics are provided, checks all available counters.
        - Confirms that all checked counters have a value of zero.

    Expected Results
    ----------------
    * Success: If all of the following conditions are met:
        - All specified peers are found in the BGP configuration.
        - All specified drop statistics counters (or all counters if none specified) are zero.
    * Failure: If any of the following occur:
        - A specified peer is not found in the BGP configuration.
        - Any checked drop statistics counter has a non-zero value.
        - A specified drop statistics counter does not exist.

    Examples
    --------
    ```yaml
    anta.tests.routing:
      bgp:
        - VerifyBGPPeerDropStats:
            bgp_peers:
              - peer_address: 172.30.11.1
                vrf: default
                drop_stats:
                  - inDropAsloop
                  - prefixEvpnDroppedUnsupportedRouteType
    ```
    """

    categories: ClassVar[list[str]] = ["bgp"]
    commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show bgp neighbors vrf all", revision=3)]

    class Input(AntaTest.Input):
        """Input model for the VerifyBGPPeerDropStats test."""

        bgp_peers: list[BgpPeer]
        """List of BGP IPv4 peers."""
        BgpPeer: ClassVar[type[BgpPeer]] = BgpPeer

    @AntaTest.anta_test
    def test(self) -> None:
        """Main test function for VerifyBGPPeerDropStats."""
        self.result.is_success()

        output = self.instance_commands[0].json_output

        for peer in self.inputs.bgp_peers:
            peer_ip = str(peer.peer_address)
            drop_stats_input = peer.drop_stats
            peer_list = get_value(output, f"vrfs.{peer.vrf}.peerList", default=[])

            # Check if the peer is found
            if (peer_data := get_item(peer_list, "peerAddress", peer_ip)) is None:
                self.result.is_failure(f"{peer} - Not found")
                continue

            # Verify BGP peers' drop stats
            drop_stats_output = peer_data["dropStats"]

            # In case drop stats not provided, It will check all drop statistics
            if not drop_stats_input:
                drop_stats_input = drop_stats_output

            # Verify BGP peer's drop stats
            for drop_stat in drop_stats_input:
                if (stat_value := drop_stats_output.get(drop_stat, 0)) != 0:
                    self.result.is_failure(f"{peer} - Non-zero NLRI drop statistics counter - {drop_stat}: {stat_value}")

VerifyBGPPeerGroup ¶

Verifies BGP peer group of BGP IPv4 peer(s).

This test performs the following checks for each specified peer:

Verifies that the peer is found in its VRF in the BGP configuration.
Confirms the peer group is correctly assigned to the specified BGP peer.

Expected Results

Success: If all of the following conditions are met:
- All specified peers are found in the BGP configuration.
- The peer group is correctly assigned to the specified BGP peer.
Failure: If any of the following occur:
- A specified peer is not found in the BGP configuration.
- The peer group is not correctly assigned to the specified BGP peer.

Examples

anta.tests.routing:
  bgp:
    - VerifyBGPPeerGroup:
        bgp_peers:
          - peer_address: 172.30.11.1
            vrf: default
            peer_group: IPv4-UNDERLAY-PEERS

Inputs¶

Name	Type	Description	Default
`bgp_peers`	`list[BgpPeer]`	List of BGP IPv4 peers.	-

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

class VerifyBGPPeerGroup(AntaTest):
    """Verifies BGP peer group of BGP IPv4 peer(s).

    This test performs the following checks for each specified peer:

      1. Verifies that the peer is found in its VRF in the BGP configuration.
      2. Confirms the peer group is correctly assigned to the specified BGP peer.

    Expected Results
    ----------------
    * Success: If all of the following conditions are met:
        - All specified peers are found in the BGP configuration.
        - The peer group is correctly assigned to the specified BGP peer.
    * Failure: If any of the following occur:
        - A specified peer is not found in the BGP configuration.
        - The peer group is not correctly assigned to the specified BGP peer.

    Examples
    --------
    ```yaml
    anta.tests.routing:
      bgp:
        - VerifyBGPPeerGroup:
            bgp_peers:
              - peer_address: 172.30.11.1
                vrf: default
                peer_group: IPv4-UNDERLAY-PEERS
    ```
    """

    categories: ClassVar[list[str]] = ["bgp"]
    commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show bgp neighbors vrf all", revision=3)]

    class Input(AntaTest.Input):
        """Input model for the VerifyBGPPeerGroup test."""

        bgp_peers: list[BgpPeer]
        """List of BGP IPv4 peers."""

        @field_validator("bgp_peers")
        @classmethod
        def validate_bgp_peers(cls, bgp_peers: list[BgpPeer]) -> list[BgpPeer]:
            """Validate that 'peer_group' field is provided in each BGP peer."""
            for peer in bgp_peers:
                if peer.peer_group is None:
                    msg = f"{peer} 'peer_group' field missing in the input"
                    raise ValueError(msg)
            return bgp_peers

    @AntaTest.anta_test
    def test(self) -> None:
        """Main test function for VerifyBGPPeerGroup."""
        self.result.is_success()

        output = self.instance_commands[0].json_output

        for peer in self.inputs.bgp_peers:
            peer_ip = str(peer.peer_address)
            peer_list = get_value(output, f"vrfs.{peer.vrf}.peerList", default=[])

            # Check if the peer is found
            if (peer_data := get_item(peer_list, "peerAddress", peer_ip)) is None:
                self.result.is_failure(f"{peer} - Not found")
                continue

            if (actual_peer_group := peer_data.get("peerGroupName", "Not Found")) != peer.peer_group:
                self.result.is_failure(f"{peer} - Incorrect peer group configured - Expected: {peer.peer_group} Actual: {actual_peer_group}")

VerifyBGPPeerMD5Auth ¶

Verifies the MD5 authentication and state of IPv4 BGP peer(s) in a specified VRF.

This test performs the following checks for each specified peer:

Verifies that the peer is found in its VRF in the BGP configuration.
Validates that the BGP session is in Established state.
Confirms that MD5 authentication is enabled for the peer.

Expected Results

Success: If all of the following conditions are met:
- All specified peers are found in the BGP configuration.
- All peers are in Established state.
- MD5 authentication is enabled for all peers.
Failure: If any of the following occur:
- A specified peer is not found in the BGP configuration.
- A peer’s session state is not Established.
- MD5 authentication is not enabled for a peer.

Examples

anta.tests.routing:
  bgp:
    - VerifyBGPPeerMD5Auth:
        bgp_peers:
          - peer_address: 172.30.11.1
            vrf: default
          - peer_address: 172.30.11.5
            vrf: default

Inputs¶

Name	Type	Description	Default
`bgp_peers`	`list[BgpPeer]`	List of IPv4 BGP peers.	-

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

class VerifyBGPPeerMD5Auth(AntaTest):
    """Verifies the MD5 authentication and state of IPv4 BGP peer(s) in a specified VRF.

    This test performs the following checks for each specified peer:

      1. Verifies that the peer is found in its VRF in the BGP configuration.
      2. Validates that the BGP session is in `Established` state.
      3. Confirms that MD5 authentication is enabled for the peer.

    Expected Results
    ----------------
    * Success: If all of the following conditions are met:
        - All specified peers are found in the BGP configuration.
        - All peers are in `Established` state.
        - MD5 authentication is enabled for all peers.
    * Failure: If any of the following occur:
        - A specified peer is not found in the BGP configuration.
        - A peer's session state is not `Established`.
        - MD5 authentication is not enabled for a peer.

    Examples
    --------
    ```yaml
    anta.tests.routing:
      bgp:
        - VerifyBGPPeerMD5Auth:
            bgp_peers:
              - peer_address: 172.30.11.1
                vrf: default
              - peer_address: 172.30.11.5
                vrf: default
    ```
    """

    categories: ClassVar[list[str]] = ["bgp"]
    commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show bgp neighbors vrf all", revision=3)]

    class Input(AntaTest.Input):
        """Input model for the VerifyBGPPeerMD5Auth test."""

        bgp_peers: list[BgpPeer]
        """List of IPv4 BGP peers."""
        BgpPeer: ClassVar[type[BgpPeer]] = BgpPeer

    @AntaTest.anta_test
    def test(self) -> None:
        """Main test function for VerifyBGPPeerMD5Auth."""
        self.result.is_success()

        output = self.instance_commands[0].json_output

        for peer in self.inputs.bgp_peers:
            peer_ip = str(peer.peer_address)
            peer_list = get_value(output, f"vrfs.{peer.vrf}.peerList", default=[])

            # Check if the peer is found
            if (peer_data := get_item(peer_list, "peerAddress", peer_ip)) is None:
                self.result.is_failure(f"{peer} - Not found")
                continue

            # Check BGP peer state and MD5 authentication
            state = peer_data.get("state")
            md5_auth_enabled = peer_data.get("md5AuthEnabled")
            if state != "Established":
                self.result.is_failure(f"{peer} - Incorrect session state - Expected: Established Actual: {state}")
            if not md5_auth_enabled:
                self.result.is_failure(f"{peer} - Session does not have MD5 authentication enabled")

VerifyBGPPeerMPCaps ¶

Verifies the multiprotocol capabilities of BGP IPv4 peer(s).

This test performs the following checks for each specified peer:

Verifies that the peer is found in its VRF in the BGP configuration.
For each specified capability:
- Validates that the capability is present in the peer configuration.
- Confirms that the capability is advertised, received, and enabled.
When strict mode is enabled (strict: true):
- Verifies that only the specified capabilities are configured.
- Ensures an exact match between configured and expected capabilities.

Expected Results

Success: If all of the following conditions are met:
- All specified peers are found in the BGP configuration.
- All specified capabilities are present and properly negotiated.
- In strict mode, only the specified capabilities are configured.
Failure: If any of the following occur:
- A specified peer is not found in the BGP configuration.
- A specified capability is not found.
- A capability is not properly negotiated (not advertised, received, or enabled).
- In strict mode, additional or missing capabilities are detected.

Examples

anta.tests.routing:
  bgp:
    - VerifyBGPPeerMPCaps:
        bgp_peers:
          - peer_address: 172.30.11.1
            vrf: default
            strict: False
            capabilities:
              - ipv4 labeled-Unicast
              - ipv4MplsVpn

Inputs¶

Name	Type	Description	Default
`bgp_peers`	`list[BgpPeer]`	List of BGP IPv4 peers.	-

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

class VerifyBGPPeerMPCaps(AntaTest):
    """Verifies the multiprotocol capabilities of BGP IPv4 peer(s).

    This test performs the following checks for each specified peer:

      1. Verifies that the peer is found in its VRF in the BGP configuration.
      2. For each specified capability:
        - Validates that the capability is present in the peer configuration.
        - Confirms that the capability is advertised, received, and enabled.
      3. When strict mode is enabled (`strict: true`):
        - Verifies that only the specified capabilities are configured.
        - Ensures an exact match between configured and expected capabilities.

    Expected Results
    ----------------
    * Success: If all of the following conditions are met:
        - All specified peers are found in the BGP configuration.
        - All specified capabilities are present and properly negotiated.
        - In strict mode, only the specified capabilities are configured.
    * Failure: If any of the following occur:
        - A specified peer is not found in the BGP configuration.
        - A specified capability is not found.
        - A capability is not properly negotiated (not advertised, received, or enabled).
        - In strict mode, additional or missing capabilities are detected.

    Examples
    --------
    ```yaml
    anta.tests.routing:
      bgp:
        - VerifyBGPPeerMPCaps:
            bgp_peers:
              - peer_address: 172.30.11.1
                vrf: default
                strict: False
                capabilities:
                  - ipv4 labeled-Unicast
                  - ipv4MplsVpn
    ```
    """

    categories: ClassVar[list[str]] = ["bgp"]
    commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show bgp neighbors vrf all", revision=3)]

    class Input(AntaTest.Input):
        """Input model for the VerifyBGPPeerMPCaps test."""

        bgp_peers: list[BgpPeer]
        """List of BGP IPv4 peers."""
        BgpPeer: ClassVar[type[BgpPeer]] = BgpPeer

        @field_validator("bgp_peers")
        @classmethod
        def validate_bgp_peers(cls, bgp_peers: list[T]) -> list[T]:
            """Validate that 'capabilities' field is provided in each BGP peer."""
            for peer in bgp_peers:
                if peer.capabilities is None:
                    msg = f"{peer} 'capabilities' field missing in the input"
                    raise ValueError(msg)
            return bgp_peers

    @AntaTest.anta_test
    def test(self) -> None:
        """Main test function for VerifyBGPPeerMPCaps."""
        self.result.is_success()

        output = self.instance_commands[0].json_output

        for peer in self.inputs.bgp_peers:
            peer_ip = str(peer.peer_address)
            peer_list = get_value(output, f"vrfs.{peer.vrf}.peerList", default=[])

            # Check if the peer is found
            if (peer_data := get_item(peer_list, "peerAddress", peer_ip)) is None:
                self.result.is_failure(f"{peer} - Not found")
                continue

            # Fetching the multiprotocol capabilities
            act_mp_caps = get_value(peer_data, "neighborCapabilities.multiprotocolCaps")

            # If strict is True, check if only the specified capabilities are configured
            if peer.strict and sorted(peer.capabilities) != sorted(act_mp_caps):
                self.result.is_failure(f"{peer} - Mismatch - Expected: {', '.join(peer.capabilities)} Actual: {', '.join(act_mp_caps)}")
                continue

            # Check each capability
            for capability in peer.capabilities:
                # Check if the capability is found
                if (capability_status := get_value(act_mp_caps, capability)) is None:
                    self.result.is_failure(f"{peer} - {capability} not found")

                # Check if the capability is advertised, received, and enabled
                elif not _check_bgp_neighbor_capability(capability_status):
                    self.result.is_failure(f"{peer} - {capability} not negotiated - {format_data(capability_status)}")

VerifyBGPPeerRouteLimit ¶

Verifies maximum routes and warning limit for BGP IPv4 peer(s).

This test performs the following checks for each specified peer:

Verifies that the peer is found in its VRF in the BGP configuration.
Confirms the maximum routes and maximum routes warning limit, if provided, match the expected value.

Expected Results

Success: If all of the following conditions are met:
- All specified peers are found in the BGP configuration.
- The maximum routes/maximum routes warning limit match the expected value for a peer.
Failure: If any of the following occur:
- A specified peer is not found in the BGP configuration.
- The maximum routes/maximum routes warning limit do not match the expected value for a peer.

Examples

anta.tests.routing:
  bgp:
    - VerifyBGPPeerRouteLimit:
        bgp_peers:
          - peer_address: 172.30.11.1
            vrf: default
            maximum_routes: 12000
            warning_limit: 10000

Inputs¶

Name	Type	Description	Default
`bgp_peers`	`list[BgpPeer]`	List of BGP IPv4 peers.	-

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

class VerifyBGPPeerRouteLimit(AntaTest):
    """Verifies maximum routes and warning limit for BGP IPv4 peer(s).

    This test performs the following checks for each specified peer:

      1. Verifies that the peer is found in its VRF in the BGP configuration.
      2. Confirms the maximum routes and maximum routes warning limit, if provided, match the expected value.

    Expected Results
    ----------------
    * Success: If all of the following conditions are met:
        - All specified peers are found in the BGP configuration.
        - The maximum routes/maximum routes warning limit match the expected value for a peer.
    * Failure: If any of the following occur:
        - A specified peer is not found in the BGP configuration.
        - The maximum routes/maximum routes warning limit do not match the expected value for a peer.

    Examples
    --------
    ```yaml
    anta.tests.routing:
      bgp:
        - VerifyBGPPeerRouteLimit:
            bgp_peers:
              - peer_address: 172.30.11.1
                vrf: default
                maximum_routes: 12000
                warning_limit: 10000
    ```
    """

    categories: ClassVar[list[str]] = ["bgp"]
    commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show bgp neighbors vrf all", revision=3)]

    class Input(AntaTest.Input):
        """Input model for the VerifyBGPPeerRouteLimit test."""

        bgp_peers: list[BgpPeer]
        """List of BGP IPv4 peers."""
        BgpPeer: ClassVar[type[BgpPeer]] = BgpPeer

        @field_validator("bgp_peers")
        @classmethod
        def validate_bgp_peers(cls, bgp_peers: list[T]) -> list[T]:
            """Validate that 'maximum_routes' field is provided in each BGP peer."""
            for peer in bgp_peers:
                if peer.maximum_routes is None:
                    msg = f"{peer} 'maximum_routes' field missing in the input"
                    raise ValueError(msg)
            return bgp_peers

    @AntaTest.anta_test
    def test(self) -> None:
        """Main test function for VerifyBGPPeerRouteLimit."""
        self.result.is_success()

        output = self.instance_commands[0].json_output

        for peer in self.inputs.bgp_peers:
            peer_ip = str(peer.peer_address)
            maximum_routes = peer.maximum_routes
            warning_limit = peer.warning_limit
            peer_list = get_value(output, f"vrfs.{peer.vrf}.peerList", default=[])

            # Check if the peer is found
            if (peer_data := get_item(peer_list, "peerAddress", peer_ip)) is None:
                self.result.is_failure(f"{peer} - Not found")
                continue

            # Verify maximum routes
            if (actual_maximum_routes := peer_data.get("maxTotalRoutes", "Not Found")) != maximum_routes:
                self.result.is_failure(f"{peer} - Maximum routes mismatch - Expected: {maximum_routes} Actual: {actual_maximum_routes}")

            # Verify warning limit if provided. By default, EOS does not have a warning limit and `totalRoutesWarnLimit` is not present in the output.
            if warning_limit is not None and (actual_warning_limit := peer_data.get("totalRoutesWarnLimit", 0)) != warning_limit:
                self.result.is_failure(f"{peer} - Maximum routes warning limit mismatch - Expected: {warning_limit} Actual: {actual_warning_limit}")

VerifyBGPPeerRouteRefreshCap ¶

Verifies the route refresh capabilities of IPv4 BGP peer(s) in a specified VRF.

This test performs the following checks for each specified peer:

Verifies that the peer is found in its VRF in the BGP configuration.
Validates that the route refresh capability is present in the peer configuration.
Confirms that the capability is advertised, received, and enabled.

Expected Results

Success: If all of the following conditions are met:
- All specified peers are found in the BGP configuration.
- The route refresh capability is present in each peer configuration.
- The capability is properly negotiated (advertised, received, and enabled) for all peers.
Failure: If any of the following occur:
- A specified peer is not found in the BGP configuration.
- The route refresh capability is not present for a peer.
- The capability is not properly negotiated (not advertised, received, or enabled) for any peer.

Examples

anta.tests.routing:
  bgp:
    - VerifyBGPPeerRouteRefreshCap:
        bgp_peers:
          - peer_address: 172.30.11.1
            vrf: default

Inputs¶

Name	Type	Description	Default
`bgp_peers`	`list[BgpPeer]`	List of BGP IPv4 peers.	-

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

class VerifyBGPPeerRouteRefreshCap(AntaTest):
    """Verifies the route refresh capabilities of IPv4 BGP peer(s) in a specified VRF.

    This test performs the following checks for each specified peer:

      1. Verifies that the peer is found in its VRF in the BGP configuration.
      2. Validates that the route refresh capability is present in the peer configuration.
      3. Confirms that the capability is advertised, received, and enabled.

    Expected Results
    ----------------
    * Success: If all of the following conditions are met:
        - All specified peers are found in the BGP configuration.
        - The route refresh capability is present in each peer configuration.
        - The capability is properly negotiated (advertised, received, and enabled) for all peers.
    * Failure: If any of the following occur:
        - A specified peer is not found in the BGP configuration.
        - The route refresh capability is not present for a peer.
        - The capability is not properly negotiated (not advertised, received, or enabled) for any peer.

    Examples
    --------
    ```yaml
    anta.tests.routing:
      bgp:
        - VerifyBGPPeerRouteRefreshCap:
            bgp_peers:
              - peer_address: 172.30.11.1
                vrf: default
    ```
    """

    categories: ClassVar[list[str]] = ["bgp"]
    commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show bgp neighbors vrf all", revision=3)]

    class Input(AntaTest.Input):
        """Input model for the VerifyBGPPeerRouteRefreshCap test."""

        bgp_peers: list[BgpPeer]
        """List of BGP IPv4 peers."""
        BgpPeer: ClassVar[type[BgpPeer]] = BgpPeer

    @AntaTest.anta_test
    def test(self) -> None:
        """Main test function for VerifyBGPPeerRouteRefreshCap."""
        self.result.is_success()

        output = self.instance_commands[0].json_output

        for peer in self.inputs.bgp_peers:
            peer_ip = str(peer.peer_address)
            peer_list = get_value(output, f"vrfs.{peer.vrf}.peerList", default=[])

            # Check if the peer is found
            if (peer_data := get_item(peer_list, "peerAddress", peer_ip)) is None:
                self.result.is_failure(f"{peer} - Not found")
                continue

            # Check if the route refresh capability is found
            if (capablity_status := get_value(peer_data, "neighborCapabilities.routeRefreshCap")) is None:
                self.result.is_failure(f"{peer} - Route refresh capability not found")
                continue

            # Check if the route refresh capability is advertised, received, and enabled
            if not _check_bgp_neighbor_capability(capablity_status):
                self.result.is_failure(f"{peer} - Route refresh capability not negotiated - {format_data(capablity_status)}")

VerifyBGPPeerSession ¶

Verifies the session state of BGP IPv4 peer(s).

This test performs the following checks for each specified peer:

Verifies that the peer is found in its VRF in the BGP configuration.
Verifies that the BGP session is Established and, if specified, has remained established for at least the duration given by minimum_established_time.
Ensures that both input and output TCP message queues are empty. Can be disabled by setting check_tcp_queues global flag to False.

Expected Results

Success: If all of the following conditions are met:
- All specified peers are found in the BGP configuration.
- All peers sessions state are Established and, if specified, has remained established for at least the duration given by minimum_established_time.
- All peers have empty TCP message queues if check_tcp_queues is True (default).
- All peers are established for specified minimum duration.
Failure: If any of the following occur:
- A specified peer is not found in the BGP configuration.
- A peer’s session state is not Established or if specified, has not remained established for at least the duration specified by the minimum_established_time.
- A peer has non-empty TCP message queues (input or output) when check_tcp_queues is True.

Examples

anta.tests.routing:
  bgp:
    - VerifyBGPPeerSession:
        minimum_established_time: 10000
        check_tcp_queues: false
        bgp_peers:
          - peer_address: 10.1.0.1
            vrf: default
          - peer_address: 10.1.0.2
            vrf: default
          - peer_address: 10.1.255.2
            vrf: DEV
          - peer_address: 10.1.255.4
            vrf: DEV

Inputs¶

Name	Type	Description	Default
`minimum_established_time`	`PositiveInt \| None`	Minimum established time (seconds) for all the BGP sessions.	`None`
`check_tcp_queues`	`bool`	Flag to check if the TCP session queues are empty for all BGP peers. Defaults to `True`.	`True`
`bgp_peers`	`list[BgpPeer]`	List of BGP IPv4 peers.	-

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

class VerifyBGPPeerSession(AntaTest):
    """Verifies the session state of BGP IPv4 peer(s).

    This test performs the following checks for each specified peer:

      1. Verifies that the peer is found in its VRF in the BGP configuration.
      2. Verifies that the BGP session is `Established` and, if specified, has remained established for at least the duration given by `minimum_established_time`.
      3. Ensures that both input and output TCP message queues are empty.
      Can be disabled by setting `check_tcp_queues` global flag to `False`.

    Expected Results
    ----------------
    * Success: If all of the following conditions are met:
        - All specified peers are found in the BGP configuration.
        - All peers sessions state are `Established` and, if specified, has remained established for at least the duration given by `minimum_established_time`.
        - All peers have empty TCP message queues if `check_tcp_queues` is `True` (default).
        - All peers are established for specified minimum duration.
    * Failure: If any of the following occur:
        - A specified peer is not found in the BGP configuration.
        - A peer's session state is not `Established` or if specified, has not remained established for at least the duration specified by
        the `minimum_established_time`.
        - A peer has non-empty TCP message queues (input or output) when `check_tcp_queues` is `True`.

    Examples
    --------
    ```yaml
    anta.tests.routing:
      bgp:
        - VerifyBGPPeerSession:
            minimum_established_time: 10000
            check_tcp_queues: false
            bgp_peers:
              - peer_address: 10.1.0.1
                vrf: default
              - peer_address: 10.1.0.2
                vrf: default
              - peer_address: 10.1.255.2
                vrf: DEV
              - peer_address: 10.1.255.4
                vrf: DEV
    ```
    """

    categories: ClassVar[list[str]] = ["bgp"]
    commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show bgp neighbors vrf all", revision=3)]

    class Input(AntaTest.Input):
        """Input model for the VerifyBGPPeerSession test."""

        minimum_established_time: PositiveInt | None = None
        """Minimum established time (seconds) for all the BGP sessions."""
        check_tcp_queues: bool = True
        """Flag to check if the TCP session queues are empty for all BGP peers. Defaults to `True`."""
        bgp_peers: list[BgpPeer]
        """List of BGP IPv4 peers."""

    @AntaTest.anta_test
    def test(self) -> None:
        """Main test function for VerifyBGPPeerSession."""
        self.result.is_success()

        output = self.instance_commands[0].json_output

        for peer in self.inputs.bgp_peers:
            peer_ip = str(peer.peer_address)
            peer_list = get_value(output, f"vrfs.{peer.vrf}.peerList", default=[])

            # Check if the peer is found
            if (peer_data := get_item(peer_list, "peerAddress", peer_ip)) is None:
                self.result.is_failure(f"{peer} - Not found")
                continue

            # Check if the BGP session is established
            if peer_data["state"] != "Established":
                self.result.is_failure(f"{peer} - Incorrect session state - Expected: Established Actual: {peer_data['state']}")
                continue

            if self.inputs.minimum_established_time and (act_time := peer_data["establishedTime"]) < self.inputs.minimum_established_time:
                self.result.is_failure(
                    f"{peer} - BGP session not established for the minimum required duration - Expected: {self.inputs.minimum_established_time}s Actual: {act_time}s"
                )

            # Check the TCP session message queues
            if self.inputs.check_tcp_queues:
                inq = peer_data["peerTcpInfo"]["inputQueueLength"]
                outq = peer_data["peerTcpInfo"]["outputQueueLength"]
                if inq != 0 or outq != 0:
                    self.result.is_failure(f"{peer} - Session has non-empty message queues - InQ: {inq} OutQ: {outq}")

VerifyBGPPeerSessionRibd ¶

Verifies the session state of BGP IPv4 peer(s).

Compatible with EOS operating in ribd routing protocol model.

This test performs the following checks for each specified peer:

Verifies that the peer is found in its VRF in the BGP configuration.
Verifies that the BGP session is Established and, if specified, has remained established for at least the duration given by minimum_established_time.
Ensures that both input and output TCP message queues are empty. Can be disabled by setting check_tcp_queues global flag to False.

Expected Results

Success: If all of the following conditions are met:
- All specified peers are found in the BGP configuration.
- All peers sessions state are Established and, if specified, has remained established for at least the duration given by minimum_established_time.
- All peers have empty TCP message queues if check_tcp_queues is True (default).
- All peers are established for specified minimum duration.
Failure: If any of the following occur:
- A specified peer is not found in the BGP configuration.
- A peer’s session state is not Established or if specified, has not remained established for at least the duration specified by the minimum_established_time.
- A peer has non-empty TCP message queues (input or output) when check_tcp_queues is True.

Examples

anta.tests.routing:
  bgp:
    - VerifyBGPPeerSessionRibd:
        minimum_established_time: 10000
        check_tcp_queues: false
        bgp_peers:
          - peer_address: 10.1.0.1
            vrf: default
          - peer_address: 10.1.0.2
            vrf: default
          - peer_address: 10.1.255.2
            vrf: DEV
          - peer_address: 10.1.255.4
            vrf: DEV

Inputs¶

Name	Type	Description	Default
`minimum_established_time`	`PositiveInt \| None`	Minimum established time (seconds) for all the BGP sessions.	`None`
`check_tcp_queues`	`bool`	Flag to check if the TCP session queues are empty for all BGP peers. Defaults to `True`.	`True`
`bgp_peers`	`list[BgpPeer]`	List of BGP IPv4 peers.	-

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

class VerifyBGPPeerSessionRibd(AntaTest):
    """Verifies the session state of BGP IPv4 peer(s).

    Compatible with EOS operating in `ribd` routing protocol model.

    This test performs the following checks for each specified peer:

      1. Verifies that the peer is found in its VRF in the BGP configuration.
      2. Verifies that the BGP session is `Established` and, if specified, has remained established for at least the duration given by `minimum_established_time`.
      3. Ensures that both input and output TCP message queues are empty.
      Can be disabled by setting `check_tcp_queues` global flag to `False`.

    Expected Results
    ----------------
    * Success: If all of the following conditions are met:
        - All specified peers are found in the BGP configuration.
        - All peers sessions state are `Established` and, if specified, has remained established for at least the duration given by `minimum_established_time`.
        - All peers have empty TCP message queues if `check_tcp_queues` is `True` (default).
        - All peers are established for specified minimum duration.
    * Failure: If any of the following occur:
        - A specified peer is not found in the BGP configuration.
        - A peer's session state is not `Established` or if specified, has not remained established for at least the duration specified by
        the `minimum_established_time`.
        - A peer has non-empty TCP message queues (input or output) when `check_tcp_queues` is `True`.

    Examples
    --------
    ```yaml
    anta.tests.routing:
      bgp:
        - VerifyBGPPeerSessionRibd:
            minimum_established_time: 10000
            check_tcp_queues: false
            bgp_peers:
              - peer_address: 10.1.0.1
                vrf: default
              - peer_address: 10.1.0.2
                vrf: default
              - peer_address: 10.1.255.2
                vrf: DEV
              - peer_address: 10.1.255.4
                vrf: DEV
    ```
    """

    categories: ClassVar[list[str]] = ["bgp"]
    commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show ip bgp neighbors vrf all", revision=2)]

    class Input(AntaTest.Input):
        """Input model for the VerifyBGPPeerSessionRibd test."""

        minimum_established_time: PositiveInt | None = None
        """Minimum established time (seconds) for all the BGP sessions."""
        check_tcp_queues: bool = True
        """Flag to check if the TCP session queues are empty for all BGP peers. Defaults to `True`."""
        bgp_peers: list[BgpPeer]
        """List of BGP IPv4 peers."""

    @AntaTest.anta_test
    def test(self) -> None:
        """Main test function for VerifyBGPPeerSessionRibd."""
        self.result.is_success()

        output = self.instance_commands[0].json_output

        for peer in self.inputs.bgp_peers:
            peer_address = str(peer.peer_address)
            peers = get_value(output, f"vrfs.{peer.vrf}.peerList", default=[])

            # Check if the peer is found
            if (peer_data := get_item(peers, "peerAddress", peer_address)) is None:
                self.result.is_failure(f"{peer} - Not found")
                continue

            # Check if the BGP session is established
            if peer_data["state"] != "Established":
                self.result.is_failure(f"{peer} - Incorrect session state - Expected: Established Actual: {peer_data['state']}")
                continue

            if self.inputs.minimum_established_time and (act_time := peer_data["establishedTime"]) < self.inputs.minimum_established_time:
                self.result.is_failure(
                    f"{peer} - BGP session not established for the minimum required duration - Expected: {self.inputs.minimum_established_time}s Actual: {act_time}s"
                )

            # Check the TCP session message queues
            if self.inputs.check_tcp_queues:
                inq_stat = peer_data["peerTcpInfo"]["inputQueueLength"]
                outq_stat = peer_data["peerTcpInfo"]["outputQueueLength"]
                if inq_stat != 0 or outq_stat != 0:
                    self.result.is_failure(f"{peer} - Session has non-empty message queues - InQ: {inq_stat} OutQ: {outq_stat}")

VerifyBGPPeerTtlMultiHops ¶

Verifies BGP TTL and max-ttl-hops count for BGP IPv4 peer(s).

This test performs the following checks for each specified BGP peer:

Verifies the specified BGP peer exists in the BGP configuration.
Verifies the TTL and max-ttl-hops attribute matches the expected value.

Expected Results

Success: The test will pass if all specified peers exist with TTL and max-ttl-hops attributes matching the expected values.
Failure: If any of the following occur:
- A specified BGP peer is not found.
- A TTL or max-ttl-hops attribute doesn’t match the expected value for any peer.

Examples

anta.tests.routing:
  bgp:
    - VerifyBGPPeerTtlMultiHops:
        bgp_peers:
            - peer_address: 172.30.11.1
              vrf: default
              ttl: 3
              max_ttl_hops: 3
            - peer_address: 172.30.11.2
              vrf: test
              ttl: 30
              max_ttl_hops: 30

Inputs¶

Name	Type	Description	Default
`bgp_peers`	`list[BgpPeer]`	List of IPv4 peer(s).	-

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

class VerifyBGPPeerTtlMultiHops(AntaTest):
    """Verifies BGP TTL and max-ttl-hops count for BGP IPv4 peer(s).

    This test performs the following checks for each specified BGP peer:

      1. Verifies the specified BGP peer exists in the BGP configuration.
      2. Verifies the TTL and max-ttl-hops attribute matches the expected value.

    Expected Results
    ----------------
    * Success: The test will pass if all specified peers exist with TTL and max-ttl-hops attributes matching the expected values.
    * Failure: If any of the following occur:
        - A specified BGP peer is not found.
        - A TTL or max-ttl-hops attribute doesn't match the expected value for any peer.

    Examples
    --------
    ```yaml
    anta.tests.routing:
      bgp:
        - VerifyBGPPeerTtlMultiHops:
            bgp_peers:
                - peer_address: 172.30.11.1
                  vrf: default
                  ttl: 3
                  max_ttl_hops: 3
                - peer_address: 172.30.11.2
                  vrf: test
                  ttl: 30
                  max_ttl_hops: 30
    ```
    """

    categories: ClassVar[list[str]] = ["bgp"]
    commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show ip bgp neighbors vrf all", revision=2)]

    class Input(AntaTest.Input):
        """Input model for the VerifyBGPPeerTtlMultiHops test."""

        bgp_peers: list[BgpPeer]
        """List of IPv4 peer(s)."""

        @field_validator("bgp_peers")
        @classmethod
        def validate_bgp_peers(cls, bgp_peers: list[BgpPeer]) -> list[BgpPeer]:
            """Validate that 'ttl' and 'max_ttl_hops' field is provided in each BGP peer."""
            for peer in bgp_peers:
                if peer.ttl is None:
                    msg = f"{peer} 'ttl' field missing in the input"
                    raise ValueError(msg)
                if peer.max_ttl_hops is None:
                    msg = f"{peer} 'max_ttl_hops' field missing in the input"
                    raise ValueError(msg)

            return bgp_peers

    @AntaTest.anta_test
    def test(self) -> None:
        """Main test function for VerifyBGPPeerTtlMultiHops."""
        self.result.is_success()
        command_output = self.instance_commands[0].json_output

        for peer in self.inputs.bgp_peers:
            peer_ip = str(peer.peer_address)
            peer_list = get_value(command_output, f"vrfs.{peer.vrf}.peerList", default=[])

            # Check if the peer is found
            if (peer_details := get_item(peer_list, "peerAddress", peer_ip)) is None:
                self.result.is_failure(f"{peer} - Not found")
                continue

            # Verify if the TTL duration matches the expected value.
            if peer_details.get("ttl") != peer.ttl:
                self.result.is_failure(f"{peer} - TTL mismatch - Expected: {peer.ttl} Actual: {peer_details.get('ttl')}")

            # Verify if the max-ttl-hops time matches the expected value.
            if peer_details.get("maxTtlHops") != peer.max_ttl_hops:
                self.result.is_failure(f"{peer} - Max TTL Hops mismatch - Expected: {peer.max_ttl_hops} Actual: {peer_details.get('maxTtlHops')}")

VerifyBGPPeerUpdateErrors ¶

Verifies BGP update error counters for the provided BGP IPv4 peer(s).

This test performs the following checks for each specified peer:

Verifies that the peer is found in its VRF in the BGP configuration.
Validates the BGP update error counters:
- If specific update error counters are provided, checks only those counters.
- If no update error counters are provided, checks all available counters.
- Confirms that all checked counters have a value of zero.

Note: For “disabledAfiSafi” error counter field, checking that it’s not “None” versus 0.

Expected Results

Success: If all of the following conditions are met:
- All specified peers are found in the BGP configuration.
- All specified update error counters (or all counters if none specified) are zero.
Failure: If any of the following occur:
- A specified peer is not found in the BGP configuration.
- Any checked update error counters has a non-zero value.
- A specified update error counters does not exist.

Examples

anta.tests.routing:
  bgp:
    - VerifyBGPPeerUpdateErrors:
        bgp_peers:
          - peer_address: 172.30.11.1
            vrf: default
            update_errors:
              - inUpdErrWithdraw

Inputs¶

Name	Type	Description	Default
`bgp_peers`	`list[BgpPeer]`	List of BGP IPv4 peers.	-

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

class VerifyBGPPeerUpdateErrors(AntaTest):
    """Verifies BGP update error counters for the provided BGP IPv4 peer(s).

    This test performs the following checks for each specified peer:

      1. Verifies that the peer is found in its VRF in the BGP configuration.
      2. Validates the BGP update error counters:
        - If specific update error counters are provided, checks only those counters.
        - If no update error counters are provided, checks all available counters.
        - Confirms that all checked counters have a value of zero.

    Note: For "disabledAfiSafi" error counter field, checking that it's not "None" versus 0.

    Expected Results
    ----------------
    * Success: If all of the following conditions are met:
        - All specified peers are found in the BGP configuration.
        - All specified update error counters (or all counters if none specified) are zero.
    * Failure: If any of the following occur:
        - A specified peer is not found in the BGP configuration.
        - Any checked update error counters has a non-zero value.
        - A specified update error counters does not exist.

    Examples
    --------
    ```yaml
    anta.tests.routing:
      bgp:
        - VerifyBGPPeerUpdateErrors:
            bgp_peers:
              - peer_address: 172.30.11.1
                vrf: default
                update_errors:
                  - inUpdErrWithdraw
    ```
    """

    categories: ClassVar[list[str]] = ["bgp"]
    commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show bgp neighbors vrf all", revision=3)]

    class Input(AntaTest.Input):
        """Input model for the VerifyBGPPeerUpdateErrors test."""

        bgp_peers: list[BgpPeer]
        """List of BGP IPv4 peers."""
        BgpPeer: ClassVar[type[BgpPeer]] = BgpPeer

    @AntaTest.anta_test
    def test(self) -> None:
        """Main test function for VerifyBGPPeerUpdateErrors."""
        self.result.is_success()

        output = self.instance_commands[0].json_output

        for peer in self.inputs.bgp_peers:
            peer_ip = str(peer.peer_address)
            update_errors_input = peer.update_errors
            peer_list = get_value(output, f"vrfs.{peer.vrf}.peerList", default=[])

            # Check if the peer is found
            if (peer_data := get_item(peer_list, "peerAddress", peer_ip)) is None:
                self.result.is_failure(f"{peer} - Not found")
                continue

            # Getting the BGP peer's error counters output.
            error_counters_output = peer_data.get("peerInUpdateErrors", {})

            # In case update error counters not provided, It will check all the update error counters.
            if not update_errors_input:
                update_errors_input = error_counters_output

            # Verify BGP peer's update error counters
            for error_counter in update_errors_input:
                if (stat_value := error_counters_output.get(error_counter, "Not Found")) != 0 and stat_value != "None":
                    self.result.is_failure(f"{peer} - Non-zero update error counter - {error_counter}: {stat_value}")

VerifyBGPPeersHealth ¶

Verifies the health of BGP peers for given address families.

This test performs the following checks for each specified address family:

Validates that the VRF is configured.
Checks if there are any peers for the given AFI/SAFI.
For each relevant peer:
- Verifies that the BGP session is Established and, if specified, has remained established for at least the duration given by minimum_established_time.
- Confirms that the AFI/SAFI state is negotiated.
- Checks that both input and output TCP message queues are empty. Can be disabled by setting check_tcp_queues to False.

Expected Results

Success: If all checks pass for all specified address families and their peers.
Failure: If any of the following occur:
- The specified VRF is not configured.
- No peers are found for a given AFI/SAFI.
- A peer’s session state is not Established or if specified, has not remained established for at least the duration specified by the minimum_established_time.
- The AFI/SAFI state is not ‘negotiated’ for any peer.
- Any TCP message queue (input or output) is not empty when check_tcp_queues is True (default).

Examples

anta.tests.routing:
  bgp:
    - VerifyBGPPeersHealth:
        minimum_established_time: 10000
        address_families:
          - afi: "evpn"
          - afi: "ipv4"
            safi: "unicast"
            vrf: "default"
          - afi: "ipv6"
            safi: "unicast"
            vrf: "DEV"
            check_tcp_queues: false

Inputs¶

Name	Type	Description	Default
`minimum_established_time`	`PositiveInt \| None`	Minimum established time (seconds) for all the BGP sessions.	`None`
`address_families`	`list[BgpAddressFamily]`	List of BGP address families.	-

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

class VerifyBGPPeersHealth(AntaTest):
    """Verifies the health of BGP peers for given address families.

    This test performs the following checks for each specified address family:

      1. Validates that the VRF is configured.
      2. Checks if there are any peers for the given AFI/SAFI.
      3. For each relevant peer:
        - Verifies that the BGP session is `Established` and, if specified, has remained established for at least the duration given by `minimum_established_time`.
        - Confirms that the AFI/SAFI state is `negotiated`.
        - Checks that both input and output TCP message queues are empty.
          Can be disabled by setting `check_tcp_queues` to `False`.

    Expected Results
    ----------------
    * Success: If all checks pass for all specified address families and their peers.
    * Failure: If any of the following occur:
        - The specified VRF is not configured.
        - No peers are found for a given AFI/SAFI.
        - A peer's session state is not `Established` or if specified, has not remained established for at least the duration specified by
        the `minimum_established_time`.
        - The AFI/SAFI state is not 'negotiated' for any peer.
        - Any TCP message queue (input or output) is not empty when `check_tcp_queues` is `True` (default).

    Examples
    --------
    ```yaml
    anta.tests.routing:
      bgp:
        - VerifyBGPPeersHealth:
            minimum_established_time: 10000
            address_families:
              - afi: "evpn"
              - afi: "ipv4"
                safi: "unicast"
                vrf: "default"
              - afi: "ipv6"
                safi: "unicast"
                vrf: "DEV"
                check_tcp_queues: false
    ```
    """

    categories: ClassVar[list[str]] = ["bgp"]
    commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show bgp neighbors vrf all", revision=3)]

    class Input(AntaTest.Input):
        """Input model for the VerifyBGPPeersHealth test."""

        minimum_established_time: PositiveInt | None = None
        """Minimum established time (seconds) for all the BGP sessions."""
        address_families: list[BgpAddressFamily]
        """List of BGP address families."""
        BgpAfi: ClassVar[type[BgpAfi]] = BgpAfi

    @AntaTest.anta_test
    def test(self) -> None:
        """Main test function for VerifyBGPPeersHealth."""
        self.result.is_success()

        output = self.instance_commands[0].json_output

        for address_family in self.inputs.address_families:
            # Check if the VRF is configured
            if (vrf_output := get_value(output, f"vrfs.{address_family.vrf}")) is None:
                self.result.is_failure(f"{address_family} - VRF not configured")
                continue

            # Check if any peers are found for this AFI/SAFI
            relevant_peers = [
                peer for peer in vrf_output.get("peerList", []) if get_value(peer, f"neighborCapabilities.multiprotocolCaps.{address_family.eos_key}") is not None
            ]

            if not relevant_peers:
                self.result.is_failure(f"{address_family} - No peers found")
                continue

            for peer in relevant_peers:
                # Check if the BGP session is established
                if peer["state"] != "Established":
                    self.result.is_failure(f"{address_family} Peer: {peer['peerAddress']} - Incorrect session state - Expected: Established Actual: {peer['state']}")
                    continue

                if self.inputs.minimum_established_time and (act_time := peer["establishedTime"]) < self.inputs.minimum_established_time:
                    msg = f"BGP session not established for the minimum required duration - Expected: {self.inputs.minimum_established_time}s Actual: {act_time}s"
                    self.result.is_failure(f"{address_family} Peer: {peer['peerAddress']} - {msg}")

                # Check if the AFI/SAFI state is negotiated
                capability_status = get_value(peer, f"neighborCapabilities.multiprotocolCaps.{address_family.eos_key}")
                if not _check_bgp_neighbor_capability(capability_status):
                    self.result.is_failure(f"{address_family} Peer: {peer['peerAddress']} - AFI/SAFI state is not negotiated - {format_data(capability_status)}")

                # Check the TCP session message queues
                if address_family.check_tcp_queues:
                    inq = peer["peerTcpInfo"]["inputQueueLength"]
                    outq = peer["peerTcpInfo"]["outputQueueLength"]
                    if inq != 0 or outq != 0:
                        self.result.is_failure(f"{address_family} Peer: {peer['peerAddress']} - Session has non-empty message queues - InQ: {inq} OutQ: {outq}")

VerifyBGPPeersHealthRibd ¶

Verifies the health of all the BGP IPv4 peer(s).

Compatible with EOS operating in ribd routing protocol model.

This test performs the following checks for all BGP IPv4 peers:

Verifies that the BGP session is in the Established state.
Checks that both input and output TCP message queues are empty. Can be disabled by setting check_tcp_queues global flag to False.

Expected Results

Success: If all checks pass for all BGP IPv4 peers.
Failure: If any of the following occur:
- Any BGP session is not in the Established state.
- Any TCP message queue (input or output) is not empty when check_tcp_queues is True (default).

Examples

anta.tests.routing:
  bgp:
    - VerifyBGPPeersHealthRibd:
        check_tcp_queues: True

Inputs¶

Name	Type	Description	Default
`check_tcp_queues`	`bool`	Flag to check if the TCP session queues are empty for all BGP peers. Defaults to `True`.	`True`

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

class VerifyBGPPeersHealthRibd(AntaTest):
    """Verifies the health of all the BGP IPv4 peer(s).

    Compatible with EOS operating in `ribd` routing protocol model.

    This test performs the following checks for all BGP IPv4 peers:

      1. Verifies that the BGP session is in the `Established` state.
      2. Checks that both input and output TCP message queues are empty.
      Can be disabled by setting `check_tcp_queues` global flag to `False`.

    Expected Results
    ----------------
    * Success: If all checks pass for all BGP IPv4 peers.
    * Failure: If any of the following occur:
        - Any BGP session is not in the `Established` state.
        - Any TCP message queue (input or output) is not empty when `check_tcp_queues` is `True` (default).

    Examples
    --------
    ```yaml
    anta.tests.routing:
      bgp:
        - VerifyBGPPeersHealthRibd:
            check_tcp_queues: True
    ```
    """

    categories: ClassVar[list[str]] = ["bgp"]
    commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show ip bgp neighbors vrf all", revision=2)]

    class Input(AntaTest.Input):
        """Input model for the VerifyBGPPeersHealthRibd test."""

        check_tcp_queues: bool = True
        """Flag to check if the TCP session queues are empty for all BGP peers. Defaults to `True`."""

    @AntaTest.anta_test
    def test(self) -> None:
        """Main test function for VerifyBGPPeersHealthRibd."""
        self.result.is_success()

        output = self.instance_commands[0].json_output

        for vrf, vrf_data in output["vrfs"].items():
            peer_list = vrf_data.get("peerList", [])

            for peer in peer_list:
                # Check if the BGP session is established
                if peer["state"] != "Established":
                    self.result.is_failure(f"Peer: {peer['peerAddress']} VRF: {vrf} - Incorrect session state - Expected: Established Actual: {peer['state']}")
                    continue

                # Check the TCP session message queues
                inq = peer["peerTcpInfo"]["inputQueueLength"]
                outq = peer["peerTcpInfo"]["outputQueueLength"]
                if self.inputs.check_tcp_queues and (inq != 0 or outq != 0):
                    self.result.is_failure(f"Peer: {peer['peerAddress']} VRF: {vrf} - Session has non-empty message queues - InQ: {inq} OutQ: {outq}")

VerifyBGPRedistribution ¶

Verifies BGP redistribution.

This test performs the following checks for each specified VRF in the BGP instance:

Ensures that the expected address-family is configured on the device.
Confirms that the redistributed route protocol, include leaked and route map match the expected values.

Expected Results

Success: If all of the following conditions are met:
- The expected address-family is configured on the device.
- The redistributed route protocol, include leaked and route map align with the expected values for the route.
Failure: If any of the following occur:
- The expected address-family is not configured on device.
- The redistributed route protocol, include leaked or route map does not match the expected values.

Examples

anta.tests.routing:
  bgp:
    - VerifyBGPRedistribution:
        vrfs:
          - vrf: default
            address_families:
              - afi_safi: ipv4multicast
                redistributed_routes:
                  - proto: Connected
                    include_leaked: True
                    route_map: RM-CONN-2-BGP
                  - proto: IS-IS
                    include_leaked: True
                    route_map: RM-CONN-2-BGP
              - afi_safi: IPv6 Unicast
                redistributed_routes:
                  - proto: User # Converted to EOS SDK
                    route_map: RM-CONN-2-BGP
                  - proto: Static
                    include_leaked: True
                    route_map: RM-CONN-2-BGP

Inputs¶

Name	Type	Description	Default
`vrfs`	`list[BgpVrf]`	List of VRFs in the BGP instance.	-

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

class VerifyBGPRedistribution(AntaTest):
    """Verifies BGP redistribution.

    This test performs the following checks for each specified VRF in the BGP instance:

      1. Ensures that the expected address-family is configured on the device.
      2. Confirms that the redistributed route protocol, include leaked and route map match the expected values.


    Expected Results
    ----------------
    * Success: If all of the following conditions are met:
        - The expected address-family is configured on the device.
        - The redistributed route protocol, include leaked and route map align with the expected values for the route.
    * Failure: If any of the following occur:
        - The expected address-family is not configured on device.
        - The redistributed route protocol, include leaked or route map does not match the expected values.

    Examples
    --------
    ```yaml
    anta.tests.routing:
      bgp:
        - VerifyBGPRedistribution:
            vrfs:
              - vrf: default
                address_families:
                  - afi_safi: ipv4multicast
                    redistributed_routes:
                      - proto: Connected
                        include_leaked: True
                        route_map: RM-CONN-2-BGP
                      - proto: IS-IS
                        include_leaked: True
                        route_map: RM-CONN-2-BGP
                  - afi_safi: IPv6 Unicast
                    redistributed_routes:
                      - proto: User # Converted to EOS SDK
                        route_map: RM-CONN-2-BGP
                      - proto: Static
                        include_leaked: True
                        route_map: RM-CONN-2-BGP
    ```
    """

    categories: ClassVar[list[str]] = ["bgp"]
    commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show bgp instance vrf all", revision=4)]

    class Input(AntaTest.Input):
        """Input model for the VerifyBGPRedistribution test."""

        vrfs: list[BgpVrf]
        """List of VRFs in the BGP instance."""

    def _validate_redistribute_route(self, vrf_data: str, addr_family: str, afi_safi_configs: list[dict[str, Any]], route_info: dict[str, Any]) -> list[Any]:
        """Validate the redstributed route details for a given address family."""
        failure_msg = []
        # If the redistributed route protocol does not match the expected value, test fails.
        if not (actual_route := get_item(afi_safi_configs.get("redistributedRoutes"), "proto", route_info.proto)):
            failure_msg.append(f"{vrf_data}, {addr_family}, Proto: {route_info.proto} - Not configured")
            return failure_msg

        # If includes leaked field applicable, and it does not matches the expected value, test fails.
        if (act_include_leaked := actual_route.get("includeLeaked", False)) != route_info.include_leaked:
            failure_msg.append(f"{vrf_data}, {addr_family}, {route_info} - Include leaked mismatch - Actual: {act_include_leaked}")

        # If route map is required and it is not matching the expected value, test fails.
        if all([route_info.route_map, (act_route_map := actual_route.get("routeMap", "Not Found")) != route_info.route_map]):
            failure_msg.append(f"{vrf_data}, {addr_family}, {route_info} - Route map mismatch - Actual: {act_route_map}")
        return failure_msg

    @AntaTest.anta_test
    def test(self) -> None:
        """Main test function for VerifyBGPRedistribution."""
        self.result.is_success()
        command_output = self.instance_commands[0].json_output

        for vrf_data in self.inputs.vrfs:
            # If the specified VRF details are not found, test fails.
            if not (instance_details := get_value(command_output, f"vrfs.{vrf_data.vrf}")):
                self.result.is_failure(f"{vrf_data} - Not configured")
                continue
            for address_family in vrf_data.address_families:
                # If the AFI-SAFI configuration details are not found, test fails.
                if not (afi_safi_configs := get_value(instance_details, f"afiSafiConfig.{address_family.afi_safi}")):
                    self.result.is_failure(f"{vrf_data}, {address_family} - Not redistributed")
                    continue

                for route_info in address_family.redistributed_routes:
                    failure_msg = self._validate_redistribute_route(str(vrf_data), str(address_family), afi_safi_configs, route_info)
                    for msg in failure_msg:
                        self.result.is_failure(msg)

VerifyBGPRouteECMP ¶

Verifies BGP IPv4 route ECMP paths.

This test performs the following checks for each specified BGP route entry:

Route exists in BGP table.
First path is a valid and active ECMP head.
Correct number of valid ECMP contributors follow the head path.
Route is installed in RIB with same amount of next-hops.

Expected Results

Success: The test will pass if all specified routes exist in both BGP and RIB tables with correct amount of ECMP paths.
Failure: The test will fail if:
- A specified route is not found in BGP table.
- A valid and active ECMP head is not found.
- ECMP contributors count does not match the expected value.
- Route is not installed in RIB table.
- BGP and RIB nexthops count do not match.

Examples

anta.tests.routing:
  bgp:
    - VerifyBGPRouteECMP:
        route_entries:
            - prefix: 10.100.0.128/31
              vrf: default
              ecmp_count: 2

Inputs¶

Name	Type	Description	Default
`route_entries`	`list[BgpRoute]`	List of BGP IPv4 route(s).	-

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

class VerifyBGPRouteECMP(AntaTest):
    """Verifies BGP IPv4 route ECMP paths.

    This test performs the following checks for each specified BGP route entry:

      1. Route exists in BGP table.
      2. First path is a valid and active ECMP head.
      3. Correct number of valid ECMP contributors follow the head path.
      4. Route is installed in RIB with same amount of next-hops.

    Expected Results
    ----------------
    * Success: The test will pass if all specified routes exist in both BGP and RIB tables with correct amount of ECMP paths.
    * Failure: The test will fail if:
        - A specified route is not found in BGP table.
        - A valid and active ECMP head is not found.
        - ECMP contributors count does not match the expected value.
        - Route is not installed in RIB table.
        - BGP and RIB nexthops count do not match.

    Examples
    --------
    ```yaml
    anta.tests.routing:
      bgp:
        - VerifyBGPRouteECMP:
            route_entries:
                - prefix: 10.100.0.128/31
                  vrf: default
                  ecmp_count: 2
    ```
    """

    categories: ClassVar[list[str]] = ["bgp"]
    commands: ClassVar[list[AntaCommand | AntaTemplate]] = [
        AntaCommand(command="show ip bgp vrf all", revision=3),
        AntaCommand(command="show ip route vrf all bgp", revision=4),
    ]

    class Input(AntaTest.Input):
        """Input model for the VerifyBGPRouteECMP test."""

        route_entries: list[BgpRoute]
        """List of BGP IPv4 route(s)."""

        @field_validator("route_entries")
        @classmethod
        def validate_route_entries(cls, route_entries: list[BgpRoute]) -> list[BgpRoute]:
            """Validate that 'ecmp_count' field is provided in each BGP route."""
            for route in route_entries:
                if route.ecmp_count is None:
                    msg = f"{route} 'ecmp_count' field missing in the input"
                    raise ValueError(msg)
            return route_entries

    @AntaTest.anta_test
    def test(self) -> None:
        """Main test function for VerifyBGPRouteECMP."""
        self.result.is_success()

        for route in self.inputs.route_entries:
            # Verify if the prefix exists in BGP table.
            if not (bgp_route_entry := get_value(self.instance_commands[0].json_output, f"vrfs..{route.vrf}..bgpRouteEntries..{route.prefix}", separator="..")):
                self.result.is_failure(f"{route} - Prefix not found in BGP table")
                continue

            route_paths = iter(bgp_route_entry["bgpRoutePaths"])
            head = next(route_paths, None)
            # Verify if the active ECMP head exists.
            if head is None or not all(head["routeType"][key] for key in ["valid", "active", "ecmpHead"]):
                self.result.is_failure(f"{route} - Valid and active ECMP head not found")
                continue

            bgp_nexthops = {head["nextHop"]}
            bgp_nexthops.update([path["nextHop"] for path in route_paths if all(path["routeType"][key] for key in ["valid", "ecmp", "ecmpContributor"])])

            # Verify ECMP count is correct.
            if len(bgp_nexthops) != route.ecmp_count:
                self.result.is_failure(f"{route} - ECMP count mismatch - Expected: {route.ecmp_count} Actual: {len(bgp_nexthops)}")
                continue

            # Verify if the prefix exists in routing table.
            if not (route_entry := get_value(self.instance_commands[1].json_output, f"vrfs..{route.vrf}..routes..{route.prefix}", separator="..")):
                self.result.is_failure(f"{route} - Prefix not found in routing table")
                continue

            # Verify BGP and RIB nexthops are same.
            if len(bgp_nexthops) != len(route_entry["vias"]):
                self.result.is_failure(f"{route} - Nexthops count mismatch - BGP: {len(bgp_nexthops)} RIB: {len(route_entry['vias'])}")

VerifyBGPRoutePaths ¶

Verifies BGP IPv4 route paths.

This test performs the following checks for each specified BGP route entry:

Verifies the specified BGP route exists in the routing table.
For each expected paths:
- Verifies a path with matching next-hop exists.
- Verifies the path’s origin attribute matches the expected value.

Expected Results

Success: The test will pass if all specified routes exist with paths matching the expected next-hops and origin attributes.
Failure: The test will fail if:
- A specified BGP route is not found.
- A path with specified next-hop is not found.
- A path’s origin attribute doesn’t match the expected value.

Examples

anta.tests.routing:
  bgp:
    - VerifyBGPRoutePaths:
        route_entries:
            - prefix: 10.100.0.128/31
              vrf: default
              paths:
                - nexthop: 10.100.0.10
                  origin: Igp
                - nexthop: 10.100.4.5
                  origin: Incomplete

Inputs¶

Name	Type	Description	Default
`route_entries`	`list[BgpRoute]`	List of BGP IPv4 route(s).	-

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

class VerifyBGPRoutePaths(AntaTest):
    """Verifies BGP IPv4 route paths.

    This test performs the following checks for each specified BGP route entry:

      1. Verifies the specified BGP route exists in the routing table.
      2. For each expected paths:
          - Verifies a path with matching next-hop exists.
          - Verifies the path's origin attribute matches the expected value.

    Expected Results
    ----------------
    * Success: The test will pass if all specified routes exist with paths matching the expected next-hops and origin attributes.
    * Failure: The test will fail if:
        - A specified BGP route is not found.
        - A path with specified next-hop is not found.
        - A path's origin attribute doesn't match the expected value.

    Examples
    --------
    ```yaml
    anta.tests.routing:
      bgp:
        - VerifyBGPRoutePaths:
            route_entries:
                - prefix: 10.100.0.128/31
                  vrf: default
                  paths:
                    - nexthop: 10.100.0.10
                      origin: Igp
                    - nexthop: 10.100.4.5
                      origin: Incomplete
    ```
    """

    categories: ClassVar[list[str]] = ["bgp"]
    commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show ip bgp vrf all", revision=3)]

    class Input(AntaTest.Input):
        """Input model for the VerifyBGPRoutePaths test."""

        route_entries: list[BgpRoute]
        """List of BGP IPv4 route(s)."""

        @field_validator("route_entries")
        @classmethod
        def validate_route_entries(cls, route_entries: list[BgpRoute]) -> list[BgpRoute]:
            """Validate that 'paths' field is provided in each BGP route."""
            for route in route_entries:
                if route.paths is None:
                    msg = f"{route} 'paths' field missing in the input"
                    raise ValueError(msg)
            return route_entries

    @AntaTest.anta_test
    def test(self) -> None:
        """Main test function for VerifyBGPRoutePaths."""
        self.result.is_success()

        for route in self.inputs.route_entries:
            # Verify if the prefix exists in BGP table
            if not (bgp_routes := get_value(self.instance_commands[0].json_output, f"vrfs..{route.vrf}..bgpRouteEntries..{route.prefix}", separator="..")):
                self.result.is_failure(f"{route} - Prefix not found")
                continue

            # Iterating over each path.
            for path in route.paths:
                nexthop = str(path.nexthop)
                origin = path.origin
                if not (route_path := get_item(bgp_routes["bgpRoutePaths"], "nextHop", nexthop)):
                    self.result.is_failure(f"{route} {path} - Path not found")
                    continue

                if (actual_origin := get_value(route_path, "routeType.origin")) != origin:
                    self.result.is_failure(f"{route} {path} - Origin mismatch - Actual: {actual_origin}")

VerifyBGPSpecificPeers ¶

Verifies the health of specific BGP peer(s) for given address families.

This test performs the following checks for each specified address family and peer:

Confirms that the specified VRF is configured.
For each specified peer:
- Verifies that the peer is found in the BGP configuration.
- Verifies that the BGP session is Established and, if specified, has remained established for at least the duration given by minimum_established_time.
- Confirms that the AFI/SAFI state is negotiated.
- Ensures that both input and output TCP message queues are empty. Can be disabled by setting check_tcp_queues to False.

Expected Results

Success: If all checks pass for all specified peers in all address families.
Failure: If any of the following occur:
- The specified VRF is not configured.
- A specified peer is not found in the BGP configuration.
- A peer’s session state is not Established or if specified, has not remained established for at least the duration specified by the minimum_established_time.
- The AFI/SAFI state is not negotiated for a peer.
- Any TCP message queue (input or output) is not empty for a peer when check_tcp_queues is True (default).

Examples

anta.tests.routing:
  bgp:
    - VerifyBGPSpecificPeers:
        minimum_established_time: 10000
        address_families:
          - afi: "evpn"
            peers:
              - 10.1.0.1
              - 10.1.0.2
          - afi: "ipv4"
            safi: "unicast"
            peers:
              - 10.1.254.1
              - 10.1.255.0
              - 10.1.255.2
              - 10.1.255.4

Inputs¶

Name	Type	Description	Default
`minimum_established_time`	`PositiveInt \| None`	Minimum established time (seconds) for all the BGP sessions.	`None`
`address_families`	`list[BgpAddressFamily]`	List of BGP address families.	-

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

class VerifyBGPSpecificPeers(AntaTest):
    """Verifies the health of specific BGP peer(s) for given address families.

    This test performs the following checks for each specified address family and peer:

      1. Confirms that the specified VRF is configured.
      2. For each specified peer:
        - Verifies that the peer is found in the BGP configuration.
        - Verifies that the BGP session is `Established` and, if specified, has remained established for at least the duration given by `minimum_established_time`.
        - Confirms that the AFI/SAFI state is `negotiated`.
        - Ensures that both input and output TCP message queues are empty.
          Can be disabled by setting `check_tcp_queues` to `False`.

    Expected Results
    ----------------
    * Success: If all checks pass for all specified peers in all address families.
    * Failure: If any of the following occur:
        - The specified VRF is not configured.
        - A specified peer is not found in the BGP configuration.
        - A peer's session state is not `Established` or if specified, has not remained established for at least the duration specified by
        the `minimum_established_time`.
        - The AFI/SAFI state is not `negotiated` for a peer.
        - Any TCP message queue (input or output) is not empty for a peer when `check_tcp_queues` is `True` (default).

    Examples
    --------
    ```yaml
    anta.tests.routing:
      bgp:
        - VerifyBGPSpecificPeers:
            minimum_established_time: 10000
            address_families:
              - afi: "evpn"
                peers:
                  - 10.1.0.1
                  - 10.1.0.2
              - afi: "ipv4"
                safi: "unicast"
                peers:
                  - 10.1.254.1
                  - 10.1.255.0
                  - 10.1.255.2
                  - 10.1.255.4
    ```
    """

    categories: ClassVar[list[str]] = ["bgp"]
    commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show bgp neighbors vrf all", revision=3)]

    class Input(AntaTest.Input):
        """Input model for the VerifyBGPSpecificPeers test."""

        minimum_established_time: PositiveInt | None = None
        """Minimum established time (seconds) for all the BGP sessions."""
        address_families: list[BgpAddressFamily]
        """List of BGP address families."""
        BgpAfi: ClassVar[type[BgpAfi]] = BgpAfi

        @field_validator("address_families")
        @classmethod
        def validate_address_families(cls, address_families: list[BgpAddressFamily]) -> list[BgpAddressFamily]:
            """Validate that 'peers' field is provided in each address family."""
            for af in address_families:
                if af.peers is None:
                    msg = f"{af} 'peers' field missing in the input"
                    raise ValueError(msg)
            return address_families

    @AntaTest.anta_test
    def test(self) -> None:
        """Main test function for VerifyBGPSpecificPeers."""
        self.result.is_success()

        output = self.instance_commands[0].json_output

        for address_family in self.inputs.address_families:
            # Check if the VRF is configured
            if (vrf_output := get_value(output, f"vrfs.{address_family.vrf}")) is None:
                self.result.is_failure(f"{address_family} - VRF not configured")
                continue

            for peer in address_family.peers:
                peer_ip = str(peer)

                # Check if the peer is found
                if (peer_data := get_item(vrf_output["peerList"], "peerAddress", peer_ip)) is None:
                    self.result.is_failure(f"{address_family} Peer: {peer_ip} - Not configured")
                    continue

                # Check if the BGP session is established
                if peer_data["state"] != "Established":
                    self.result.is_failure(f"{address_family} Peer: {peer_ip} - Incorrect session state - Expected: Established Actual: {peer_data['state']}")
                    continue

                if self.inputs.minimum_established_time and (act_time := peer_data["establishedTime"]) < self.inputs.minimum_established_time:
                    msg = f"BGP session not established for the minimum required duration - Expected: {self.inputs.minimum_established_time}s Actual: {act_time}s"
                    self.result.is_failure(f"{address_family} Peer: {peer_ip} - {msg}")

                # Check if the AFI/SAFI state is negotiated
                capability_status = get_value(peer_data, f"neighborCapabilities.multiprotocolCaps.{address_family.eos_key}")
                if not capability_status:
                    self.result.is_failure(f"{address_family} Peer: {peer_ip} - AFI/SAFI state is not negotiated")

                if capability_status and not _check_bgp_neighbor_capability(capability_status):
                    self.result.is_failure(f"{address_family} Peer: {peer_ip} - AFI/SAFI state is not negotiated - {format_data(capability_status)}")

                # Check the TCP session message queues
                inq = peer_data["peerTcpInfo"]["inputQueueLength"]
                outq = peer_data["peerTcpInfo"]["outputQueueLength"]
                if address_family.check_tcp_queues and (inq != 0 or outq != 0):
                    self.result.is_failure(f"{address_family} Peer: {peer_ip} - Session has non-empty message queues - InQ: {inq} OutQ: {outq}")

VerifyBGPTimers ¶

Verifies the timers of BGP IPv4 peer(s).

This test performs the following checks for each specified peer:

Verifies that the peer is found in its VRF in the BGP configuration.
Confirms the BGP session hold time/keepalive timers match the expected value.

Expected Results

Success: If all of the following conditions are met:
- All specified peers are found in the BGP configuration.
- The hold time/keepalive timers match the expected value for each peer.
Failure: If any of the following occur:
- A specified peer is not found in the BGP configuration.
- The hold time/keepalive timers do not match the expected value for a peer.

Examples

anta.tests.routing:
  bgp:
    - VerifyBGPTimers:
        bgp_peers:
          - peer_address: 172.30.11.1
            vrf: default
            hold_time: 180
            keep_alive_time: 60
          - peer_address: 172.30.11.5
            vrf: default
            hold_time: 180
            keep_alive_time: 60

Inputs¶

Name	Type	Description	Default
`bgp_peers`	`list[BgpPeer]`	List of BGP IPv4 peers.	-

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

class VerifyBGPTimers(AntaTest):
    """Verifies the timers of BGP IPv4 peer(s).

    This test performs the following checks for each specified peer:

      1. Verifies that the peer is found in its VRF in the BGP configuration.
      2. Confirms the BGP session hold time/keepalive timers match the expected value.

    Expected Results
    ----------------
    * Success: If all of the following conditions are met:
        - All specified peers are found in the BGP configuration.
        - The hold time/keepalive timers match the expected value for each peer.
    * Failure: If any of the following occur:
        - A specified peer is not found in the BGP configuration.
        - The hold time/keepalive timers do not match the expected value for a peer.

    Examples
    --------
    ```yaml
    anta.tests.routing:
      bgp:
        - VerifyBGPTimers:
            bgp_peers:
              - peer_address: 172.30.11.1
                vrf: default
                hold_time: 180
                keep_alive_time: 60
              - peer_address: 172.30.11.5
                vrf: default
                hold_time: 180
                keep_alive_time: 60
    ```
    """

    categories: ClassVar[list[str]] = ["bgp"]
    commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show bgp neighbors vrf all", revision=3)]

    class Input(AntaTest.Input):
        """Input model for the VerifyBGPTimers test."""

        bgp_peers: list[BgpPeer]
        """List of BGP IPv4 peers."""
        BgpPeer: ClassVar[type[BgpPeer]] = BgpPeer

        @field_validator("bgp_peers")
        @classmethod
        def validate_bgp_peers(cls, bgp_peers: list[T]) -> list[T]:
            """Validate that 'hold_time' or 'keep_alive_time'  field is provided in each BGP peer."""
            for peer in bgp_peers:
                if peer.hold_time is None or peer.keep_alive_time is None:
                    msg = f"{peer} 'hold_time' or 'keep_alive_time' field missing in the input"
                    raise ValueError(msg)
            return bgp_peers

    @AntaTest.anta_test
    def test(self) -> None:
        """Main test function for VerifyBGPTimers."""
        self.result.is_success()

        output = self.instance_commands[0].json_output

        for peer in self.inputs.bgp_peers:
            peer_ip = str(peer.peer_address)
            peer_list = get_value(output, f"vrfs.{peer.vrf}.peerList", default=[])

            # Check if the peer is found
            if (peer_data := get_item(peer_list, "peerAddress", peer_ip)) is None:
                self.result.is_failure(f"{peer} - Not found")
                continue

            # Check BGP peer timers
            if peer_data["holdTime"] != peer.hold_time:
                self.result.is_failure(f"{peer} - Hold time mismatch - Expected: {peer.hold_time} Actual: {peer_data['holdTime']}")
            if peer_data["keepaliveTime"] != peer.keep_alive_time:
                self.result.is_failure(f"{peer} - Keepalive time mismatch - Expected: {peer.keep_alive_time} Actual: {peer_data['keepaliveTime']}")

VerifyBgpRouteMaps ¶

Verifies BGP inbound and outbound route-maps of BGP IPv4 peer(s).

This test performs the following checks for each specified peer:

Verifies that the peer is found in its VRF in the BGP configuration.
Validates the correct BGP route maps are applied in the correct direction (inbound or outbound).

Expected Results

Success: If all of the following conditions are met:
- All specified peers are found in the BGP configuration.
- All specified peers has correct BGP route maps are applied in the correct direction (inbound or outbound).
Failure: If any of the following occur:
- A specified peer is not found in the BGP configuration.
- A incorrect or missing route map in either the inbound or outbound direction.

Examples

anta.tests.routing:
  bgp:
    - VerifyBgpRouteMaps:
        bgp_peers:
          - peer_address: 172.30.11.1
            vrf: default
            inbound_route_map: RM-MLAG-PEER-IN
            outbound_route_map: RM-MLAG-PEER-OUT

Inputs¶

Name	Type	Description	Default
`bgp_peers`	`list[BgpPeer]`	List of BGP IPv4 peers.	-

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

class VerifyBgpRouteMaps(AntaTest):
    """Verifies BGP inbound and outbound route-maps of BGP IPv4 peer(s).

    This test performs the following checks for each specified peer:

      1. Verifies that the peer is found in its VRF in the BGP configuration.
      2. Validates the correct BGP route maps are applied in the correct direction (inbound or outbound).

    Expected Results
    ----------------
    * Success: If all of the following conditions are met:
        - All specified peers are found in the BGP configuration.
        - All specified peers has correct BGP route maps are applied in the correct direction (inbound or outbound).
    * Failure: If any of the following occur:
        - A specified peer is not found in the BGP configuration.
        - A incorrect or missing route map in either the inbound or outbound direction.

    Examples
    --------
    ```yaml
    anta.tests.routing:
      bgp:
        - VerifyBgpRouteMaps:
            bgp_peers:
              - peer_address: 172.30.11.1
                vrf: default
                inbound_route_map: RM-MLAG-PEER-IN
                outbound_route_map: RM-MLAG-PEER-OUT
    ```
    """

    categories: ClassVar[list[str]] = ["bgp"]
    commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show bgp neighbors vrf all", revision=3)]

    class Input(AntaTest.Input):
        """Input model for the VerifyBgpRouteMaps test."""

        bgp_peers: list[BgpPeer]
        """List of BGP IPv4 peers."""
        BgpPeer: ClassVar[type[BgpPeer]] = BgpPeer

        @field_validator("bgp_peers")
        @classmethod
        def validate_bgp_peers(cls, bgp_peers: list[T]) -> list[T]:
            """Validate that 'inbound_route_map' or 'outbound_route_map' field is provided in each BGP peer."""
            for peer in bgp_peers:
                if not (peer.inbound_route_map or peer.outbound_route_map):
                    msg = f"{peer} 'inbound_route_map' or 'outbound_route_map' field missing in the input"
                    raise ValueError(msg)
            return bgp_peers

    @AntaTest.anta_test
    def test(self) -> None:
        """Main test function for VerifyBgpRouteMaps."""
        self.result.is_success()

        output = self.instance_commands[0].json_output

        for peer in self.inputs.bgp_peers:
            peer_ip = str(peer.peer_address)
            inbound_route_map = peer.inbound_route_map
            outbound_route_map = peer.outbound_route_map
            peer_list = get_value(output, f"vrfs.{peer.vrf}.peerList", default=[])

            # Check if the peer is found
            if (peer_data := get_item(peer_list, "peerAddress", peer_ip)) is None:
                self.result.is_failure(f"{peer} - Not found")
                continue

            # Verify Inbound route-map
            if inbound_route_map and (inbound_map := peer_data.get("routeMapInbound", "Not Configured")) != inbound_route_map:
                self.result.is_failure(f"{peer} - Inbound route-map mismatch - Expected: {inbound_route_map} Actual: {inbound_map}")

            # Verify Outbound route-map
            if outbound_route_map and (outbound_map := peer_data.get("routeMapOutbound", "Not Configured")) != outbound_route_map:
                self.result.is_failure(f"{peer} - Outbound route-map mismatch - Expected: {outbound_route_map} Actual: {outbound_map}")

VerifyEVPNType2Route ¶

Verifies the EVPN Type-2 routes for a given IPv4 or MAC address and VNI.

This test performs the following checks for each specified VXLAN endpoint:

Verifies that the endpoint exists in the BGP EVPN table.
Confirms that at least one EVPN Type-2 route with a valid and active path exists.

Expected Results

Success: If all of the following conditions are met:
- All specified VXLAN endpoints are found in the BGP EVPN table.
- Each endpoint has at least one EVPN Type-2 route with a valid and active path.
Failure: If any of the following occur:
- A VXLAN endpoint is not found in the BGP EVPN table.
- No EVPN Type-2 route with a valid and active path exists for an endpoint.

Examples

anta.tests.routing:
  bgp:
    - VerifyEVPNType2Route:
        vxlan_endpoints:
          - address: 192.168.20.102
            vni: 10020
          - address: aac1.ab5d.b41e
            vni: 10010

Inputs¶

Name	Type	Description	Default
`vxlan_endpoints`	`list[VxlanEndpoint]`	List of VXLAN endpoints to verify.	-

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

class VerifyEVPNType2Route(AntaTest):
    """Verifies the EVPN Type-2 routes for a given IPv4 or MAC address and VNI.

    This test performs the following checks for each specified VXLAN endpoint:

      1. Verifies that the endpoint exists in the BGP EVPN table.
      2. Confirms that at least one EVPN Type-2 route with a valid and active path exists.

    Expected Results
    ----------------
    * Success: If all of the following conditions are met:
        - All specified VXLAN endpoints are found in the BGP EVPN table.
        - Each endpoint has at least one EVPN Type-2 route with a valid and active path.
    * Failure: If any of the following occur:
        - A VXLAN endpoint is not found in the BGP EVPN table.
        - No EVPN Type-2 route with a valid and active path exists for an endpoint.

    Examples
    --------
    ```yaml
    anta.tests.routing:
      bgp:
        - VerifyEVPNType2Route:
            vxlan_endpoints:
              - address: 192.168.20.102
                vni: 10020
              - address: aac1.ab5d.b41e
                vni: 10010
    ```
    """

    categories: ClassVar[list[str]] = ["bgp"]
    commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaTemplate(template="show bgp evpn route-type mac-ip {address} vni {vni}", revision=2)]

    class Input(AntaTest.Input):
        """Input model for the VerifyEVPNType2Route test."""

        vxlan_endpoints: list[VxlanEndpoint]
        """List of VXLAN endpoints to verify."""
        VxlanEndpoint: ClassVar[type[VxlanEndpoint]] = VxlanEndpoint

    def render(self, template: AntaTemplate) -> list[AntaCommand]:
        """Render the template for each VXLAN endpoint in the input list."""
        return [template.render(address=str(endpoint.address), vni=endpoint.vni) for endpoint in self.inputs.vxlan_endpoints]

    @AntaTest.anta_test
    def test(self) -> None:
        """Main test function for VerifyEVPNType2Route."""
        self.result.is_success()

        for command, endpoint in zip(self.instance_commands, self.inputs.vxlan_endpoints):
            # Verify that the VXLAN endpoint is in the BGP EVPN table
            evpn_routes = command.json_output["evpnRoutes"]
            if not evpn_routes:
                self.result.is_failure(f"{endpoint} - No EVPN Type-2 route")
                continue

            # Verify that at least one EVPN Type-2 route has at least one active and valid path across all learned routes from all RDs combined
            has_active_path = False
            for route_data in evpn_routes.values():
                for path in route_data.get("evpnRoutePaths", []):
                    route_type = path.get("routeType", {})
                    if route_type.get("active") and route_type.get("valid"):
                        has_active_path = True
                        break
            if not has_active_path:
                self.result.is_failure(f"{endpoint} - No valid and active path")

Input models¶

Module containing input models for routing BGP tests.

AFI_SAFI_MAPPINGS `module-attribute` ¶

AFI_SAFI_MAPPINGS = {
    "v4u": "IPv4 Unicast",
    "v4m": "IPv4 Multicast",
    "v6u": "IPv6 Unicast",
    "v6m": "IPv6 Multicast",
}

Dictionary mapping AFI/SAFI to EOS key representation for BGP redistributed route protocol.

IPV4_MULTICAST_SUPPORTED_PROTO `module-attribute` ¶

IPV4_MULTICAST_SUPPORTED_PROTO = [
    "AttachedHost",
    "Connected",
    "IS-IS",
    "OSPF Internal",
    "OSPF External",
    "OSPF Nssa-External",
    "OSPFv3 Internal",
    "OSPFv3 External",
    "OSPFv3 Nssa-External",
    "Static",
]

List of BGP redistributed route protocol, supported for IPv4 multicast address family.

IPV6_MULTICAST_SUPPORTED_PROTO `module-attribute` ¶

IPV6_MULTICAST_SUPPORTED_PROTO = [
    proto
    for proto in IPV4_MULTICAST_SUPPORTED_PROTO
    if proto != "AttachedHost"
]

List of BGP redistributed route protocol, supported for IPv6 multicast address family.

AddressFamilyConfig ¶

Model representing a BGP address family configuration.

Name	Type	Description	Default
`afi_safi`	`RedistributedAfiSafi`	AFI/SAFI abbreviation per EOS.	-
`redistributed_routes`	`list[RedistributedRouteConfig]`	List of redistributed route configuration.	-

validate_afi_safi_supported_routes ¶

validate_afi_safi_supported_routes() -> Self

Validate each address family supported redistributed protocol.

Following table shows the supported redistributed routes for each address family.

IPv4 Unicast	IPv6 Unicast	IPv4 Multicast	IPv6 Multicast
AttachedHost	AttachedHost	AttachedHost	Connected
Bgp	Bgp	Connected	IS-IS
Connected	Connected	IS-IS	OSPF Internal
Dynamic	DHCP	OSPF Internal	OSPF External
IS-IS	Dynamic	OSPF External	OSPF Nssa-External
OSPF Internal	IS-IS	OSPF Nssa-External	OSPFv3 Internal
OSPF External	OSPFv3 Internal	OSPFv3 Internal	OSPFv3 External
OSPF Nssa-External	OSPFv3 External	OSPFv3 External	OSPFv3 Nssa-External
OSPFv3 Internal	OSPFv3 Nssa-External	OSPFv3 Nssa-External	Static
OSPFv3 External	Static	Static
OSPFv3 Nssa-External	User
RIP
Static
User

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

@model_validator(mode="after")
def validate_afi_safi_supported_routes(self) -> Self:
    """Validate each address family supported redistributed protocol.

    Following table shows the supported redistributed routes for each address family.

    |    IPv4 Unicast         |    IPv6 Unicast         |   IPv4 Multicast       |   IPv6 Multicast       |
    | ------------------------|-------------------------|------------------------|------------------------|
    |    AttachedHost         |    AttachedHost         |   AttachedHost         |   Connected            |
    |    Bgp                  |    Bgp                  |   Connected            |   IS-IS                |
    |    Connected            |    Connected            |   IS-IS                |   OSPF Internal        |
    |    Dynamic              |    DHCP                 |   OSPF Internal        |   OSPF External        |
    |    IS-IS                |    Dynamic              |   OSPF External        |   OSPF Nssa-External   |
    |    OSPF Internal        |    IS-IS                |   OSPF Nssa-External   |   OSPFv3 Internal      |
    |    OSPF External        |    OSPFv3 Internal      |   OSPFv3 Internal      |   OSPFv3 External      |
    |    OSPF Nssa-External   |    OSPFv3 External      |   OSPFv3 External      |   OSPFv3 Nssa-External |
    |    OSPFv3 Internal      |    OSPFv3 Nssa-External |   OSPFv3 Nssa-External |   Static               |
    |    OSPFv3 External      |    Static               |   Static               |                        |
    |    OSPFv3 Nssa-External |    User                 |                        |                        |
    |    RIP                  |                         |                        |                        |
    |    Static               |                         |                        |                        |
    |    User                 |                         |                        |                        |
    """
    for routes_data in self.redistributed_routes:
        if all([self.afi_safi == "v4u", routes_data.proto == "DHCP"]):
            msg = f"Redistributed protocol 'DHCP' is not supported for address-family '{AFI_SAFI_MAPPINGS[self.afi_safi]}'"
            raise ValueError(msg)

        if self.afi_safi == "v6u" and routes_data.proto in ["OSPF Internal", "OSPF External", "OSPF Nssa-External", "RIP"]:
            msg = f"Redistributed protocol '{routes_data.proto}' is not supported for address-family '{AFI_SAFI_MAPPINGS[self.afi_safi]}'"
            raise ValueError(msg)

        if self.afi_safi == "v4m" and routes_data.proto not in IPV4_MULTICAST_SUPPORTED_PROTO:
            msg = f"Redistributed protocol '{routes_data.proto}' is not supported for address-family '{AFI_SAFI_MAPPINGS[self.afi_safi]}'"
            raise ValueError(msg)

        if self.afi_safi == "v6m" and routes_data.proto not in IPV6_MULTICAST_SUPPORTED_PROTO:
            msg = f"Redistributed protocol '{routes_data.proto}' is not supported for address-family '{AFI_SAFI_MAPPINGS[self.afi_safi]}'"
            raise ValueError(msg)

    return self

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

class AddressFamilyConfig(BaseModel):
    """Model representing a BGP address family configuration."""

    afi_safi: RedistributedAfiSafi
    """AFI/SAFI abbreviation per EOS."""
    redistributed_routes: list[RedistributedRouteConfig]
    """List of redistributed route configuration."""

    @model_validator(mode="after")
    def validate_afi_safi_supported_routes(self) -> Self:
        """Validate each address family supported redistributed protocol.

        Following table shows the supported redistributed routes for each address family.

        |    IPv4 Unicast         |    IPv6 Unicast         |   IPv4 Multicast       |   IPv6 Multicast       |
        | ------------------------|-------------------------|------------------------|------------------------|
        |    AttachedHost         |    AttachedHost         |   AttachedHost         |   Connected            |
        |    Bgp                  |    Bgp                  |   Connected            |   IS-IS                |
        |    Connected            |    Connected            |   IS-IS                |   OSPF Internal        |
        |    Dynamic              |    DHCP                 |   OSPF Internal        |   OSPF External        |
        |    IS-IS                |    Dynamic              |   OSPF External        |   OSPF Nssa-External   |
        |    OSPF Internal        |    IS-IS                |   OSPF Nssa-External   |   OSPFv3 Internal      |
        |    OSPF External        |    OSPFv3 Internal      |   OSPFv3 Internal      |   OSPFv3 External      |
        |    OSPF Nssa-External   |    OSPFv3 External      |   OSPFv3 External      |   OSPFv3 Nssa-External |
        |    OSPFv3 Internal      |    OSPFv3 Nssa-External |   OSPFv3 Nssa-External |   Static               |
        |    OSPFv3 External      |    Static               |   Static               |                        |
        |    OSPFv3 Nssa-External |    User                 |                        |                        |
        |    RIP                  |                         |                        |                        |
        |    Static               |                         |                        |                        |
        |    User                 |                         |                        |                        |
        """
        for routes_data in self.redistributed_routes:
            if all([self.afi_safi == "v4u", routes_data.proto == "DHCP"]):
                msg = f"Redistributed protocol 'DHCP' is not supported for address-family '{AFI_SAFI_MAPPINGS[self.afi_safi]}'"
                raise ValueError(msg)

            if self.afi_safi == "v6u" and routes_data.proto in ["OSPF Internal", "OSPF External", "OSPF Nssa-External", "RIP"]:
                msg = f"Redistributed protocol '{routes_data.proto}' is not supported for address-family '{AFI_SAFI_MAPPINGS[self.afi_safi]}'"
                raise ValueError(msg)

            if self.afi_safi == "v4m" and routes_data.proto not in IPV4_MULTICAST_SUPPORTED_PROTO:
                msg = f"Redistributed protocol '{routes_data.proto}' is not supported for address-family '{AFI_SAFI_MAPPINGS[self.afi_safi]}'"
                raise ValueError(msg)

            if self.afi_safi == "v6m" and routes_data.proto not in IPV6_MULTICAST_SUPPORTED_PROTO:
                msg = f"Redistributed protocol '{routes_data.proto}' is not supported for address-family '{AFI_SAFI_MAPPINGS[self.afi_safi]}'"
                raise ValueError(msg)

        return self

    def __str__(self) -> str:
        """Return a human-readable string representation of the AddressFamilyConfig for reporting.

        Examples
        --------
        - AFI-SAFI: IPv4 Unicast
        """
        return f"AFI-SAFI: {AFI_SAFI_MAPPINGS[self.afi_safi]}"

BgpAddressFamily ¶

Model for a BGP address family.

Name	Type	Description	Default
`afi`	`Afi`	BGP Address Family Identifier (AFI).	-
`safi`	`Safi \| None`	BGP Subsequent Address Family Identifier (SAFI). Required when `afi` is `ipv4` or `ipv6`.	`None`
`vrf`	`str`	Optional VRF when `afi` is `ipv4` or `ipv6`. Defaults to `default`. If the input `afi` is NOT `ipv4` or `ipv6` (e.g. `evpn`, `vpn-ipv4`, etc.), the `vrf` must be `default`. These AFIs operate at a global level and do not use the VRF concept in the same way as IPv4/IPv6.	`'default'`
`num_peers`	`PositiveInt \| None`	Number of expected established BGP peers with negotiated AFI/SAFI. Required field in the `VerifyBGPPeerCount` test.	`None`
`peers`	`list[IPv4Address \| IPv6Address] \| None`	List of expected IPv4/IPv6 BGP peers supporting the AFI/SAFI. Required field in the `VerifyBGPSpecificPeers` test.	`None`
`check_tcp_queues`	`bool`	Flag to check if the TCP session queues are empty for a BGP peer. Defaults to `True`. Can be disabled in the `VerifyBGPPeersHealth` and `VerifyBGPSpecificPeers` tests.	`True`
`check_peer_state`	`bool`	Flag to check if the peers are established with negotiated AFI/SAFI. Defaults to `False`. Can be enabled in the `VerifyBGPPeerCount` tests.	`False`

validate_inputs ¶

validate_inputs() -> Self

Validate the inputs provided to the BgpAddressFamily 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/input_models/routing/bgp.py

@model_validator(mode="after")
def validate_inputs(self) -> Self:
    """Validate the inputs provided to the BgpAddressFamily 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:
            msg = "'safi' must be provided when afi is ipv4 or ipv6"
            raise ValueError(msg)
    elif self.safi is not None:
        msg = "'safi' must not be provided when afi is not ipv4 or ipv6"
        raise ValueError(msg)
    elif self.vrf != "default":
        msg = "'vrf' must be default when afi is not ipv4 or ipv6"
        raise ValueError(msg)
    return self

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

class BgpAddressFamily(BaseModel):
    """Model for a BGP address family."""

    model_config = ConfigDict(extra="forbid")
    afi: Afi
    """BGP Address Family Identifier (AFI)."""
    safi: Safi | None = None
    """BGP Subsequent Address Family Identifier (SAFI). Required when `afi` is `ipv4` or `ipv6`."""
    vrf: str = "default"
    """Optional VRF when `afi` is `ipv4` or `ipv6`. Defaults to `default`.

    If the input `afi` is NOT `ipv4` or `ipv6` (e.g. `evpn`, `vpn-ipv4`, etc.), the `vrf` must be `default`.

    These AFIs operate at a global level and do not use the VRF concept in the same way as IPv4/IPv6.
    """
    num_peers: PositiveInt | None = None
    """Number of expected established BGP peers with negotiated AFI/SAFI. Required field in the `VerifyBGPPeerCount` test."""
    peers: list[IPv4Address | IPv6Address] | None = None
    """List of expected IPv4/IPv6 BGP peers supporting the AFI/SAFI. Required field in the `VerifyBGPSpecificPeers` test."""
    check_tcp_queues: bool = True
    """Flag to check if the TCP session queues are empty for a BGP peer. Defaults to `True`.

    Can be disabled in the `VerifyBGPPeersHealth` and `VerifyBGPSpecificPeers` tests.
    """
    check_peer_state: bool = False
    """Flag to check if the peers are established with negotiated AFI/SAFI. Defaults to `False`.

    Can be enabled in the `VerifyBGPPeerCount` tests."""

    @model_validator(mode="after")
    def validate_inputs(self) -> Self:
        """Validate the inputs provided to the BgpAddressFamily 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:
                msg = "'safi' must be provided when afi is ipv4 or ipv6"
                raise ValueError(msg)
        elif self.safi is not None:
            msg = "'safi' must not be provided when afi is not ipv4 or ipv6"
            raise ValueError(msg)
        elif self.vrf != "default":
            msg = "'vrf' must be default when afi is not ipv4 or ipv6"
            raise ValueError(msg)
        return self

    @property
    def eos_key(self) -> str:
        """AFI/SAFI EOS key representation."""
        # Pydantic handles the validation of the AFI/SAFI combination, so we can ignore error handling here.
        return AFI_SAFI_EOS_KEY[(self.afi, self.safi)]

    def __str__(self) -> str:
        """Return a human-readable string representation of the BgpAddressFamily for reporting.

        Examples
        --------
        - AFI:ipv4 SAFI:unicast VRF:default
        - AFI:evpn
        """
        base_string = f"AFI: {self.afi}"
        if self.safi is not None:
            base_string += f" SAFI: {self.safi}"
        if self.afi in ["ipv4", "ipv6"]:
            base_string += f" VRF: {self.vrf}"
        return base_string

BgpNeighbor ¶

Alias for the BgpPeer model to maintain backward compatibility.

When initialised, it will emit a deprecation warning and call the BgpPeer model.

TODO: Remove this class in ANTA v2.0.0.

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

class BgpNeighbor(BgpPeer):  # pragma: no cover
    """Alias for the BgpPeer model to maintain backward compatibility.

    When initialised, it will emit a deprecation warning and call the BgpPeer model.

    TODO: Remove this class in ANTA v2.0.0.
    """

    def __init__(self, **data: Any) -> None:  # noqa: ANN401
        """Initialize the BgpPeer class, emitting a depreciation warning."""
        warn(
            message="BgpNeighbor model is deprecated and will be removed in ANTA v2.0.0. Use the BgpPeer model instead.",
            category=DeprecationWarning,
            stacklevel=2,
        )
        super().__init__(**data)

BgpPeer ¶

Model for a BGP peer.

Only IPv4 peers are supported for now.

Name	Type	Description	Default
`peer_address`	`IPv4Address`	IPv4 address of the BGP peer.	-
`vrf`	`str`	Optional VRF for the BGP peer. Defaults to `default`.	`'default'`
`peer_group`	`str \| None`	Peer group of the BGP peer. Required field in the `VerifyBGPPeerGroup` test.	`None`
`advertised_routes`	`list[IPv4Network] \| None`	List of advertised routes in CIDR format. Required field in the `VerifyBGPExchangedRoutes` test.	`None`
`received_routes`	`list[IPv4Network] \| None`	List of received routes in CIDR format. Required field in the `VerifyBGPExchangedRoutes` test.	`None`
`capabilities`	`list[MultiProtocolCaps] \| None`	List of BGP multiprotocol capabilities. Required field in the `VerifyBGPPeerMPCaps`, `VerifyBGPNlriAcceptance` tests.	`None`
`strict`	`bool`	If True, requires exact match of the provided BGP multiprotocol capabilities. Optional field in the `VerifyBGPPeerMPCaps` test. Defaults to False.	`False`
`hold_time`	`int \| None`	BGP hold time in seconds. Required field in the `VerifyBGPTimers` test.	`Field(default=None, ge=3, le=7200)`
`keep_alive_time`	`int \| None`	BGP keepalive time in seconds. Required field in the `VerifyBGPTimers` test.	`Field(default=None, ge=0, le=3600)`
`drop_stats`	`list[BgpDropStats] \| None`	List of drop statistics to be verified. Optional field in the `VerifyBGPPeerDropStats` test. If not provided, the test will verifies all drop statistics.	`None`
`update_errors`	`list[BgpUpdateError] \| None`	List of update error counters to be verified. Optional field in the `VerifyBGPPeerUpdateErrors` test. If not provided, the test will verifies all the update error counters.	`None`
`inbound_route_map`	`str \| None`	Inbound route map applied, defaults to None. Required field in the `VerifyBgpRouteMaps` test.	`None`
`outbound_route_map`	`str \| None`	Outbound route map applied, defaults to None. Required field in the `VerifyBgpRouteMaps` test.	`None`
`maximum_routes`	`int \| None`	The maximum allowable number of BGP routes. `0` means unlimited. Required field in the `VerifyBGPPeerRouteLimit` test	`Field(default=None, ge=0, le=4294967294)`
`warning_limit`	`int \| None`	The warning limit for the maximum routes. `0` means no warning. Optional field in the `VerifyBGPPeerRouteLimit` test. If not provided, the test will not verify the warning limit.	`Field(default=None, ge=0, le=4294967294)`
`ttl`	`int \| None`	The Time-To-Live (TTL). Required field in the `VerifyBGPPeerTtlMultiHops` test.	`Field(default=None, ge=1, le=255)`
`max_ttl_hops`	`int \| None`	The Max TTL hops. Required field in the `VerifyBGPPeerTtlMultiHops` test.	`Field(default=None, ge=1, le=255)`

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

class BgpPeer(BaseModel):
    """Model for a BGP peer.

    Only IPv4 peers are supported for now.
    """

    model_config = ConfigDict(extra="forbid")
    peer_address: IPv4Address
    """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
    """List of received routes in CIDR format. Required field in the `VerifyBGPExchangedRoutes` test."""
    capabilities: list[MultiProtocolCaps] | None = None
    """List of BGP multiprotocol capabilities. Required field in the `VerifyBGPPeerMPCaps`, `VerifyBGPNlriAcceptance` tests."""
    strict: bool = False
    """If True, requires exact match of the provided BGP multiprotocol capabilities.

    Optional field in the `VerifyBGPPeerMPCaps` test. Defaults to False."""
    hold_time: int | None = Field(default=None, ge=3, le=7200)
    """BGP hold time in seconds. Required field in the `VerifyBGPTimers` test."""
    keep_alive_time: int | None = Field(default=None, ge=0, le=3600)
    """BGP keepalive time in seconds. Required field in the `VerifyBGPTimers` test."""
    drop_stats: list[BgpDropStats] | None = None
    """List of drop statistics to be verified.

    Optional field in the `VerifyBGPPeerDropStats` test. If not provided, the test will verifies all drop statistics."""
    update_errors: list[BgpUpdateError] | None = None
    """List of update error counters to be verified.

    Optional field in the `VerifyBGPPeerUpdateErrors` test. If not provided, the test will verifies all the update error counters."""
    inbound_route_map: str | None = None
    """Inbound route map applied, defaults to None. Required field in the `VerifyBgpRouteMaps` test."""
    outbound_route_map: str | None = None
    """Outbound route map applied, defaults to None. Required field in the `VerifyBgpRouteMaps` test."""
    maximum_routes: int | None = Field(default=None, ge=0, le=4294967294)
    """The maximum allowable number of BGP routes. `0` means unlimited. Required field in the `VerifyBGPPeerRouteLimit` test"""
    warning_limit: int | None = Field(default=None, ge=0, le=4294967294)
    """The warning limit for the maximum routes. `0` means no warning.

    Optional field in the `VerifyBGPPeerRouteLimit` test. If not provided, the test will not verify the warning limit."""
    ttl: int | None = Field(default=None, ge=1, le=255)
    """The Time-To-Live (TTL). Required field in the `VerifyBGPPeerTtlMultiHops` test."""
    max_ttl_hops: int | None = Field(default=None, ge=1, le=255)
    """The Max TTL hops. Required field in the `VerifyBGPPeerTtlMultiHops` test."""

    def __str__(self) -> str:
        """Return a human-readable string representation of the BgpPeer for reporting."""
        return f"Peer: {self.peer_address} VRF: {self.vrf}"

BgpRoute ¶

Model representing BGP routes.

Only IPv4 prefixes are supported for now.

Name	Type	Description	Default
`prefix`	`IPv4Network`	The IPv4 network address.	-
`vrf`	`str`	Optional VRF for the BGP peer. Defaults to `default`.	`'default'`
`paths`	`list[BgpRoutePath] \| None`	A list of paths for the BGP route. Required field in the `VerifyBGPRoutePaths` test.	`None`
`ecmp_count`	`int \| None`	The expected number of ECMP paths for the BGP route. Required field in the `VerifyBGPRouteECMP` test.	`None`

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

class BgpRoute(BaseModel):
    """Model representing BGP routes.

    Only IPv4 prefixes are supported for now.
    """

    model_config = ConfigDict(extra="forbid")
    prefix: IPv4Network
    """The IPv4 network address."""
    vrf: str = "default"
    """Optional VRF for the BGP peer. Defaults to `default`."""
    paths: list[BgpRoutePath] | None = None
    """A list of paths for the BGP route. Required field in the `VerifyBGPRoutePaths` test."""
    ecmp_count: int | None = None
    """The expected number of ECMP paths for the BGP route. Required field in the `VerifyBGPRouteECMP` test."""

    def __str__(self) -> str:
        """Return a human-readable string representation of the BgpRoute for reporting.

        Examples
        --------
        - Prefix: 192.168.66.100/24 VRF: default
        """
        return f"Prefix: {self.prefix} VRF: {self.vrf}"

BgpRoutePath ¶

Model representing a BGP route path.

Name	Type	Description	Default
`nexthop`	`IPv4Address`	The next-hop IPv4 address for the path.	-
`origin`	`Literal['Igp', 'Egp', 'Incomplete']`	The BGP origin attribute of the route.	-

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

class BgpRoutePath(BaseModel):
    """Model representing a BGP route path."""

    model_config = ConfigDict(extra="forbid")
    nexthop: IPv4Address
    """The next-hop IPv4 address for the path."""
    origin: Literal["Igp", "Egp", "Incomplete"]
    """The BGP origin attribute of the route."""

    def __str__(self) -> str:
        """Return a human-readable string representation of the RoutePath for reporting.

        Examples
        --------
        - Next-hop: 192.168.66.101 Origin: Igp
        """
        return f"Next-hop: {self.nexthop} Origin: {self.origin}"

BgpVrf ¶

Model representing a VRF in a BGP instance.

Name	Type	Description	Default
`vrf`	`str`	VRF context.	`'default'`
`address_families`	`list[AddressFamilyConfig]`	List of address family configuration.	-

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

class BgpVrf(BaseModel):
    """Model representing a VRF in a BGP instance."""

    vrf: str = "default"
    """VRF context."""
    address_families: list[AddressFamilyConfig]
    """List of address family configuration."""

    def __str__(self) -> str:
        """Return a human-readable string representation of the BgpVrf for reporting.

        Examples
        --------
        - VRF: default
        """
        return f"VRF: {self.vrf}"

RedistributedRouteConfig ¶

Model representing a BGP redistributed route configuration.

Name	Type	Description	Default
`proto`	`RedistributedProtocol`	The redistributed protocol.	-
`include_leaked`	`bool`	Flag to include leaked routes of the redistributed protocol while redistributing.	`False`
`route_map`	`str \| None`	Optional route map applied to the redistribution.	`None`

validate_inputs ¶

validate_inputs() -> Self

Validate that ‘include_leaked’ is not set when the redistributed protocol is AttachedHost, User, Dynamic, or RIP.

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

@model_validator(mode="after")
def validate_inputs(self) -> Self:
    """Validate that 'include_leaked' is not set when the redistributed protocol is AttachedHost, User, Dynamic, or RIP."""
    if self.include_leaked and self.proto in ["AttachedHost", "EOS SDK", "Dynamic", "RIP"]:
        msg = f"'include_leaked' field is not supported for redistributed protocol '{self.proto}'"
        raise ValueError(msg)
    return self

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

class RedistributedRouteConfig(BaseModel):
    """Model representing a BGP redistributed route configuration."""

    proto: RedistributedProtocol
    """The redistributed protocol."""
    include_leaked: bool = False
    """Flag to include leaked routes of the redistributed protocol while redistributing."""
    route_map: str | None = None
    """Optional route map applied to the redistribution."""

    @model_validator(mode="after")
    def validate_inputs(self) -> Self:
        """Validate that 'include_leaked' is not set when the redistributed protocol is AttachedHost, User, Dynamic, or RIP."""
        if self.include_leaked and self.proto in ["AttachedHost", "EOS SDK", "Dynamic", "RIP"]:
            msg = f"'include_leaked' field is not supported for redistributed protocol '{self.proto}'"
            raise ValueError(msg)
        return self

    def __str__(self) -> str:
        """Return a human-readable string representation of the RedistributedRouteConfig for reporting.

        Examples
        --------
        - Proto: Connected, Include Leaked: True, Route Map: RM-CONN-2-BGP
        """
        base_string = f"Proto: {self.proto}"
        if self.include_leaked:
            base_string += f", Include Leaked: {self.include_leaked}"
        if self.route_map:
            base_string += f", Route Map: {self.route_map}"
        return base_string

VxlanEndpoint ¶

Model for a VXLAN endpoint.

Name	Type	Description	Default
`address`	`IPv4Address \| MacAddress`	IPv4 or MAC address of the VXLAN endpoint.	-
`vni`	`Vni`	VNI of the VXLAN endpoint.	-

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

class VxlanEndpoint(BaseModel):
    """Model for a VXLAN endpoint."""

    address: IPv4Address | MacAddress
    """IPv4 or MAC address of the VXLAN endpoint."""
    vni: Vni
    """VNI of the VXLAN endpoint."""

    def __str__(self) -> str:
        """Return a human-readable string representation of the VxlanEndpoint for reporting."""
        return f"Address: {self.address} VNI: {self.vni}"

ANTA catalog for BGP tests

Tests¶

VerifyBGPAdvCommunities ¶

Inputs¶

VerifyBGPExchangedRoutes ¶

Inputs¶

VerifyBGPNlriAcceptance ¶

Inputs¶

VerifyBGPPeerASNCap ¶

Inputs¶

VerifyBGPPeerCount ¶

Inputs¶

VerifyBGPPeerDropStats ¶

Inputs¶

VerifyBGPPeerGroup ¶

Inputs¶

VerifyBGPPeerMD5Auth ¶

Inputs¶

VerifyBGPPeerMPCaps ¶

Inputs¶

VerifyBGPPeerRouteLimit ¶

Inputs¶

VerifyBGPPeerRouteRefreshCap ¶

Inputs¶

VerifyBGPPeerSession ¶

Inputs¶

VerifyBGPPeerSessionRibd ¶

Inputs¶

VerifyBGPPeerTtlMultiHops ¶

Inputs¶

VerifyBGPPeerUpdateErrors ¶

Inputs¶

VerifyBGPPeersHealth ¶

Inputs¶

VerifyBGPPeersHealthRibd ¶

Inputs¶

VerifyBGPRedistribution ¶

Inputs¶

VerifyBGPRouteECMP ¶

Inputs¶

VerifyBGPRoutePaths ¶

Inputs¶

VerifyBGPSpecificPeers ¶

Inputs¶

VerifyBGPTimers ¶

Inputs¶

VerifyBgpRouteMaps ¶

Inputs¶

VerifyEVPNType2Route ¶

Inputs¶

Input models¶

AFI_SAFI_MAPPINGS module-attribute ¶

IPV4_MULTICAST_SUPPORTED_PROTO module-attribute ¶

IPV6_MULTICAST_SUPPORTED_PROTO module-attribute ¶

AddressFamilyConfig ¶

validate_afi_safi_supported_routes ¶

BgpAddressFamily ¶

validate_inputs ¶

BgpNeighbor ¶

BgpPeer ¶

BgpRoute ¶

BgpRoutePath ¶

BgpVrf ¶

RedistributedRouteConfig ¶

validate_inputs ¶

VxlanEndpoint ¶

AFI_SAFI_MAPPINGS `module-attribute` ¶

IPV4_MULTICAST_SUPPORTED_PROTO `module-attribute` ¶

IPV6_MULTICAST_SUPPORTED_PROTO `module-attribute` ¶