From de6bbd1223995f4e011e98a6b91a64fc96f1d614 Mon Sep 17 00:00:00 2001
From: Quentin Young <qlyoung@qlyoung.net>
Date: Wed, 9 Aug 2023 15:30:39 -0400
Subject: [PATCH] tests: amalgamate check_ping
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

❌✂️📋

Signed-off-by: Quentin Young <qlyoung@qlyoung.net>
---
 .../test_bgp_srv6l3vpn_over_ipv6.py           |  29 +---
 .../test_bgp_srv6l3vpn_sid.py                 |  81 +++++-----
 .../test_bgp_srv6l3vpn_to_bgp_vrf.py          |  61 +++-----
 .../test_bgp_srv6l3vpn_to_bgp_vrf2.py         |  36 ++---
 .../test_bgp_srv6l3vpn_to_bgp_vrf3.py         | 138 +++++++-----------
 .../bgp_vpnv4_asbr/test_bgp_vpnv4_asbr.py     |  33 +----
 tests/topotests/lib/checkping.py              |  33 +++++
 7 files changed, 167 insertions(+), 244 deletions(-)
 create mode 100644 tests/topotests/lib/checkping.py

diff --git a/tests/topotests/bgp_srv6l3vpn_over_ipv6/test_bgp_srv6l3vpn_over_ipv6.py b/tests/topotests/bgp_srv6l3vpn_over_ipv6/test_bgp_srv6l3vpn_over_ipv6.py
index 9e529059e743..14b9ba84988c 100755
--- a/tests/topotests/bgp_srv6l3vpn_over_ipv6/test_bgp_srv6l3vpn_over_ipv6.py
+++ b/tests/topotests/bgp_srv6l3vpn_over_ipv6/test_bgp_srv6l3vpn_over_ipv6.py
@@ -24,6 +24,7 @@
 from lib.topogen import Topogen, TopoRouter, get_topogen
 from lib.topolog import logger
 from lib.common_config import required_linux_kernel_version
+from lib.checkping import check_ping
 
 pytestmark = [pytest.mark.bgpd]
 
@@ -101,31 +102,15 @@ def teardown_module(mod):
     tgen.stop_topology()
 
 
-def check_ping4(name, dest_addr, expected):
-    def _check(name, dest_addr, match):
-        tgen = get_topogen()
-        output = tgen.gears[name].run("ping {} -c 1 -w 1".format(dest_addr))
-        logger.info(output)
-        if match not in output:
-            return "ping fail"
-
-    match = ", {} packet loss".format("0%" if expected else "100%")
-    logger.info("[+] check {} {} {}".format(name, dest_addr, match))
-    tgen = get_topogen()
-    func = functools.partial(_check, name, dest_addr, match)
-    success, result = topotest.run_and_expect(func, None, count=10, wait=1)
-    assert result is None, "Failed"
-
-
 def test_ping():
     tgen = get_topogen()
 
-    check_ping4("c11", "192.168.2.1", True)
-    check_ping4("c11", "192.168.3.1", True)
-    check_ping4("c12", "192.168.2.1", True)
-    check_ping4("c12", "192.168.3.1", True)
-    check_ping4("c21", "192.168.3.1", True)
-    check_ping4("c22", "192.168.3.1", True)
+    check_ping("c11", "192.168.2.1", True, 10, 1)
+    check_ping("c11", "192.168.3.1", True, 10, 1)
+    check_ping("c12", "192.168.2.1", True, 10, 1)
+    check_ping("c12", "192.168.3.1", True, 10, 1)
+    check_ping("c21", "192.168.3.1", True, 10, 1)
+    check_ping("c22", "192.168.3.1", True, 10, 1)
 
 
 if __name__ == "__main__":
diff --git a/tests/topotests/bgp_srv6l3vpn_sid/test_bgp_srv6l3vpn_sid.py b/tests/topotests/bgp_srv6l3vpn_sid/test_bgp_srv6l3vpn_sid.py
index cddcf6a9a1f6..189d2baf17aa 100755
--- a/tests/topotests/bgp_srv6l3vpn_sid/test_bgp_srv6l3vpn_sid.py
+++ b/tests/topotests/bgp_srv6l3vpn_sid/test_bgp_srv6l3vpn_sid.py
@@ -35,6 +35,7 @@
 from lib.topogen import Topogen, TopoRouter, get_topogen
 from lib.topolog import logger
 from lib.common_config import required_linux_kernel_version
+from lib.checkping import check_ping
 
 
 def build_topo(tgen):
@@ -167,22 +168,6 @@ def open_json_file(filename):
         assert False, "Could not read file {}".format(filename)
 
 
-def check_ping(name, dest_addr, expect_connected):
-    def _check(name, dest_addr, match):
-        tgen = get_topogen()
-        output = tgen.gears[name].run("ping {} -c 1 -w 1".format(dest_addr))
-        logger.info(output)
-        if match not in output:
-            return True
-
-    match = ", {} packet loss".format("0%" if expect_connected else "100%")
-    logger.info("[+] check {} {} {}".format(name, dest_addr, match))
-    tgen = get_topogen()
-    func = functools.partial(_check, name, dest_addr, match)
-    success, result = topotest.run_and_expect(func, None, count=10, wait=0.5)
-    assert result is None, "Failed"
-
-
 def check_rib(name, cmd, expected_file):
     def _check(name, cmd, expected_file):
         logger.info("polling")
@@ -215,21 +200,21 @@ def test_rib():
 
 
 def test_ping():
