Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

feat(anta): Added testcase to verify the BGP Redistributed Routes #993

Open
wants to merge 12 commits into
base: main
Choose a base branch
from
Open
Show file tree
Hide file tree
Changes from 7 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions anta/custom_types.py
Original file line number Diff line number Diff line change
Expand Up @@ -263,3 +263,4 @@ def validate_regex(value: str) -> str:
SnmpHashingAlgorithm = Literal["MD5", "SHA", "SHA-224", "SHA-256", "SHA-384", "SHA-512"]
SnmpEncryptionAlgorithm = Literal["AES-128", "AES-192", "AES-256", "DES"]
DynamicVlanSource = Literal["dmf", "dot1x", "dynvtep", "evpn", "mlag", "mlagsync", "mvpn", "swfwd", "vccbfd"]
RedistributedProtocol = Literal["AttachedHost", "Bgp", "Connected", "Dynamic", "IS-IS", "OSPF Internal", "OSPFv3 Internal", "RIP", "Static", "User"]
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We need to add OSPF External, OSPF Nssa-External, same for OSPFv3. We can also redistribute specific IS-IS levels; level-1, level-2, level-1-2 so please double check the options.

18 changes: 15 additions & 3 deletions anta/input_models/routing/bgp.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,7 @@
from pydantic import BaseModel, ConfigDict, Field, PositiveInt, model_validator
from pydantic_extra_types.mac_address import MacAddress

from anta.custom_types import Afi, BgpDropStats, BgpUpdateError, MultiProtocolCaps, Safi, Vni
from anta.custom_types import Afi, BgpDropStats, BgpUpdateError, MultiProtocolCaps, RedistributedProtocol, Safi, Vni

if TYPE_CHECKING:
import sys
Expand Down Expand Up @@ -40,6 +40,10 @@
}
"""Dictionary mapping AFI/SAFI to EOS key representation."""

AFI_SAFI_REDISTRIBUTED_ROUTE_KEY = {"ipv4Unicast": "v4u", "ipv4Multicast": "v4m", "ipv6Unicast": "v6u", "ipv6Multicast": "v6m"}
vitthalmagadum marked this conversation as resolved.
Show resolved Hide resolved

"""Dictionary mapping of AFI/SAFI to redistributed routes key representation."""


class BgpAddressFamily(BaseModel):
"""Model for a BGP address family."""
Expand Down Expand Up @@ -68,8 +72,11 @@ class BgpAddressFamily(BaseModel):
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.
"""
Can be enabled in the `VerifyBGPPeerCount` tests."""
redistributed_route_protocol: RedistributedProtocol | None = None
"""Specify redistributed route protocol. Required field in the `VerifyBGPRedistributedRoutes` test."""
route_map: str | None = None
"""Specify redistributed route protocol route map. Required field in the `VerifyBGPRedistributedRoutes` test."""
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Let's use separate models for the show bgp instance related tests. Something like this:

class RedistributedRoute(BaseModel):
    proto: RedistributedProtocol  # Custom type that you already created
    include_leaked: bool | None = None  # We should also check this for protocols that it applies
    route_map: str | None = None  # This is optional, we can redistribute without a route-map
class AfiSafiConfig(BaseModel):
    name: Literal["v4u", "v4m"]  # Here we should also support other formats like `IPv4 Unicast` or `ipv4Unicast`
    redistributed_routes: list[RedistributedRoute]
class BgpVrf(BaseModel):
    name: str = "default"
    address_families: list[AfiSafiConfig]
    # With this structure we can easily add more fields in the future for other tests.
    # router_id: str
    # local_as: int

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Actually AddressFamilyConfig might be more appropriate versus AfiSafiConfig

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Hi @carl-baillargeon.
Thank you for providing the input model. I've implemented the test case based on that. Here's the current input model:

- VerifyBGPRedistribution:
    vrfs:
       - vrf: default
          address_families:
            - afi_safi: ipv4Unicast
               redistributed_routes:
                 - proto: Connected
                     include_leaked: True
                     route_map: RM-CONN-2-BGP
                  - proto: Static
                     include_leaked: True
                     route_map: RM-CONN-2-BGP
            - afi_safi: IPv6 Unicast
               redistributed_routes:
                 - proto: Dynamic
                    route_map: RM-CONN-2-BGP
                 - proto: Static
                    include_leaked: True
                    route_map: RM-CONN-2-BGP

However, I'm encountering an issue with Cognitive Complexity. To resolve this, I've proposed an alternative input model approach:

