From aee05e20fd06ba702c85de47b20dd6e4730d7abd Mon Sep 17 00:00:00 2001 From: Eugene Bogomazov Date: Tue, 21 Jun 2022 13:43:02 +0300 Subject: [PATCH] bgpd: Update topotests for RFC 9234 In the previous version, the time.sleep function was included to wait for the moment when the routes were sent to all routers. Changed this function to topotest.run_and_expect for more deterministic behavior. Signed-off-by: Eugene Bogomazov --- .../test_bgp_roles_capability.py | 75 +++++++++++++------ .../test_bgp_roles_filtering.py | 35 +++++---- 2 files changed, 72 insertions(+), 38 deletions(-) diff --git a/tests/topotests/bgp_roles_capability/test_bgp_roles_capability.py b/tests/topotests/bgp_roles_capability/test_bgp_roles_capability.py index 28219da07b..33c77fece3 100644 --- a/tests/topotests/bgp_roles_capability/test_bgp_roles_capability.py +++ b/tests/topotests/bgp_roles_capability/test_bgp_roles_capability.py @@ -31,7 +31,6 @@ import os import sys import functools import pytest -import time CWD = os.path.dirname(os.path.realpath(__file__)) sys.path.append(os.path.join(CWD, "../")) @@ -56,7 +55,6 @@ def tgen(request): router.load_config(TopoRouter.RD_ZEBRA, "zebra.conf") router.load_config(TopoRouter.RD_BGP, "bgpd.conf") tgen.start_router() - time.sleep(1) yield tgen tgen.stop_topology() @@ -67,6 +65,16 @@ def skip_on_failure(tgen): pytest.skip("skipped because of previous test failure") +def find_neighbor_status(router, neighbor_ip): + return json.loads(router.vtysh_cmd(f"show bgp neighbors {neighbor_ip} json"))[ + neighbor_ip + ] + + +def check_role_mismatch(router, neighbor_ip): + return is_role_mismatch(find_neighbor_status(router, neighbor_ip)) + + def is_role_mismatch(neighbor_status): return ( neighbor_status["bgpState"] != "Established" @@ -75,15 +83,26 @@ def is_role_mismatch(neighbor_status): ) +def check_session_established(router, neighbor_ip): + neighbor_status = find_neighbor_status(router, neighbor_ip) + return neighbor_status["bgpState"] == "Established" + + def test_correct_pair(tgen): # provider-customer pair + router = tgen.gears["r1"] neighbor_ip = "192.168.2.2" - neighbor_status = json.loads( - tgen.gears["r1"].vtysh_cmd(f"show bgp neighbors {neighbor_ip} json") - )[neighbor_ip] + check_r2_established = functools.partial( + check_session_established, router, neighbor_ip + ) + success, result = topotest.run_and_expect( + check_r2_established, True, count=20, wait=3 + ) + assert success, "Session with r2 is not Established" + + neighbor_status = find_neighbor_status(router, neighbor_ip) assert neighbor_status["localRole"] == "provider" assert neighbor_status["remoteRole"] == "customer" - assert neighbor_status["bgpState"] == "Established" assert ( neighbor_status["neighborCapabilities"].get("role") == "advertisedAndReceived" ) @@ -91,44 +110,56 @@ def test_correct_pair(tgen): def test_role_pair_mismatch(tgen): # provider-peer mistmatch + router = tgen.gears["r1"] neighbor_ip = "192.168.3.2" - neighbor_status = json.loads( - tgen.gears["r1"].vtysh_cmd(f"show bgp neighbors {neighbor_ip} json") - )[neighbor_ip] - assert is_role_mismatch(neighbor_status) + check_r3_mismatch = functools.partial(check_role_mismatch, router, neighbor_ip) + success, result = topotest.run_and_expect(check_r3_mismatch, True, count=20, wait=3) + assert success, "Session with r3 was not correctly closed" def test_single_role_advertising(tgen): # provider-undefined pair; we set role + router = tgen.gears["r1"] neighbor_ip = "192.168.4.2" - neighbor_status = json.loads( - tgen.gears["r1"].vtysh_cmd(f"show bgp neighbors {neighbor_ip} json") - )[neighbor_ip] + check_r4_established = functools.partial( + check_session_established, router, neighbor_ip + ) + success, result = topotest.run_and_expect( + check_r4_established, True, count=20, wait=3 + ) + assert success, "Session with r4 is not Established" + + neighbor_status = find_neighbor_status(router, neighbor_ip) assert neighbor_status["localRole"] == "provider" assert neighbor_status["remoteRole"] == "undefined" - assert neighbor_status["bgpState"] == "Established" assert neighbor_status["neighborCapabilities"].get("role") == "advertised" def test_single_role_receiving(tgen): # provider-undefined pair; we receive role + router = tgen.gears["r4"] neighbor_ip = "192.168.4.1" - neighbor_status = json.loads( - tgen.gears["r4"].vtysh_cmd(f"show bgp neighbors {neighbor_ip} json") - )[neighbor_ip] + check_r1_established = functools.partial( + check_session_established, router, neighbor_ip + ) + success, result = topotest.run_and_expect( + check_r1_established, True, count=20, wait=3 + ) + assert success, "Session with r1 is not Established" + + neighbor_status = find_neighbor_status(router, neighbor_ip) assert neighbor_status["localRole"] == "undefined" assert neighbor_status["remoteRole"] == "provider" - assert neighbor_status["bgpState"] == "Established" assert neighbor_status["neighborCapabilities"].get("role") == "received" def test_role_strict_mode(tgen): # provider-undefined pair with strict-mode + router = tgen.gears["r1"] neighbor_ip = "192.168.5.2" - neighbor_status = json.loads( - tgen.gears["r1"].vtysh_cmd(f"show bgp neighbors {neighbor_ip} json") - ) - assert is_role_mismatch(neighbor_status[neighbor_ip]) + check_r5_mismatch = functools.partial(check_role_mismatch, router, neighbor_ip) + success, result = topotest.run_and_expect(check_r5_mismatch, True, count=20, wait=3) + assert success, "Session with r5 was not correctly closed" if __name__ == "__main__": diff --git a/tests/topotests/bgp_roles_filtering/test_bgp_roles_filtering.py b/tests/topotests/bgp_roles_filtering/test_bgp_roles_filtering.py index c739509b50..c5827d7a9e 100644 --- a/tests/topotests/bgp_roles_filtering/test_bgp_roles_filtering.py +++ b/tests/topotests/bgp_roles_filtering/test_bgp_roles_filtering.py @@ -31,7 +31,6 @@ import os import sys import functools import pytest -import time CWD = os.path.dirname(os.path.realpath(__file__)) sys.path.append(os.path.join(CWD, "../")) @@ -57,10 +56,6 @@ def tgen(request): router.load_config(TopoRouter.RD_ZEBRA, "zebra.conf") router.load_config(TopoRouter.RD_BGP, "bgpd.conf") tgen.start_router() - BGP_CONVERGENCE = verify_bgp_convergence_from_running_config(tgen) - assert BGP_CONVERGENCE, f"setup_module :Failed \n Error: {BGP_CONVERGENCE}" - # Todo: What is the indented way to wait for convergence without json?! - time.sleep(5) yield tgen tgen.stop_topology() @@ -73,17 +68,25 @@ def skip_on_failure(tgen): def test_r10_routes(tgen): # provider-undefine pair bur strict-mode was set - routes = json.loads(tgen.gears["r10"].vtysh_cmd("show bgp ipv4 json"))["routes"] - route_list = sorted(routes.keys()) - assert route_list == [ - "192.0.2.1/32", - "192.0.2.2/32", - "192.0.2.3/32", - "192.0.2.4/32", - "192.0.2.5/32", - "192.0.2.6/32", - "192.0.2.7/32", - ] + def _routes_half_converged(): + routes = json.loads(tgen.gears["r10"].vtysh_cmd("show bgp ipv4 json"))["routes"] + output = sorted(routes.keys()) + expected = [ + "192.0.2.1/32", + "192.0.2.2/32", + "192.0.2.3/32", + "192.0.2.4/32", + "192.0.2.5/32", + "192.0.2.6/32", + "192.0.2.7/32", + ] + return output == expected + + success, result = topotest.run_and_expect( + _routes_half_converged, True, count=20, wait=3 + ) + assert success, "Routes did not converged" + routes_with_otc = list() for number in range(1, 8): prefix = f"192.0.2.{number}/32"