-    check_ping("ce1", "2001:2::2", True)
-    check_ping("ce1", "2001:3::2", True)
-    check_ping("ce1", "2001:4::2", False)
-    check_ping("ce1", "2001:5::2", False)
-    check_ping("ce1", "2001:6::2", False)
-    check_ping("ce4", "2001:1::2", False)
-    check_ping("ce4", "2001:2::2", False)
-    check_ping("ce4", "2001:3::2", False)
-    check_ping("ce4", "2001:5::2", True)
-    check_ping("ce4", "2001:6::2", True)
+    check_ping("ce1", "2001:2::2", True, 10, 0.5)
+    check_ping("ce1", "2001:3::2", True, 10, 0.5)
+    check_ping("ce1", "2001:4::2", False, 10, 0.5)
+    check_ping("ce1", "2001:5::2", False, 10, 0.5)
+    check_ping("ce1", "2001:6::2", False, 10, 0.5)
+    check_ping("ce4", "2001:1::2", False, 10, 0.5)
+    check_ping("ce4", "2001:2::2", False, 10, 0.5)
+    check_ping("ce4", "2001:3::2", False, 10, 0.5)
+    check_ping("ce4", "2001:5::2", True, 10, 0.5)
+    check_ping("ce4", "2001:6::2", True, 10, 0.5)
 
 
 def test_sid_per_afv6_auto():
     check_rib("r1", "show ipv6 route vrf vrf10 json", "r1/vrf10_afv6_auto_sid_rib.json")
