diff --git a/anta/input_models/routing/bgp.py b/anta/input_models/routing/bgp.py index 5abbe1911..f75135381 100644 --- a/anta/input_models/routing/bgp.py +++ b/anta/input_models/routing/bgp.py @@ -178,7 +178,7 @@ def __str__(self) -> str: return f"Peer: {self.peer_address} VRF: {self.vrf}" -class BgpNeighbor(BgpPeer): +class BgpNeighbor(BgpPeer): # pragma: no cover """Alias for the BgpPeer model to maintain backward compatibility. When initialized, it will emit a depreciation warning and call the BgpPeer model. diff --git a/anta/tests/routing/bgp.py b/anta/tests/routing/bgp.py index 51efdbfac..cab7c61ce 100644 --- a/anta/tests/routing/bgp.py +++ b/anta/tests/routing/bgp.py @@ -383,7 +383,7 @@ class Input(AntaTest.Input): @field_validator("bgp_peers") @classmethod def validate_bgp_peers(cls, bgp_peers: list[BgpPeer]) -> list[BgpPeer]: - """Validate that 'peers' field is provided in each address family.""" + """Validate that 'advertised_routes' or 'received_routes' field is provided in each address family.""" for peer in bgp_peers: if peer.advertised_routes is None or peer.received_routes is None: msg = f"{peer} 'advertised_routes' or 'received_routes' field missing in the input" @@ -486,7 +486,7 @@ class Input(AntaTest.Input): @field_validator("bgp_peers") @classmethod def validate_bgp_peers(cls, bgp_peers: list[T]) -> list[T]: - """Validate that 'peers' field is provided in each address family.""" + """Validate that 'capabilities' field is provided in each address family.""" for peer in bgp_peers: if peer.capabilities is None: msg = f"{peer} 'capabilities' field missing in the input" @@ -927,7 +927,7 @@ class Input(AntaTest.Input): @field_validator("bgp_peers") @classmethod def validate_bgp_peers(cls, bgp_peers: list[T]) -> list[T]: - """Validate that 'peers' field is provided in each address family.""" + """Validate that 'hold_time' or 'keep_alive_time' field is provided in each address family.""" 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" diff --git a/tests/units/input_models/routing/test_bgp.py b/tests/units/input_models/routing/test_bgp.py index aabc8f293..66c37af61 100644 --- a/tests/units/input_models/routing/test_bgp.py +++ b/tests/units/input_models/routing/test_bgp.py @@ -11,8 +11,16 @@ import pytest from pydantic import ValidationError -from anta.input_models.routing.bgp import BgpAddressFamily -from anta.tests.routing.bgp import VerifyBGPPeerCount, VerifyBGPSpecificPeers +from anta.input_models.routing.bgp import BgpAddressFamily, BgpPeer +from anta.tests.routing.bgp import ( + VerifyBGPExchangedRoutes, + VerifyBGPPeerCount, + VerifyBGPPeerMPCaps, + VerifyBGPPeerRouteLimit, + VerifyBgpRouteMaps, + VerifyBGPSpecificPeers, + VerifyBGPTimers, +) if TYPE_CHECKING: from anta.custom_types import Afi, Safi @@ -96,3 +104,135 @@ def test_invalid(self, address_families: list[BgpAddressFamily]) -> None: """Test VerifyBGPSpecificPeers.Input invalid inputs.""" with pytest.raises(ValidationError): VerifyBGPSpecificPeers.Input(address_families=address_families) + + +class TestVerifyBGPExchangedRoutesInput: + """Test anta.tests.routing.bgp.VerifyBGPExchangedRoutes.Input.""" + + @pytest.mark.parametrize( + ("bgp_peers"), + [ + pytest.param( + [{"peer_address": "172.30.255.5", "vrf": "default", "advertised_routes": ["192.0.254.5/32"], "received_routes": ["192.0.255.4/32"]}], + id="valid_both_received_advertised", + ), + ], + ) + def test_valid(self, bgp_peers: list[BgpPeer]) -> None: + """Test VerifyBGPExchangedRoutes.Input valid inputs.""" + VerifyBGPExchangedRoutes.Input(bgp_peers=bgp_peers) + + @pytest.mark.parametrize( + ("bgp_peers"), + [ + pytest.param([{"peer_address": "172.30.255.5", "vrf": "default"}], id="invalid"), + pytest.param([{"peer_address": "172.30.255.5", "vrf": "default", "advertised_routes": ["192.0.254.5/32"]}], id="invalid_received_route"), + pytest.param([{"peer_address": "172.30.255.5", "vrf": "default", "received_routes": ["192.0.254.5/32"]}], id="invalid_advertised_route"), + ], + ) + def test_invalid(self, bgp_peers: list[BgpPeer]) -> None: + """Test VerifyBGPExchangedRoutes.Input invalid inputs.""" + with pytest.raises(ValidationError): + VerifyBGPExchangedRoutes.Input(bgp_peers=bgp_peers) + + +class TestVerifyBGPPeerMPCapsInput: + """Test anta.tests.routing.bgp.VerifyBGPPeerMPCaps.Input.""" + + @pytest.mark.parametrize( + ("bgp_peers"), + [ + pytest.param([{"peer_address": "172.30.255.5", "vrf": "default", "capabilities": ["ipv4Unicast"]}], id="valid"), + ], + ) + def test_valid(self, bgp_peers: list[BgpPeer]) -> None: + """Test VerifyBGPPeerMPCaps.Input valid inputs.""" + VerifyBGPPeerMPCaps.Input(bgp_peers=bgp_peers) + + @pytest.mark.parametrize( + ("bgp_peers"), + [ + pytest.param([{"peer_address": "172.30.255.5", "vrf": "default"}], id="invalid"), + ], + ) + def test_invalid(self, bgp_peers: list[BgpPeer]) -> None: + """Test VerifyBGPPeerMPCaps.Input invalid inputs.""" + with pytest.raises(ValidationError): + VerifyBGPPeerMPCaps.Input(bgp_peers=bgp_peers) + + +class TestVerifyBGPTimersInput: + """Test anta.tests.routing.bgp.VerifyBGPTimers.Input.""" + + @pytest.mark.parametrize( + ("bgp_peers"), + [ + pytest.param([{"peer_address": "172.30.255.5", "vrf": "default", "hold_time": 180, "keep_alive_time": 60}], id="valid"), + ], + ) + def test_valid(self, bgp_peers: list[BgpPeer]) -> None: + """Test VerifyBGPTimers.Input valid inputs.""" + VerifyBGPTimers.Input(bgp_peers=bgp_peers) + + @pytest.mark.parametrize( + ("bgp_peers"), + [ + pytest.param([{"peer_address": "172.30.255.5", "vrf": "default"}], id="invalid"), + pytest.param([{"peer_address": "172.30.255.5", "vrf": "default", "hold_time": 180}], id="invalid_keep_alive"), + pytest.param([{"peer_address": "172.30.255.5", "vrf": "default", "keep_alive_time": 180}], id="invalid_hold_time"), + ], + ) + def test_invalid(self, bgp_peers: list[BgpPeer]) -> None: + """Test VerifyBGPTimers.Input invalid inputs.""" + with pytest.raises(ValidationError): + VerifyBGPTimers.Input(bgp_peers=bgp_peers) + + +class TestVerifyBgpRouteMapsInput: + """Test anta.tests.routing.bgp.VerifyBgpRouteMaps.Input.""" + + @pytest.mark.parametrize( + ("bgp_peers"), + [ + pytest.param([{"peer_address": "172.30.255.5", "vrf": "default", "inbound_route_map": "Test", "outbound_route_map": "Test"}], id="valid"), + ], + ) + def test_valid(self, bgp_peers: list[BgpPeer]) -> None: + """Test VerifyBgpRouteMaps.Input valid inputs.""" + VerifyBgpRouteMaps.Input(bgp_peers=bgp_peers) + + @pytest.mark.parametrize( + ("bgp_peers"), + [ + pytest.param([{"peer_address": "172.30.255.5", "vrf": "default"}], id="invalid"), + ], + ) + def test_invalid(self, bgp_peers: list[BgpPeer]) -> None: + """Test VerifyBgpRouteMaps.Input invalid inputs.""" + with pytest.raises(ValidationError): + VerifyBgpRouteMaps.Input(bgp_peers=bgp_peers) + + +class TestVerifyBGPPeerRouteLimitInput: + """Test anta.tests.routing.bgp.VerifyBGPPeerRouteLimit.Input.""" + + @pytest.mark.parametrize( + ("bgp_peers"), + [ + pytest.param([{"peer_address": "172.30.255.5", "vrf": "default", "maximum_routes": 10000}], id="valid"), + ], + ) + def test_valid(self, bgp_peers: list[BgpPeer]) -> None: + """Test VerifyBGPPeerRouteLimit.Input valid inputs.""" + VerifyBGPPeerRouteLimit.Input(bgp_peers=bgp_peers) + + @pytest.mark.parametrize( + ("bgp_peers"), + [ + pytest.param([{"peer_address": "172.30.255.5", "vrf": "default"}], id="invalid"), + ], + ) + def test_invalid(self, bgp_peers: list[BgpPeer]) -> None: + """Test VerifyBGPPeerRouteLimit.Input invalid inputs.""" + with pytest.raises(ValidationError): + VerifyBGPPeerRouteLimit.Input(bgp_peers=bgp_peers)