tests: use new helper object for mcast-tester and iperf

- Decrease igmp query interval to fix pim test run times

Signed-off-by: Christian Hopps <chopps@labn.net>

11 files changed, 420 insertions, 707 deletions

diff --git a/tests/topotests/msdp_mesh_topo1/test_msdp_mesh_topo1.py b/tests/topotests/msdp_mesh_topo1/test_msdp_mesh_topo1.py
index 121f4a7fdc..c32800c936 100644
--- a/tests/topotests/msdp_mesh_topo1/test_msdp_mesh_topo1.py
+++ b/tests/topotests/msdp_mesh_topo1/test_msdp_mesh_topo1.py
@@ -45,73 +45,11 @@ from lib.micronet_compat import Topo
 from lib.topogen import Topogen, TopoRouter, get_topogen
 from lib.topolog import logger
 
+from lib.pim import McastTesterHelper
 
 pytestmark = [pytest.mark.bgpd, pytest.mark.ospfd, pytest.mark.pimd]
 
-#
-# Test global variables:
-# They are used to handle communicating with external application.
-#
-HELPER_APP_PATH = os.path.join(CWD, "../lib/mcast-tester.py")
-app_listener = None
-app_clients = {}
-app_procs = []
-
-
-def get_app_sock_path():
-    tgen = get_topogen()
-    return os.path.join(tgen.logdir, "apps.sock")
-
-
-def listen_to_applications():
-    "Start listening socket to connect with applications."
-    # Remove old socket.
-    app_sock_path = get_app_sock_path()
-    try:
-        os.unlink(app_sock_path)
-    except OSError:
-        pass
-
-    sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM, 0)
-    sock.bind(app_sock_path)
-    sock.listen(10)
-    global app_listener
-    app_listener = sock
-
-def accept_host(host):
-    "Accept connection from application running in hosts."
-    global app_listener, app_clients
-    conn = app_listener.accept()
-    app_clients[host] = {
-        'fd': conn[0],
-        'address': conn[1]
-    }
-
-def close_applications():
-    "Signal applications to stop and close all sockets."
-    global app_listener, app_clients
-
-    # Close listening socket.
-    app_listener.close()
-
-    app_sock_path = get_app_sock_path()
-
-    # Remove old socket.
-    try:
-        os.unlink(app_sock_path)
-    except OSError:
-        pass
-
-    # Close all host connections.
-    for host in ["h1", "h2"]:
-        if app_clients.get(host) is None:
-            continue
-        app_clients["h1"]["fd"].close()
-
-    for p in app_procs:
-        p.terminate()
-        p.wait()
-
+app_helper = McastTesterHelper()
 
 def build_topo(tgen):
     "Build function"
@@ -166,8 +104,8 @@ def setup_module(mod):
     # Initialize all routers.
     tgen.start_router()
 
-    # Start applications socket.
-    listen_to_applications()
+    app_helper.init(tgen)
+
 
 
 def test_wait_ospf_convergence():
@@ -228,21 +166,12 @@ def test_wait_msdp_convergence():
         assertmsg = '"{}" MSDP connection failure'.format(router)
         assert result is None, assertmsg
 
-    app_sock_path = get_app_sock_path()
-
-
-    python3_path = tgen.net.get_exec_path(["python3", "python"])
-    ph_base = [python3_path, HELPER_APP_PATH, app_sock_path]
-
-    ph1_cmd = ph_base + ["--send=0.7", "229.0.1.10", "h1-eth0"]
-    ph1 = tgen.gears["h1"].popen(ph1_cmd)
-    app_procs.append(ph1)
-    accept_host("h1")
+    mcastaddr = "229.0.1.10"
+    logger.info("Starting helper1")
+    app_helper.run("h1", ["--send=0.7", mcastaddr, "h1-eth0"])
 
-    ph2_cmd = ph_base + ["229.0.1.10", "h2-eth0"]
-    ph2 = tgen.gears["h2"].popen(ph2_cmd)
-    app_procs.append(ph2)
-    accept_host("h2")
+    logger.info("Starting helper2")
+    app_helper.run("h2", [mcastaddr, "h2-eth0"])
 
     # R1 peers.
     expect_msdp_peer("r1", "10.254.254.2")
@@ -295,7 +224,7 @@ def test_msdp_sa_configuration():
 def teardown_module(_mod):
     "Teardown the pytest environment"
     tgen = get_topogen()
-    close_applications()
+    app_helper.cleanup()
     tgen.stop_topology()
 
 
diff --git a/tests/topotests/msdp_topo1/test_msdp_topo1.py b/tests/topotests/msdp_topo1/test_msdp_topo1.py
index 8fad62db79..93a4f62fce 100755
--- a/tests/topotests/msdp_topo1/test_msdp_topo1.py
+++ b/tests/topotests/msdp_topo1/test_msdp_topo1.py
@@ -46,71 +46,11 @@ from lib.micronet_compat import Topo
 from lib.topogen import Topogen, TopoRouter, get_topogen
 from lib.topolog import logger
 
+from lib.pim import McastTesterHelper
 
 pytestmark = [pytest.mark.bgpd, pytest.mark.pimd]
 
-#
-# Test global variables:
-# They are used to handle communicating with external application.
-#
-HELPER_APP_PATH = os.path.join(CWD, "../lib/mcast-tester.py")
-app_listener = None
-app_clients = {}
-
-
-def get_app_sock_path():
-    tgen = get_topogen()
-    return os.path.join(tgen.logdir, "apps.sock")
-
-
-def listen_to_applications():
-    "Start listening socket to connect with applications."
-
-    app_sock_path = get_app_sock_path()
-    # Remove old socket.
-    try:
-        os.unlink(app_sock_path)
-    except OSError:
-        pass
-
-    sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM, 0)
-    # Do not block forever
-    sock.settimeout(10)
-    sock.bind(app_sock_path)
-    sock.listen(10)
-    global app_listener
-    app_listener = sock
-
-
-def accept_host(host):
-    "Accept connection from application running in hosts."
-    global app_listener, app_clients
-    conn = app_listener.accept()
-    app_clients[host] = {
-        'fd': conn[0],
-        'address': conn[1]
-    }
-
-
-def close_applications():
-    "Signal applications to stop and close all sockets."
-    global app_listener, app_clients
-
-    # Close listening socket.
-    app_listener.close()
-
-    app_sock_path = get_app_sock_path()
-    # Remove old socket.
-    try:
-        os.unlink(app_sock_path)
-    except OSError:
-        pass
-
-    # Close all host connections.
-    for host in ["h1", "h2"]:
-        if app_clients.get(host) is None:
-            continue
-        app_clients[host]["fd"].close()
+app_helper = McastTesterHelper()
 
 
 def build_topo(tgen):
@@ -172,14 +112,12 @@ def setup_module(mod):
     # Initialize all routers.
     tgen.start_router()
 
-    # Start applications socket.
-    listen_to_applications()
-
+    app_helper.init(tgen)
 
 def teardown_module(mod):
     "Teardown the pytest environment"
     tgen = get_topogen()
-    close_applications()
+    app_helper.cleanup()
     tgen.stop_topology()
 
 
@@ -368,34 +306,14 @@ def test_mroute_install():
     if tgen.routers_have_failure():
         pytest.skip(tgen.errors)
 
-    ph1 = ph2 = None
+    logger.info("Starting helper1")
+    mcastaddr = "229.1.2.3"
+    app_helper.run("h1", [mcastaddr, "h1-eth0"])
 