- VerifyBGPRedistribution:
     address_families:
        - afi_safi: ipv4Unicast
           vrf: default
            redistributed_routes:
              - proto: Connected
                 include_leaked: True
                 route_map: RM-CONN-2-BGP
               - proto: Static
                  include_leaked: True
                   route_map: RM-CONN-2-BGP
        - afi_safi: IPv4 Unicast
           vrf: test
            redistributed_routes:
              - proto: Dynamic
                 route_map: RM-CONN-2-BGP
               - proto: Static
                  include_leaked: True
                  route_map: RM-CONN-2-BGP  
         - afi_safi: IPv4Unicast
            vrf: mgmt
            redistributed_routes:
               - proto: Dynamic
                  route_map: RM-CONN-2-BGP
                - proto: Static
                   include_leaked: True
                   route_map: RM-CONN-2-BGP

Key Reasons for the Proposed Change

  • It aligns with the structure already used in other BGP tests, ensuring consistency.
  • It eliminates unnecessary looping and conditional logic.
  • It reduces code complexity, making the test case easier to maintain.

As discussed with @gmuloc we will revisit this once Carl returns and then finalize the approach, In the meantime, I’m putting this on hold.

Thanks,
Geetanjali


@model_validator(mode="after")
def validate_inputs(self) -> Self:
Expand Down Expand Up @@ -97,6 +104,11 @@ def eos_key(self) -> str:
# 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)]

@property
def redistributed_route_key(self) -> str:
"""AFI/SAFI Redistributed route key representation."""
return AFI_SAFI_REDISTRIBUTED_ROUTE_KEY[self.eos_key]

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

Expand Down
89 changes: 89 additions & 0 deletions anta/tests/routing/bgp.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,9 @@
# Using a TypeVar for the BgpPeer model since mypy thinks it's a ClassVar and not a valid type when used in field validators
T = TypeVar("T", bound=BgpPeer)

# pylint: disable=C0302
# TODO: Refactor to reduce the number of lines in this module later


def _check_bgp_neighbor_capability(capability_status: dict[str, bool]) -> bool:
"""Check if a BGP neighbor capability is advertised, received, and enabled.
Expand Down Expand Up @@ -1685,3 +1688,89 @@ def test(self) -> None:

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


class VerifyBGPRedistributedRoutes(AntaTest):
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I would call this test VerifyBGPRedistribution instead since we are not checking anything related to redistributed routes.

"""Verifies BGP redistributed routes protocol and route-map.

This test performs the following checks for each specified route:

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

Note: For "User" redistributed_route_protocol field, checking that it's "EOS SDK" versus User.

Expected Results
----------------
* Success: If all of the following conditions are met:
- The expected address-family is configured on the device.
- The redistributed route protocol 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 or route map does not match the expected value for a route.