-    check_ping("ce1", "2001:2::2", True)
+    check_ping("ce1", "2001:2::2", True, 10, 0.5)
     get_topogen().gears["r2"].vtysh_cmd(
         """
         configure terminal
@@ -241,7 +226,7 @@ def test_sid_per_afv6_auto():
     check_rib(
         "r1", "show ipv6 route vrf vrf10 json", "r1/vrf10_afv6_auto_no_sid_rib.json"
     )
-    check_ping("ce1", "2001:2::2", False)
+    check_ping("ce1", "2001:2::2", False, 10, 0.5)
 
     get_topogen().gears["r2"].vtysh_cmd(
         """
@@ -252,7 +237,7 @@ def test_sid_per_afv6_auto():
         """
     )
     check_rib("r1", "show ipv6 route vrf vrf10 json", "r1/vrf10_afv6_auto_sid_rib.json")
-    check_ping("ce1", "2001:2::2", True)
+    check_ping("ce1", "2001:2::2", True, 10, 0.5)
 
     get_topogen().gears["r2"].vtysh_cmd(
         """
@@ -265,14 +250,14 @@ def test_sid_per_afv6_auto():
     check_rib(
         "r1", "show ipv6 route vrf vrf10 json", "r1/vrf10_afv6_auto_no_sid_rib.json"
     )
-    check_ping("ce1", "2001:2::2", False)
+    check_ping("ce1", "2001:2::2", False, 10, 0.5)
 
 
 def test_sid_per_afv6_manual():
     check_rib(
         "r1", "show ipv6 route vrf vrf10 json", "r1/vrf10_afv6_manual_no_sid_rib.json"
     )
-    check_ping("ce1", "2001:2::2", False)
+    check_ping("ce1", "2001:2::2", False, 10, 0.5)
 
     get_topogen().gears["r2"].vtysh_cmd(
         """
@@ -286,7 +271,7 @@ def test_sid_per_afv6_manual():
     check_rib(
         "r1", "show ipv6 route vrf vrf10 json", "r1/vrf10_afv6_manual_sid_rib.json"
     )
-    check_ping("ce1", "2001:2::2", True)
+    check_ping("ce1", "2001:2::2", True, 10, 0.5)
 
     get_topogen().gears["r2"].vtysh_cmd(
         """
@@ -299,12 +284,12 @@ def test_sid_per_afv6_manual():
     check_rib(
         "r1", "show ipv6 route vrf vrf10 json", "r1/vrf10_afv6_manual_no_sid_rib.json"
     )
-    check_ping("ce1", "2001:2::2", False)
+    check_ping("ce1", "2001:2::2", False, 10, 0.5)
 
 
 def test_sid_per_afv4_auto():
     check_rib("r1", "show ip route vrf vrf10 json", "r1/vrf10_afv4_auto_sid_rib.json")
-    check_ping("ce1", "192.168.2.2", True)
+    check_ping("ce1", "192.168.2.2", True, 10, 0.5)
     get_topogen().gears["r2"].vtysh_cmd(
         """
         configure terminal
@@ -317,7 +302,7 @@ def test_sid_per_afv4_auto():
     check_rib(
         "r1", "show ip route vrf vrf10 json", "r1/vrf10_afv4_auto_no_sid_rib.json"
     )
-    check_ping("ce1", "192.168.2.2", False)
+    check_ping("ce1", "192.168.2.2", False, 10, 0.5)
 
     get_topogen().gears["r2"].vtysh_cmd(
         """
@@ -329,7 +314,7 @@ def test_sid_per_afv4_auto():
     )
 
     check_rib("r1", "show ip route vrf vrf10 json", "r1/vrf10_afv4_auto_sid_rib.json")
-    check_ping("ce1", "192.168.2.2", True)
+    check_ping("ce1", "192.168.2.2", True, 10, 0.5)
 
     get_topogen().gears["r2"].vtysh_cmd(
         """
@@ -342,14 +327,14 @@ def test_sid_per_afv4_auto():
     check_rib(
         "r1", "show ip route vrf vrf10 json", "r1/vrf10_afv4_auto_no_sid_rib.json"
     )
-    check_ping("ce1", "192.168.2.2", False)
+    check_ping("ce1", "192.168.2.2", False, 10, 0.5)
 
 
 def test_sid_per_afv4_manual():
     check_rib(
         "r1", "show ip route vrf vrf10 json", "r1/vrf10_afv4_manual_no_sid_rib.json"
     )
-    check_ping("ce1", "192.168.2.2", False)
+    check_ping("ce1", "192.168.2.2", False, 10, 0.5)
     get_topogen().gears["r2"].vtysh_cmd(
         """
         configure terminal
@@ -360,7 +345,7 @@ def test_sid_per_afv4_manual():
     )
 
     check_rib("r1", "show ip route vrf vrf10 json", "r1/vrf10_afv4_manual_sid_rib.json")
-    check_ping("ce1", "192.168.2.2", True)
+    check_ping("ce1", "192.168.2.2", True, 10, 0.5)
 
     get_topogen().gears["r2"].vtysh_cmd(
         """
@@ -370,7 +355,7 @@ def test_sid_per_afv4_manual():
            no sid vpn export 8
         """
     )
-    check_ping("ce1", "192.168.2.2", False)
+    check_ping("ce1", "192.168.2.2", False, 10, 0.5)
     check_rib(
         "r1", "show ip route vrf vrf10 json", "r1/vrf10_afv4_manual_no_sid_rib.json"
     )
@@ -380,7 +365,7 @@ def test_sid_per_vrf_auto():
     check_rib(
         "r1", "show ipv6 route vrf vrf10 json", "r1/vrf10_pervrf_auto_no_sid_rib.json"
     )
-    check_ping("ce1", "2001:2::2", False)
+    check_ping("ce1", "2001:2::2", False, 10, 0.5)
     get_topogen().gears["r2"].vtysh_cmd(
         """
         configure terminal
@@ -392,11 +377,11 @@ def test_sid_per_vrf_auto():
     check_rib(
         "r1", "show ipv6 route vrf vrf10 json", "r1/vrf10_pervrf6_auto_sid_rib.json"
     )
-    check_ping("ce1", "2001:2::2", True)
+    check_ping("ce1", "2001:2::2", True, 10, 0.5)
     check_rib(
         "r1", "show ip route vrf vrf10 json", "r1/vrf10_pervrf4_auto_sid_rib.json"
     )
-    check_ping("ce1", "192.168.2.2", True)
+    check_ping("ce1", "192.168.2.2", True, 10, 0.5)
 
     get_topogen().gears["r2"].vtysh_cmd(
         """
@@ -409,14 +394,14 @@ def test_sid_per_vrf_auto():
     check_rib(
         "r1", "show ipv6 route vrf vrf10 json", "r1/vrf10_pervrf_auto_no_sid_rib.json"
     )
-    check_ping("ce1", "2001:2::2", False)
+    check_ping("ce1", "2001:2::2", False, 10, 0.5)
 
 
 def test_sid_per_vrf_manual():
     check_rib(
         "r1", "show ip route vrf vrf10 json", "r1/vrf10_pervrf_manual_no_sid_rib.json"
     )
-    check_ping("ce1", "192.168.2.2", False)
+    check_ping("ce1", "192.168.2.2", False, 10, 0.5)
     get_topogen().gears["r2"].vtysh_cmd(
         """
         configure terminal
@@ -428,11 +413,11 @@ def test_sid_per_vrf_manual():
     check_rib(
         "r1", "show ipv6 route vrf vrf10 json", "r1/vrf10_pervrf6_manual_sid_rib.json"
     )
-    check_ping("ce1", "2001:2::2", True)
+    check_ping("ce1", "2001:2::2", True, 10, 0.5)
     check_rib(
         "r1", "show ip route vrf vrf10 json", "r1/vrf10_pervrf4_manual_sid_rib.json"
     )
-    check_ping("ce1", "192.168.2.2", True)
+    check_ping("ce1", "192.168.2.2", True, 10, 0.5)
 
     get_topogen().gears["r2"].vtysh_cmd(
         """
@@ -445,7 +430,7 @@ def test_sid_per_vrf_manual():
     check_rib(
         "r1", "show ip route vrf vrf10 json", "r1/vrf10_pervrf_manual_no_sid_rib.json"
     )
-    check_ping("ce1", "192.168.2.2", False)
+    check_ping("ce1", "192.168.2.2", False, 10, 0.5)
 
 
 if __name__ == "__main__":
diff --git a/tests/topotests/bgp_srv6l3vpn_to_bgp_vrf/test_bgp_srv6l3vpn_to_bgp_vrf.py b/tests/topotests/bgp_srv6l3vpn_to_bgp_vrf/test_bgp_srv6l3vpn_to_bgp_vrf.py
index fdc1f4eec315..5a9193f18e0d 100755
--- a/tests/topotests/bgp_srv6l3vpn_to_bgp_vrf/test_bgp_srv6l3vpn_to_bgp_vrf.py
+++ b/tests/topotests/bgp_srv6l3vpn_to_bgp_vrf/test_bgp_srv6l3vpn_to_bgp_vrf.py
@@ -23,6 +23,7 @@
 from lib.topogen import Topogen, TopoRouter, get_topogen
 from lib.topolog import logger
 from lib.common_config import required_linux_kernel_version
+from lib.checkping import check_ping
 
 pytestmark = [pytest.mark.bgpd]
 
@@ -134,22 +135,6 @@ def open_json_file(filename):
         assert False, "Could not read file {}".format(filename)
 
 
-def check_ping(name, dest_addr, expect_connected):
-    def _check(name, dest_addr, match):
-        tgen = get_topogen()
-        output = tgen.gears[name].run("ping6 {} -c 1 -w 1".format(dest_addr))
-        logger.info(output)
-        if match not in output:
-            return "ping fail"
-
-    match = ", {} packet loss".format("0%" if expect_connected else "100%")
-    logger.info("[+] check {} {} {}".format(name, dest_addr, match))
-    tgen = get_topogen()
-    func = functools.partial(_check, name, dest_addr, match)
-    success, result = topotest.run_and_expect(func, None, count=10, wait=1)
-    assert result is None, "Failed"
-
-
 def check_rib(name, cmd, expected_file):
     def _check(name, cmd, expected_file):
         logger.info("polling")
@@ -182,20 +167,20 @@ def test_rib():
 
 
 def test_ping():
-    check_ping("ce1", "2001:2::2", True)
-    check_ping("ce1", "2001:3::2", True)
-    check_ping("ce1", "2001:4::2", False)
-    check_ping("ce1", "2001:5::2", False)
-    check_ping("ce1", "2001:6::2", False)
-    check_ping("ce4", "2001:1::2", False)
-    check_ping("ce4", "2001:2::2", False)
-    check_ping("ce4", "2001:3::2", False)
-    check_ping("ce4", "2001:5::2", True)
-    check_ping("ce4", "2001:6::2", True)
+    check_ping("ce1", "2001:2::2", True, 10, 1)
+    check_ping("ce1", "2001:3::2", True, 10, 1)
+    check_ping("ce1", "2001:4::2", False, 10, 1)
+    check_ping("ce1", "2001:5::2", False, 10, 1)
+    check_ping("ce1", "2001:6::2", False, 10, 1)
+    check_ping("ce4", "2001:1::2", False, 10, 1)
+    check_ping("ce4", "2001:2::2", False, 10, 1)
+    check_ping("ce4", "2001:3::2", False, 10, 1)
+    check_ping("ce4", "2001:5::2", True, 10, 1)
+    check_ping("ce4", "2001:6::2", True, 10, 1)
 
 
 def test_locator_delete():
-    check_ping("ce1", "2001:2::2", True)
+    check_ping("ce1", "2001:2::2", True, 10, 1)
     get_topogen().gears["r1"].vtysh_cmd(
         """
         configure terminal
@@ -207,11 +192,11 @@ def test_locator_delete():
     )
     check_rib("r1", "show bgp ipv6 vpn json", "r1/vpnv6_rib_locator_deleted.json")
     check_rib("r2", "show bgp ipv6 vpn json", "r2/vpnv6_rib_locator_deleted.json")
-    check_ping("ce1", "2001:2::2", False)
+    check_ping("ce1", "2001:2::2", False, 10, 1)
 
 
 def test_locator_recreate():
-    check_ping("ce1", "2001:2::2", False)
+    check_ping("ce1", "2001:2::2", False, 10, 1)
     get_topogen().gears["r1"].vtysh_cmd(
         """
         configure terminal
@@ -224,11 +209,11 @@ def test_locator_recreate():
     )
     check_rib("r1", "show bgp ipv6 vpn json", "r1/vpnv6_rib_locator_recreated.json")
     check_rib("r2", "show bgp ipv6 vpn json", "r2/vpnv6_rib_locator_recreated.json")
-    check_ping("ce1", "2001:2::2", True)
+    check_ping("ce1", "2001:2::2", True, 10, 1)
 
 
 def test_bgp_locator_unset():
-    check_ping("ce1", "2001:2::2", True)
+    check_ping("ce1", "2001:2::2", True, 10, 1)
     get_topogen().gears["r1"].vtysh_cmd(
         """
         configure terminal
@@ -239,11 +224,11 @@ def test_bgp_locator_unset():
     )
     check_rib("r1", "show bgp ipv6 vpn json", "r1/vpnv6_rib_locator_deleted.json")
     check_rib("r2", "show bgp ipv6 vpn json", "r2/vpnv6_rib_locator_deleted.json")
-    check_ping("ce1", "2001:2::2", False)
+    check_ping("ce1", "2001:2::2", False, 10, 1)
 
 
 def test_bgp_locator_reset():
-    check_ping("ce1", "2001:2::2", False)
+    check_ping("ce1", "2001:2::2", False, 10, 1)
     get_topogen().gears["r1"].vtysh_cmd(
         """
         configure terminal
@@ -254,11 +239,11 @@ def test_bgp_locator_reset():
     )
     check_rib("r1", "show bgp ipv6 vpn json", "r1/vpnv6_rib_locator_recreated.json")
     check_rib("r2", "show bgp ipv6 vpn json", "r2/vpnv6_rib_locator_recreated.json")
-    check_ping("ce1", "2001:2::2", True)
+    check_ping("ce1", "2001:2::2", True, 10, 1)
 
 
 def test_bgp_srv6_unset():
-    check_ping("ce1", "2001:2::2", True)
+    check_ping("ce1", "2001:2::2", True, 10 ,1)
     get_topogen().gears["r1"].vtysh_cmd(
         """
         configure terminal
@@ -268,11 +253,11 @@ def test_bgp_srv6_unset():
     )
     check_rib("r1", "show bgp ipv6 vpn json", "r1/vpnv6_rib_locator_deleted.json")
     check_rib("r2", "show bgp ipv6 vpn json", "r2/vpnv6_rib_locator_deleted.json")
-    check_ping("ce1", "2001:2::2", False)
+    check_ping("ce1", "2001:2::2", False, 10, 1)
 
 
 def test_bgp_srv6_reset():
-    check_ping("ce1", "2001:2::2", False)
+    check_ping("ce1", "2001:2::2", False, 10, 1)
     get_topogen().gears["r1"].vtysh_cmd(
         """
         configure terminal
@@ -283,7 +268,7 @@ def test_bgp_srv6_reset():
     )
     check_rib("r1", "show bgp ipv6 vpn json", "r1/vpnv6_rib_locator_recreated.json")
     check_rib("r2", "show bgp ipv6 vpn json", "r2/vpnv6_rib_locator_recreated.json")
-    check_ping("ce1", "2001:2::2", True)
+    check_ping("ce1", "2001:2::2", True, 10, 1)
 
 
 if __name__ == "__main__":
diff --git a/tests/topotests/bgp_srv6l3vpn_to_bgp_vrf2/test_bgp_srv6l3vpn_to_bgp_vrf2.py b/tests/topotests/bgp_srv6l3vpn_to_bgp_vrf2/test_bgp_srv6l3vpn_to_bgp_vrf2.py
index a1202f5d93df..914c29f0c1dd 100755
--- a/tests/topotests/bgp_srv6l3vpn_to_bgp_vrf2/test_bgp_srv6l3vpn_to_bgp_vrf2.py
+++ b/tests/topotests/bgp_srv6l3vpn_to_bgp_vrf2/test_bgp_srv6l3vpn_to_bgp_vrf2.py
@@ -24,6 +24,7 @@
 from lib.topogen import Topogen, TopoRouter, get_topogen
 from lib.topolog import logger
 from lib.common_config import required_linux_kernel_version
+from lib.checkping import check_ping
 
 pytestmark = [pytest.mark.bgpd]
 
@@ -96,21 +97,6 @@ def open_json_file(filename):
         assert False, "Could not read file {}".format(filename)
 
 
-def check_ping(name, dest_addr, expect_connected):
-    def _check(name, dest_addr, match):
-        tgen = get_topogen()
-        output = tgen.gears[name].run("ping {} -c 1 -w 1".format(dest_addr))
-        logger.info(output)
-        assert match in output, "ping fail"
-
-    match = ", {} packet loss".format("0%" if expect_connected else "100%")
-    logger.info("[+] check {} {} {}".format(name, dest_addr, match))
-    tgen = get_topogen()
-    func = functools.partial(_check, name, dest_addr, match)
-    success, result = topotest.run_and_expect(func, None, count=10, wait=0.5)
-    assert result is None, "Failed"
-
-
 def check_rib(name, cmd, expected_file):
     def _check(name, dest_addr, match):
         logger.info("polling")
@@ -143,16 +129,16 @@ def test_rib():
 
 
 def test_ping():
-    check_ping("ce1", "192.168.2.2", True)
-    check_ping("ce1", "192.168.3.2", True)
-    check_ping("ce1", "192.168.4.2", False)
-    check_ping("ce1", "192.168.5.2", False)
-    check_ping("ce1", "192.168.6.2", False)
-    check_ping("ce4", "192.168.1.2", False)
-    check_ping("ce4", "192.168.2.2", False)
-    check_ping("ce4", "192.168.3.2", False)
-    check_ping("ce4", "192.168.5.2", True)
-    check_ping("ce4", "192.168.6.2", True)
+    check_ping("ce1", "192.168.2.2", True, 10, 0.5)
+    check_ping("ce1", "192.168.3.2", True, 10, 0.5)
+    check_ping("ce1", "192.168.4.2", False, 10, 0.5)
+    check_ping("ce1", "192.168.5.2", False, 10, 0.5)
+    check_ping("ce1", "192.168.6.2", False, 10, 0.5)
+    check_ping("ce4", "192.168.1.2", False, 10, 0.5)
+    check_ping("ce4", "192.168.2.2", False, 10, 0.5)
+    check_ping("ce4", "192.168.3.2", False, 10, 0.5)
+    check_ping("ce4", "192.168.5.2", True, 10, 0.5)
+    check_ping("ce4", "192.168.6.2", True, 10, 0.5)
 
 
 if __name__ == "__main__":
diff --git a/tests/topotests/bgp_srv6l3vpn_to_bgp_vrf3/test_bgp_srv6l3vpn_to_bgp_vrf3.py b/tests/topotests/bgp_srv6l3vpn_to_bgp_vrf3/test_bgp_srv6l3vpn_to_bgp_vrf3.py
index 7c2c7cfdaa43..03601ddd44c2 100644
--- a/tests/topotests/bgp_srv6l3vpn_to_bgp_vrf3/test_bgp_srv6l3vpn_to_bgp_vrf3.py
+++ b/tests/topotests/bgp_srv6l3vpn_to_bgp_vrf3/test_bgp_srv6l3vpn_to_bgp_vrf3.py
@@ -21,6 +21,7 @@
 from lib.topogen import Topogen, TopoRouter, get_topogen
 from lib.topolog import logger
 from lib.common_config import required_linux_kernel_version
+from lib.checkping import check_ping, check_ping
 
 pytestmark = [pytest.mark.bgpd]
 
@@ -93,37 +94,6 @@ def open_json_file(filename):
         assert False, "Could not read file {}".format(filename)
 
 
-def check_ping4(name, dest_addr, expect_connected):
-    def _check(name, dest_addr, match):
-        tgen = get_topogen()
-        output = tgen.gears[name].run("ping {} -c 1 -w 1".format(dest_addr))
-        logger.info(output)
-        if match not in output:
-            return "ping fail"
-
-    match = ", {} packet loss".format("0%" if expect_connected else "100%")
-    logger.info("[+] check {} {} {}".format(name, dest_addr, match))
-    tgen = get_topogen()
-    func = functools.partial(_check, name, dest_addr, match)
-    success, result = topotest.run_and_expect(func, None, count=10, wait=0.5)
-    assert result is None, "Failed"
-
-
-def check_ping6(name, dest_addr, expect_connected):
-    def _check(name, dest_addr, match):
-        tgen = get_topogen()
-        output = tgen.gears[name].run("ping6 {} -c 1 -w 1".format(dest_addr))
-        logger.info(output)
-        if match not in output:
-            return "ping fail"
-
-    match = "{} packet loss".format("0%" if expect_connected else "100%")
-    logger.info("[+] check {} {} {}".format(name, dest_addr, match))
-    tgen = get_topogen()
-    func = functools.partial(_check, name, dest_addr, match)
-    success, result = topotest.run_and_expect(func, None, count=10, wait=1)
-    assert result is None, "Failed"
-
 
 def check_rib(name, cmd, expected_file):
     def _check(name, dest_addr, match):
@@ -170,32 +140,32 @@ def test_rib():
 
 
 def test_ping():
-    check_ping4("ce1", "192.168.2.2", True)
-    check_ping4("ce1", "192.168.3.2", True)
-    check_ping4("ce1", "192.168.4.2", False)
-    check_ping4("ce1", "192.168.5.2", False)
-    check_ping4("ce1", "192.168.6.2", False)
-    check_ping4("ce4", "192.168.1.2", False)
-    check_ping4("ce4", "192.168.2.2", False)
-    check_ping4("ce4", "192.168.3.2", False)
-    check_ping4("ce4", "192.168.5.2", True)
-    check_ping4("ce4", "192.168.6.2", True)
-
-    check_ping6("ce1", "2001:2::2", True)
-    check_ping6("ce1", "2001:3::2", True)
-    check_ping6("ce1", "2001:4::2", False)
-    check_ping6("ce1", "2001:5::2", False)
-    check_ping6("ce1", "2001:6::2", False)
-    check_ping6("ce4", "2001:1::2", False)
-    check_ping6("ce4", "2001:2::2", False)
-    check_ping6("ce4", "2001:3::2", False)
-    check_ping6("ce4", "2001:5::2", True)
-    check_ping6("ce4", "2001:6::2", True)
+    check_ping("ce1", "192.168.2.2", True, 10, 0.5)
+    check_ping("ce1", "192.168.3.2", True, 10, 0.5)
+    check_ping("ce1", "192.168.4.2", False, 10, 0.5)
+    check_ping("ce1", "192.168.5.2", False, 10, 0.5)
+    check_ping("ce1", "192.168.6.2", False, 10, 0.5)
+    check_ping("ce4", "192.168.1.2", False, 10, 0.5)
+    check_ping("ce4", "192.168.2.2", False, 10, 0.5)
+    check_ping("ce4", "192.168.3.2", False, 10, 0.5)
+    check_ping("ce4", "192.168.5.2", True, 10, 0.5)
+    check_ping("ce4", "192.168.6.2", True, 10, 0.5)
+
+    check_ping("ce1", "2001:2::2", True, 10, 1)
+    check_ping("ce1", "2001:3::2", True, 10, 1)
+    check_ping("ce1", "2001:4::2", False, 10, 1)
+    check_ping("ce1", "2001:5::2", False, 10, 1)
+    check_ping("ce1", "2001:6::2", False, 10, 1)
+    check_ping("ce4", "2001:1::2", False, 10, 1)
+    check_ping("ce4", "2001:2::2", False, 10, 1)
+    check_ping("ce4", "2001:3::2", False, 10, 1)
+    check_ping("ce4", "2001:5::2", True, 10, 1)
+    check_ping("ce4", "2001:6::2", True, 10, 1)
 
 
 def test_bgp_sid_vpn_export_disable():
-    check_ping4("ce1", "192.168.2.2", True)
-    check_ping6("ce1", "2001:2::2", True)
+    check_ping("ce1", "192.168.2.2", True, 10, 0.5)
+    check_ping("ce1", "2001:2::2", True, 10, 1)
     get_topogen().gears["r1"].vtysh_cmd(
         """
         configure terminal
@@ -216,13 +186,13 @@ def test_bgp_sid_vpn_export_disable():
     check_rib(
         "r2", "show bgp ipv6 vpn json", "r2/vpnv6_rib_sid_vpn_export_disabled.json"
     )
-    check_ping4("ce1", "192.168.2.2", False)
-    check_ping6("ce1", "2001:2::2", False)
+    check_ping("ce1", "192.168.2.2", False, 10, 0.5)
+    check_ping("ce1", "2001:2::2", False, 10, 1)
 
 
 def test_bgp_sid_vpn_export_reenable():
-    check_ping4("ce1", "192.168.2.2", False)
-    check_ping6("ce1", "2001:2::2", False)
+    check_ping("ce1", "192.168.2.2", False, 10, 0.5)
+    check_ping("ce1", "2001:2::2", False, 10, 1)
     get_topogen().gears["r1"].vtysh_cmd(
         """
         configure terminal
@@ -243,13 +213,13 @@ def test_bgp_sid_vpn_export_reenable():
     check_rib(
         "r2", "show bgp ipv6 vpn json", "r2/vpnv6_rib_sid_vpn_export_reenabled.json"
     )
-    check_ping4("ce1", "192.168.2.2", True)
-    check_ping6("ce1", "2001:2::2", True)
+    check_ping("ce1", "192.168.2.2", True, 10, 0.5)
+    check_ping("ce1", "2001:2::2", True, 10, 1)
 
 
 def test_locator_delete():
-    check_ping4("ce1", "192.168.2.2", True)
-    check_ping6("ce1", "2001:2::2", True)
+    check_ping("ce1", "192.168.2.2", True, 10, 0.5)
+    check_ping("ce1", "2001:2::2", True, 10, 1)
     get_topogen().gears["r1"].vtysh_cmd(
         """
         configure terminal
@@ -263,13 +233,13 @@ def test_locator_delete():
     check_rib("r2", "show bgp ipv4 vpn json", "r2/vpnv4_rib_locator_deleted.json")
     check_rib("r1", "show bgp ipv6 vpn json", "r1/vpnv6_rib_locator_deleted.json")
     check_rib("r2", "show bgp ipv6 vpn json", "r2/vpnv6_rib_locator_deleted.json")
-    check_ping4("ce1", "192.168.2.2", False)
-    check_ping6("ce1", "2001:2::2", False)
+    check_ping("ce1", "192.168.2.2", False, 10, 0.5)
+    check_ping("ce1", "2001:2::2", False, 10, 1)
 
 
 def test_locator_recreate():
-    check_ping4("ce1", "192.168.2.2", False)
-    check_ping6("ce1", "2001:2::2", False)
+    check_ping("ce1", "192.168.2.2", False, 10, 0.5)
+    check_ping("ce1", "2001:2::2", False, 10, 1)
     get_topogen().gears["r1"].vtysh_cmd(
         """
         configure terminal
@@ -284,13 +254,13 @@ def test_locator_recreate():
     check_rib("r2", "show bgp ipv4 vpn json", "r2/vpnv4_rib_locator_recreated.json")
     check_rib("r1", "show bgp ipv6 vpn json", "r1/vpnv6_rib_locator_recreated.json")
     check_rib("r2", "show bgp ipv6 vpn json", "r2/vpnv6_rib_locator_recreated.json")
-    check_ping4("ce1", "192.168.2.2", True)
-    check_ping6("ce1", "2001:2::2", True)
+    check_ping("ce1", "192.168.2.2", True, 10, 0.5)
+    check_ping("ce1", "2001:2::2", True, 10, 1)
 
 
 def test_bgp_locator_unset():
-    check_ping4("ce1", "192.168.2.2", True)
-    check_ping6("ce1", "2001:2::2", True)
+    check_ping("ce1", "192.168.2.2", True, 10, 0.5)
+    check_ping("ce1", "2001:2::2", True, 10, 1)
     get_topogen().gears["r1"].vtysh_cmd(
         """
         configure terminal
@@ -303,13 +273,13 @@ def test_bgp_locator_unset():
     check_rib("r2", "show bgp ipv4 vpn json", "r2/vpnv4_rib_locator_deleted.json")
     check_rib("r1", "show bgp ipv6 vpn json", "r1/vpnv6_rib_locator_deleted.json")
     check_rib("r2", "show bgp ipv6 vpn json", "r2/vpnv6_rib_locator_deleted.json")
-    check_ping4("ce1", "192.168.2.2", False)
-    check_ping6("ce1", "2001:2::2", False)
+    check_ping("ce1", "192.168.2.2", False, 10, 0.5)
+    check_ping("ce1", "2001:2::2", False, 10, 1)
 
 
 def test_bgp_locator_reset():
-    check_ping4("ce1", "192.168.2.2", False)
-    check_ping6("ce1", "2001:2::2", False)
+    check_ping("ce1", "192.168.2.2", False, 10, 0.5)
+    check_ping("ce1", "2001:2::2", False, 10, 1)
     get_topogen().gears["r1"].vtysh_cmd(
         """
         configure terminal
@@ -322,13 +292,13 @@ def test_bgp_locator_reset():
     check_rib("r2", "show bgp ipv4 vpn json", "r2/vpnv4_rib_locator_recreated.json")
     check_rib("r1", "show bgp ipv6 vpn json", "r1/vpnv6_rib_locator_recreated.json")
     check_rib("r2", "show bgp ipv6 vpn json", "r2/vpnv6_rib_locator_recreated.json")
-    check_ping4("ce1", "192.168.2.2", True)
-    check_ping6("ce1", "2001:2::2", True)
+    check_ping("ce1", "192.168.2.2", True, 10, 0.5)
+    check_ping("ce1", "2001:2::2", True, 10, 1)
 
 
 def test_bgp_srv6_unset():
-    check_ping4("ce1", "192.168.2.2", True)
-    check_ping6("ce1", "2001:2::2", True)
+    check_ping("ce1", "192.168.2.2", True, 10, 0.5)
+    check_ping("ce1", "2001:2::2", True, 10, 1)
     get_topogen().gears["r1"].vtysh_cmd(
         """
         configure terminal
@@ -340,13 +310,13 @@ def test_bgp_srv6_unset():
     check_rib("r2", "show bgp ipv4 vpn json", "r2/vpnv4_rib_locator_deleted.json")
     check_rib("r1", "show bgp ipv6 vpn json", "r1/vpnv6_rib_locator_deleted.json")
     check_rib("r2", "show bgp ipv6 vpn json", "r2/vpnv6_rib_locator_deleted.json")
-    check_ping4("ce1", "192.168.2.2", False)
-    check_ping6("ce1", "2001:2::2", False)
+    check_ping("ce1", "192.168.2.2", False, 10, 0.5)
+    check_ping("ce1", "2001:2::2", False, 10, 1)
 
 
 def test_bgp_srv6_reset():
-    check_ping4("ce1", "192.168.2.2", False)
-    check_ping6("ce1", "2001:2::2", False)
+    check_ping("ce1", "192.168.2.2", False, 10, 0.5)
+    check_ping("ce1", "2001:2::2", False, 10, 1)
     get_topogen().gears["r1"].vtysh_cmd(
         """
         configure terminal
@@ -359,8 +329,8 @@ def test_bgp_srv6_reset():
     check_rib("r2", "show bgp ipv4 vpn json", "r2/vpnv4_rib_locator_recreated.json")
     check_rib("r1", "show bgp ipv6 vpn json", "r1/vpnv6_rib_locator_recreated.json")
     check_rib("r2", "show bgp ipv6 vpn json", "r2/vpnv6_rib_locator_recreated.json")
-    check_ping4("ce1", "192.168.2.2", True)
-    check_ping6("ce1", "2001:2::2", True)
+    check_ping("ce1", "192.168.2.2", True, 10, 0.5)
+    check_ping("ce1", "2001:2::2", True, 10, 1)
 
 
 if __name__ == "__main__":
diff --git a/tests/topotests/bgp_vpnv4_asbr/test_bgp_vpnv4_asbr.py b/tests/topotests/bgp_vpnv4_asbr/test_bgp_vpnv4_asbr.py
index c0e57930a310..a908e74cc63e 100644
--- a/tests/topotests/bgp_vpnv4_asbr/test_bgp_vpnv4_asbr.py
+++ b/tests/topotests/bgp_vpnv4_asbr/test_bgp_vpnv4_asbr.py
@@ -50,6 +50,8 @@
 from lib import topotest
 from lib.topogen import Topogen, TopoRouter, get_topogen
 from lib.topolog import logger
+from lib.checkping import check_ping
+
 
 # Required to instantiate the topology builder class.
 
@@ -257,29 +259,6 @@ def mpls_table_check_entry(router, out_label, out_nexthop):
     )
 
 
-def check_ping(name, dest_addr, expect_connected):
-    """
-    Assert that ping to dest_addr is expected
-    * 'name': the router to set the ping from
-    * 'dest_addr': The destination ip address to ping
-    * 'expect_connected': True if ping is expected to pass
-    """
-
-    def _check(name, dest_addr, match):
-        tgen = get_topogen()
-        output = tgen.gears[name].run("ping {} -c 1 -w 1".format(dest_addr))
-        logger.info(output)
-        if match not in output:
-            return "ping fail"
-
-    match = ", {} packet loss".format("0%" if expect_connected else "100%")
-    logger.info("[+] check {} {} {}".format(name, dest_addr, match))
-    tgen = get_topogen()
-    func = functools.partial(_check, name, dest_addr, match)
-    success, result = topotest.run_and_expect(func, None, count=20, wait=0.5)
-    assert result is None, "Failed"
-
-
 def check_show_bgp_vpn_prefix_found(
     router, ipversion, prefix, rd, label=None, nexthop=None
 ):
@@ -493,8 +472,8 @@ def test_mpls_setup_ok():
     check_show_bgp_vpn_ok(router, vpnv4_checks)
 
     logger.info("h1, check that ping from h1 to (h2,h3) is ok")
-    check_ping("h1", "172.31.1.10", True)
-    check_ping("h1", "172.31.2.10", True)
+    check_ping("h1", "172.31.1.10", True, 20, 0.5)
+    check_ping("h1", "172.31.2.10", True, 20, 0.5)
 
 
 def test_r3_prefixes_removed():
@@ -751,8 +730,8 @@ def test_reconfigure_nexthop_change_nexthop_self():
     check_show_bgp_vpn_ok(router, vpnv4_checks)
 
     logger.info("h1, check that ping from h1 to (h2,h3) is ok")
-    check_ping("h1", "172.31.1.10", True)
-    check_ping("h1", "172.31.2.10", True)
+    check_ping("h1", "172.31.1.10", True, 20, 0.5)
+    check_ping("h1", "172.31.2.10", True, 20, 0.5)
     # diagnostic
     logger.info("Dumping mplsvpn nexthop table")
     router.vtysh_cmd("show bgp mplsvpn-nh-label-bind detail", isjson=False)
diff --git a/tests/topotests/lib/checkping.py b/tests/topotests/lib/checkping.py
new file mode 100644
index 000000000000..aaa6164dd45f
--- /dev/null
+++ b/tests/topotests/lib/checkping.py
@@ -0,0 +1,33 @@
+# SPDX-License-Identifier: GPL-2.0-or-later
+#
+# Copyright 2023 Quentin Young
+
+import functools
+from lib.topogen import get_topogen
+from lib.topolog import logger
+from lib import topotest
+
+
+def check_ping(name, dest_addr, expect_connected, count, wait):
+    """
+    Assert that ping to dest_addr is expected
+    * 'name': the router to set the ping from
+    * 'dest_addr': The destination ip address to ping
+    * 'expect_connected': True if ping is expected to pass
+    * 'count': how many echos to send
+    * 'wait': how long ping should wait to receive all replies
+    """
+
+    def _check(name, dest_addr, match):
+        tgen = get_topogen()
+        output = tgen.gears[name].run("ping {} -c 1 -w 1".format(dest_addr))
+        logger.info(output)
+        if match not in output:
+            return "ping fail"
+
+    match = ", {} packet loss".format("0%" if expect_connected else "100%")
+    logger.info("[+] check {} {} {}".format(name, dest_addr, match))
+    tgen = get_topogen()
+    func = functools.partial(_check, name, dest_addr, match)
+    success, result = topotest.run_and_expect(func, None, count=count, wait=wait)
+    assert result is None, "Failed"