diff --git a/anta/input_models/routing/bgp.py b/anta/input_models/routing/bgp.py index 04ced3853..2eb14e4bb 100644 --- a/anta/input_models/routing/bgp.py +++ b/anta/input_models/routing/bgp.py @@ -169,9 +169,11 @@ class BgpPeer(BaseModel): 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""" + """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) - """Optional maximum routes warning limit. If not provided, it defaults to `0` meaning no warning limit.""" + """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.""" def __str__(self) -> str: """Return a human-readable string representation of the BgpPeer for reporting.""" diff --git a/anta/tests/routing/bgp.py b/anta/tests/routing/bgp.py index fb17cbc8c..a2863c6e0 100644 --- a/anta/tests/routing/bgp.py +++ b/anta/tests/routing/bgp.py @@ -330,8 +330,88 @@ def test(self) -> None: self.result.is_failure(f"{address_family} Peer: {peer_ip} - Session has non-empty message queues - InQ: {inq}, OutQ: {outq}") +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. Checks that the BGP session is in the `Established` state. + 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`. + - All peers have empty TCP message queues if `check_tcp_queues` is `True` (default). + * 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`. + - A peer has non-empty TCP message queues (input or output) when `check_tcp_queues` is `True`. + + Examples + -------- + ```yaml + anta.tests.routing: + bgp: + - VerifyBGPPeerSession: + 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.""" + + 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} - Session state is not established - State: {peer_data['state']}") + continue + + # 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}") + + class VerifyBGPExchangedRoutes(AntaTest): - """Verifies the advertised and received routes of BGP peers. + """Verifies the advertised and received routes of BGP IPv4 peer(s). This test performs the following checks for each specified peer: @@ -381,13 +461,13 @@ class Input(AntaTest.Input): """Input model for the VerifyBGPExchangedRoutes test.""" bgp_peers: list[BgpPeer] - """List of BGP peers.""" + """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 address family.""" + """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 or peer.received_routes is None: msg = f"{peer} 'advertised_routes' or 'received_routes' field missing in the input" @@ -435,28 +515,25 @@ def test(self) -> None: class VerifyBGPPeerMPCaps(AntaTest): - """Verifies the multiprotocol capabilities of BGP peers. + """Verifies the multiprotocol capabilities of BGP IPv4 peer(s). This test performs the following checks for each specified peer: - 1. Confirms that the specified VRF is configured. - 2. Verifies that the peer exists in the BGP configuration. - 3. For each specified capability: + 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. - 4. When strict mode is enabled (`strict: true`): + 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: - - The specified VRF is configured. - 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: - - The specified VRF is not configured. - 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). @@ -484,13 +561,13 @@ class Input(AntaTest.Input): """Input model for the VerifyBGPPeerMPCaps test.""" bgp_peers: list[BgpPeer] - """List of BGP peers""" + """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 address family.""" + """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" @@ -506,14 +583,10 @@ def test(self) -> None: for peer in self.inputs.bgp_peers: peer_ip = str(peer.peer_address) - - # Check if the VRF is configured - if (vrf_output := get_value(output, f"vrfs.{peer.vrf}")) is None: - self.result.is_failure(f"{peer} - VRF not configured") - continue + peer_list = get_value(output, f"vrfs.{peer.vrf}.peerList", default=[]) # Check if the peer is found - if (peer_data := get_item(vrf_output["peerList"], "peerAddress", peer_ip)) is None: + if (peer_data := get_item(peer_list, "peerAddress", peer_ip)) is None: self.result.is_failure(f"{peer} - Not found") continue @@ -537,14 +610,13 @@ def test(self) -> None: class VerifyBGPPeerASNCap(AntaTest): - """Verifies the four octet ASN capability of BGP peers. + """Verifies the four octet ASN capability of BGP IPv4 peer(s). This test performs the following checks for each specified peer: - 1. Confirms that the specified VRF is configured. - 2. Verifies that the peer exists in the BGP configuration. - 3. Validates that the capability is present in the peer configuration. - 4. Confirms that the capability is advertised, received, and enabled. + 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 ---------------- @@ -576,7 +648,7 @@ class Input(AntaTest.Input): """Input model for the VerifyBGPPeerASNCap test.""" bgp_peers: list[BgpPeer] - """List of BGP peers.""" + """List of BGP IPv4 peers.""" BgpPeer: ClassVar[type[BgpPeer]] = BgpPeer @AntaTest.anta_test @@ -606,14 +678,13 @@ def test(self) -> None: class VerifyBGPPeerRouteRefreshCap(AntaTest): - """Verifies the route refresh capabilities of a BGP peer in a specified VRF. + """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. Confirms that the specified VRF is configured. - 2. Verifies that the peer exists in the BGP configuration. - 3. Validates that the route refresh capability is present in the peer configuration. - 4. Confirms that the capability is advertised, received, and enabled. + 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 ---------------- @@ -645,7 +716,7 @@ class Input(AntaTest.Input): """Input model for the VerifyBGPPeerRouteRefreshCap test.""" bgp_peers: list[BgpPeer] - """List of BGP peers""" + """List of BGP IPv4 peers.""" BgpPeer: ClassVar[type[BgpPeer]] = BgpPeer @AntaTest.anta_test @@ -675,14 +746,13 @@ def test(self) -> None: class VerifyBGPPeerMD5Auth(AntaTest): - """Verifies the MD5 authentication and state of IPv4 BGP peers in a specified VRF. + """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. Confirms that the specified VRF is configured. - 2. Verifies that the peer exists in the BGP configuration. - 3. Validates that the BGP session is in `Established` state. - 4. Confirms that MD5 authentication is enabled for the 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 ---------------- @@ -814,13 +884,12 @@ def test(self) -> None: class VerifyBGPAdvCommunities(AntaTest): - """Verifies that advertised communities are standard, extended and large for BGP peers. + """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. Confirms that the specified VRF is configured. - 2. Verifies that the peer exists in the BGP configuration. - 3. Validates that all required community types are advertised: + 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 @@ -855,7 +924,7 @@ class Input(AntaTest.Input): """Input model for the VerifyBGPAdvCommunities test.""" bgp_peers: list[BgpPeer] - """List of BGP peers.""" + """List of BGP IPv4 peers.""" BgpPeer: ClassVar[type[BgpPeer]] = BgpPeer @AntaTest.anta_test @@ -880,13 +949,12 @@ def test(self) -> None: class VerifyBGPTimers(AntaTest): - """Verifies the timers of BGP peers. + """Verifies the timers of BGP IPv4 peer(s). This test performs the following checks for each specified peer: - 1. Confirms that the specified VRF is configured. - 2. Verifies that the peer exists in the BGP configuration. - 3. Confirms the BGP session hold time/keepalive timers match the expected value. + 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 ---------------- @@ -922,13 +990,13 @@ class Input(AntaTest.Input): """Input model for the VerifyBGPTimers test.""" bgp_peers: list[BgpPeer] - """List of BGP peers""" + """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 address family.""" + """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" @@ -963,9 +1031,8 @@ class VerifyBGPPeerDropStats(AntaTest): This test performs the following checks for each specified peer: - 1. Confirms that the specified VRF is configured. - 2. Verifies that the peer exists in the BGP configuration. - 3. Validates the BGP drop statistics: + 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. @@ -1002,7 +1069,7 @@ class Input(AntaTest.Input): """Input model for the VerifyBGPPeerDropStats test.""" bgp_peers: list[BgpPeer] - """List of BGP peers""" + """List of BGP IPv4 peers.""" BgpPeer: ClassVar[type[BgpPeer]] = BgpPeer @AntaTest.anta_test @@ -1040,9 +1107,8 @@ class VerifyBGPPeerUpdateErrors(AntaTest): This test performs the following checks for each specified peer: - 1. Confirms that the specified VRF is configured. - 2. Verifies that the peer exists in the BGP configuration. - 3. Validates the BGP update error counters: + 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. @@ -1080,7 +1146,7 @@ class Input(AntaTest.Input): """Input model for the VerifyBGPPeerUpdateErrors test.""" bgp_peers: list[BgpPeer] - """List of BGP peers""" + """List of BGP IPv4 peers.""" BgpPeer: ClassVar[type[BgpPeer]] = BgpPeer @AntaTest.anta_test @@ -1118,9 +1184,8 @@ class VerifyBgpRouteMaps(AntaTest): This test performs the following checks for each specified peer: - 1. Confirms that the specified VRF is configured. - 2. Verifies that the peer exists in the BGP configuration. - 3. Validates the correct BGP route maps are applied in the correct direction (inbound or outbound). + 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 ---------------- @@ -1152,16 +1217,13 @@ class Input(AntaTest.Input): """Input model for the VerifyBgpRouteMaps test.""" bgp_peers: list[BgpPeer] - """List of BGP peers""" + """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 'peers' field is provided in each address family. - - At least one of 'inbound' or 'outbound' route-map must be provided. - """ + """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}; At least one of 'inbound_route_map' or 'outbound_route_map' must be provided." @@ -1196,22 +1258,21 @@ def test(self) -> None: class VerifyBGPPeerRouteLimit(AntaTest): - """Verifies maximum routes and outbound route-maps of BGP IPv4 peer(s). + """Verifies maximum routes and warning limit for BGP IPv4 peer(s). This test performs the following checks for each specified peer: - 1. Confirms that the specified VRF is configured. - 2. Verifies that the peer exists in the BGP configuration. - 3. Confirms the Maximum routes and maximum routes warning limit, if provided match the expected value. + 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 routese/maximum routes warning limit match the expected value for a peer. + - 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 routese/maximum routes warning limit do not match the expected value for a peer. + - The maximum routes/maximum routes warning limit do not match the expected value for a peer. Examples -------- @@ -1234,13 +1295,13 @@ class Input(AntaTest.Input): """Input model for the VerifyBGPPeerRouteLimit test.""" bgp_peers: list[BgpPeer] - """List of BGP peers""" + """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 'peers' field is provided in each address family.""" + """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" @@ -1265,10 +1326,10 @@ def test(self) -> None: self.result.is_failure(f"{peer} - Not found") continue - # Verify maximum routes configured. - if (actual_routes := peer_data.get("maxTotalRoutes", "Not Found")) != maximum_routes: - self.result.is_failure(f"{peer} - Maximum routes mismatch - Expected: {maximum_routes}, Actual: {actual_routes}") + # 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 given. - if warning_limit and (actual_warning_limit := peer_data.get("totalRoutesWarnLimit", "Not Found")) != warning_limit: - self.result.is_failure(f"{peer} - Maximum route warning limit mismatch - Expected: {warning_limit}, Actual: {actual_warning_limit}") + # 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}") diff --git a/examples/tests.yaml b/examples/tests.yaml index 6e1fc5b85..a4bc1fabf 100644 --- a/examples/tests.yaml +++ b/examples/tests.yaml @@ -366,14 +366,14 @@ anta.tests.ptp: # Verifies the PTP interfaces state. anta.tests.routing.bgp: - VerifyBGPAdvCommunities: - # Verifies that advertised communities are standard, extended and large for BGP peers. + # Verifies that advertised communities are standard, extended and large for BGP IPv4 peer(s). bgp_peers: - peer_address: 172.30.11.17 vrf: default - peer_address: 172.30.11.21 vrf: default - VerifyBGPExchangedRoutes: - # Verifies the advertised and received routes of BGP peers. + # Verifies the advertised and received routes of BGP IPv4 peer(s). bgp_peers: - peer_address: 172.30.255.5 vrf: default @@ -389,7 +389,7 @@ anta.tests.routing.bgp: received_routes: - 192.0.254.3/32 - VerifyBGPPeerASNCap: - # Verifies the four octet ASN capability of BGP peers. + # Verifies the four octet ASN capability of BGP IPv4 peer(s). bgp_peers: - peer_address: 172.30.11.1 vrf: default @@ -419,14 +419,14 @@ anta.tests.routing.bgp: - inDropAsloop - prefixEvpnDroppedUnsupportedRouteType - VerifyBGPPeerMD5Auth: - # Verifies the MD5 authentication and state of IPv4 BGP peers in a specified VRF. + # Verifies the MD5 authentication and state of IPv4 BGP peer(s) in a specified VRF. bgp_peers: - peer_address: 172.30.11.1 vrf: default - peer_address: 172.30.11.5 vrf: default - VerifyBGPPeerMPCaps: - # Verifies the multiprotocol capabilities of BGP peers. + # Verifies the multiprotocol capabilities of BGP IPv4 peer(s). bgp_peers: - peer_address: 172.30.11.1 vrf: default @@ -434,17 +434,29 @@ anta.tests.routing.bgp: capabilities: - ipv4Unicast - VerifyBGPPeerRouteLimit: - # Verifies maximum routes and outbound route-maps of BGP IPv4 peer(s). + # Verifies maximum routes and warning limit for BGP IPv4 peer(s). bgp_peers: - peer_address: 172.30.11.1 vrf: default maximum_routes: 12000 warning_limit: 10000 - VerifyBGPPeerRouteRefreshCap: - # Verifies the route refresh capabilities of a BGP peer in a specified VRF. + # Verifies the route refresh capabilities of IPv4 BGP peer(s) in a specified VRF. bgp_peers: - peer_address: 172.30.11.1 vrf: default + - VerifyBGPPeerSession: + # Verifies the session state of BGP IPv4 peer(s). + 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 - VerifyBGPPeerUpdateErrors: # Verifies BGP update error counters for the provided BGP IPv4 peer(s). bgp_peers: @@ -478,7 +490,7 @@ anta.tests.routing.bgp: - 10.1.255.2 - 10.1.255.4 - VerifyBGPTimers: - # Verifies the timers of BGP peers. + # Verifies the timers of BGP IPv4 peer(s). bgp_peers: - peer_address: 172.30.11.1 vrf: default diff --git a/tests/units/anta_tests/routing/test_bgp.py b/tests/units/anta_tests/routing/test_bgp.py index d58054170..4d9e3c026 100644 --- a/tests/units/anta_tests/routing/test_bgp.py +++ b/tests/units/anta_tests/routing/test_bgp.py @@ -21,6 +21,7 @@ VerifyBGPPeerMPCaps, VerifyBGPPeerRouteLimit, VerifyBGPPeerRouteRefreshCap, + VerifyBGPPeerSession, VerifyBGPPeersHealth, VerifyBGPPeerUpdateErrors, VerifyBgpRouteMaps, @@ -1389,50 +1390,6 @@ def test_check_bgp_neighbor_capability(input_dict: dict[str, bool], expected: bo }, "expected": {"result": "success"}, }, - { - "name": "failure-no-vrf", - "test": VerifyBGPPeerMPCaps, - "eos_data": [ - { - "vrfs": { - "default": { - "peerList": [ - { - "peerAddress": "172.30.11.1", - "neighborCapabilities": { - "multiprotocolCaps": { - "ipv4Unicast": { - "advertised": True, - "received": True, - "enabled": True, - }, - "ipv4MplsVpn": { - "advertised": True, - "received": True, - "enabled": True, - }, - } - }, - } - ] - } - } - } - ], - "inputs": { - "bgp_peers": [ - { - "peer_address": "172.30.11.1", - "vrf": "MGMT", - "capabilities": ["ipv4 Unicast", "ipv4mplslabels"], - } - ] - }, - "expected": { - "result": "failure", - "messages": ["Peer: 172.30.11.1 VRF: MGMT - VRF not configured"], - }, - }, { "name": "failure-no-peer", "test": VerifyBGPPeerMPCaps, @@ -3740,6 +3697,39 @@ def test_check_bgp_neighbor_capability(input_dict: dict[str, bool], expected: bo }, "expected": {"result": "success"}, }, + { + "name": "success-no-warning-limit", + "test": VerifyBGPPeerRouteLimit, + "eos_data": [ + { + "vrfs": { + "default": { + "peerList": [ + { + "peerAddress": "10.100.0.8", + "maxTotalRoutes": 12000, + } + ] + }, + "MGMT": { + "peerList": [ + { + "peerAddress": "10.100.0.9", + "maxTotalRoutes": 10000, + } + ] + }, + }, + }, + ], + "inputs": { + "bgp_peers": [ + {"peer_address": "10.100.0.8", "vrf": "default", "maximum_routes": 12000, "warning_limit": 0}, + {"peer_address": "10.100.0.9", "vrf": "MGMT", "maximum_routes": 10000}, + ] + }, + "expected": {"result": "success"}, + }, { "name": "failure-peer-not-found", "test": VerifyBGPPeerRouteLimit, @@ -3802,9 +3792,9 @@ def test_check_bgp_neighbor_capability(input_dict: dict[str, bool], expected: bo "result": "failure", "messages": [ "Peer: 10.100.0.8 VRF: default - Maximum routes mismatch - Expected: 12000, Actual: 13000", - "Peer: 10.100.0.8 VRF: default - Maximum route warning limit mismatch - Expected: 10000, Actual: 11000", + "Peer: 10.100.0.8 VRF: default - Maximum routes warning limit mismatch - Expected: 10000, Actual: 11000", "Peer: 10.100.0.9 VRF: MGMT - Maximum routes mismatch - Expected: 10000, Actual: 11000", - "Peer: 10.100.0.9 VRF: MGMT - Maximum route warning limit mismatch - Expected: 9000, Actual: 10000", + "Peer: 10.100.0.9 VRF: MGMT - Maximum routes warning limit mismatch - Expected: 9000, Actual: 10000", ], }, }, @@ -3826,6 +3816,7 @@ def test_check_bgp_neighbor_capability(input_dict: dict[str, bool], expected: bo "peerList": [ { "peerAddress": "10.100.0.9", + "maxTotalRoutes": 10000, } ] }, @@ -3841,9 +3832,219 @@ def test_check_bgp_neighbor_capability(input_dict: dict[str, bool], expected: bo "expected": { "result": "failure", "messages": [ - "Peer: 10.100.0.8 VRF: default - Maximum route warning limit mismatch - Expected: 10000, Actual: Not Found", - "Peer: 10.100.0.9 VRF: MGMT - Maximum routes mismatch - Expected: 10000, Actual: Not Found", - "Peer: 10.100.0.9 VRF: MGMT - Maximum route warning limit mismatch - Expected: 9000, Actual: Not Found", + "Peer: 10.100.0.8 VRF: default - Maximum routes warning limit mismatch - Expected: 10000, Actual: 0", + "Peer: 10.100.0.9 VRF: MGMT - Maximum routes warning limit mismatch - Expected: 9000, Actual: 0", + ], + }, + }, + { + "name": "success-no-check-tcp-queues", + "test": VerifyBGPPeerSession, + "eos_data": [ + { + "vrfs": { + "default": { + "peerList": [ + { + "peerAddress": "10.100.0.8", + "state": "Established", + "peerTcpInfo": { + "outputQueueLength": 10, + "inputQueueLength": 5, + }, + } + ] + }, + "MGMT": { + "peerList": [ + { + "peerAddress": "10.100.0.9", + "state": "Established", + "peerTcpInfo": { + "outputQueueLength": 10, + "inputQueueLength": 5, + }, + } + ] + }, + }, + }, + ], + "inputs": { + "check_tcp_queues": False, + "bgp_peers": [ + {"peer_address": "10.100.0.8", "vrf": "default"}, + {"peer_address": "10.100.0.9", "vrf": "MGMT"}, + ], + }, + "expected": {"result": "success"}, + }, + { + "name": "success-check-tcp-queues", + "test": VerifyBGPPeerSession, + "eos_data": [ + { + "vrfs": { + "default": { + "peerList": [ + { + "peerAddress": "10.100.0.8", + "state": "Established", + "peerTcpInfo": { + "outputQueueLength": 0, + "inputQueueLength": 0, + }, + } + ] + }, + "MGMT": { + "peerList": [ + { + "peerAddress": "10.100.0.9", + "state": "Established", + "peerTcpInfo": { + "outputQueueLength": 0, + "inputQueueLength": 0, + }, + } + ] + }, + }, + }, + ], + "inputs": { + "check_tcp_queues": True, + "bgp_peers": [ + {"peer_address": "10.100.0.8", "vrf": "default"}, + {"peer_address": "10.100.0.9", "vrf": "MGMT"}, + ], + }, + "expected": {"result": "success"}, + }, + { + "name": "failure-peer-not-found", + "test": VerifyBGPPeerSession, + "eos_data": [ + { + "vrfs": { + "default": { + "peerList": [ + { + "peerAddress": "10.100.0.8", + "state": "Established", + "peerTcpInfo": { + "outputQueueLength": 0, + "inputQueueLength": 0, + }, + } + ] + }, + }, + }, + ], + "inputs": { + "bgp_peers": [ + {"peer_address": "10.100.0.8", "vrf": "default"}, + {"peer_address": "10.100.0.9", "vrf": "MGMT"}, + ] + }, + "expected": { + "result": "failure", + "messages": [ + "Peer: 10.100.0.9 VRF: MGMT - Not found", + ], + }, + }, + { + "name": "failure-not-established", + "test": VerifyBGPPeerSession, + "eos_data": [ + { + "vrfs": { + "default": { + "peerList": [ + { + "peerAddress": "10.100.0.8", + "state": "Active", + "peerTcpInfo": { + "outputQueueLength": 0, + "inputQueueLength": 0, + }, + } + ] + }, + "MGMT": { + "peerList": [ + { + "peerAddress": "10.100.0.9", + "state": "Active", + "peerTcpInfo": { + "outputQueueLength": 0, + "inputQueueLength": 0, + }, + } + ] + }, + }, + }, + ], + "inputs": { + "bgp_peers": [ + {"peer_address": "10.100.0.8", "vrf": "default"}, + {"peer_address": "10.100.0.9", "vrf": "MGMT"}, + ] + }, + "expected": { + "result": "failure", + "messages": [ + "Peer: 10.100.0.8 VRF: default - Session state is not established - State: Active", + "Peer: 10.100.0.9 VRF: MGMT - Session state is not established - State: Active", + ], + }, + }, + { + "name": "failure-check-tcp-queues", + "test": VerifyBGPPeerSession, + "eos_data": [ + { + "vrfs": { + "default": { + "peerList": [ + { + "peerAddress": "10.100.0.8", + "state": "Established", + "peerTcpInfo": { + "outputQueueLength": 10, + "inputQueueLength": 5, + }, + } + ] + }, + "MGMT": { + "peerList": [ + { + "peerAddress": "10.100.0.9", + "state": "Established", + "peerTcpInfo": { + "outputQueueLength": 0, + "inputQueueLength": 0, + }, + } + ] + }, + }, + }, + ], + "inputs": { + "bgp_peers": [ + {"peer_address": "10.100.0.8", "vrf": "default"}, + {"peer_address": "10.100.0.9", "vrf": "MGMT"}, + ] + }, + "expected": { + "result": "failure", + "messages": [ + "Peer: 10.100.0.8 VRF: default - Session has non-empty message queues - InQ: 5, OutQ: 10", ], }, },