Examples
--------
```yaml
anta.tests.routing:
bgp:
- VerifyBGPRedistributedRoutes:
address_families:
- afi: "ipv4"
safi: "unicast"
vrf: default
redistributed_route_protocol: Connected
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 VerifyBGPRedistributedRoutes test."""

address_families: list[BgpAddressFamily]
"""List of BGP address families."""

@field_validator("address_families")
@classmethod
def validate_address_families(cls, address_families: list[BgpAddressFamily]) -> list[BgpAddressFamily]:
"""Validate that all required fields are provided in each address family."""
for address_family in address_families:
if address_family.afi not in ["ipv4", "ipv6"] or address_family.safi not in ["unicast", "multicast"]:
msg = f"{address_family}; redistributed route protocol is not supported for address family `{address_family.eos_key}`"
raise ValueError(msg)
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This check could go in the AfiSafiConfig model directly.

if address_family.redistributed_route_protocol is None:
msg = f"{address_family}; 'redistributed_route_protocol' field missing in the input"
raise ValueError(msg)
if address_family.route_map is None:
msg = f"{address_family}; 'route_map' field missing in the input"
raise ValueError(msg)
return address_families

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

# If the specified VRF, AFI-SAFI details are not found, or if the redistributed route protocol or route map do not match the expected values, the test fails.
for address_family in self.inputs.address_families:
vrf = address_family.vrf
redistributed_route_protocol = "EOS SDK" if address_family.redistributed_route_protocol == "User" else address_family.redistributed_route_protocol
route_map = address_family.route_map
afi_safi_key = address_family.redistributed_route_key

if not (afi_safi_configs := get_value(cmd_output, f"vrfs.{vrf}.afiSafiConfig.{afi_safi_key}")):
self.result.is_failure(f"{address_family} - Not found")
continue

if not (route := get_item(afi_safi_configs.get("redistributedRoutes"), "proto", redistributed_route_protocol)):
self.result.is_failure(f"{address_family} Protocol: {address_family.redistributed_route_protocol} - Not Found")
continue

if (act_route_map := route.get("routeMap", "Not Found")) != route_map:
self.result.is_failure(
f"{address_family} Protocol: {address_family.redistributed_route_protocol} - Route map mismatch - Expected: {route_map} Actual: {act_route_map}"
)
8 changes: 8 additions & 0 deletions examples/tests.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -506,6 +506,14 @@ anta.tests.routing.bgp:
- VerifyBGPPeersHealthRibd:
# Verifies the health of all the BGP IPv4 peer(s).
check_tcp_queues: True
- VerifyBGPRedistributedRoutes:
# Verifies BGP redistributed routes protocol and route-map.
address_families:
- afi: "ipv4"
safi: "unicast"
vrf: default
redistributed_route_protocol: Connected
route_map: RM-CONN-2-BGP
- VerifyBGPRoutePaths:
# Verifies BGP IPv4 route paths.
route_entries:
Expand Down
157 changes: 157 additions & 0 deletions tests/units/anta_tests/routing/test_bgp.py
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,7 @@
VerifyBGPPeersHealth,
VerifyBGPPeersHealthRibd,
VerifyBGPPeerUpdateErrors,
VerifyBGPRedistributedRoutes,
VerifyBgpRouteMaps,
VerifyBGPRoutePaths,
VerifyBGPSpecificPeers,
Expand Down Expand Up @@ -5100,4 +5101,160 @@ def test_check_bgp_neighbor_capability(input_dict: dict[str, bool], expected: bo
"messages": ["Prefix: 10.100.0.128/31 VRF: default - prefix not found", "Prefix: 10.100.0.130/31 VRF: MGMT - prefix not found"],
},
},
{
"name": "success",
"test": VerifyBGPRedistributedRoutes,
"eos_data": [
{
"vrfs": {
"default": {"afiSafiConfig": {"v4u": {"redistributedRoutes": [{"proto": "Connected", "routeMap": "RM-CONN-2-BGP"}]}}},
"test": {"afiSafiConfig": {"v6u": {"redistributedRoutes": [{"proto": "Connected", "routeMap": "RM-CONN-2-BGP"}]}}},
}
}
],
"inputs": {
"address_families": [
{
"vrf": "default",
"redistributed_route_protocol": "Connected",
"route_map": "RM-CONN-2-BGP",
"afi": "ipv4",
"safi": "unicast",
},
{
"vrf": "test",
"redistributed_route_protocol": "Connected",
"route_map": "RM-CONN-2-BGP",
"afi": "ipv6",
"safi": "unicast",
},
]
},
"expected": {"result": "success"},
},
{
"name": "failure-afi-safi-config-not-found",
"test": VerifyBGPRedistributedRoutes,
"eos_data": [
{
"vrfs": {
"default": {"afiSafiConfig": {"v4u": {}}},
"test": {"afiSafiConfig": {"v6m": {"redistributedRoutes": [{"proto": "Connected", "routeMap": "RM-CONN-2-BGP"}]}}},
}
}
],
"inputs": {
"address_families": [
{
"vrf": "default",
"redistributed_route_protocol": "Connected",
"route_map": "RM-CONN-2-BGP",
"afi": "ipv4",
"safi": "unicast",
},
{
"vrf": "test",
"redistributed_route_protocol": "Connected",
"route_map": "RM-CONN-2-BGP",
"afi": "ipv6",
"safi": "multicast",
},
]
},
"expected": {"result": "failure", "messages": ["AFI: ipv4 SAFI: unicast VRF: default - Not found"]},
},
{
"name": "failure-expected-proto-not-found",
"test": VerifyBGPRedistributedRoutes,
"eos_data": [
{
"vrfs": {
"default": {
"afiSafiConfig": {
"v4m": {"redistributedRoutes": [{"proto": "RIP", "routeMap": "RM-CONN-2-BGP"}, {"proto": "IS-IS", "routeMap": "RM-MLAG-PEER-IN"}]}
}
},
"test": {
"afiSafiConfig": {
"v6m": {
"redistributedRoutes": [{"proto": "Static", "routeMap": "RM-CONN-2-BGP"}],
}
}
},
}
}
],
"inputs": {
"address_families": [
{
"vrf": "default",
"redistributed_route_protocol": "Connected",
"route_map": "RM-CONN-2-BGP",
"afi": "ipv4",
"safi": "multicast",
},
{
"vrf": "test",
"redistributed_route_protocol": "User",
"route_map": "RM-CONN-2-BGP",
"afi": "ipv6",
"safi": "multicast",
},
]
},
"expected": {
"result": "failure",
"messages": [
"AFI: ipv4 SAFI: multicast VRF: default Protocol: Connected - Not Found",
"AFI: ipv6 SAFI: multicast VRF: test Protocol: User - Not Found",
],
},
},
{
"name": "failure-route-map-not-found",
"test": VerifyBGPRedistributedRoutes,
"eos_data": [
{
"vrfs": {
"default": {
"afiSafiConfig": {
"v4m": {"redistributedRoutes": [{"proto": "Connected", "routeMap": "RM-CONN-10-BGP"}, {"proto": "IS-IS", "routeMap": "RM-MLAG-PEER-IN"}]}
}
},
"test": {
"afiSafiConfig": {
"v6u": {
"redistributedRoutes": [{"proto": "Connected", "routeMap": "RM-MLAG-PEER-IN"}],
}
}
},
}
}
],
"inputs": {
"address_families": [
{
"vrf": "default",
"redistributed_route_protocol": "Connected",
"route_map": "RM-CONN-2-BGP",
"afi": "ipv4",
"safi": "multicast",
},
{
"vrf": "test",
"redistributed_route_protocol": "Connected",
"route_map": "RM-CONN-2-BGP",
"afi": "ipv6",
"safi": "unicast",
},
]
},
"expected": {
"result": "failure",
"messages": [
"AFI: ipv4 SAFI: multicast VRF: default Protocol: Connected - Route map mismatch - Expected: RM-CONN-2-BGP Actual: RM-CONN-10-BGP",
"AFI: ipv6 SAFI: unicast VRF: test Protocol: Connected - Route map mismatch - Expected: RM-CONN-2-BGP Actual: RM-MLAG-PEER-IN",
],
},
},
]
46 changes: 46 additions & 0 deletions tests/units/input_models/routing/test_bgp.py
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@
VerifyBGPPeerGroup,
VerifyBGPPeerMPCaps,
VerifyBGPPeerRouteLimit,
VerifyBGPRedistributedRoutes,
VerifyBgpRouteMaps,
VerifyBGPSpecificPeers,
VerifyBGPTimers,
Expand Down Expand Up @@ -288,3 +289,48 @@ def test_invalid(self, bgp_peers: list[BgpPeer]) -> None:
"""Test VerifyBGPNlriAcceptance.Input invalid inputs."""
with pytest.raises(ValidationError):
VerifyBGPNlriAcceptance.Input(bgp_peers=bgp_peers)


class TestVerifyBGPRedistributedRoutes:
"""Test anta.tests.routing.bgp.VerifyBGPRedistributedRoutes.Input."""

@pytest.mark.parametrize(
("address_families"),
[
pytest.param(
[{"afi": "ipv4", "safi": "unicast", "vrf": "default", "redistributed_route_protocol": "Connected", "route_map": "RM-CONN-2-BGP"}], id="ipv4-valid"
),
pytest.param(
[{"afi": "ipv6", "safi": "multicast", "vrf": "default", "redistributed_route_protocol": "Connected", "route_map": "RM-CONN-2-BGP"}], id="ipv6-valid"
),
],
)
def test_valid(self, address_families: list[BgpAddressFamily]) -> None:
"""Test VerifyBGPRedistributedRoutes.Input valid inputs."""
VerifyBGPRedistributedRoutes.Input(address_families=address_families)

@pytest.mark.parametrize(
("address_families"),
[
pytest.param(
[{"afi": "ipv4", "safi": "unicast", "vrf": "default", "redistributed_route_protocol": None, "route_map": "RM-CONN-2-BGP"}],
id="invalid-redistributed-route-protocol",
),
pytest.param(
[{"afi": "ipv6", "safi": "multicast", "vrf": "default", "redistributed_route_protocol": "Connected", "route_map": None}],
id="invalid-route-map",
),
pytest.param(
[{"afi": "evpn", "safi": "unicast", "vrf": "default", "redistributed_route_protocol": "Connected", "route_map": "RM-CONN-2-BGP"}],
id="invalid-afi-for-redistributed-route",
),
pytest.param(
[{"afi": "ipv6", "safi": "sr-te", "vrf": "default", "redistributed_route_protocol": "Connected", "route_map": "RM-CONN-2-BGP"}],
id="invalid-safi-for-redistributed-route",
),
],
)
def test_invalid(self, address_families: list[BgpAddressFamily]) -> None:
"""Test VerifyBGPRedistributedRoutes.Input invalid inputs."""
with pytest.raises(ValidationError):
VerifyBGPRedistributedRoutes.Input(address_families=address_families)
Loading