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"
# Initialize all routers.
tgen.start_router()
- # Start applications socket.
- listen_to_applications()
+ app_helper.init(tgen)
+
def test_wait_ospf_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")
def teardown_module(_mod):
"Teardown the pytest environment"
tgen = get_topogen()
- close_applications()
+ app_helper.cleanup()
tgen.stop_topology()
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):
# 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()
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():
"""
step,
addKernelRoute,
create_static_routes,
- iperfSendIGMPJoin,
stop_router,
start_router,
shutdown_bringup_interface,
reset_config_on_routers,
do_countdown,
apply_raw_config,
- kill_iperf,
run_frr_cmd,
required_linux_kernel_version,
topo_daemons,
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
# 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")
tgen = get_topogen()
- # Kill any iperfs we left running.
- kill_iperf(tgen)
+ app_helper.cleanup()
# Stop toplogy and Remove tmp files
tgen.stop_topology()
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)
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)
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)
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)
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
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
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)
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
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)
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
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)
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
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)
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)
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
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)
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
step,
addKernelRoute,
create_static_routes,
- iperfSendIGMPJoin,
stop_router,
start_router,
shutdown_bringup_interface,
reset_config_on_routers,
do_countdown,
apply_raw_config,
- kill_iperf,
run_frr_cmd,
required_linux_kernel_version,
topo_daemons,
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
# 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")
tgen = get_topogen()
- # Kill any iperfs we left running.
- kill_iperf(tgen)
+ app_helper.cleanup()
# Stop toplogy and Remove tmp files
tgen.stop_topology()
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)
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
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)
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
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)
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
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)
]
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
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)
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"
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)
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
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)
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"
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,
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
# 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")
tgen = get_topogen()
- # Kill any iperfs we left running.
- kill_iperf(tgen)
+ app_helper.cleanup()
# Stop toplogy and Remove tmp files
tgen.stop_topology()
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)
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")
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(
)
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"
)
)
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(
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)
)
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)")
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(
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)
)
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")
)
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")
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)
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)
)
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")
)
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'")
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)
)
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)
)
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(
)
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' ")
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)
)
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")
)
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
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)
)
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")
)
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)
)
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)
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)
)
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")
)
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(
"l1": {
"igmp": {
"interfaces": {
- "l1-i1-eth1": {"igmp": {"query": {"query-interval": 100}}}
+ "l1-i1-eth1": {"igmp": {"query": {"query-interval": 20}}}
}
}
}
"l1": {
"igmp": {
"interfaces": {
- "l1-i1-eth1": {"igmp": {"query": {"query-interval": 200}}}
+ "l1-i1-eth1": {"igmp": {"query": {"query-interval": 25}}}
}
}
}
"l1": {
"igmp": {
"interfaces": {
- "l1-i1-eth1": {"igmp": {"query": {"query-interval": 300}}}
+ "l1-i1-eth1": {"igmp": {"query": {"query-interval": 30}}}
}
}
}
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)
)
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")
)
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(
)
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"
)
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": {
"l1-i1-eth1": {
"igmp": {
"version": "2",
- "query": {"query-max-response-time": 110},
+ "query": {"query-max-response-time": 105},
}
}
}
"l1-i1-eth1": {
"igmp": {
"version": "2",
- "query": {"query-max-response-time": 120},
+ "query": {"query-max-response-time": 110},
}
}
}
"l1-i1-eth1": {
"igmp": {
"version": "2",
- "query": {"query-max-response-time": 140},
+ "query": {"query-max-response-time": 115},
}
}
}
"l1-i1-eth1": {
"igmp": {
"version": "2",
- "query": {"query-max-response-time": 150},
+ "query": {"query-max-response-time": 120},
}
}
}
write_test_header,
write_test_footer,
step,
- iperfSendIGMPJoin,
addKernelRoute,
reset_config_on_routers,
- iperfSendTraffic,
- kill_iperf,
shutdown_bringup_interface,
kill_router_daemons,
start_router,
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
# 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")
tgen = get_topogen()
- # Kill any iperfs we left running.
- kill_iperf(tgen)
+ app_helper.cleanup()
# Stop toplogy and Remove tmp files
tgen.stop_topology()
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)
)
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)
)
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(
)
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
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)
)
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)
)
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(
)
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")
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)
)
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")
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(
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, "
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)
)
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)
)
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(
)
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(
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(
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)
)
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(
)
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(
" 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)
)
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'")
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)
"(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(
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),"
# 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"
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)
"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)
)
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(
)
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"
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)
)
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)
)
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")
)
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(
write_test_header,
write_test_footer,
step,
- iperfSendIGMPJoin,
addKernelRoute,
reset_config_on_routers,
- iperfSendTraffic,
- kill_iperf,
shutdown_bringup_interface,
kill_router_daemons,
start_router,
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
# 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")
tgen = get_topogen()
- # Kill any iperfs we left running.
- kill_iperf(tgen)
+ app_helper.cleanup()
# Stop toplogy and Remove tmp files
tgen.stop_topology()
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)
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)
)
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")
)
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]
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)
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)
)
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")
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)
)
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")
)
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(
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)
)
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")
)
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(
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)
)
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)")
)
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)")
)
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]
)
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)
)
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)
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)
)
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)")
)
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)")
)
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]
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)
)
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)")
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)
)
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)")
)
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]
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)
)
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)")
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)
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)
)
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")
)
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]
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]}}
)
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:
)
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]
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)
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)
)
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")
)
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]
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"
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"
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)
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)
)
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")
)
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]
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)
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)
)
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)")
)
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(
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)
)
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)")
)
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(
write_test_header,
write_test_footer,
step,
- iperfSendIGMPJoin,
addKernelRoute,
reset_config_on_routers,
- iperfSendTraffic,
- kill_iperf,
shutdown_bringup_interface,
start_router,
stop_router,
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
# 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")
tgen = get_topogen()
- # Kill any iperfs we left running.
- kill_iperf(tgen)
+ app_helper.cleanup()
# Stop toplogy and Remove tmp files
tgen.stop_topology()
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)
)
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")
)
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]
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)
)
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")
)
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]
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)
)
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")
)
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]
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,
clear_ip_pim_interfaces,
clear_ip_mroute,
clear_ip_mroute_verify,
+ McastTesterHelper,
)
pytestmark = [pytest.mark.pimd, pytest.mark.staticd]
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")
tgen = get_topogen()
- # Kill any iperfs we left running.
- kill_iperf(tgen)
+ app_helper.cleanup()
# Stop toplogy and Remove tmp files
tgen.stop_topology()
)
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")
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)
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")
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")
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)
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")
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)
)
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")
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)
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")
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)
)
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")
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)
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")
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")
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)
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")
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)
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")
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")
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)
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")
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)
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")
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")
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)
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")
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")
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)
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")
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")
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)
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")
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")
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)
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")
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")
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)
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")
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)
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")
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)
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")
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"])
#####################################################
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
def teardown_module(module):
tgen = get_topogen()
tgen.stop_topology()
- close_applications()
def test_ospf_convergence():
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
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"])
#####################################################
#
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 = [
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():
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():
projects / mirror_frr.git / commitdiff