-    app_sock_path = get_app_sock_path()
-    try:
-        logger.info("Starting helper1")
-        ph1 = tgen.gears["h1"].popen(
-            "{} '{}' '{}' '{}'".format(
-                HELPER_APP_PATH, app_sock_path, "229.1.2.3", "h1-eth0"
-            )
-        )
-        logger.info("Accepting helper1")
-        accept_host("h1")
-
-        logger.info("Starting helper2")
-        ph2 = tgen.gears["h2"].popen(
-            "{} --send='0.7' '{}' '{}' '{}'".format(
-                HELPER_APP_PATH, app_sock_path, "229.1.2.3", "h2-eth0"
-            )
-        )
-        accept_host("h2")
-
-        _test_mroute_install()
-    finally:
-        if ph1:
-            ph1.terminate()
-            ph1.wait()
-            ph2.terminate()
-            ph2.wait()
+    logger.info("Starting helper2")
+    app_helper.run("h2", ["--send=0.7", mcastaddr, "h2-eth0"])
+
+    _test_mroute_install()
 
 def test_msdp():
     """
diff --git a/tests/topotests/multicast_pim_bsm_topo1/test_mcast_pim_bsmp_01.py b/tests/topotests/multicast_pim_bsm_topo1/test_mcast_pim_bsmp_01.py
index df1cce654f..b346e3d33b 100644
--- a/tests/topotests/multicast_pim_bsm_topo1/test_mcast_pim_bsmp_01.py
+++ b/tests/topotests/multicast_pim_bsm_topo1/test_mcast_pim_bsmp_01.py
@@ -75,7 +75,6 @@ from lib.common_config import (
     step,
     addKernelRoute,
     create_static_routes,
-    iperfSendIGMPJoin,
     stop_router,
     start_router,
     shutdown_bringup_interface,
@@ -84,7 +83,6 @@ from lib.common_config import (
     reset_config_on_routers,
     do_countdown,
     apply_raw_config,
-    kill_iperf,
     run_frr_cmd,
     required_linux_kernel_version,
     topo_daemons,
@@ -109,6 +107,7 @@ from lib.pim import (
     clear_ip_mroute,
     clear_ip_pim_interface_traffic,
     verify_pim_interface_traffic,
+    McastTesterHelper,
 )
 from lib.topolog import logger
 from lib.topojson import build_topo_from_json, build_config_from_json
@@ -183,6 +182,10 @@ def setup_module(mod):
     # Creating configuration from JSON
     build_config_from_json(tgen, topo)
 
+    # XXX Replace this using "with McastTesterHelper()... " in each test if possible.
+    global app_helper
+    app_helper = McastTesterHelper(tgen)
+
     logger.info("Running setup_module() done")
 
 
@@ -193,8 +196,7 @@ def teardown_module():
 
     tgen = get_topogen()
 
-    # Kill any iperfs we left running.
-    kill_iperf(tgen)
+    app_helper.cleanup()
 
     # Stop toplogy and Remove tmp files
     tgen.stop_topology()
@@ -388,7 +390,7 @@ def test_BSR_higher_prefer_ip_p0(request):
     if tgen.routers_have_failure():
         pytest.skip(tgen.errors)
 
-    kill_iperf(tgen)
+    app_helper.stop_all_hosts()
     clear_ip_mroute(tgen)
     reset_config_on_routers(tgen)
     clear_ip_pim_interface_traffic(tgen, topo)
@@ -465,7 +467,7 @@ def test_BSR_higher_prefer_ip_p0(request):
     result = scapy_send_bsr_raw_packet(tgen, topo, "b1", "f1", "packet9")
     assert result is True, "Testcase {} :Failed \n Error {}".format(tc_name, result)
 
-    result = iperfSendIGMPJoin(tgen, "r1", GROUP_ADDRESS, join_interval=1)
+    result = app_helper.run_join("r1", GROUP_ADDRESS, "l1")
     assert result is True, "Testcase {}:Failed \n Error: {}".format(tc_name, result)
     do_countdown(5)
 
@@ -594,7 +596,7 @@ def test_BSR_CRP_with_blackhole_address_p1(request):
     if tgen.routers_have_failure():
         pytest.skip(tgen.errors)
 
-    kill_iperf(tgen)
+    app_helper.stop_all_hosts()
     clear_ip_mroute(tgen)
     reset_config_on_routers(tgen)
     clear_ip_pim_interface_traffic(tgen, topo)
@@ -771,7 +773,7 @@ def test_new_router_fwd_p0(request):
     if tgen.routers_have_failure():
         pytest.skip(tgen.errors)
 
-    kill_iperf(tgen)
+    app_helper.stop_all_hosts()
     clear_ip_mroute(tgen)
     reset_config_on_routers(tgen)
     clear_ip_pim_interface_traffic(tgen, topo)
@@ -797,7 +799,7 @@ def test_new_router_fwd_p0(request):
     bsr_ip = topo["routers"]["b1"]["bsm"]["bsr_packets"]["packet1"]["bsr"].split("/")[0]
     time.sleep(1)
 
-    result = iperfSendIGMPJoin(tgen, "r1", GROUP_ADDRESS, join_interval=1)
+    result = app_helper.run_join("r1", GROUP_ADDRESS, "l1")
     assert result is True, "Testcase {}:Failed \n Error: {}".format(tc_name, result)
 
     # Verify bsr state in FHR
@@ -850,7 +852,7 @@ def test_new_router_fwd_p0(request):
     stop_router(tgen, "i1")
     start_router(tgen, "i1")
 
-    result = iperfSendIGMPJoin(tgen, "r1", GROUP_ADDRESS, join_interval=1)
+    result = app_helper.run_join("r1", GROUP_ADDRESS, "l1")
     assert result is True, "Testcase {}:Failed \n Error: {}".format(tc_name, result)
 
     # Verify again if BSR is installed from bsm forwarded by f1
@@ -908,7 +910,7 @@ def test_int_bsm_config_p1(request):
     if tgen.routers_have_failure():
         pytest.skip(tgen.errors)
 
-    kill_iperf(tgen)
+    app_helper.stop_all_hosts()
     clear_ip_mroute(tgen)
     reset_config_on_routers(tgen)
     clear_ip_pim_interface_traffic(tgen, topo)
@@ -930,7 +932,7 @@ def test_int_bsm_config_p1(request):
     bsr_ip = topo["routers"]["b1"]["bsm"]["bsr_packets"]["packet1"]["bsr"].split("/")[0]
     time.sleep(1)
 
-    result = iperfSendIGMPJoin(tgen, "r1", GROUP_ADDRESS, join_interval=1)
+    result = app_helper.run_join("r1", GROUP_ADDRESS, "l1")
     assert result is True, "Testcase {}:Failed \n Error: {}".format(tc_name, result)
 
     # Use scapy to send pre-defined packet from senser to receiver
@@ -1069,7 +1071,7 @@ def test_static_rp_override_p1(request):
     if tgen.routers_have_failure():
         pytest.skip(tgen.errors)
 
-    kill_iperf(tgen)
+    app_helper.stop_all_hosts()
     clear_ip_mroute(tgen)
     reset_config_on_routers(tgen)
     clear_ip_pim_interface_traffic(tgen, topo)
@@ -1094,7 +1096,7 @@ def test_static_rp_override_p1(request):
     bsr_ip = topo["routers"]["b1"]["bsm"]["bsr_packets"]["packet1"]["bsr"].split("/")[0]
     time.sleep(1)
 
-    result = iperfSendIGMPJoin(tgen, "r1", GROUP_ADDRESS, join_interval=1)
+    result = app_helper.run_join("r1", GROUP_ADDRESS, "l1")
     assert result is True, "Testcase {}:Failed \n Error: {}".format(tc_name, result)
 
     # Verify bsr state in FHR
@@ -1220,7 +1222,7 @@ def test_bsmp_stress_add_del_restart_p2(request):
     if tgen.routers_have_failure():
         pytest.skip(tgen.errors)
 
-    kill_iperf(tgen)
+    app_helper.stop_all_hosts()
     clear_ip_mroute(tgen)
     reset_config_on_routers(tgen)
     clear_ip_pim_interface_traffic(tgen, topo)
@@ -1247,7 +1249,7 @@ def test_bsmp_stress_add_del_restart_p2(request):
     bsr_ip = topo["routers"]["b1"]["bsm"]["bsr_packets"]["packet1"]["bsr"].split("/")[0]
     time.sleep(1)
 
-    result = iperfSendIGMPJoin(tgen, "r1", GROUP_ADDRESS, join_interval=1)
+    result = app_helper.run_join("r1", GROUP_ADDRESS, "l1")
     assert result is True, "Testcase {}:Failed \n Error: {}".format(tc_name, result)
 
     # Verify bsr state in FHR
@@ -1341,7 +1343,7 @@ def test_bsmp_stress_add_del_restart_p2(request):
     assert result is True, "Testcase {}:Failed \n Error: {}".format(tc_name, result)
 
     # Send IGMP join to LHR
-    result = iperfSendIGMPJoin(tgen, "r1", GROUP_ADDRESS, join_interval=1)
+    result = app_helper.run_join("r1", GROUP_ADDRESS, "l1")
     assert result is True, "Testcase {}:Failed \n Error: {}".format(tc_name, result)
 
     do_countdown(5)
@@ -1388,7 +1390,7 @@ def test_BSM_timeout_p0(request):
     if tgen.routers_have_failure():
         pytest.skip(tgen.errors)
 
-    kill_iperf(tgen)
+    app_helper.stop_all_hosts()
     clear_ip_mroute(tgen)
     reset_config_on_routers(tgen)
     clear_ip_pim_interface_traffic(tgen, topo)
@@ -1413,7 +1415,7 @@ def test_BSM_timeout_p0(request):
     assert result is True, "Testcase {} :Failed \n Error {}".format(tc_name, result)
 
     # Send IGMP join for group 225.1.1.1 from receiver
-    result = iperfSendIGMPJoin(tgen, "r1", GROUP_ADDRESS, join_interval=1)
+    result = app_helper.run_join("r1", GROUP_ADDRESS, "l1")
     assert result is True, "Testcase {}:Failed \n Error: {}".format(tc_name, result)
 
     # Verify bsr state in FHR
@@ -1545,7 +1547,7 @@ def test_iif_join_state_p0(request):
     if tgen.routers_have_failure():
         pytest.skip(tgen.errors)
 
-    kill_iperf(tgen)
+    app_helper.stop_all_hosts()
     clear_ip_mroute(tgen)
     reset_config_on_routers(tgen)
     clear_ip_pim_interface_traffic(tgen, topo)
@@ -1571,7 +1573,7 @@ def test_iif_join_state_p0(request):
     bsr_ip = topo["routers"]["b1"]["bsm"]["bsr_packets"]["packet1"]["bsr"].split("/")[0]
     time.sleep(1)
 
-    result = iperfSendIGMPJoin(tgen, "r1", GROUP_ADDRESS, join_interval=1)
+    result = app_helper.run_join("r1", GROUP_ADDRESS, "l1")
     assert result is True, "Testcase {}:Failed \n Error: {}".format(tc_name, result)
 
     # Verify bsr state in FHR
diff --git a/tests/topotests/multicast_pim_bsm_topo2/test_mcast_pim_bsmp_02.py b/tests/topotests/multicast_pim_bsm_topo2/test_mcast_pim_bsmp_02.py
index 64c284f7f1..69b105cc49 100644
--- a/tests/topotests/multicast_pim_bsm_topo2/test_mcast_pim_bsmp_02.py
+++ b/tests/topotests/multicast_pim_bsm_topo2/test_mcast_pim_bsmp_02.py
@@ -66,7 +66,6 @@ from lib.common_config import (
     step,
     addKernelRoute,
     create_static_routes,
-    iperfSendIGMPJoin,
     stop_router,
     start_router,
     shutdown_bringup_interface,
@@ -75,7 +74,6 @@ from lib.common_config import (
     reset_config_on_routers,
     do_countdown,
     apply_raw_config,
-    kill_iperf,
     run_frr_cmd,
     required_linux_kernel_version,
     topo_daemons,
@@ -100,6 +98,7 @@ from lib.pim import (
     clear_ip_mroute,
     clear_ip_pim_interface_traffic,
     verify_pim_interface_traffic,
+    McastTesterHelper,
 )
 from lib.topolog import logger
 from lib.topojson import build_topo_from_json, build_config_from_json
@@ -175,6 +174,10 @@ def setup_module(mod):
     # Creating configuration from JSON
     build_config_from_json(tgen, topo)
 
+    # XXX Replace this using "with McastTesterHelper()... " in each test if possible.
+    global app_helper
+    app_helper = McastTesterHelper(tgen)
+
     logger.info("Running setup_module() done")
 
 
@@ -185,8 +188,7 @@ def teardown_module():
 
     tgen = get_topogen()
 
-    # Kill any iperfs we left running.
-    kill_iperf(tgen)
+    app_helper.cleanup()
 
     # Stop toplogy and Remove tmp files
     tgen.stop_topology()
@@ -342,7 +344,7 @@ def test_starg_mroute_p0(request):
     if tgen.routers_have_failure():
         pytest.skip(tgen.errors)
 
-    kill_iperf(tgen)
+    app_helper.stop_all_hosts()
     clear_ip_mroute(tgen)
     reset_config_on_routers(tgen)
     clear_ip_pim_interface_traffic(tgen, topo)
@@ -368,7 +370,7 @@ def test_starg_mroute_p0(request):
     bsr_ip = topo["routers"]["b1"]["bsm"]["bsr_packets"]["packet1"]["bsr"].split("/")[0]
     time.sleep(1)
 
-    result = iperfSendIGMPJoin(tgen, "r1", GROUP_ADDRESS, join_interval=1)
+    result = app_helper.run_join("r1", GROUP_ADDRESS, "l1")
     assert result is True, "Testcase {}:Failed \n Error: {}".format(tc_name, result)
 
     # Verify bsr state in FHR
@@ -494,7 +496,7 @@ def test_overlapping_group_p0(request):
     if tgen.routers_have_failure():
         pytest.skip(tgen.errors)
 
-    kill_iperf(tgen)
+    app_helper.stop_all_hosts()
     clear_ip_mroute(tgen)
     reset_config_on_routers(tgen)
     clear_ip_pim_interface_traffic(tgen, topo)
@@ -521,7 +523,7 @@ def test_overlapping_group_p0(request):
     bsr_ip = topo["routers"]["b1"]["bsm"]["bsr_packets"]["packet1"]["bsr"].split("/")[0]
     time.sleep(1)
 
-    result = iperfSendIGMPJoin(tgen, "r1", GROUP_ADDRESS, join_interval=1)
+    result = app_helper.run_join("r1", GROUP_ADDRESS, "l1")
     assert result is True, "Testcase {}:Failed \n Error: {}".format(tc_name, result)
 
     # Verify bsr state in FHR
@@ -600,7 +602,7 @@ def test_RP_priority_p0(request):
     if tgen.routers_have_failure():
         pytest.skip(tgen.errors)
 
-    kill_iperf(tgen)
+    app_helper.stop_all_hosts()
     clear_ip_mroute(tgen)
     reset_config_on_routers(tgen)
     clear_ip_pim_interface_traffic(tgen, topo)
@@ -627,7 +629,7 @@ def test_RP_priority_p0(request):
     bsr_ip = topo["routers"]["b1"]["bsm"]["bsr_packets"]["packet1"]["bsr"].split("/")[0]
     time.sleep(1)
 
-    result = iperfSendIGMPJoin(tgen, "r1", GROUP_ADDRESS, join_interval=1)
+    result = app_helper.run_join("r1", GROUP_ADDRESS, "l1")
     assert result is True, "Testcase {}:Failed \n Error: {}".format(tc_name, result)
 
     # Verify bsr state in FHR
@@ -729,7 +731,7 @@ def test_BSR_election_p0(request):
     tc_name = request.node.name
     write_test_header(tc_name)
 
-    kill_iperf(tgen)
+    app_helper.stop_all_hosts()
     clear_ip_mroute(tgen)
     reset_config_on_routers(tgen)
     clear_ip_pim_interface_traffic(tgen, topo)
@@ -762,7 +764,7 @@ def test_BSR_election_p0(request):
     ]
     time.sleep(1)
 
-    result = iperfSendIGMPJoin(tgen, "r1", GROUP_ADDRESS, join_interval=1)
+    result = app_helper.run_join("r1", GROUP_ADDRESS, "l1")
     assert result is True, "Testcase {}:Failed \n Error: {}".format(tc_name, result)
 
     # Verify bsr state in FHR
@@ -849,7 +851,7 @@ def test_RP_hash_p0(request):
     if tgen.routers_have_failure():
         pytest.skip(tgen.errors)
 
-    kill_iperf(tgen)
+    app_helper.stop_all_hosts()
     clear_ip_mroute(tgen)
     reset_config_on_routers(tgen)
     clear_ip_pim_interface_traffic(tgen, topo)
@@ -875,7 +877,7 @@ def test_RP_hash_p0(request):
     bsr_ip = topo["routers"]["b1"]["bsm"]["bsr_packets"]["packet1"]["bsr"].split("/")[0]
     time.sleep(1)
 
-    result = iperfSendIGMPJoin(tgen, "r1", GROUP_ADDRESS, join_interval=1)
+    result = app_helper.run_join("r1", GROUP_ADDRESS, "l1")
     assert result is True, "Testcase {}:Failed \n Error: {}".format(tc_name, result)
 
     dut = "l1"
@@ -942,7 +944,7 @@ def test_BSM_fragmentation_p1(request):
     if tgen.routers_have_failure():
         pytest.skip(tgen.errors)
 
-    kill_iperf(tgen)
+    app_helper.stop_all_hosts()
     clear_ip_mroute(tgen)
     reset_config_on_routers(tgen)
     clear_ip_pim_interface_traffic(tgen, topo)
@@ -999,7 +1001,7 @@ def test_BSM_fragmentation_p1(request):
     result = scapy_send_bsr_raw_packet(tgen, topo, "b1", "f1", "packet2")
     assert result is True, "Testcase {} :Failed \n Error {}".format(tc_name, result)
 
-    result = iperfSendIGMPJoin(tgen, "r1", GROUP_ADDRESS, join_interval=1)
+    result = app_helper.run_join("r1", GROUP_ADDRESS, "l1")
     assert result is True, "Testcase {}:Failed \n Error: {}".format(tc_name, result)
 
     # Verify bsr state in FHR
@@ -1060,7 +1062,7 @@ def test_RP_with_all_ip_octet_p1(request):
     if tgen.routers_have_failure():
         pytest.skip(tgen.errors)
 
-    kill_iperf(tgen)
+    app_helper.stop_all_hosts()
     clear_ip_mroute(tgen)
     reset_config_on_routers(tgen)
     clear_ip_pim_interface_traffic(tgen, topo)
@@ -1082,7 +1084,7 @@ def test_RP_with_all_ip_octet_p1(request):
     bsr_ip = topo["routers"]["b1"]["bsm"]["bsr_packets"]["packet8"]["bsr"].split("/")[0]
     time.sleep(1)
 
-    result = iperfSendIGMPJoin(tgen, "r1", GROUP_ADDRESS, join_interval=1)
+    result = app_helper.run_join("r1", GROUP_ADDRESS, "l1")
     assert result is True, "Testcase {}:Failed \n Error: {}".format(tc_name, result)
 
     dut = "l1"
diff --git a/tests/topotests/multicast_pim_sm_topo1/test_multicast_pim_sm_topo1.py b/tests/topotests/multicast_pim_sm_topo1/test_multicast_pim_sm_topo1.py
index 9eb3d8cc58..e6c9a30b3d 100755
--- a/tests/topotests/multicast_pim_sm_topo1/test_multicast_pim_sm_topo1.py
+++ b/tests/topotests/multicast_pim_sm_topo1/test_multicast_pim_sm_topo1.py
@@ -76,12 +76,9 @@ from lib.common_config import (
     write_test_header,
     write_test_footer,
     step,
-    iperfSendIGMPJoin,
     addKernelRoute,
     apply_raw_config,
     reset_config_on_routers,
-    iperfSendTraffic,
-    kill_iperf,
     shutdown_bringup_interface,
     kill_router_daemons,
     start_router,
@@ -104,6 +101,7 @@ from lib.pim import (
     clear_ip_pim_interface_traffic,
     verify_igmp_config,
     clear_ip_mroute_verify,
+    McastTesterHelper,
 )
 from lib.topolog import logger
 from lib.topojson import build_topo_from_json, build_config_from_json
@@ -202,6 +200,10 @@ def setup_module(mod):
     # Creating configuration from JSON
     build_config_from_json(tgen, tgen.json_topo)
 
+    # XXX Replace this using "with McastTesterHelper()... " in each test if possible.
+    global app_helper
+    app_helper = McastTesterHelper(tgen)
+
     logger.info("Running setup_module() done")
 
 
@@ -212,8 +214,7 @@ def teardown_module():
 
     tgen = get_topogen()
 
-    # Kill any iperfs we left running.
-    kill_iperf(tgen)
+    app_helper.cleanup()
 
     # Stop toplogy and Remove tmp files
     tgen.stop_topology()
@@ -321,9 +322,8 @@ def test_multicast_data_traffic_static_RP_send_join_then_traffic_p0(request):
         pytest.skip(tgen.errors)
 
     step("Enable IGMP on FRR1 interface and send IGMP join (225.1.1.1)")
-    intf_i1_l1 = topo["routers"]["i1"]["links"]["l1"]["interface"]
 
-    step("joinRx value before join sent")
+    step("get joinRx value before join")
     intf_r2_l1 = topo["routers"]["r2"]["links"]["l1"]["interface"]
     state_dict = {"r2": {intf_r2_l1: ["joinRx"]}}
     state_before = verify_pim_interface_traffic(tgen, state_dict)
@@ -333,7 +333,7 @@ def test_multicast_data_traffic_static_RP_send_join_then_traffic_p0(request):
         tc_name, state_before
     )
 
-    result = iperfSendIGMPJoin(tgen, "i1", ["{}%{}".format(IGMP_JOIN, intf_i1_l1)], join_interval=1)
+    result = app_helper.run_join("i1", IGMP_JOIN, "l1")
     assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
     step("Send the IGMP join first and then start the traffic")
@@ -359,7 +359,7 @@ def test_multicast_data_traffic_static_RP_send_join_then_traffic_p0(request):
     assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
 
     step("Send multicast traffic from FRR3 to 225.1.1.1 receiver")
-    result = iperfSendTraffic(tgen, "i2", IGMP_JOIN, 32, 2500, bindToIntf="f1")
+    result = app_helper.run_traffic("i2", IGMP_JOIN, "f1")
     assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
 
     step(
@@ -388,6 +388,7 @@ def test_multicast_data_traffic_static_RP_send_join_then_traffic_p0(request):
         )
         assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
 
+
     step(
         "Verify 'show ip pim upstream' showing correct OIL and IIF" " on all the nodes"
     )
@@ -397,6 +398,7 @@ def test_multicast_data_traffic_static_RP_send_join_then_traffic_p0(request):
         )
         assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
 
+
     step("joinRx value after join sent")
     state_after = verify_pim_interface_traffic(tgen, state_dict)
     assert isinstance(
@@ -437,7 +439,7 @@ def test_multicast_data_traffic_static_RP_send_traffic_then_join_p0(request):
         pytest.skip(tgen.errors)
 
     # Creating configuration from JSON
-    kill_iperf(tgen)
+    app_helper.stop_all_hosts()
     clear_ip_mroute(tgen)
     reset_config_on_routers(tgen)
     clear_ip_pim_interface_traffic(tgen, topo)
@@ -470,7 +472,7 @@ def test_multicast_data_traffic_static_RP_send_traffic_then_join_p0(request):
     )
     assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
-    result = iperfSendTraffic(tgen, "i2", IGMP_JOIN, 32, 2500)
+    result = app_helper.run_traffic("i2", IGMP_JOIN, "f1")
     assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
 
     step("Enable IGMP on FRR1 interface and send IGMP join (225.1.1.1)")
@@ -488,7 +490,7 @@ def test_multicast_data_traffic_static_RP_send_traffic_then_join_p0(request):
         tc_name, result
     )
 
-    result = iperfSendIGMPJoin(tgen, "i1", IGMP_JOIN, join_interval=1)
+    result = app_helper.run_join("i1", IGMP_JOIN, "l1")
     assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
     step(
@@ -564,7 +566,7 @@ def test_clear_pim_neighbors_and_mroute_p0(request):
         pytest.skip(tgen.errors)
 
     # Creating configuration from JSON
-    kill_iperf(tgen)
+    app_helper.stop_all_hosts()
     clear_ip_mroute(tgen)
     reset_config_on_routers(tgen)
     clear_ip_pim_interface_traffic(tgen, topo)
@@ -597,7 +599,7 @@ def test_clear_pim_neighbors_and_mroute_p0(request):
     )
     assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
-    result = iperfSendIGMPJoin(tgen, "i1", IGMP_JOIN_RANGE_1, join_interval=1)
+    result = app_helper.run_join("i1", IGMP_JOIN_RANGE_1, "l1")
     assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
     step("Send multicast traffic from FRR3, wait for SPT switchover")
@@ -606,7 +608,7 @@ def test_clear_pim_neighbors_and_mroute_p0(request):
     )
     assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
-    result = iperfSendTraffic(tgen, "i2", IGMP_JOIN_RANGE_1, 32, 2500)
+    result = app_helper.run_traffic("i2", IGMP_JOIN_RANGE_1, "f1")
     assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
 
     step("Clear the mroute on l1, wait for 5 sec")
@@ -658,7 +660,7 @@ def test_verify_mroute_when_same_receiver_in_FHR_LHR_and_RP_p0(request):
         pytest.skip(tgen.errors)
 
     # Creating configuration from JSON
-    kill_iperf(tgen)
+    app_helper.stop_all_hosts()
     clear_ip_mroute(tgen)
     reset_config_on_routers(tgen)
     clear_ip_pim_interface_traffic(tgen, topo)
@@ -686,8 +688,8 @@ def test_verify_mroute_when_same_receiver_in_FHR_LHR_and_RP_p0(request):
     step("Enable IGMP on FRR1 interface and send IGMP join " "(225.1.1.1) to R1")
 
     input_dict = {
-        "f1": {"igmp": {"interfaces": {"f1-i8-eth2": {"igmp": {"version": "2"}}}}},
-        "r2": {"igmp": {"interfaces": {"r2-i3-eth1": {"igmp": {"version": "2"}}}}},
+        "f1": {"igmp": {"interfaces": {"f1-i8-eth2": {"igmp": {"version": "2", "query": {"query-interval": 15} }}}}},
+        "r2": {"igmp": {"interfaces": {"r2-i3-eth1": {"igmp": {"version": "2", "query": {"query-interval": 15} }}}}},
     }
     result = create_igmp_config(tgen, topo, input_dict)
     assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
@@ -700,7 +702,7 @@ def test_verify_mroute_when_same_receiver_in_FHR_LHR_and_RP_p0(request):
         )
         assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
-        result = iperfSendIGMPJoin(tgen, recvr, IGMP_JOIN, join_interval=1)
+        result = app_helper.run_join(recvr, IGMP_JOIN, join_intf=recvr_intf)
         assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
     step("Send multicast traffic from R3 to 225.1.1.1 receiver")
@@ -709,7 +711,7 @@ def test_verify_mroute_when_same_receiver_in_FHR_LHR_and_RP_p0(request):
     )
     assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
-    result = iperfSendTraffic(tgen, "i2", IGMP_JOIN, 32, 2500)
+    result = app_helper.run_traffic("i2", IGMP_JOIN, "f1")
     assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
 
     step("IGMP is received on FRR1 , FRR2 , FRR3, using " "'show ip igmp groups'")
@@ -754,7 +756,7 @@ def test_verify_mroute_when_same_receiver_joining_5_diff_sources_p0(request):
         pytest.skip(tgen.errors)
 
     # Creating configuration from JSON
-    kill_iperf(tgen)
+    app_helper.stop_all_hosts()
     clear_ip_mroute(tgen)
     reset_config_on_routers(tgen)
     clear_ip_pim_interface_traffic(tgen, topo)
@@ -792,11 +794,11 @@ def test_verify_mroute_when_same_receiver_joining_5_diff_sources_p0(request):
     )
     assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
-    result = iperfSendIGMPJoin(tgen, "i1", _IGMP_JOIN_RANGE, join_interval=1)
+    result = app_helper.run_join("i1", _IGMP_JOIN_RANGE, "l1")
     assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
     input_dict = {
-        "f1": {"igmp": {"interfaces": {"f1-i8-eth2": {"igmp": {"version": "2"}}}}}
+        "f1": {"igmp": {"interfaces": {"f1-i8-eth2": {"igmp": {"version": "2", "query": {"query-interval": 15} }}}}}
     }
     result = create_igmp_config(tgen, topo, input_dict)
     assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
@@ -806,7 +808,7 @@ def test_verify_mroute_when_same_receiver_joining_5_diff_sources_p0(request):
     )
     assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
-    result = iperfSendIGMPJoin(tgen, "i8", _IGMP_JOIN_RANGE, join_interval=1)
+    result = app_helper.run_join("i8", _IGMP_JOIN_RANGE, "f1")
     assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
     step(
@@ -828,7 +830,7 @@ def test_verify_mroute_when_same_receiver_joining_5_diff_sources_p0(request):
         )
         assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
-        result = iperfSendTraffic(tgen, src, _IGMP_JOIN_RANGE, 32, 2500)
+        result = app_helper.run_traffic(src, _IGMP_JOIN_RANGE, bind_intf=src_intf)
         assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
 
     step("Verify (*,G) are created on FRR1 and FRR3 node " " 'show ip mroute' ")
@@ -1078,7 +1080,7 @@ def test_verify_mroute_when_frr_is_transit_router_p2(request):
         pytest.skip(tgen.errors)
 
     # Creating configuration from JSON
-    kill_iperf(tgen)
+    app_helper.stop_all_hosts()
     clear_ip_mroute(tgen)
     reset_config_on_routers(tgen)
     clear_ip_pim_interface_traffic(tgen, topo)
@@ -1108,7 +1110,7 @@ def test_verify_mroute_when_frr_is_transit_router_p2(request):
     )
     assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
-    result = iperfSendIGMPJoin(tgen, "i1", IGMP_JOIN_RANGE_1, join_interval=1)
+    result = app_helper.run_join("i1", IGMP_JOIN_RANGE_1, "l1")
     assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
     step("Send multicast traffic from FRR3 to 225.1.1.1-5 receivers")
@@ -1117,7 +1119,7 @@ def test_verify_mroute_when_frr_is_transit_router_p2(request):
     )
     assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
-    result = iperfSendTraffic(tgen, "i2", IGMP_JOIN_RANGE_1, 32, 2500)
+    result = app_helper.run_traffic("i2", IGMP_JOIN_RANGE_1, "f1")
     assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
 
     # Stop r2 router to make r2 router disabled from topology
@@ -1191,7 +1193,7 @@ def test_verify_mroute_when_RP_unreachable_p1(request):
         pytest.skip(tgen.errors)
 
     # Creating configuration from JSON
-    kill_iperf(tgen)
+    app_helper.stop_all_hosts()
     clear_ip_mroute(tgen)
     reset_config_on_routers(tgen)
     clear_ip_pim_interface_traffic(tgen, topo)
@@ -1223,7 +1225,7 @@ def test_verify_mroute_when_RP_unreachable_p1(request):
     )
     assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
-    result = iperfSendIGMPJoin(tgen, "i1", IGMP_JOIN, join_interval=1)
+    result = app_helper.run_join("i1", IGMP_JOIN, "l1")
     assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
     step("Send multicast traffic from FRR3 to 225.1.1.1 receiver")
@@ -1232,12 +1234,12 @@ def test_verify_mroute_when_RP_unreachable_p1(request):
     )
     assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
-    result = iperfSendTraffic(tgen, "i2", IGMP_JOIN, 32, 2500)
+    result = app_helper.run_traffic("i2", IGMP_JOIN, "f1")
     assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
 
     step("Configure one IGMP interface on FRR3 node and send IGMP" " join (225.1.1.1)")
     input_dict = {
-        "f1": {"igmp": {"interfaces": {"f1-i8-eth2": {"igmp": {"version": "2"}}}}}
+        "f1": {"igmp": {"interfaces": {"f1-i8-eth2": {"igmp": {"version": "2", "query": {"query-interval": 15} }}}}}
     }
     result = create_igmp_config(tgen, topo, input_dict)
     assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
@@ -1247,7 +1249,7 @@ def test_verify_mroute_when_RP_unreachable_p1(request):
     )
     assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
-    result = iperfSendIGMPJoin(tgen, "i8", IGMP_JOIN, join_interval=1)
+    result = app_helper.run_join("i8", IGMP_JOIN, "f1")
     assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
     # Verify mroutes are present in FRR3(f1)
@@ -1315,7 +1317,7 @@ def test_modify_igmp_query_timer_p0(request):
         pytest.skip(tgen.errors)
 
     # Creating configuration from JSON
-    kill_iperf(tgen)
+    app_helper.stop_all_hosts()
     clear_ip_mroute(tgen)
     reset_config_on_routers(tgen)
     clear_ip_pim_interface_traffic(tgen, topo)
@@ -1326,7 +1328,7 @@ def test_modify_igmp_query_timer_p0(request):
     )
     assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
-    result = iperfSendIGMPJoin(tgen, "i1", IGMP_JOIN, join_interval=1)
+    result = app_helper.run_join("i1", IGMP_JOIN, "l1")
     assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
     step("Configure RP on R2 (loopback interface) for the" " group range 225.0.0.0/8")
@@ -1355,7 +1357,7 @@ def test_modify_igmp_query_timer_p0(request):
     )
     assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
-    result = iperfSendTraffic(tgen, "i2", IGMP_JOIN, 32, 2500)
+    result = app_helper.run_traffic("i2", IGMP_JOIN, "f1")
     assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
 
     step(
@@ -1389,7 +1391,7 @@ def test_modify_igmp_query_timer_p0(request):
         "l1": {
             "igmp": {
                 "interfaces": {
-                    "l1-i1-eth1": {"igmp": {"query": {"query-interval": 100}}}
+                    "l1-i1-eth1": {"igmp": {"query": {"query-interval": 20}}}
                 }
             }
         }
@@ -1404,7 +1406,7 @@ def test_modify_igmp_query_timer_p0(request):
         "l1": {
             "igmp": {
                 "interfaces": {
-                    "l1-i1-eth1": {"igmp": {"query": {"query-interval": 200}}}
+                    "l1-i1-eth1": {"igmp": {"query": {"query-interval": 25}}}
                 }
             }
         }
@@ -1419,7 +1421,7 @@ def test_modify_igmp_query_timer_p0(request):
         "l1": {
             "igmp": {
                 "interfaces": {
-                    "l1-i1-eth1": {"igmp": {"query": {"query-interval": 300}}}
+                    "l1-i1-eth1": {"igmp": {"query": {"query-interval": 30}}}
                 }
             }
         }
@@ -1454,7 +1456,7 @@ def test_modify_igmp_max_query_response_timer_p0(request):
         pytest.skip(tgen.errors)
 
     # Creating configuration from JSON
-    kill_iperf(tgen)
+    app_helper.stop_all_hosts()
     clear_ip_mroute(tgen)
     reset_config_on_routers(tgen)
     clear_ip_pim_interface_traffic(tgen, topo)
@@ -1465,7 +1467,7 @@ def test_modify_igmp_max_query_response_timer_p0(request):
     )
     assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
-    result = iperfSendIGMPJoin(tgen, "i1", IGMP_JOIN, join_interval=1)
+    result = app_helper.run_join("i1", IGMP_JOIN, "l1")
     assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
     step("Configure IGMP query response time to 10 deci-sec on FRR1")
@@ -1515,7 +1517,7 @@ def test_modify_igmp_max_query_response_timer_p0(request):
     )
     assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
-    result = iperfSendTraffic(tgen, "i2", IGMP_JOIN, 32, 2500)
+    result = app_helper.run_traffic("i2", IGMP_JOIN, "f1")
     assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
 
     step(
@@ -1535,6 +1537,7 @@ def test_modify_igmp_max_query_response_timer_p0(request):
         )
         assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
 
+
     step(
         "Verify 'show ip pim upstream' showing correct OIL and IIF" " on all the nodes"
     )
@@ -1573,7 +1576,7 @@ def test_modify_igmp_max_query_response_timer_p0(request):
     result = create_pim_config(tgen, topo["routers"])
     assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
-    step("Configure max query response timer 100sec on FRR1")
+    step("Configure max query response timer 100 decisec on FRR1")
     input_dict_3 = {
         "l1": {
             "igmp": {
@@ -1607,7 +1610,7 @@ def test_modify_igmp_max_query_response_timer_p0(request):
                     "l1-i1-eth1": {
                         "igmp": {
                             "version": "2",
-                            "query": {"query-max-response-time": 110},
+                            "query": {"query-max-response-time": 105},
                         }
                     }
                 }
@@ -1627,7 +1630,7 @@ def test_modify_igmp_max_query_response_timer_p0(request):
                     "l1-i1-eth1": {
                         "igmp": {
                             "version": "2",
-                            "query": {"query-max-response-time": 120},
+                            "query": {"query-max-response-time": 110},
                         }
                     }
                 }
@@ -1647,7 +1650,7 @@ def test_modify_igmp_max_query_response_timer_p0(request):
                     "l1-i1-eth1": {
                         "igmp": {
                             "version": "2",
-                            "query": {"query-max-response-time": 140},
+                            "query": {"query-max-response-time": 115},
                         }
                     }
                 }
@@ -1667,7 +1670,7 @@ def test_modify_igmp_max_query_response_timer_p0(request):
                     "l1-i1-eth1": {
                         "igmp": {
                             "version": "2",
-                            "query": {"query-max-response-time": 150},
+                            "query": {"query-max-response-time": 120},
                         }
                     }
                 }
diff --git a/tests/topotests/multicast_pim_sm_topo2/test_multicast_pim_sm_topo2.py b/tests/topotests/multicast_pim_sm_topo2/test_multicast_pim_sm_topo2.py
index 96886b0d95..27abfae9f4 100755
--- a/tests/topotests/multicast_pim_sm_topo2/test_multicast_pim_sm_topo2.py
+++ b/tests/topotests/multicast_pim_sm_topo2/test_multicast_pim_sm_topo2.py
@@ -72,11 +72,8 @@ from lib.common_config import (
     write_test_header,
     write_test_footer,
     step,
-    iperfSendIGMPJoin,
     addKernelRoute,
     reset_config_on_routers,
-    iperfSendTraffic,
-    kill_iperf,
     shutdown_bringup_interface,
     kill_router_daemons,
     start_router,
@@ -98,6 +95,7 @@ from lib.pim import (
     clear_ip_mroute,
     clear_ip_pim_interface_traffic,
     verify_igmp_config,
+    McastTesterHelper,
 )
 from lib.topolog import logger
 from lib.topojson import build_topo_from_json, build_config_from_json
@@ -195,6 +193,10 @@ def setup_module(mod):
     # Creating configuration from JSON
     build_config_from_json(tgen, topo)
 
+    # XXX Replace this using "with McastTesterHelper()... " in each test if possible.
+    global app_helper
+    app_helper = McastTesterHelper(tgen)
+
     logger.info("Running setup_module() done")
 
 
@@ -205,8 +207,7 @@ def teardown_module():
 
     tgen = get_topogen()
 
-    # Kill any iperfs we left running.
-    kill_iperf(tgen)
+    app_helper.cleanup()
 
     # Stop toplogy and Remove tmp files
     tgen.stop_topology()
@@ -318,7 +319,7 @@ def test_verify_mroute_and_traffic_when_pimd_restarted_p2(request):
         pytest.skip(tgen.errors)
 
     # Creating configuration from JSON
-    kill_iperf(tgen)
+    app_helper.stop_all_hosts()
     clear_ip_mroute(tgen)
     reset_config_on_routers(tgen)
     clear_ip_pim_interface_traffic(tgen, topo)
@@ -369,7 +370,7 @@ def test_verify_mroute_and_traffic_when_pimd_restarted_p2(request):
     )
 
     input_dict = {
-        "f1": {"igmp": {"interfaces": {"f1-i8-eth2": {"igmp": {"version": "2"}}}}}
+        "f1": {"igmp": {"interfaces": {"f1-i8-eth2": {"igmp": {"version": "2", "query": {"query-interval": 15}}}}}}
     }
     result = create_igmp_config(tgen, topo, input_dict)
     assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
@@ -382,7 +383,7 @@ def test_verify_mroute_and_traffic_when_pimd_restarted_p2(request):
         )
         assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
-        result = iperfSendIGMPJoin(tgen, recvr, _IGMP_JOIN_RANGE, join_interval=1)
+        result = app_helper.run_join(recvr, _IGMP_JOIN_RANGE, join_intf=recvr_intf)
         assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
     step(
@@ -402,7 +403,7 @@ def test_verify_mroute_and_traffic_when_pimd_restarted_p2(request):
         )
         assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
-        result = iperfSendTraffic(tgen, src, _IGMP_JOIN_RANGE, 32, 2500)
+        result = app_helper.run_traffic(src, _IGMP_JOIN_RANGE, bind_intf=src_intf)
         assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
 
     # Verifying mroutes before PIMd restart, fetching uptime
@@ -530,7 +531,7 @@ def test_verify_mroute_and_traffic_when_frr_restarted_p2(request):
         pytest.skip(tgen.errors)
 
     # Creating configuration from JSON
-    kill_iperf(tgen)
+    app_helper.stop_all_hosts()
     clear_ip_mroute(tgen)
     reset_config_on_routers(tgen)
     clear_ip_pim_interface_traffic(tgen, topo)
@@ -581,7 +582,7 @@ def test_verify_mroute_and_traffic_when_frr_restarted_p2(request):
     )
 
     input_dict = {
-        "f1": {"igmp": {"interfaces": {"f1-i8-eth2": {"igmp": {"version": "2"}}}}}
+        "f1": {"igmp": {"interfaces": {"f1-i8-eth2": {"igmp": {"version": "2", "query": {"query-interval": 15}}}}}}
     }
     result = create_igmp_config(tgen, topo, input_dict)
     assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
@@ -594,7 +595,7 @@ def test_verify_mroute_and_traffic_when_frr_restarted_p2(request):
         )
         assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
-        result = iperfSendIGMPJoin(tgen, recvr, _IGMP_JOIN_RANGE, join_interval=1)
+        result = app_helper.run_join(recvr, _IGMP_JOIN_RANGE, join_intf=recvr_intf)
         assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
     step(
@@ -614,7 +615,7 @@ def test_verify_mroute_and_traffic_when_frr_restarted_p2(request):
         )
         assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
-        result = iperfSendTraffic(tgen, src, _IGMP_JOIN_RANGE, 32, 2500)
+        result = app_helper.run_traffic(src, _IGMP_JOIN_RANGE, bind_intf=src_intf)
         assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
 
     step("Verifying mroutes before FRR restart, fetching uptime")
@@ -741,7 +742,7 @@ def test_verify_SPT_switchover_when_RPT_and_SPT_path_is_different_p0(request):
         pytest.skip(tgen.errors)
 
     # Creating configuration from JSON
-    kill_iperf(tgen)
+    app_helper.stop_all_hosts()
     clear_ip_mroute(tgen)
     reset_config_on_routers(tgen)
     clear_ip_pim_interface_traffic(tgen, topo)
@@ -779,7 +780,7 @@ def test_verify_SPT_switchover_when_RPT_and_SPT_path_is_different_p0(request):
     )
     assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
-    result = iperfSendIGMPJoin(tgen, "i1", _IGMP_JOIN_RANGE, join_interval=1)
+    result = app_helper.run_join("i1", _IGMP_JOIN_RANGE, "l1")
     assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
     step("Send multicast traffic from FRR3 to '226.1.1.1-5'" ", '232.1.1.1-5' receiver")
@@ -798,7 +799,7 @@ def test_verify_SPT_switchover_when_RPT_and_SPT_path_is_different_p0(request):
         tc_name, result
     )
 
-    result = iperfSendTraffic(tgen, "i2", _IGMP_JOIN_RANGE, 32, 2500)
+    result = app_helper.run_traffic("i2", _IGMP_JOIN_RANGE, "f1")
     assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
 
     step(
@@ -851,7 +852,7 @@ def test_verify_SPT_switchover_when_RPT_and_SPT_path_is_different_p0(request):
 
     step("Stop the traffic to all the receivers")
 
-    kill_iperf(tgen, "i2", "remove_traffic")
+    app_helper.stop_host("i2")
 
     step(
         "Null register packet being send periodically from FRR3 to RP, "
@@ -903,7 +904,7 @@ def test_verify_mroute_after_shut_noshut_of_upstream_interface_p1(request):
         pytest.skip(tgen.errors)
 
     # Creating configuration from JSON
-    kill_iperf(tgen)
+    app_helper.stop_all_hosts()
     clear_ip_mroute(tgen)
     reset_config_on_routers(tgen)
     clear_ip_pim_interface_traffic(tgen, topo)
@@ -954,7 +955,7 @@ def test_verify_mroute_after_shut_noshut_of_upstream_interface_p1(request):
     )
 
     input_dict = {
-        "f1": {"igmp": {"interfaces": {"f1-i8-eth2": {"igmp": {"version": "2"}}}}}
+        "f1": {"igmp": {"interfaces": {"f1-i8-eth2": {"igmp": {"version": "2", "query": {"query-interval": 15}}}}}}
     }
     result = create_igmp_config(tgen, topo, input_dict)
     assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
@@ -967,7 +968,7 @@ def test_verify_mroute_after_shut_noshut_of_upstream_interface_p1(request):
         )
         assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
-        result = iperfSendIGMPJoin(tgen, recvr, _IGMP_JOIN_RANGE, join_interval=1)
+        result = app_helper.run_join(recvr, _IGMP_JOIN_RANGE, join_intf=recvr_intf)
         assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
     step(
@@ -987,7 +988,7 @@ def test_verify_mroute_after_shut_noshut_of_upstream_interface_p1(request):
         )
         assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
-        result = iperfSendTraffic(tgen, src, _IGMP_JOIN_RANGE, 32, 2500)
+        result = app_helper.run_traffic(src, _IGMP_JOIN_RANGE, bind_intf=src_intf)
         assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
 
     step(
@@ -1139,7 +1140,7 @@ def test_verify_mroute_after_shut_noshut_of_upstream_interface_p1(request):
         assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
 
     step("Stop the traffic to all the receivers")
-    kill_iperf(tgen)
+    app_helper.stop_all_hosts()
 
     for data in input_dict:
         result = verify_ip_mroutes(
@@ -1176,7 +1177,7 @@ def test_verify_mroute_when_receiver_is_outside_frr_p0(request):
         pytest.skip(tgen.errors)
 
     # Creating configuration from JSON
-    kill_iperf(tgen)
+    app_helper.stop_all_hosts()
     clear_ip_mroute(tgen)
     reset_config_on_routers(tgen)
     clear_ip_pim_interface_traffic(tgen, topo)
@@ -1213,7 +1214,7 @@ def test_verify_mroute_when_receiver_is_outside_frr_p0(request):
     )
     assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
-    result = iperfSendIGMPJoin(tgen, "i1", _IGMP_JOIN_RANGE, join_interval=1)
+    result = app_helper.run_join("i1", _IGMP_JOIN_RANGE, "l1")
     assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
     step(
@@ -1225,7 +1226,7 @@ def test_verify_mroute_when_receiver_is_outside_frr_p0(request):
     )
     assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
-    result = iperfSendTraffic(tgen, "i2", _IGMP_JOIN_RANGE, 32, 2500)
+    result = app_helper.run_traffic("i2", _IGMP_JOIN_RANGE, "f1")
     assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
 
     step(
@@ -1233,7 +1234,7 @@ def test_verify_mroute_when_receiver_is_outside_frr_p0(request):
         " join (226.1.1.1-5) and (232.1.1.1-5)"
     )
     input_dict = {
-        "c2": {"igmp": {"interfaces": {"c2-i5-eth2": {"igmp": {"version": "2"}}}}}
+        "c2": {"igmp": {"interfaces": {"c2-i5-eth2": {"igmp": {"version": "2", "query": {"query-interval": 15}}}}}}
     }
     result = create_igmp_config(tgen, topo, input_dict)
     assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
@@ -1243,7 +1244,7 @@ def test_verify_mroute_when_receiver_is_outside_frr_p0(request):
     )
     assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
-    result = iperfSendIGMPJoin(tgen, "i5", _IGMP_JOIN_RANGE, join_interval=1)
+    result = app_helper.run_join("i5", _IGMP_JOIN_RANGE, "c2")
     assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
     step("FRR1 has 10 (*.G) and 10 (S,G) verify using 'show ip mroute count'")
@@ -1316,7 +1317,7 @@ def test_verify_mroute_when_FRR_is_FHR_and_LHR_p0(request):
         pytest.skip(tgen.errors)
 
     # Creating configuration from JSON
-    kill_iperf(tgen)
+    app_helper.stop_all_hosts()
     clear_ip_mroute(tgen)
     reset_config_on_routers(tgen)
     clear_ip_pim_interface_traffic(tgen, topo)
@@ -1356,22 +1357,22 @@ def test_verify_mroute_when_FRR_is_FHR_and_LHR_p0(request):
         "(226.1.1.1-5) and (232.1.1.1-5)"
     )
 
-    result = config_to_send_igmp_join_and_traffic(
-        tgen, topo, tc_name, "i1", "i1-l1-eth0", _GROUP_RANGE, join=True
-    )
-    assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
+    # result = config_to_send_igmp_join_and_traffic(
+    #     tgen, topo, tc_name, "i1", "i1-l1-eth0", _GROUP_RANGE, join=True
+    # )
+    # assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
-    result = config_to_send_igmp_join_and_traffic(
-        tgen, topo, tc_name, "i2", "i2-f1-eth0", _GROUP_RANGE, traffic=True
-    )
-    assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
+    # result = config_to_send_igmp_join_and_traffic(
+    #     tgen, topo, tc_name, "i2", "i2-f1-eth0", _GROUP_RANGE, traffic=True
+    # )
+    # assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
     step("Send IGMP join (226.1.1.1-5, 232.1.1.1-5) to LHR(l1)")
-    result = iperfSendIGMPJoin(tgen, "i1", _IGMP_JOIN_RANGE, join_interval=1)
+    result = app_helper.run_join("i1", _IGMP_JOIN_RANGE, "l1")
     assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
     step("Send multicast traffic from FRR3 to '226.1.1.1-5'" ", '232.1.1.1-5' receiver")
-    result = iperfSendTraffic(tgen, "i2", _IGMP_JOIN_RANGE, 32, 2500)
+    result = app_helper.run_traffic("i2", _IGMP_JOIN_RANGE, "f1")
     assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
 
     step(
@@ -1381,17 +1382,17 @@ def test_verify_mroute_when_FRR_is_FHR_and_LHR_p0(request):
 
     step("Configure one IGMP interface on f1 node and send IGMP" " join (225.1.1.1)")
     input_dict = {
-        "f1": {"igmp": {"interfaces": {"f1-i8-eth2": {"igmp": {"version": "2"}}}}}
+        "f1": {"igmp": {"interfaces": {"f1-i8-eth2": {"igmp": {"version": "2", "query": {"query-interval": 15}}}}}}
     }
     result = create_igmp_config(tgen, topo, input_dict)
     assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
-    result = config_to_send_igmp_join_and_traffic(
-        tgen, topo, tc_name, "i8", "i8-f1-eth0", _GROUP_RANGE, join=True
-    )
-    assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
+    # result = config_to_send_igmp_join_and_traffic(
+    #     tgen, topo, tc_name, "i8", "i8-f1-eth0", _GROUP_RANGE, join=True
+    # )
+    # assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
-    result = iperfSendIGMPJoin(tgen, "i8", _IGMP_JOIN_RANGE, join_interval=1)
+    result = app_helper.run_join("i8", _IGMP_JOIN_RANGE, "f1")
     assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
     step(
         "l1 and f1 has 10 IGMP groups (226.1.1.1-5, 232.1.1.1-5),"
@@ -1446,7 +1447,7 @@ def test_verify_mroute_when_FRR_is_FHR_and_LHR_p0(request):
 
     # Stop the multicast traffic
     step("Stop the traffic to all the receivers")
-    kill_iperf(tgen)
+    app_helper.stop_all_hosts()
 
     step(
         "After traffic stopped , verify (*,G) entries are not flushed"
@@ -1512,7 +1513,7 @@ def test_verify_mroute_when_5_different_receiver_joining_same_sources_p0(request
         pytest.skip(tgen.errors)
 
     # Creating configuration from JSON
-    kill_iperf(tgen)
+    app_helper.stop_all_hosts()
     clear_ip_mroute(tgen)
     reset_config_on_routers(tgen)
     clear_ip_pim_interface_traffic(tgen, topo)
@@ -1570,12 +1571,12 @@ def test_verify_mroute_when_5_different_receiver_joining_same_sources_p0(request
         "f1": {
             "igmp": {
                 "interfaces": {
-                    "f1-i8-eth2": {"igmp": {"version": "2"}},
-                    "f1-i2-eth1": {"igmp": {"version": "2"}},
+                    "f1-i8-eth2": {"igmp": {"version": "2", "query": {"query-interval": 15}}},
+                    "f1-i2-eth1": {"igmp": {"version": "2", "query": {"query-interval": 15}}},
                 }
             }
         },
-        "l1": {"igmp": {"interfaces": {"l1-i6-eth2": {"igmp": {"version": "2"}}}}},
+        "l1": {"igmp": {"interfaces": {"l1-i6-eth2": {"igmp": {"version": "2", "query": {"query-interval": 15}}}}}},
     }
     result = create_igmp_config(tgen, topo, input_dict)
     assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
@@ -1593,7 +1594,7 @@ def test_verify_mroute_when_5_different_receiver_joining_same_sources_p0(request
         )
         assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
-        result = iperfSendIGMPJoin(tgen, recvr, _IGMP_JOIN_RANGE, join_interval=1)
+        result = app_helper.run_join(recvr, _IGMP_JOIN_RANGE, join_intf=recvr_intf)
         assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
     step("Configure one source in FRR2 , one in c1")
     step(
@@ -1609,7 +1610,7 @@ def test_verify_mroute_when_5_different_receiver_joining_same_sources_p0(request
         )
         assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
-        result = iperfSendTraffic(tgen, src, _IGMP_JOIN_RANGE, 32, 2500)
+        result = app_helper.run_traffic(src, _IGMP_JOIN_RANGE, bind_intf=src_intf)
         assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
     step(
         "After all the IGMP groups received with correct port using"
@@ -1743,7 +1744,7 @@ def test_verify_oil_iif_for_mroute_after_shut_noshut_source_interface_p1(request
         pytest.skip(tgen.errors)
 
     # Creating configuration from JSON
-    kill_iperf(tgen)
+    app_helper.stop_all_hosts()
     clear_ip_mroute(tgen)
     reset_config_on_routers(tgen)
     clear_ip_pim_interface_traffic(tgen, topo)
@@ -1790,7 +1791,7 @@ def test_verify_oil_iif_for_mroute_after_shut_noshut_source_interface_p1(request
     )
 
     input_dict = {
-        "f1": {"igmp": {"interfaces": {"f1-i8-eth2": {"igmp": {"version": "2"}}}}}
+        "f1": {"igmp": {"interfaces": {"f1-i8-eth2": {"igmp": {"version": "2", "query": {"query-interval": 15}}}}}}
     }
     result = create_igmp_config(tgen, topo, input_dict)
     assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
@@ -1803,7 +1804,7 @@ def test_verify_oil_iif_for_mroute_after_shut_noshut_source_interface_p1(request
         )
         assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
-        result = iperfSendIGMPJoin(tgen, recvr, _IGMP_JOIN_RANGE, join_interval=1)
+        result = app_helper.run_join(recvr, _IGMP_JOIN_RANGE, join_intf=recvr_intf)
         assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
     step("Configure 1 source in FRR1 , 1 in FRR3")
@@ -1820,7 +1821,7 @@ def test_verify_oil_iif_for_mroute_after_shut_noshut_source_interface_p1(request
         )
         assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
-        result = iperfSendTraffic(tgen, src, _IGMP_JOIN_RANGE, 32, 2500)
+        result = app_helper.run_traffic(src, _IGMP_JOIN_RANGE, bind_intf=src_intf)
         assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
 
     step(
diff --git a/tests/topotests/multicast_pim_sm_topo3/test_multicast_pim_sm_topo3.py b/tests/topotests/multicast_pim_sm_topo3/test_multicast_pim_sm_topo3.py
index 7c04dba066..afefbfed10 100755
--- a/tests/topotests/multicast_pim_sm_topo3/test_multicast_pim_sm_topo3.py
+++ b/tests/topotests/multicast_pim_sm_topo3/test_multicast_pim_sm_topo3.py
@@ -74,11 +74,8 @@ from lib.common_config import (
     write_test_header,
     write_test_footer,
     step,
-    iperfSendIGMPJoin,
     addKernelRoute,
     reset_config_on_routers,
-    iperfSendTraffic,
-    kill_iperf,
     shutdown_bringup_interface,
     kill_router_daemons,
     start_router,
@@ -109,6 +106,7 @@ from lib.pim import (
     verify_pim_rp_info,
     get_refCount_for_mroute,
     verify_multicast_flag_state,
+    McastTesterHelper,
 )
 from lib.topolog import logger
 from lib.topojson import build_topo_from_json, build_config_from_json
@@ -212,6 +210,10 @@ def setup_module(mod):
     # Creating configuration from JSON
     build_config_from_json(tgen, topo)
 
+    # XXX Replace this using "with McastTesterHelper()... " in each test if possible.
+    global app_helper
+    app_helper = McastTesterHelper(tgen)
+
     logger.info("Running setup_module() done")
 
 
@@ -222,8 +224,7 @@ def teardown_module():
 
     tgen = get_topogen()
 
-    # Kill any iperfs we left running.
-    kill_iperf(tgen)
+    app_helper.cleanup()
 
     # Stop toplogy and Remove tmp files
     tgen.stop_topology()
@@ -433,7 +434,7 @@ def test_verify_oil_when_join_prune_sent_scenario_1_p1(request):
         pytest.skip(tgen.errors)
 
     # Creating configuration from JSON
-    kill_iperf(tgen)
+    app_helper.stop_all_hosts()
     clear_ip_mroute(tgen)
     reset_config_on_routers(tgen)
     clear_ip_pim_interface_traffic(tgen, topo)
@@ -451,7 +452,7 @@ def test_verify_oil_when_join_prune_sent_scenario_1_p1(request):
 
     intf_f1_i8 = topo["routers"]["f1"]["links"]["i8"]["interface"]
     input_dict = {
-        "f1": {"igmp": {"interfaces": {intf_f1_i8: {"igmp": {"version": "2"}}}}}
+        "f1": {"igmp": {"interfaces": {intf_f1_i8: {"igmp": {"version": "2", "query": {"query-interval": 15}}}}}}
     }
     result = create_igmp_config(tgen, topo, input_dict)
     assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
@@ -467,7 +468,7 @@ def test_verify_oil_when_join_prune_sent_scenario_1_p1(request):
         )
         assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
-        result = iperfSendIGMPJoin(tgen, recvr, IGMP_JOIN_RANGE_1, join_interval=1)
+        result = app_helper.run_join(recvr, IGMP_JOIN_RANGE_1, join_intf=recvr_intf)
         assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
     step("Configure static RP for (226.1.1.1-5) in R2")
@@ -502,7 +503,7 @@ def test_verify_oil_when_join_prune_sent_scenario_1_p1(request):
         )
         assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
-        result = iperfSendTraffic(tgen, src, IGMP_JOIN_RANGE_1, 32, 2500)
+        result = app_helper.run_traffic(src, IGMP_JOIN_RANGE_1, bind_intf=src_intf)
         assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
 
     source_i2 = topo["routers"]["i2"]["links"]["f1"]["ipv4"].split("/")[0]
@@ -808,7 +809,7 @@ def test_verify_oil_when_join_prune_sent_scenario_2_p1(request):
         pytest.skip(tgen.errors)
 
     # Creating configuration from JSON
-    kill_iperf(tgen)
+    app_helper.stop_all_hosts()
     clear_ip_mroute(tgen)
     reset_config_on_routers(tgen)
     clear_ip_pim_interface_traffic(tgen, topo)
@@ -835,7 +836,7 @@ def test_verify_oil_when_join_prune_sent_scenario_2_p1(request):
 
     intf_r2_i3 = topo["routers"]["r2"]["links"]["i3"]["interface"]
     input_dict = {
-        "r2": {"igmp": {"interfaces": {intf_r2_i3: {"igmp": {"version": "2"}}}}}
+        "r2": {"igmp": {"interfaces": {intf_r2_i3: {"igmp": {"version": "2", "query": {"query-interval": 15}}}}}}
     }
     result = create_igmp_config(tgen, topo, input_dict)
     assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
@@ -851,7 +852,7 @@ def test_verify_oil_when_join_prune_sent_scenario_2_p1(request):
         )
         assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
-        result = iperfSendIGMPJoin(tgen, recvr, IGMP_JOIN_RANGE_1, join_interval=1)
+        result = app_helper.run_join(recvr, IGMP_JOIN_RANGE_1, join_intf=recvr_intf)
         assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
     step("Configure static RP for (226.1.1.1-5) in R2")
@@ -1050,7 +1051,7 @@ def test_shut_noshut_source_interface_when_upstream_cleared_from_LHR_p1(request)
         pytest.skip(tgen.errors)
 
     # Creating configuration from JSON
-    kill_iperf(tgen)
+    app_helper.stop_all_hosts()
     clear_ip_mroute(tgen)
     reset_config_on_routers(tgen)
     clear_ip_pim_interface_traffic(tgen, topo)
@@ -1067,7 +1068,7 @@ def test_shut_noshut_source_interface_when_upstream_cleared_from_LHR_p1(request)
         )
         assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
-        result = iperfSendIGMPJoin(tgen, recvr, IGMP_JOIN_RANGE_1, join_interval=1)
+        result = app_helper.run_join(recvr, IGMP_JOIN_RANGE_1, join_intf=recvr_intf)
         assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
     step("Configure RP on R2 (loopback interface) for " "the group range 225.0.0.0/8")
@@ -1100,7 +1101,7 @@ def test_shut_noshut_source_interface_when_upstream_cleared_from_LHR_p1(request)
         )
         assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
-        result = iperfSendTraffic(tgen, src, IGMP_JOIN_RANGE_1, 32, 2500)
+        result = app_helper.run_traffic(src, IGMP_JOIN_RANGE_1, bind_intf=src_intf)
         assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
 
     step(
@@ -1267,7 +1268,7 @@ def test_shut_noshut_receiver_interface_when_upstream_cleared_from_LHR_p1(reques
         pytest.skip(tgen.errors)
 
     # Creating configuration from JSON
-    kill_iperf(tgen)
+    app_helper.stop_all_hosts()
     clear_ip_mroute(tgen)
     reset_config_on_routers(tgen)
     clear_ip_pim_interface_traffic(tgen, topo)
@@ -1284,7 +1285,7 @@ def test_shut_noshut_receiver_interface_when_upstream_cleared_from_LHR_p1(reques
         )
         assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
-        result = iperfSendIGMPJoin(tgen, recvr, IGMP_JOIN_RANGE_1, join_interval=1)
+        result = app_helper.run_join(recvr, IGMP_JOIN_RANGE_1, join_intf=recvr_intf)
         assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
     step("Configure RP on R2 (loopback interface) for " "the group range 225.0.0.0/8")
@@ -1317,7 +1318,7 @@ def test_shut_noshut_receiver_interface_when_upstream_cleared_from_LHR_p1(reques
         )
         assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
-        result = iperfSendTraffic(tgen, src, IGMP_JOIN_RANGE_1, 32, 2500)
+        result = app_helper.run_traffic(src, IGMP_JOIN_RANGE_1, bind_intf=src_intf)
         assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
 
     step(
@@ -1469,7 +1470,7 @@ def test_verify_remove_add_igmp_config_to_receiver_interface_p0(request):
         pytest.skip(tgen.errors)
 
     # Creating configuration from JSON
-    kill_iperf(tgen)
+    app_helper.stop_all_hosts()
     clear_ip_mroute(tgen)
     reset_config_on_routers(tgen)
     clear_ip_pim_interface_traffic(tgen, topo)
@@ -1486,7 +1487,7 @@ def test_verify_remove_add_igmp_config_to_receiver_interface_p0(request):
         )
         assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
-        result = iperfSendIGMPJoin(tgen, recvr, IGMP_JOIN_RANGE_1, join_interval=1)
+        result = app_helper.run_join(recvr, IGMP_JOIN_RANGE_1, join_intf=recvr_intf)
         assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
     step("Configure RP for (226.1.1.1-5) and (232.1.1.1-5) in cisco-1(f1)")
@@ -1519,7 +1520,7 @@ def test_verify_remove_add_igmp_config_to_receiver_interface_p0(request):
         )
         assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
-        result = iperfSendTraffic(tgen, src, IGMP_JOIN_RANGE_1, 32, 2500)
+        result = app_helper.run_traffic(src, IGMP_JOIN_RANGE_1, bind_intf=src_intf)
         assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
 
     step("Configure source on FRR1 and start the traffic for" " (225.1.1.1-225.1.1.10)")
@@ -1532,7 +1533,7 @@ def test_verify_remove_add_igmp_config_to_receiver_interface_p0(request):
         )
         assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
-        result = iperfSendTraffic(tgen, src, IGMP_JOIN_RANGE_1, 32, 2500)
+        result = app_helper.run_traffic(src, IGMP_JOIN_RANGE_1, bind_intf=src_intf)
         assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
 
     source_i6 = topo["routers"]["i6"]["links"]["l1"]["ipv4"].split("/")[0]
@@ -1639,7 +1640,7 @@ def test_verify_remove_add_igmp_config_to_receiver_interface_p0(request):
     )
 
     input_dict_2 = {
-        "l1": {"igmp": {"interfaces": {intf_l1_i1: {"igmp": {"version": "2"}}}}}
+        "l1": {"igmp": {"interfaces": {intf_l1_i1: {"igmp": {"version": "2", "query": {"query-interval": 15}}}}}}
     }
     result = create_igmp_config(tgen, topo, input_dict_2)
     assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
@@ -1722,7 +1723,7 @@ def test_verify_remove_add_igmp_config_to_receiver_interface_p0(request):
     )
 
     input_dict_2 = {
-        "l1": {"igmp": {"interfaces": {intf_l1_i1: {"igmp": {"version": "2"}}}}}
+        "l1": {"igmp": {"interfaces": {intf_l1_i1: {"igmp": {"version": "2", "query": {"query-interval": 15}}}}}}
     }
     result = create_igmp_config(tgen, topo, input_dict_2)
     assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
@@ -1845,7 +1846,7 @@ def test_verify_remove_add_igmp_commands_when_pim_configured_p0(request):
         pytest.skip(tgen.errors)
 
     # Creating configuration from JSON
-    kill_iperf(tgen)
+    app_helper.stop_all_hosts()
     clear_ip_mroute(tgen)
     reset_config_on_routers(tgen)
     clear_ip_pim_interface_traffic(tgen, topo)
@@ -1862,7 +1863,7 @@ def test_verify_remove_add_igmp_commands_when_pim_configured_p0(request):
         )
         assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
-        result = iperfSendIGMPJoin(tgen, recvr, IGMP_JOIN_RANGE_1, join_interval=1)
+        result = app_helper.run_join(recvr, IGMP_JOIN_RANGE_1, join_intf=recvr_intf)
         assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
     step("Configure RP for (226.1.1.1-5) and (232.1.1.1-5) in cisco-1(f1)")
@@ -1895,7 +1896,7 @@ def test_verify_remove_add_igmp_commands_when_pim_configured_p0(request):
         )
         assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
-        result = iperfSendTraffic(tgen, src, IGMP_JOIN_RANGE_1, 32, 2500)
+        result = app_helper.run_traffic(src, IGMP_JOIN_RANGE_1, bind_intf=src_intf)
         assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
 
     step("Configure source on FRR1 and start the traffic for" " (225.1.1.1-225.1.1.10)")
@@ -1908,7 +1909,7 @@ def test_verify_remove_add_igmp_commands_when_pim_configured_p0(request):
         )
         assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
-        result = iperfSendTraffic(tgen, src, IGMP_JOIN_RANGE_1, 32, 2500)
+        result = app_helper.run_traffic(src, IGMP_JOIN_RANGE_1, bind_intf=src_intf)
         assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
 
     source_i6 = topo["routers"]["i6"]["links"]["l1"]["ipv4"].split("/")[0]
@@ -2144,7 +2145,7 @@ def test_verify_remove_add_pim_commands_when_igmp_configured_p1(request):
         pytest.skip(tgen.errors)
 
     # Creating configuration from JSON
-    kill_iperf(tgen)
+    app_helper.stop_all_hosts()
     clear_ip_mroute(tgen)
     reset_config_on_routers(tgen)
     clear_ip_pim_interface_traffic(tgen, topo)
@@ -2162,7 +2163,7 @@ def test_verify_remove_add_pim_commands_when_igmp_configured_p1(request):
         )
         assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
-        result = iperfSendIGMPJoin(tgen, recvr, IGMP_JOIN_RANGE_1, join_interval=1)
+        result = app_helper.run_join(recvr, IGMP_JOIN_RANGE_1, join_intf=recvr_intf)
         assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
     step("Configure RP for (226.1.1.1-5) and (232.1.1.1-5) in cisco-1(f1)")
@@ -2339,7 +2340,7 @@ def test_pim_dr_priority_p0(request):
         pytest.skip(tgen.errors)
 
     # Creating configuration from JSON
-    kill_iperf(tgen)
+    app_helper.stop_all_hosts()
     clear_ip_mroute(tgen)
     reset_config_on_routers(tgen)
     clear_ip_pim_interface_traffic(tgen, topo)
@@ -2357,7 +2358,7 @@ def test_pim_dr_priority_p0(request):
         )
         assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
-        result = iperfSendIGMPJoin(tgen, recvr, IGMP_JOIN_RANGE_1, join_interval=1)
+        result = app_helper.run_join(recvr, IGMP_JOIN_RANGE_1, join_intf=recvr_intf)
         assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
     step("Configure RP for (226.1.1.1-5) and (232.1.1.1-5) in cisco-1(f1)")
@@ -2388,7 +2389,7 @@ def test_pim_dr_priority_p0(request):
         )
         assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
-        result = iperfSendTraffic(tgen, src, IGMP_JOIN_RANGE_1, 32, 2500)
+        result = app_helper.run_traffic(src, IGMP_JOIN_RANGE_1, bind_intf=src_intf)
         assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
 
     source_i2 = topo["routers"]["i2"]["links"]["f1"]["ipv4"].split("/")[0]
@@ -2622,7 +2623,7 @@ def test_pim_hello_timer_p1(request):
         pytest.skip(tgen.errors)
 
     # Creating configuration from JSON
-    kill_iperf(tgen)
+    app_helper.stop_all_hosts()
     clear_ip_mroute(tgen)
     reset_config_on_routers(tgen)
     clear_ip_pim_interface_traffic(tgen, topo)
@@ -2640,7 +2641,7 @@ def test_pim_hello_timer_p1(request):
         )
         assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
-        result = iperfSendIGMPJoin(tgen, recvr, IGMP_JOIN_RANGE_1, join_interval=1)
+        result = app_helper.run_join(recvr, IGMP_JOIN_RANGE_1, join_intf=recvr_intf)
         assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
     step("Configure RP for (226.1.1.1-5) and (232.1.1.1-5) in cisco-1(f1)")
@@ -2742,7 +2743,7 @@ def test_mroute_after_removing_RP_sending_IGMP_prune_p2(request):
         pytest.skip(tgen.errors)
 
     # Creating configuration from JSON
-    kill_iperf(tgen)
+    app_helper.stop_all_hosts()
     clear_ip_mroute(tgen)
     reset_config_on_routers(tgen)
     clear_ip_pim_interface_traffic(tgen, topo)
@@ -2766,7 +2767,7 @@ def test_mroute_after_removing_RP_sending_IGMP_prune_p2(request):
 
     intf_f1_i8 = topo["routers"]["f1"]["links"]["i8"]["interface"]
     input_dict = {
-        "f1": {"igmp": {"interfaces": {intf_f1_i8: {"igmp": {"version": "2"}}}}}
+        "f1": {"igmp": {"interfaces": {intf_f1_i8: {"igmp": {"version": "2", "query": {"query-interval": 15}}}}}}
     }
     result = create_igmp_config(tgen, topo, input_dict)
     assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
@@ -2779,7 +2780,7 @@ def test_mroute_after_removing_RP_sending_IGMP_prune_p2(request):
         )
         assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
-        result = iperfSendIGMPJoin(tgen, recvr, IGMP_JOIN_RANGE_1, join_interval=1)
+        result = app_helper.run_join(recvr, IGMP_JOIN_RANGE_1, join_intf=recvr_intf)
         assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
     step("Configure static RP for (225.1.1.1-5) as R2")
@@ -2815,7 +2816,7 @@ def test_mroute_after_removing_RP_sending_IGMP_prune_p2(request):
         )
         assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
-        result = iperfSendTraffic(tgen, src, IGMP_JOIN_RANGE_1, 32, 2500)
+        result = app_helper.run_traffic(src, IGMP_JOIN_RANGE_1, bind_intf=src_intf)
         assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
 
     source_i2 = topo["routers"]["i6"]["links"]["l1"]["ipv4"].split("/")[0]
@@ -2934,7 +2935,7 @@ def test_mroute_after_removing_RP_sending_IGMP_prune_p2(request):
         assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
 
     step("Send prune from receiver-1 (using ctrl+c) on iperf interface")
-    kill_iperf(tgen)
+    app_helper.stop_all_hosts()
 
     intf_f1_i8 = topo["routers"]["f1"]["links"]["i8"]["interface"]
     input_traffic = {"f1": {"traffic_sent": [intf_f1_i8]}}
@@ -3005,7 +3006,7 @@ def test_mroute_after_removing_RP_sending_IGMP_prune_p2(request):
         )
         assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
-        result = iperfSendIGMPJoin(tgen, recvr, IGMP_JOIN_RANGE_1, join_interval=1)
+        result = app_helper.run_join(recvr, IGMP_JOIN_RANGE_1, join_intf=recvr_intf)
         assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
     for data in input_dict_starg:
@@ -3027,7 +3028,7 @@ def test_mroute_after_removing_RP_sending_IGMP_prune_p2(request):
         )
         assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
-        result = iperfSendTraffic(tgen, src, IGMP_JOIN_RANGE_1, 32, 2500)
+        result = app_helper.run_traffic(src, IGMP_JOIN_RANGE_1, bind_intf=src_intf)
         assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
 
     source_i2 = topo["routers"]["i6"]["links"]["l1"]["ipv4"].split("/")[0]
@@ -3060,7 +3061,7 @@ def test_prune_sent_to_LHR_and_FHR_when_PIMnbr_down_p2(request):
         pytest.skip(tgen.errors)
 
     # Creating configuration from JSON
-    kill_iperf(tgen)
+    app_helper.stop_all_hosts()
     clear_ip_mroute(tgen)
     reset_config_on_routers(tgen)
     clear_ip_pim_interface_traffic(tgen, topo)
@@ -3084,7 +3085,7 @@ def test_prune_sent_to_LHR_and_FHR_when_PIMnbr_down_p2(request):
 
     intf_f1_i8 = topo["routers"]["f1"]["links"]["i8"]["interface"]
     input_dict = {
-        "f1": {"igmp": {"interfaces": {intf_f1_i8: {"igmp": {"version": "2"}}}}}
+        "f1": {"igmp": {"interfaces": {intf_f1_i8: {"igmp": {"version": "2", "query": {"query-interval": 15}}}}}}
     }
     result = create_igmp_config(tgen, topo, input_dict)
     assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
@@ -3097,7 +3098,7 @@ def test_prune_sent_to_LHR_and_FHR_when_PIMnbr_down_p2(request):
         )
         assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
-        result = iperfSendIGMPJoin(tgen, recvr, IGMP_JOIN_RANGE_1, join_interval=1)
+        result = app_helper.run_join(recvr, IGMP_JOIN_RANGE_1, join_intf=recvr_intf)
         assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
     step("Configure static RP for (225.1.1.1-5) as R2")
@@ -3136,7 +3137,7 @@ def test_prune_sent_to_LHR_and_FHR_when_PIMnbr_down_p2(request):
         )
         assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
-        result = iperfSendTraffic(tgen, src, IGMP_JOIN_RANGE_1, 32, 2500)
+        result = app_helper.run_traffic(src, IGMP_JOIN_RANGE_1, bind_intf=src_intf)
         assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
 
     source_i2 = topo["routers"]["i6"]["links"]["l1"]["ipv4"].split("/")[0]
@@ -3390,7 +3391,7 @@ def test_prune_sent_to_LHR_and_FHR_when_PIMnbr_down_p2(request):
     intf_r2_l1 = topo["routers"]["r2"]["links"]["l1"]["interface"]
     shutdown_bringup_interface(tgen, "r2", intf_r2_l1, False)
 
-    kill_iperf(tgen, dut="i2", action="remove_traffic")
+    app_helper.stop_host("i2")
 
     step("Verify RP info after Shut the link from FHR to RP from RP node")
     dut = "l1"
@@ -3542,7 +3543,7 @@ def test_prune_sent_to_LHR_and_FHR_when_PIMnbr_down_p2(request):
 
     step("Verify PIM Nbrs after Shut the link from FHR to RP from FHR node")
 
-    kill_iperf(tgen, dut="i6", action="remove_traffic")
+    app_helper.stop_host("i6")
 
     step("Verify RP info after Shut the link from FHR to RP from FHR node")
     dut = "l1"
@@ -3702,7 +3703,7 @@ def test_mroute_flags_p1(request):
         pytest.skip(tgen.errors)
 
     # Creating configuration from JSON
-    kill_iperf(tgen)
+    app_helper.stop_all_hosts()
     clear_ip_mroute(tgen)
     reset_config_on_routers(tgen)
     clear_ip_pim_interface_traffic(tgen, topo)
@@ -3726,7 +3727,7 @@ def test_mroute_flags_p1(request):
 
     intf_f1_i8 = topo["routers"]["f1"]["links"]["i8"]["interface"]
     input_dict = {
-        "f1": {"igmp": {"interfaces": {intf_f1_i8: {"igmp": {"version": "2"}}}}}
+        "f1": {"igmp": {"interfaces": {intf_f1_i8: {"igmp": {"version": "2", "query": {"query-interval": 15}}}}}}
     }
     result = create_igmp_config(tgen, topo, input_dict)
     assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
@@ -3739,7 +3740,7 @@ def test_mroute_flags_p1(request):
         )
         assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
-        result = iperfSendIGMPJoin(tgen, recvr, IGMP_JOIN_RANGE_1, join_interval=1)
+        result = app_helper.run_join(recvr, IGMP_JOIN_RANGE_1, join_intf=recvr_intf)
         assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
     step("Configure static RP for (225.1.1.1-5) as R2")
@@ -3778,7 +3779,7 @@ def test_mroute_flags_p1(request):
         )
         assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
-        result = iperfSendTraffic(tgen, src, IGMP_JOIN_RANGE_1, 32, 2500)
+        result = app_helper.run_traffic(src, IGMP_JOIN_RANGE_1, bind_intf=src_intf)
         assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
 
     source_i2 = topo["routers"]["i6"]["links"]["l1"]["ipv4"].split("/")[0]
@@ -3865,7 +3866,7 @@ def test_verify_multicast_traffic_when_LHR_connected_to_RP_p1(request):
         pytest.skip(tgen.errors)
 
     # Creating configuration from JSON
-    kill_iperf(tgen)
+    app_helper.stop_all_hosts()
     clear_ip_mroute(tgen)
     reset_config_on_routers(tgen)
     clear_ip_pim_interface_traffic(tgen, topo)
@@ -3907,7 +3908,7 @@ def test_verify_multicast_traffic_when_LHR_connected_to_RP_p1(request):
 
     intf_r2_i3 = topo["routers"]["r2"]["links"]["i3"]["interface"]
     input_dict = {
-        "r2": {"igmp": {"interfaces": {intf_r2_i3: {"igmp": {"version": "2"}}}}}
+        "r2": {"igmp": {"interfaces": {intf_r2_i3: {"igmp": {"version": "2", "query": {"query-interval": 15}}}}}}
     }
     result = create_igmp_config(tgen, topo, input_dict)
     assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
@@ -3923,7 +3924,7 @@ def test_verify_multicast_traffic_when_LHR_connected_to_RP_p1(request):
         )
         assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
-        result = iperfSendIGMPJoin(tgen, recvr, _IGMP_JOIN_RANGE, join_interval=1)
+        result = app_helper.run_join(recvr, _IGMP_JOIN_RANGE, join_intf=recvr_intf)
         assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
     step("Configure RP for (226.1.1.1-5) and (232.1.1.1-5) in (f1)")
@@ -3956,7 +3957,7 @@ def test_verify_multicast_traffic_when_LHR_connected_to_RP_p1(request):
         )
         assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
-        result = iperfSendTraffic(tgen, src, _IGMP_JOIN_RANGE, 32, 2500)
+        result = app_helper.run_traffic(src, _IGMP_JOIN_RANGE, bind_intf=src_intf)
         assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
 
     step(
@@ -4327,7 +4328,7 @@ def test_verify_multicast_traffic_when_FHR_connected_to_RP_p1(request):
         pytest.skip(tgen.errors)
 
     # Creating configuration from JSON
-    kill_iperf(tgen)
+    app_helper.stop_all_hosts()
     clear_ip_mroute(tgen)
     reset_config_on_routers(tgen)
     clear_ip_pim_interface_traffic(tgen, topo)
@@ -4357,7 +4358,7 @@ def test_verify_multicast_traffic_when_FHR_connected_to_RP_p1(request):
         )
         assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
-        result = iperfSendIGMPJoin(tgen, recvr, _IGMP_JOIN_RANGE, join_interval=1)
+        result = app_helper.run_join(recvr, _IGMP_JOIN_RANGE, join_intf=recvr_intf)
         assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
     step("Configure RP for (225.1.1.1-5) in (f1)")
@@ -4390,7 +4391,7 @@ def test_verify_multicast_traffic_when_FHR_connected_to_RP_p1(request):
         )
         assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
-        result = iperfSendTraffic(tgen, src, _IGMP_JOIN_RANGE, 32, 2500)
+        result = app_helper.run_traffic(src, _IGMP_JOIN_RANGE, bind_intf=src_intf)
         assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
 
     step(
diff --git a/tests/topotests/multicast_pim_sm_topo3/test_multicast_pim_sm_topo4.py b/tests/topotests/multicast_pim_sm_topo3/test_multicast_pim_sm_topo4.py
index 445dab8373..dbd1dcec47 100755
--- a/tests/topotests/multicast_pim_sm_topo3/test_multicast_pim_sm_topo4.py
+++ b/tests/topotests/multicast_pim_sm_topo3/test_multicast_pim_sm_topo4.py
@@ -68,11 +68,8 @@ from lib.common_config import (
     write_test_header,
     write_test_footer,
     step,
-    iperfSendIGMPJoin,
     addKernelRoute,
     reset_config_on_routers,
-    iperfSendTraffic,
-    kill_iperf,
     shutdown_bringup_interface,
     start_router,
     stop_router,
@@ -97,6 +94,7 @@ from lib.pim import (
     verify_pim_rp_info,
     verify_pim_interface_traffic,
     verify_igmp_interface,
+    McastTesterHelper,
 )
 from lib.topolog import logger
 from lib.topojson import build_topo_from_json, build_config_from_json
@@ -180,6 +178,10 @@ def setup_module(mod):
     # Creating configuration from JSON
     build_config_from_json(tgen, topo)
 
+    # XXX Replace this using "with McastTesterHelper()... " in each test if possible.
+    global app_helper
+    app_helper = McastTesterHelper(tgen)
+
     logger.info("Running setup_module() done")
 
 
@@ -190,8 +192,7 @@ def teardown_module():
 
     tgen = get_topogen()
 
-    # Kill any iperfs we left running.
-    kill_iperf(tgen)
+    app_helper.cleanup()
 
     # Stop toplogy and Remove tmp files
     tgen.stop_topology()
@@ -311,7 +312,7 @@ def test_mroute_when_RP_reachable_default_route_p2(request):
         pytest.skip(tgen.errors)
 
     # Creating configuration from JSON
-    kill_iperf(tgen)
+    app_helper.stop_all_hosts()
     clear_ip_mroute(tgen)
     reset_config_on_routers(tgen)
     clear_ip_pim_interface_traffic(tgen, topo)
@@ -347,7 +348,7 @@ def test_mroute_when_RP_reachable_default_route_p2(request):
         )
         assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
-        result = iperfSendIGMPJoin(tgen, recvr, IGMP_JOIN_RANGE_1, join_interval=1)
+        result = app_helper.run_join(recvr, IGMP_JOIN_RANGE_1, join_intf=recvr_intf)
         assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
     step("Configure static RP for (225.1.1.1-5) as R2")
@@ -380,7 +381,7 @@ def test_mroute_when_RP_reachable_default_route_p2(request):
         )
         assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
-        result = iperfSendTraffic(tgen, src, IGMP_JOIN_RANGE_1, 32, 2500)
+        result = app_helper.run_traffic(src, IGMP_JOIN_RANGE_1, bind_intf=src_intf)
         assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
 
     source_i4 = topo["routers"]["i4"]["links"]["c1"]["ipv4"].split("/")[0]
@@ -610,7 +611,7 @@ def test_mroute_with_RP_default_route_all_nodes_p2(request):
         pytest.skip(tgen.errors)
 
     # Creating configuration from JSON
-    kill_iperf(tgen)
+    app_helper.stop_all_hosts()
     clear_ip_mroute(tgen)
     reset_config_on_routers(tgen)
     clear_ip_pim_interface_traffic(tgen, topo)
@@ -646,7 +647,7 @@ def test_mroute_with_RP_default_route_all_nodes_p2(request):
         )
         assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
-        result = iperfSendIGMPJoin(tgen, recvr, IGMP_JOIN_RANGE_1, join_interval=1)
+        result = app_helper.run_join(recvr, IGMP_JOIN_RANGE_1, join_intf=recvr_intf)
         assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
     step("Configure static RP for (225.1.1.1-5) as R2")
@@ -679,7 +680,7 @@ def test_mroute_with_RP_default_route_all_nodes_p2(request):
         )
         assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
-        result = iperfSendTraffic(tgen, src, IGMP_JOIN_RANGE_1, 32, 2500)
+        result = app_helper.run_traffic(src, IGMP_JOIN_RANGE_1, bind_intf=src_intf)
         assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
 
     source_i5 = topo["routers"]["i5"]["links"]["c2"]["ipv4"].split("/")[0]
@@ -898,7 +899,7 @@ def test_PIM_hello_tx_rx_p1(request):
         pytest.skip(tgen.errors)
 
     # Creating configuration from JSON
-    kill_iperf(tgen)
+    app_helper.stop_all_hosts()
     clear_ip_mroute(tgen)
     reset_config_on_routers(tgen)
     clear_ip_pim_interface_traffic(tgen, topo)
@@ -934,7 +935,7 @@ def test_PIM_hello_tx_rx_p1(request):
         )
         assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
-        result = iperfSendIGMPJoin(tgen, recvr, IGMP_JOIN_RANGE_1, join_interval=1)
+        result = app_helper.run_join(recvr, IGMP_JOIN_RANGE_1, join_intf=recvr_intf)
         assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
     step("Configure static RP for (225.1.1.1-5) as R2")
@@ -967,7 +968,7 @@ def test_PIM_hello_tx_rx_p1(request):
         )
         assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 
-        result = iperfSendTraffic(tgen, src, IGMP_JOIN_RANGE_1, 32, 2500)
+        result = app_helper.run_traffic(src, IGMP_JOIN_RANGE_1, bind_intf=src_intf)
         assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
 
     source_i5 = topo["routers"]["i5"]["links"]["c2"]["ipv4"].split("/")[0]
diff --git a/tests/topotests/multicast_pim_static_rp_topo1/test_multicast_pim_static_rp.py b/tests/topotests/multicast_pim_static_rp_topo1/test_multicast_pim_static_rp.py
index b1ffd37865..d0ab12e5cc 100755
--- a/tests/topotests/multicast_pim_static_rp_topo1/test_multicast_pim_static_rp.py
+++ b/tests/topotests/multicast_pim_static_rp_topo1/test_multicast_pim_static_rp.py
@@ -126,15 +126,13 @@ from lib.common_config import (
     write_test_footer,
     reset_config_on_routers,
     step,
-    iperfSendIGMPJoin,
-    iperfSendTraffic,
     addKernelRoute,
     shutdown_bringup_interface,
     kill_router_daemons,
     start_router_daemons,
     create_static_routes,
-    kill_iperf,
     topo_daemons,
+
 )
 from lib.pim import (
     create_pim_config,
@@ -151,6 +149,7 @@ from lib.pim import (
     clear_ip_pim_interfaces,
     clear_ip_mroute,
     clear_ip_mroute_verify,
+    McastTesterHelper,
 )
 
 pytestmark = [pytest.mark.pimd, pytest.mark.staticd]
@@ -246,6 +245,10 @@ def setup_module(mod):
     result = verify_pim_neighbors(tgen, TOPO)
     assert result is True, "setup_module :Failed \n Error:" " {}".format(result)
 
+    # XXX Replace this using "with McastTesterHelper()... " in each test if possible.
+    global app_helper
+    app_helper = McastTesterHelper(tgen)
+
     logger.info("Running setup_module() done")
 
 
@@ -256,8 +259,7 @@ def teardown_module():
 
     tgen = get_topogen()
 
-    # Kill any iperfs we left running.
-    kill_iperf(tgen)
+    app_helper.cleanup()
 
     # Stop toplogy and Remove tmp files
     tgen.stop_topology()
@@ -432,7 +434,7 @@ def test_add_delete_static_RP_p0(request):
     )
 
     step("r0 : Send IGMP join")
-    result = iperfSendIGMPJoin(tgen, "r0", GROUP_ADDRESS, join_interval=1)
+    result = app_helper.run_join("r0", GROUP_ADDRESS, "r1")
     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 
     step("r1: Verify IGMP groups")
@@ -566,7 +568,7 @@ def test_SPT_RPT_path_same_p1(request):
 
     step("Creating configuration from JSON")
     reset_config_on_routers(tgen)
-    kill_iperf(tgen)
+    app_helper.stop_all_hosts()
     clear_ip_mroute(tgen)
     clear_ip_pim_interface_traffic(tgen, TOPO)
 
@@ -601,7 +603,7 @@ def test_SPT_RPT_path_same_p1(request):
     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 
     step("r0: Send IGMP join")
-    result = iperfSendIGMPJoin(tgen, "r0", GROUP_ADDRESS, join_interval=1)
+    result = app_helper.run_join("r0", GROUP_ADDRESS, "r1")
     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 
     step("r1: Verify IGMP groups")
@@ -611,7 +613,7 @@ def test_SPT_RPT_path_same_p1(request):
     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 
     step("r5: Send multicast traffic for group 225.1.1.1")
-    result = iperfSendTraffic(tgen, "r5", GROUP_ADDRESS, 32, 2500)
+    result = app_helper.run_traffic("r5", GROUP_ADDRESS, "r3")
     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 
     step("r1: Verify (*, G) upstream IIF interface")
@@ -722,7 +724,7 @@ def test_not_reachable_static_RP_p0(request):
 
     step("Creating configuration from JSON")
     reset_config_on_routers(tgen)
-    kill_iperf(tgen)
+    app_helper.stop_all_hosts()
     clear_ip_mroute(tgen)
     clear_ip_pim_interface_traffic(tgen, TOPO)
 
@@ -755,7 +757,7 @@ def test_not_reachable_static_RP_p0(request):
     step("Enable PIM between r1 and r2")
 
     step("r0 : Send IGMP join")
-    result = iperfSendIGMPJoin(tgen, "r0", GROUP_ADDRESS, join_interval=1)
+    result = app_helper.run_join("r0", GROUP_ADDRESS, "r1")
     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 
     step("r1 : Verify rp info")
@@ -888,7 +890,7 @@ def test_add_RP_after_join_received_p1(request):
 
     step("Creating configuration from JSON")
     reset_config_on_routers(tgen)
-    kill_iperf(tgen)
+    app_helper.stop_all_hosts()
     clear_ip_mroute(tgen)
     clear_ip_pim_interface_traffic(tgen, TOPO)
 
@@ -942,7 +944,7 @@ def test_add_RP_after_join_received_p1(request):
     )
 
     step("r0 : Send IGMP join (225.1.1.1) to r1, when rp is not configured" "in r1")
-    result = iperfSendIGMPJoin(tgen, "r0", GROUP_ADDRESS, join_interval=1)
+    result = app_helper.run_join("r0", GROUP_ADDRESS, "r1")
     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 
     step("r1: IGMP group is received on R1 verify using show ip igmp groups")
@@ -1058,7 +1060,7 @@ def test_reachable_static_RP_after_join_p0(request):
 
     step("Creating configuration from JSON")
     reset_config_on_routers(tgen)
-    kill_iperf(tgen)
+    app_helper.stop_all_hosts()
     clear_ip_mroute(tgen)
     clear_ip_pim_interface_traffic(tgen, TOPO)
 
@@ -1096,7 +1098,7 @@ def test_reachable_static_RP_after_join_p0(request):
     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 
     step("r1 : Send IGMP join for 225.1.1.1")
-    result = iperfSendIGMPJoin(tgen, "r0", GROUP_ADDRESS, join_interval=1)
+    result = app_helper.run_join("r0", GROUP_ADDRESS, "r1")
     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 
     step("r1 : Verify IGMP groups")
@@ -1225,7 +1227,7 @@ def test_send_join_on_higher_preffered_rp_p1(request):
 
     step("Creating configuration from JSON")
     reset_config_on_routers(tgen)
-    kill_iperf(tgen)
+    app_helper.stop_all_hosts()
     clear_ip_mroute(tgen)
     clear_ip_pim_interface_traffic(tgen, TOPO)
 
@@ -1269,7 +1271,7 @@ def test_send_join_on_higher_preffered_rp_p1(request):
     )
 
     step("r0 : Send IGMP join for 225.1.1.1")
-    result = iperfSendIGMPJoin(tgen, "r0", GROUP_ADDRESS, join_interval=1)
+    result = app_helper.run_join("r0", GROUP_ADDRESS, "r1")
     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 
     step("r1 : Verify IGMP groups")
@@ -1466,7 +1468,7 @@ def test_RP_configured_as_LHR_1_p1(request):
 
     step("Creating configuration from JSON")
     reset_config_on_routers(tgen)
-    kill_iperf(tgen)
+    app_helper.stop_all_hosts()
     clear_ip_mroute(tgen)
     clear_ip_pim_interface_traffic(tgen, TOPO)
 
@@ -1593,7 +1595,7 @@ def test_RP_configured_as_LHR_1_p1(request):
     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 
     step("r0: Send IGMP join")
-    result = iperfSendIGMPJoin(tgen, "r0", GROUP_ADDRESS, join_interval=1)
+    result = app_helper.run_join("r0", GROUP_ADDRESS, "r1")
     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 
     step("r1: Verify IGMP groups")
@@ -1602,7 +1604,7 @@ def test_RP_configured_as_LHR_1_p1(request):
     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 
     step("r5: Send multicast traffic for group 225.1.1.1")
-    result = iperfSendTraffic(tgen, "r5", GROUP_ADDRESS, 32, 2500)
+    result = app_helper.run_traffic("r5", GROUP_ADDRESS, "r3")
     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 
     step("r1: Verify (*, G) upstream IIF interface")
@@ -1682,7 +1684,7 @@ def test_RP_configured_as_LHR_2_p1(request):
 
     step("Creating configuration from JSON")
     reset_config_on_routers(tgen)
-    kill_iperf(tgen)
+    app_helper.stop_all_hosts()
     clear_ip_mroute(tgen)
     clear_ip_pim_interface_traffic(tgen, TOPO)
 
@@ -1802,11 +1804,11 @@ def test_RP_configured_as_LHR_2_p1(request):
     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 
     step("r5: Send multicast traffic for group 225.1.1.1")
-    result = iperfSendTraffic(tgen, "r5", GROUP_ADDRESS, 32, 2500)
+    result = app_helper.run_traffic("r5", GROUP_ADDRESS, "r3")
     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 
     step("r0: Send IGMP join")
-    result = iperfSendIGMPJoin(tgen, "r0", GROUP_ADDRESS, join_interval=1)
+    result = app_helper.run_join("r0", GROUP_ADDRESS, "r1")
     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 
     step("r1: Verify IGMP groups")
@@ -1892,7 +1894,7 @@ def test_RP_configured_as_FHR_1_p1(request):
 
     step("Creating configuration from JSON")
     reset_config_on_routers(tgen)
-    kill_iperf(tgen)
+    app_helper.stop_all_hosts()
     clear_ip_mroute(tgen)
     clear_ip_pim_interface_traffic(tgen, TOPO)
 
@@ -2013,7 +2015,7 @@ def test_RP_configured_as_FHR_1_p1(request):
     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 
     step("r0: Send IGMP join")
-    result = iperfSendIGMPJoin(tgen, "r0", GROUP_ADDRESS, join_interval=1)
+    result = app_helper.run_join("r0", GROUP_ADDRESS, "r1")
     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 
     step("r0: Verify IGMP groups")
@@ -2022,7 +2024,7 @@ def test_RP_configured_as_FHR_1_p1(request):
     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 
     step("r5: Send multicast traffic for group 225.1.1.1")
-    result = iperfSendTraffic(tgen, "r5", GROUP_ADDRESS, 32, 2500)
+    result = app_helper.run_traffic("r5", GROUP_ADDRESS, "r3")
     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 
     step("r1: Verify (*, G) upstream IIF interface")
@@ -2101,7 +2103,7 @@ def test_RP_configured_as_FHR_2_p2(request):
 
     step("Creating configuration from JSON")
     reset_config_on_routers(tgen)
-    kill_iperf(tgen)
+    app_helper.stop_all_hosts()
     clear_ip_mroute(tgen)
     clear_ip_pim_interface_traffic(tgen, TOPO)
 
@@ -2223,11 +2225,11 @@ def test_RP_configured_as_FHR_2_p2(request):
     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 
     step("r5: Send multicast traffic for group 225.1.1.1")
-    result = iperfSendTraffic(tgen, "r5", GROUP_ADDRESS, 32, 2500)
+    result = app_helper.run_traffic("r5", GROUP_ADDRESS, "r3")
     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 
     step("r0: Send IGMP join")
-    result = iperfSendIGMPJoin(tgen, "r0", GROUP_ADDRESS, join_interval=1)
+    result = app_helper.run_join("r0", GROUP_ADDRESS, "r1")
     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 
     step("r0: Verify IGMP groups")
@@ -2314,7 +2316,7 @@ def test_SPT_RPT_path_different_p1(request):
 
     step("Creating configuration from JSON")
     reset_config_on_routers(tgen)
-    kill_iperf(tgen)
+    app_helper.stop_all_hosts()
     clear_ip_mroute(tgen)
     clear_ip_pim_interface_traffic(tgen, TOPO)
 
@@ -2335,7 +2337,7 @@ def test_SPT_RPT_path_different_p1(request):
     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 
     step("r0: Send IGMP join")
-    result = iperfSendIGMPJoin(tgen, "r0", GROUP_ADDRESS, join_interval=1)
+    result = app_helper.run_join("r0", GROUP_ADDRESS, "r1")
     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 
     step("r1: Verify IGMP groups")
@@ -2345,7 +2347,7 @@ def test_SPT_RPT_path_different_p1(request):
     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 
     step("r5: Send multicast traffic for group 225.1.1.1")
-    result = iperfSendTraffic(tgen, "r5", GROUP_ADDRESS, 32, 2500)
+    result = app_helper.run_traffic("r5", GROUP_ADDRESS, "r3")
     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 
     step("r1: Verify (*, G) upstream IIF interface")
@@ -2469,7 +2471,7 @@ def test_clear_pim_configuration_p1(request):
 
     step("Creating configuration from JSON")
     reset_config_on_routers(tgen)
-    kill_iperf(tgen)
+    app_helper.stop_all_hosts()
     clear_ip_mroute(tgen)
     clear_ip_pim_interface_traffic(tgen, TOPO)
 
@@ -2493,7 +2495,7 @@ def test_clear_pim_configuration_p1(request):
     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 
     step("r0: Send IGMP join")
-    result = iperfSendIGMPJoin(tgen, "r0", GROUP_ADDRESS, join_interval=1)
+    result = app_helper.run_join("r0", GROUP_ADDRESS, "r1")
     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 
     step("r1: Verify IGMP groups")
@@ -2503,7 +2505,7 @@ def test_clear_pim_configuration_p1(request):
     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 
     step("r5: Send multicast traffic for group 225.1.1.1")
-    result = iperfSendTraffic(tgen, "r5", GROUP_ADDRESS, 32, 2500)
+    result = app_helper.run_traffic("r5", GROUP_ADDRESS, "r3")
     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 
     step("r1: Verify (*, G) upstream IIF interface")
@@ -2566,7 +2568,7 @@ def test_restart_pimd_process_p2(request):
 
     step("Creating configuration from JSON")
     reset_config_on_routers(tgen)
-    kill_iperf(tgen)
+    app_helper.stop_all_hosts()
     clear_ip_mroute(tgen)
     clear_ip_pim_interface_traffic(tgen, TOPO)
 
@@ -2590,7 +2592,7 @@ def test_restart_pimd_process_p2(request):
     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 
     step("r0: Send IGMP join")
-    result = iperfSendIGMPJoin(tgen, "r0", GROUP_ADDRESS, join_interval=1)
+    result = app_helper.run_join("r0", GROUP_ADDRESS, "r1")
     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 
     step("r1: Verify IGMP groups")
@@ -2600,7 +2602,7 @@ def test_restart_pimd_process_p2(request):
     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 
     step("r5: Send multicast traffic for group 225.1.1.1")
-    result = iperfSendTraffic(tgen, "r5", GROUP_ADDRESS, 32, 2500)
+    result = app_helper.run_traffic("r5", GROUP_ADDRESS, "r3")
     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 
     step("r1: Verify (*, G) upstream IIF interface")
@@ -2726,7 +2728,7 @@ def test_multiple_groups_same_RP_address_p2(request):
 
     step("Creating configuration from JSON")
     reset_config_on_routers(tgen)
-    kill_iperf(tgen)
+    app_helper.stop_all_hosts()
     clear_ip_mroute(tgen)
     clear_ip_pim_interface_traffic(tgen, TOPO)
 
@@ -2757,7 +2759,7 @@ def test_multiple_groups_same_RP_address_p2(request):
 
     group_address_list = GROUP_ADDRESS_LIST_1 + GROUP_ADDRESS_LIST_2
     step("r0: Send IGMP join for 10 groups")
-    result = iperfSendIGMPJoin(tgen, "r0", group_address_list, join_interval=1)
+    result = app_helper.run_join("r0", group_address_list, "r1")
     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 
     step("r1: Verify IGMP groups")
@@ -2767,7 +2769,7 @@ def test_multiple_groups_same_RP_address_p2(request):
     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 
     step("r5: Send multicast traffic for group 225.1.1.1")
-    result = iperfSendTraffic(tgen, "r5", group_address_list, 32, 2500)
+    result = app_helper.run_traffic("r5", group_address_list, "r3")
     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 
     step("r1: Verify (*, G) upstream IIF interface")
@@ -3035,7 +3037,7 @@ def test_multiple_groups_different_RP_address_p2(request):
 
     step("Creating configuration from JSON")
     reset_config_on_routers(tgen)
-    kill_iperf(tgen)
+    app_helper.stop_all_hosts()
     clear_ip_mroute(tgen)
     clear_ip_pim_interface_traffic(tgen, TOPO)
 
@@ -3104,7 +3106,7 @@ def test_multiple_groups_different_RP_address_p2(request):
 
     group_address_list = GROUP_ADDRESS_LIST_1 + GROUP_ADDRESS_LIST_2
     step("r0: Send IGMP join")
-    result = iperfSendIGMPJoin(tgen, "r0", group_address_list, join_interval=1)
+    result = app_helper.run_join("r0", group_address_list, "r1")
     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 
     step("r1: Verify IGMP groups")
@@ -3114,7 +3116,7 @@ def test_multiple_groups_different_RP_address_p2(request):
     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 
     step("r5: Send multicast traffic for group 225.1.1.1")
-    result = iperfSendTraffic(tgen, "r5", group_address_list, 32, 2500)
+    result = app_helper.run_traffic("r5", group_address_list, "r3")
     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 
     step("r1: Verify (*, G) upstream IIF interface")
@@ -3606,7 +3608,7 @@ def test_shutdown_primary_path_p1(request):
 
     step("Creating configuration from JSON")
     reset_config_on_routers(tgen)
-    kill_iperf(tgen)
+    app_helper.stop_all_hosts()
     clear_ip_mroute(tgen)
     clear_ip_pim_interface_traffic(tgen, TOPO)
 
@@ -3632,7 +3634,7 @@ def test_shutdown_primary_path_p1(request):
     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 
     step("r0: Send IGMP join")
-    result = iperfSendIGMPJoin(tgen, "r0", GROUP_ADDRESS, join_interval=1)
+    result = app_helper.run_join("r0", GROUP_ADDRESS, "r1")
     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 
     step("r1: Verify IGMP groups")
@@ -3799,7 +3801,7 @@ def test_delete_RP_shut_noshut_upstream_interface_p1(request):
 
     step("Creating configuration from JSON")
     reset_config_on_routers(tgen)
-    kill_iperf(tgen)
+    app_helper.stop_all_hosts()
     clear_ip_mroute(tgen)
     clear_ip_pim_interface_traffic(tgen, TOPO)
 
@@ -3823,7 +3825,7 @@ def test_delete_RP_shut_noshut_upstream_interface_p1(request):
     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 
     step("r0: Send IGMP join")
-    result = iperfSendIGMPJoin(tgen, "r0", GROUP_ADDRESS, join_interval=1)
+    result = app_helper.run_join("r0", GROUP_ADDRESS, "r1")
     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 
     step("r1: Verify IGMP groups")
@@ -3932,7 +3934,7 @@ def test_delete_RP_shut_noshut_RP_interface_p1(request):
 
     step("Creating configuration from JSON")
     reset_config_on_routers(tgen)
-    kill_iperf(tgen)
+    app_helper.stop_all_hosts()
     clear_ip_mroute(tgen)
     clear_ip_pim_interface_traffic(tgen, TOPO)
 
@@ -3956,7 +3958,7 @@ def test_delete_RP_shut_noshut_RP_interface_p1(request):
     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 
     step("r0: Send IGMP join")
-    result = iperfSendIGMPJoin(tgen, "r0", GROUP_ADDRESS, join_interval=1)
+    result = app_helper.run_join("r0", GROUP_ADDRESS, "r1")
     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 
     step("r1: Verify IGMP groups")
diff --git a/tests/topotests/pim_acl/test_pim_acl.py b/tests/topotests/pim_acl/test_pim_acl.py
index 4414713d8e..23b0efe4b5 100755
--- a/tests/topotests/pim_acl/test_pim_acl.py
+++ b/tests/topotests/pim_acl/test_pim_acl.py
@@ -125,96 +125,34 @@ from lib.pim import McastTesterHelper
 pytestmark = [pytest.mark.pimd, pytest.mark.ospfd]
 
 
-#
-# Test global variables:
-# They are used to handle communicating with external application.
-#
-APP_SOCK_PATH = '/tmp/topotests/apps.sock'
-HELPER_APP_PATH = os.path.join(CWD, "../lib/mcast-tester.py")
-app_listener = None
-app_clients = {}
-
-def listen_to_applications():
-    "Start listening socket to connect with applications."
-    # Remove old socket.
-    try:
-        os.unlink(APP_SOCK_PATH)
-    except OSError:
-        pass
-
-    sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM, 0)
-    sock.bind(APP_SOCK_PATH)
-    sock.listen(10)
-    global app_listener
-    app_listener = sock
-
-def accept_host(host):
-    "Accept connection from application running in hosts."
-    global app_listener, app_clients
-    conn = app_listener.accept()
-    app_clients[host] = {
-        'fd': conn[0],
-        'address': conn[1]
-    }
-
-def close_applications():
-    "Signal applications to stop and close all sockets."
-    global app_listener, app_clients
-
-    if app_listener:
-        # Close listening socket.
-        app_listener.close()
-
-        # Remove old socket.
-        try:
-            os.unlink(APP_SOCK_PATH)
-        except OSError:
-            pass
-
-    # Close all host connections.
-    for host in ["h1", "h2"]:
-        if app_clients.get(host) is None:
-            continue
-        app_clients[host]["fd"].close()
-
-    # Reset listener and clients data struct
-    app_listener = None
-    app_clients = {}
-
-
-class PIMACLTopo(Topo):
-    "PIM ACL Test Topology"
-
-    def build(self):
-        tgen = get_topogen(self)
-
-        # Create the hosts
-        for hostNum in range(1,3):
-            tgen.add_router("h{}".format(hostNum))
-
-        # Create the main router
-        tgen.add_router("r1")
-
-        # Create the PIM RP routers
-        for rtrNum in range(11, 16):
-            tgen.add_router("r{}".format(rtrNum))
-
-        # Setup Switches and connections
-        for swNum in range(1, 3):
-            tgen.add_switch("sw{}".format(swNum))
-
-        # Add connections H1 to R1 switch sw1
-        tgen.gears["h1"].add_link(tgen.gears["sw1"])
-        tgen.gears["r1"].add_link(tgen.gears["sw1"])
-
-        # Add connections R1 to R1x switch sw2
-        tgen.gears["r1"].add_link(tgen.gears["sw2"])
-        tgen.gears["h2"].add_link(tgen.gears["sw2"])
-        tgen.gears["r11"].add_link(tgen.gears["sw2"])
-        tgen.gears["r12"].add_link(tgen.gears["sw2"])
-        tgen.gears["r13"].add_link(tgen.gears["sw2"])
-        tgen.gears["r14"].add_link(tgen.gears["sw2"])
-        tgen.gears["r15"].add_link(tgen.gears["sw2"])
+
+def build_topo(tgen):
+    for hostNum in range(1,3):
+        tgen.add_router("h{}".format(hostNum))
+
+    # Create the main router
+    tgen.add_router("r1")
+
+    # Create the PIM RP routers
+    for rtrNum in range(11, 16):
+        tgen.add_router("r{}".format(rtrNum))
+
+    # Setup Switches and connections
+    for swNum in range(1, 3):
+        tgen.add_switch("sw{}".format(swNum))
+
+    # Add connections H1 to R1 switch sw1
+    tgen.gears["h1"].add_link(tgen.gears["sw1"])
+    tgen.gears["r1"].add_link(tgen.gears["sw1"])
+
+    # Add connections R1 to R1x switch sw2
+    tgen.gears["r1"].add_link(tgen.gears["sw2"])
+    tgen.gears["h2"].add_link(tgen.gears["sw2"])
+    tgen.gears["r11"].add_link(tgen.gears["sw2"])
+    tgen.gears["r12"].add_link(tgen.gears["sw2"])
+    tgen.gears["r13"].add_link(tgen.gears["sw2"])
+    tgen.gears["r14"].add_link(tgen.gears["sw2"])
+    tgen.gears["r15"].add_link(tgen.gears["sw2"])
 
 
 #####################################################
@@ -226,7 +164,7 @@ class PIMACLTopo(Topo):
 def setup_module(module):
     logger.info("PIM RP ACL Topology: \n {}".format(TOPOLOGY))
 
-    tgen = Topogen(PIMACLTopo, module.__name__)
+    tgen = Topogen(build_topo, module.__name__)
     tgen.start_topology()
 
     # Starting Routers
@@ -251,7 +189,6 @@ def setup_module(module):
 def teardown_module(module):
     tgen = get_topogen()
     tgen.stop_topology()
-    close_applications()
 
 
 def test_ospf_convergence():
@@ -305,46 +242,38 @@ def check_mcast_entry(entry, mcastaddr, pimrp):
 
     logger.info("Testing PIM RP selection for ACL {} entry using {}".format(entry, mcastaddr));
 
-    # Start applications socket.
-    listen_to_applications()
-
-    tgen.gears["h2"].run("{} --send='0.7' '{}' '{}' '{}' &".format(
-        HELPER_APP_PATH, APP_SOCK_PATH, mcastaddr, 'h2-eth0'))
-    accept_host("h2")
+    with McastTesterHelper(tgen) as helper:
+        helper.run("h2", ["--send=0.7", mcastaddr, "h2-eth0"])
+        helper.run("h1", [mcastaddr, "h1-eth0"])
 
-    tgen.gears["h1"].run("{} '{}' '{}' '{}' &".format(
-        HELPER_APP_PATH, APP_SOCK_PATH, mcastaddr, 'h1-eth0'))
-    accept_host("h1")
+        logger.info("mcast join and source for {} started".format(mcastaddr))
 
-    logger.info("mcast join and source for {} started".format(mcastaddr))
+        # tgen.mininet_cli()
 
-    # tgen.mininet_cli()
-
-    router = tgen.gears["r1"]
-    reffile = os.path.join(CWD, "r1/acl_{}_pim_join.json".format(entry))
-    expected = json.loads(open(reffile).read())
+        router = tgen.gears["r1"]
+        reffile = os.path.join(CWD, "r1/acl_{}_pim_join.json".format(entry))
+        expected = json.loads(open(reffile).read())
 
-    logger.info("verifying pim join on r1 for {}".format(mcastaddr))
-    test_func = functools.partial(
-        topotest.router_json_cmp, router, "show ip pim join json", expected
-    )
-    _, res = topotest.run_and_expect(test_func, None, count=60, wait=2)
-    assertmsg = "PIM router r1 did not show join status"
-    assert res is None, assertmsg
+        logger.info("verifying pim join on r1 for {}".format(mcastaddr))
+        test_func = functools.partial(
+            topotest.router_json_cmp, router, "show ip pim join json", expected
+        )
+        _, res = topotest.run_and_expect(test_func, None, count=60, wait=2)
+        assertmsg = "PIM router r1 did not show join status"
+        assert res is None, assertmsg
 
-    logger.info("verifying pim join on PIM RP {} for {}".format(pimrp, mcastaddr))
-    router = tgen.gears[pimrp]
-    reffile = os.path.join(CWD, "{}/acl_{}_pim_join.json".format(pimrp, entry))
-    expected = json.loads(open(reffile).read())
+        logger.info("verifying pim join on PIM RP {} for {}".format(pimrp, mcastaddr))
+        router = tgen.gears[pimrp]
+        reffile = os.path.join(CWD, "{}/acl_{}_pim_join.json".format(pimrp, entry))
+        expected = json.loads(open(reffile).read())
 
-    test_func = functools.partial(
-        topotest.router_json_cmp, router, "show ip pim join json", expected
-    )
-    _, res = topotest.run_and_expect(test_func, None, count=60, wait=2)
-    assertmsg = "PIM router {} did not get selected as the PIM RP".format(pimrp)
-    assert res is None, assertmsg
+        test_func = functools.partial(
+            topotest.router_json_cmp, router, "show ip pim join json", expected
+        )
+        _, res = topotest.run_and_expect(test_func, None, count=60, wait=2)
+        assertmsg = "PIM router {} did not get selected as the PIM RP".format(pimrp)
+        assert res is None, assertmsg
 
-    close_applications()
     return
 
 
diff --git a/tests/topotests/pim_igmp_vrf/test_pim_vrf.py b/tests/topotests/pim_igmp_vrf/test_pim_vrf.py
index a0f9e87b2a..badd224199 100755
--- a/tests/topotests/pim_igmp_vrf/test_pim_vrf.py
+++ b/tests/topotests/pim_igmp_vrf/test_pim_vrf.py
@@ -110,116 +110,53 @@ from lib.topolog import logger
 from lib.topotest import iproute2_is_vrf_capable
 from lib.common_config import (
     required_linux_kernel_version)
+from lib.pim import McastTesterHelper
 
-# Required to instantiate the topology builder class.
-from mininet.topo import Topo
 
 pytestmark = [pytest.mark.ospfd, pytest.mark.pimd]
 
 
-#
-# Test global variables:
-# They are used to handle communicating with external application.
-#
-APP_SOCK_PATH = '/tmp/topotests/apps.sock'
-HELPER_APP_PATH = os.path.join(CWD, "../lib/mcast-tester.py")
-app_listener = None
-app_clients = {}
-
-def listen_to_applications():
-    "Start listening socket to connect with applications."
-    # Remove old socket.
-    try:
-        os.unlink(APP_SOCK_PATH)
-    except OSError:
-        pass
-
-    sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM, 0)
-    sock.bind(APP_SOCK_PATH)
-    sock.listen(10)
-    global app_listener
-    app_listener = sock
-
-def accept_host(host):
-    "Accept connection from application running in hosts."
-    global app_listener, app_clients
-    conn = app_listener.accept()
-    app_clients[host] = {
-        'fd': conn[0],
-        'address': conn[1]
-    }
-
-def close_applications():
-    "Signal applications to stop and close all sockets."
-    global app_listener, app_clients
-
-    if app_listener:
-        # Close listening socket.
-        app_listener.close()
-
-        # Remove old socket.
-        try:
-            os.unlink(APP_SOCK_PATH)
-        except OSError:
-            pass
-
-    # Close all host connections.
-    for host in ["h1", "h2"]:
-        if app_clients.get(host) is None:
-            continue
-        app_clients[host]["fd"].close()
-
-    # Reset listener and clients data struct
-    app_listener = None
-    app_clients = {}
-
-
-class PIMVRFTopo(Topo):
-    "PIM VRF Test Topology"
-
-    def build(self):
-        tgen = get_topogen(self)
-
-        # Create the hosts
-        for hostNum in range(1,5):
-            tgen.add_router("h{}".format(hostNum))
-
-        # Create the main router
-        tgen.add_router("r1")
-
-        # Create the PIM RP routers
-        for rtrNum in range(11, 13):
-            tgen.add_router("r{}".format(rtrNum))
-
-        # Setup Switches and connections
-        for swNum in range(1, 5):
-            tgen.add_switch("sw{}".format(swNum))
-        
-        ################
-        # 1st set of connections to routers for VRF red
-        ################
-
-        # Add connections H1 to R1 switch sw1
-        tgen.gears["h1"].add_link(tgen.gears["sw1"])
-        tgen.gears["r1"].add_link(tgen.gears["sw1"])
-
-        # Add connections R1 to R1x switch sw2
-        tgen.gears["r1"].add_link(tgen.gears["sw2"])
-        tgen.gears["h2"].add_link(tgen.gears["sw2"])
-        tgen.gears["r11"].add_link(tgen.gears["sw2"])
-
-        ################
-        # 2nd set of connections to routers for vrf blue
-        ################
-
-        # Add connections H1 to R1 switch sw1
-        tgen.gears["h3"].add_link(tgen.gears["sw3"])
-        tgen.gears["r1"].add_link(tgen.gears["sw3"])
-
-        # Add connections R1 to R1x switch sw2
-        tgen.gears["r1"].add_link(tgen.gears["sw4"])
-        tgen.gears["h4"].add_link(tgen.gears["sw4"])
-        tgen.gears["r12"].add_link(tgen.gears["sw4"])
+
+def build_topo(tgen):
+    for hostNum in range(1,5):
+        tgen.add_router("h{}".format(hostNum))
+
+    # Create the main router
+    tgen.add_router("r1")
+
+    # Create the PIM RP routers
+    for rtrNum in range(11, 13):
+        tgen.add_router("r{}".format(rtrNum))
+
+    # Setup Switches and connections
+    for swNum in range(1, 5):
+        tgen.add_switch("sw{}".format(swNum))
+
+    ################
+    # 1st set of connections to routers for VRF red
+    ################
+
+    # Add connections H1 to R1 switch sw1
+    tgen.gears["h1"].add_link(tgen.gears["sw1"])
+    tgen.gears["r1"].add_link(tgen.gears["sw1"])
+
+    # Add connections R1 to R1x switch sw2
+    tgen.gears["r1"].add_link(tgen.gears["sw2"])
+    tgen.gears["h2"].add_link(tgen.gears["sw2"])
+    tgen.gears["r11"].add_link(tgen.gears["sw2"])
+
+    ################
+    # 2nd set of connections to routers for vrf blue
+    ################
+
+    # Add connections H1 to R1 switch sw1
+    tgen.gears["h3"].add_link(tgen.gears["sw3"])
+    tgen.gears["r1"].add_link(tgen.gears["sw3"])
+
+    # Add connections R1 to R1x switch sw2
+    tgen.gears["r1"].add_link(tgen.gears["sw4"])
+    tgen.gears["h4"].add_link(tgen.gears["sw4"])
+    tgen.gears["r12"].add_link(tgen.gears["sw4"])
 
 #####################################################
 #
@@ -230,7 +167,7 @@ class PIMVRFTopo(Topo):
 def setup_module(module):
     logger.info("PIM IGMP VRF Topology: \n {}".format(TOPOLOGY))
 
-    tgen = Topogen(PIMVRFTopo, module.__name__)
+    tgen = Topogen(build_topo, module.__name__)
     tgen.start_topology()
 
     vrf_setup_cmds = [
@@ -264,13 +201,13 @@ def setup_module(module):
             router.load_config(
                 TopoRouter.RD_PIM, os.path.join(CWD, "{}/pimd.conf".format(rname))
             )
+
     tgen.start_router()
 
 
 def teardown_module(module):
     tgen = get_topogen()
     tgen.stop_topology()
-    close_applications()
 
 
 def test_ospf_convergence():
@@ -394,48 +331,36 @@ def check_mcast_entry(mcastaddr, pimrp, receiver, sender, vrf):
 
     logger.info("Testing PIM for VRF {} entry using {}".format(vrf, mcastaddr));
 
-    # Start applications socket.
-    listen_to_applications()
-
-    tgen.gears[sender].run("{} --send='0.7' '{}' '{}' '{}' &".format(
-        HELPER_APP_PATH, APP_SOCK_PATH, mcastaddr, '{}-eth0'.format(sender)))
-    accept_host(sender)
+    with McastTesterHelper(tgen) as helper:
+        helper.run(sender, ["--send=0.7", mcastaddr, str(sender) + "-eth0"])
+        helper.run(receiver, [mcastaddr, str(receiver) + "-eth0"])
 
-    tgen.gears[receiver].run("{} '{}' '{}' '{}' &".format(
-        HELPER_APP_PATH, APP_SOCK_PATH, mcastaddr, '{}-eth0'.format(receiver)))
-    accept_host(receiver)
+        logger.info("mcast join and source for {} started".format(mcastaddr))
 
-    logger.info("mcast join and source for {} started".format(mcastaddr))
+        router = tgen.gears["r1"]
+        reffile = os.path.join(CWD, "r1/pim_{}_join.json".format(vrf))
+        expected = json.loads(open(reffile).read())
 
-    # tgen.mininet_cli()
-
-    router = tgen.gears["r1"]
-    reffile = os.path.join(CWD, "r1/pim_{}_join.json".format(vrf))
-    expected = json.loads(open(reffile).read())
-
-    logger.info("verifying pim join on r1 for {} on VRF {}".format(mcastaddr, vrf))
-    test_func = functools.partial(
-        topotest.router_json_cmp, router, "show ip pim vrf {} join json".format(vrf),
-        expected
-    )
-    _, res = topotest.run_and_expect(test_func, None, count=10, wait=2)
-    assertmsg = "PIM router r1 did not show join status on VRF".format(vrf)
-    assert res is None, assertmsg
+        logger.info("verifying pim join on r1 for {} on VRF {}".format(mcastaddr, vrf))
+        test_func = functools.partial(
+            topotest.router_json_cmp, router, "show ip pim vrf {} join json".format(vrf),
+            expected
+        )
+        _, res = topotest.run_and_expect(test_func, None, count=10, wait=2)
+        assertmsg = "PIM router r1 did not show join status on VRF {}".format(vrf)
+        assert res is None, assertmsg
 
-    logger.info("verifying pim join on PIM RP {} for {}".format(pimrp, mcastaddr))
-    router = tgen.gears[pimrp]
-    reffile = os.path.join(CWD, "{}/pim_{}_join.json".format(pimrp, vrf))
-    expected = json.loads(open(reffile).read())
+        logger.info("verifying pim join on PIM RP {} for {}".format(pimrp, mcastaddr))
+        router = tgen.gears[pimrp]
+        reffile = os.path.join(CWD, "{}/pim_{}_join.json".format(pimrp, vrf))
+        expected = json.loads(open(reffile).read())
 
-    test_func = functools.partial(
-        topotest.router_json_cmp, router, "show ip pim join json", expected
-    )
-    _, res = topotest.run_and_expect(test_func, None, count=10, wait=2)
-    assertmsg = "PIM router {} did not get selected as the PIM RP for VRF {}".format(pimrp, vrf)
-    assert res is None, assertmsg
-
-    close_applications()
-    return
+        test_func = functools.partial(
+            topotest.router_json_cmp, router, "show ip pim join json", expected
+        )
+        _, res = topotest.run_and_expect(test_func, None, count=10, wait=2)
+        assertmsg = "PIM router {} did not get selected as the PIM RP for VRF {}".format(pimrp, vrf)
+        assert res is None, assertmsg
 
 
 def test_mcast_vrf_blue():