Skip to content

Commit

Permalink
Updated unit tests for input models
Browse files Browse the repository at this point in the history
  • Loading branch information
@vitthalmagadum
vitthalmagadum committed Nov 28, 2024
1 parent cdc7685 commit b157eb0
Show file tree
Hide file tree
Showing 3 changed files with 146 additions and 6 deletions.
2 changes: 1 addition & 1 deletion anta/input_models/routing/bgp.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down
6 changes: 3 additions & 3 deletions anta/tests/routing/bgp.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down Expand Up @@ -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"
Expand Down Expand Up @@ -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"
Expand Down
144 changes: 142 additions & 2 deletions tests/units/input_models/routing/test_bgp.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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)

0 comments on commit b157eb0

Please sign in to comment.