topotests: improve test reliability

Decrease the protocol timers, wait for peers to connect (and test it)
then finally wait a bit more for SAs to be propagated.

Signed-off-by: Rafael Zalamena <rzalamena@opensourcerouting.org>
This commit is contained in:
Rafael Zalamena 2025-01-10 10:17:35 -03:00
parent 35c0c827f2
commit dda4f00f65
3 changed files with 28 additions and 3 deletions

View File

@ -27,5 +27,6 @@ router pim
msdp originator-id 10.254.254.1 msdp originator-id 10.254.254.1
msdp log sa-events msdp log sa-events
msdp peer 192.168.1.2 source 192.168.1.1 msdp peer 192.168.1.2 source 192.168.1.1
msdp timers 10 20 3
rp 192.168.1.1 rp 192.168.1.1
! !

View File

@ -24,5 +24,6 @@ router bgp 65200
router pim router pim
msdp log sa-events msdp log sa-events
msdp peer 192.168.1.1 source 192.168.1.2 msdp peer 192.168.1.1 source 192.168.1.2
msdp timers 10 20 3
rp 192.168.1.2 rp 192.168.1.2
! !

View File

@ -121,6 +121,29 @@ def test_bgp_convergence():
expect_loopback_route("r2", "ip", "10.254.254.1/32", "bgp") expect_loopback_route("r2", "ip", "10.254.254.1/32", "bgp")
def test_msdp_connect():
"Test that the MSDP peers have connected."
tgen = get_topogen()
if tgen.routers_have_failure():
pytest.skip(tgen.errors)
def msdp_is_connected(router, peer):
logger.info(f"waiting MSDP peer {peer} in router {router}")
expected = {peer: {"state": "established"}}
test_func = partial(
topotest.router_json_cmp,
tgen.gears[router],
"show ip msdp peer json",
expected,
)
_, result = topotest.run_and_expect(test_func, None, count=40, wait=2)
assertmsg = '"{}" convergence failure'.format(router)
assert result is None, assertmsg
msdp_is_connected("r1", "192.168.1.2")
msdp_is_connected("r2", "192.168.1.1")
def test_sa_learn(): def test_sa_learn():
""" """
Test that the learned SA uses the configured originator ID instead Test that the learned SA uses the configured originator ID instead
@ -145,10 +168,10 @@ def test_sa_learn():
"local": "no", "local": "no",
} }
} }
} },
) )
_, result = topotest.run_and_expect(test_func, None, count=100, wait=1) _, result = topotest.run_and_expect(test_func, None, count=80, wait=2)
assert result is None, 'r2 SA convergence failure' assert result is None, "r2 SA convergence failure"
def test_memory_leak(): def test_memory_leak():