1. Modified pim APIs name to generic one, same APIs would be used for PIMv4 and PIMv6
verifications
2. Modified all affacted scripts and ran multiple times locally to avoid CI failures
Signed-off-by: Kuldeep Kashyap <kashyapk@vmware.com>
@retry(retry_timeout=120, diag_pct=0)
-def verify_ip_mroutes(
+def verify_mroutes(
tgen,
dut,
src_address,
):
"""
Verify ip mroutes and make sure (*, G)/(S, G) is present in mroutes
- by running "show ip pim upstream" cli
+ by running "show ip/ipv6 mroute" cli
Parameters
----------
-----
dut = "r1"
group_address = "225.1.1.1"
- result = verify_ip_mroutes(tgen, dut, src_address, group_address)
+ result = verify_mroutes(tgen, dut, src_address, group_address)
Returns
-------
return True
-def clear_ip_pim_interface_traffic(tgen, topo):
+def clear_pim_interface_traffic(tgen, topo):
"""
- Clear ip pim interface traffice by running
- "clear ip pim interface traffic" cli
+ Clear ip/ipv6 pim interface traffice by running
+ "clear ip/ipv6 pim interface traffic" cli
Parameters
----------
Usage
-----
- result = clear_ip_pim_interface_traffic(tgen, topo)
+ result = clear_pim_interface_traffic(tgen, topo)
Returns
-------
return True
-def clear_ip_pim_interfaces(tgen, dut):
+def clear_pim_interfaces(tgen, dut):
"""
- Clear ip pim interface by running
- "clear ip pim interfaces" cli
+ Clear ip/ipv6 pim interface by running
+ "clear ip/ipv6 pim interfaces" cli
Parameters
----------
Usage
-----
- result = clear_ip_pim_interfaces(tgen, dut)
+ result = clear_pim_interfaces(tgen, dut)
Returns
-------
return True
-def clear_ip_igmp_interfaces(tgen, dut):
+def clear_igmp_interfaces(tgen, dut):
"""
- Clear ip igmp interfaces by running
- "clear ip igmp interfaces" cli
+ Clear ip/ipv6 igmp interfaces by running
+ "clear ip/ipv6 igmp interfaces" cli
Parameters
----------
Usage
-----
dut = "r1"
- result = clear_ip_igmp_interfaces(tgen, dut)
+ result = clear_igmp_interfaces(tgen, dut)
Returns
-------
errormsg(str) or True
@retry(retry_timeout=20)
-def clear_ip_mroute_verify(tgen, dut, expected=True):
+def clear_mroute_verify(tgen, dut, expected=True):
"""
- Clear ip mroute by running "clear ip mroute" cli and verify
+ Clear ip/ipv6 mroute by running "clear ip/ipv6 mroute" cli and verify
mroutes are up again after mroute clear
Parameters
Usage
-----
- result = clear_ip_mroute_verify(tgen, dut)
+ result = clear_mroute_verify(tgen, dut)
Returns
-------
return True
-def clear_ip_mroute(tgen, dut=None):
+def clear_mroute(tgen, dut=None):
"""
- Clear ip mroute by running "clear ip mroute" cli
+ Clear ip/ipv6 mroute by running "clear ip mroute" cli
Parameters
----------
Usage
-----
- clear_ip_mroute(tgen, dut)
+ clear_mroute(tgen, dut)
"""
logger.debug("Entering lib API: {}".format(sys._getframe().f_code.co_name))
@retry(retry_timeout=60, diag_pct=0)
-def verify_ip_pim_upstream_rpf(
+def verify_pim_upstream_rpf(
tgen, topo, dut, interface, group_addresses, rp=None, expected=True
):
"""
- Verify IP PIM upstream rpf, config is verified
- using "show ip pim neighbor" cli
+ Verify IP/IPv6 PIM upstream rpf, config is verified
+ using "show ip/ipv6 pim neighbor" cli
Parameters
----------
Usage
-----
- result = verify_ip_pim_upstream_rpf(gen, topo, dut, interface,
+ result = verify_pim_upstream_rpf(gen, topo, dut, interface,
group_addresses, rp=None)
Returns
@retry(retry_timeout=60, diag_pct=0)
-def verify_ip_pim_join(
+def verify_pim_join(
tgen, topo, dut, interface, group_addresses, src_address=None, expected=True
):
"""
- Verify ip pim join by running "show ip pim join" cli
+ Verify ip/ipv6 pim join by running "show ip/ipv6 pim join" cli
Parameters
----------
dut = "r1"
interface = "r1-r0-eth0"
group_address = "225.1.1.1"
- result = verify_ip_pim_join(tgen, dut, star, group_address, interface)
+ result = verify_pim_join(tgen, dut, star, group_address, interface)
Returns
-------
find_rp_from_bsrp_info,
verify_pim_grp_rp_source,
verify_pim_bsr,
- verify_ip_mroutes,
+ verify_mroutes,
verify_join_state_and_timer,
verify_pim_state,
verify_upstream_iif,
verify_igmp_groups,
- verify_ip_pim_upstream_rpf,
+ verify_pim_upstream_rpf,
enable_disable_pim_unicast_bsm,
enable_disable_pim_bsm,
- clear_ip_mroute,
- clear_ip_pim_interface_traffic,
+ clear_mroute,
+ clear_pim_interface_traffic,
get_pim_interface_traffic,
McastTesterHelper,
)
pytest.skip(tgen.errors)
app_helper.stop_all_hosts()
- clear_ip_mroute(tgen)
+ clear_mroute(tgen)
reset_config_on_routers(tgen)
- clear_ip_pim_interface_traffic(tgen, topo)
+ clear_pim_interface_traffic(tgen, topo)
reset_config_on_routers(tgen)
step("pre-configure BSM packet")
pytest.skip(tgen.errors)
app_helper.stop_all_hosts()
- clear_ip_mroute(tgen)
+ clear_mroute(tgen)
reset_config_on_routers(tgen)
- clear_ip_pim_interface_traffic(tgen, topo)
+ clear_pim_interface_traffic(tgen, topo)
reset_config_on_routers(tgen)
step("pre-configure BSM packet")
pytest.skip(tgen.errors)
app_helper.stop_all_hosts()
- clear_ip_mroute(tgen)
+ clear_mroute(tgen)
reset_config_on_routers(tgen)
- clear_ip_pim_interface_traffic(tgen, topo)
+ clear_pim_interface_traffic(tgen, topo)
reset_config_on_routers(tgen)
oil = "l1-r1-eth1"
step("Verify mroute populated on l1")
- result = verify_ip_mroutes(tgen, "l1", src_addr, GROUP_ADDRESS, iif, oil)
+ result = verify_mroutes(tgen, "l1", src_addr, GROUP_ADDRESS, iif, oil)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
# Reload i1 and l1
# Verify ip mroute populated again
step("Verify mroute again on l1 (lhr)")
- result = verify_ip_mroutes(tgen, "l1", src_addr, GROUP_ADDRESS, iif, oil)
+ result = verify_mroutes(tgen, "l1", src_addr, GROUP_ADDRESS, iif, oil)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("clear BSM database before moving to next case")
pytest.skip(tgen.errors)
app_helper.stop_all_hosts()
- clear_ip_mroute(tgen)
+ clear_mroute(tgen)
reset_config_on_routers(tgen)
- clear_ip_pim_interface_traffic(tgen, topo)
+ clear_pim_interface_traffic(tgen, topo)
reset_config_on_routers(tgen)
src_addr = "*"
oil = "i1-l1-eth1"
- result = verify_ip_mroutes(tgen, "i1", src_addr, GROUP_ADDRESS, iif, oil)
+ result = verify_mroutes(tgen, "i1", src_addr, GROUP_ADDRESS, iif, oil)
assert result is True, "Testcase {}:Failed \n Error: {}".format(tc_name, result)
# wait till bsm rp age out
# check if mroute uninstalled because of rp age out
step("check if mroute uninstalled because of rp age out in i1")
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen, "i1", src_addr, GROUP_ADDRESS, iif, oil, expected=False
)
assert (
# check if mroute still not installed because of rp not available
step("check if mroute still not installed because of rp not available")
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen, "i1", src_addr, GROUP_ADDRESS, iif, oil, expected=False
)
assert result is not True, (
# verify ip mroute populated
step("Verify ip mroute")
- result = verify_ip_mroutes(tgen, "i1", src_addr, GROUP_ADDRESS, iif, oil)
+ result = verify_mroutes(tgen, "i1", src_addr, GROUP_ADDRESS, iif, oil)
assert result is True, "Testcase {}:Failed \n Error: {}".format(tc_name, result)
# Shut/No shut the bsm rpf interface and check mroute on lhr(l1)
iif = "l1-i1-eth0"
oil = "l1-r1-eth1"
- result = verify_ip_mroutes(tgen, "l1", src_addr, GROUP_ADDRESS, iif, oil)
+ result = verify_mroutes(tgen, "l1", src_addr, GROUP_ADDRESS, iif, oil)
assert result is True, "Testcase {}:Failed \n Error: {}".format(tc_name, result)
step("clear BSM database before moving to next case")
pytest.skip(tgen.errors)
app_helper.stop_all_hosts()
- clear_ip_mroute(tgen)
+ clear_mroute(tgen)
reset_config_on_routers(tgen)
- clear_ip_pim_interface_traffic(tgen, topo)
+ clear_pim_interface_traffic(tgen, topo)
reset_config_on_routers(tgen)
iif = "l1-i1-eth0"
# Verify upstream rpf for 225.1.1.1 is chosen as rp1
step("Verify upstream rpf for 225.1.1.1 is chosen as bsrp")
- result = verify_ip_pim_upstream_rpf(tgen, topo, dut, iif, GROUP_ADDRESS, rp[group])
+ result = verify_pim_upstream_rpf(tgen, topo, dut, iif, GROUP_ADDRESS, rp[group])
assert result is True, "Testcase {}:Failed \n Error: {}".format(tc_name, result)
# Configure a static rp for the group 225.1.1.1/32
# Verify if upstream also reflects the static rp
step("Verify upstream rpf for 225.1.1.1 is chosen as static in l1")
- result = verify_ip_pim_upstream_rpf(tgen, topo, dut, iif, GROUP_ADDRESS, static_rp)
+ result = verify_pim_upstream_rpf(tgen, topo, dut, iif, GROUP_ADDRESS, static_rp)
assert result is True, "Testcase {}:Failed \n Error: {}".format(tc_name, result)
# delete static rp for the group 225.1.1.1/32
# Verify upstream rpf for 225.1.1.1 is chosen as bsrp
step("Verify upstream rpf for 225.1.1.1 is chosen as bsrp in l1")
- result = verify_ip_pim_upstream_rpf(tgen, topo, dut, iif, GROUP_ADDRESS, rp[group])
+ result = verify_pim_upstream_rpf(tgen, topo, dut, iif, GROUP_ADDRESS, rp[group])
assert result is True, "Testcase {}:Failed \n Error: {}".format(tc_name, result)
step("clear BSM database before moving to next case")
pytest.skip(tgen.errors)
app_helper.stop_all_hosts()
- clear_ip_mroute(tgen)
+ clear_mroute(tgen)
reset_config_on_routers(tgen)
- clear_ip_pim_interface_traffic(tgen, topo)
+ clear_pim_interface_traffic(tgen, topo)
reset_config_on_routers(tgen)
iif = "l1-i1-eth0"
src_addr = "*"
oil = "l1-r1-eth1"
- result = verify_ip_mroutes(tgen, "l1", src_addr, GROUP_ADDRESS, iif, oil)
+ result = verify_mroutes(tgen, "l1", src_addr, GROUP_ADDRESS, iif, oil)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
write_test_footer(tc_name)
pytest.skip(tgen.errors)
app_helper.stop_all_hosts()
- clear_ip_mroute(tgen)
+ clear_mroute(tgen)
reset_config_on_routers(tgen)
- clear_ip_pim_interface_traffic(tgen, topo)
+ clear_pim_interface_traffic(tgen, topo)
result = pre_config_to_bsm(
tgen, topo, tc_name, "b1", "s1", "r1", "f1", "i1", "l1", "packet1"
iif = "l1-i1-eth0"
src_addr = "*"
oil = "l1-r1-eth1"
- result = verify_ip_mroutes(tgen, dut, src_addr, GROUP_ADDRESS, iif, oil)
+ result = verify_mroutes(tgen, dut, src_addr, GROUP_ADDRESS, iif, oil)
assert result is True, "Testcase {}:Failed \n Error: {}".format(tc_name, result)
# Verify join state and join timer
# Verify ip mroute is not installed
step("Verify mroute not installed in l1")
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen, dut, src_addr, GROUP_ADDRESS, iif, oil, expected=False
)
assert (
pytest.skip(tgen.errors)
app_helper.stop_all_hosts()
- clear_ip_mroute(tgen)
+ clear_mroute(tgen)
reset_config_on_routers(tgen)
- clear_ip_pim_interface_traffic(tgen, topo)
+ clear_pim_interface_traffic(tgen, topo)
reset_config_on_routers(tgen)
# Verify ip mroute
src_addr = "*"
step("Verify ip mroute in l1")
- result = verify_ip_mroutes(tgen, dut, src_addr, GROUP_ADDRESS, iif, oil)
+ result = verify_mroutes(tgen, dut, src_addr, GROUP_ADDRESS, iif, oil)
assert result is True, "Testcase {}:Failed \n Error: {}".format(tc_name, result)
# Make RP unreachanble in LHR
# Verify mroute not installed
step("Verify mroute not installed")
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen, dut, src_addr, GROUP_ADDRESS, iif, oil, expected=False
)
assert (
# Verify that (*,G) installed in mroute again
iif = "l1-i1-eth0"
- result = verify_ip_mroutes(tgen, dut, src_addr, GROUP_ADDRESS, iif, oil)
+ result = verify_mroutes(tgen, dut, src_addr, GROUP_ADDRESS, iif, oil)
assert result is True, "Testcase {}:Failed \n Error: {}".format(tc_name, result)
step("clear BSM database before moving to next case")
find_rp_from_bsrp_info,
verify_pim_grp_rp_source,
verify_pim_bsr,
- verify_ip_mroutes,
+ verify_mroutes,
verify_join_state_and_timer,
verify_pim_state,
verify_upstream_iif,
verify_igmp_groups,
- verify_ip_pim_upstream_rpf,
- clear_ip_mroute,
- clear_ip_pim_interface_traffic,
+ verify_pim_upstream_rpf,
+ clear_mroute,
+ clear_pim_interface_traffic,
McastTesterHelper,
)
from lib.topolog import logger
pytest.skip(tgen.errors)
app_helper.stop_all_hosts()
- clear_ip_mroute(tgen)
+ clear_mroute(tgen)
reset_config_on_routers(tgen)
- clear_ip_pim_interface_traffic(tgen, topo)
+ clear_pim_interface_traffic(tgen, topo)
reset_config_on_routers(tgen)
# Verify ip mroute
step("Verify ip mroute in l1")
src_addr = "*"
- result = verify_ip_mroutes(tgen, dut, src_addr, GROUP_ADDRESS, iif, oil)
+ result = verify_mroutes(tgen, dut, src_addr, GROUP_ADDRESS, iif, oil)
assert result is True, "Testcase {}:Failed \n Error: {}".format(tc_name, result)
# Remove the group rp mapping and send bsm
# Verify mroute not installed
step("Verify mroute not installed in l1")
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen, dut, src_addr, GROUP_ADDRESS, iif, oil, retry_timeout=20, expected=False
)
assert (
# Verify that (*,G) installed in mroute again
iif = "l1-i1-eth0"
- result = verify_ip_mroutes(tgen, dut, src_addr, GROUP_ADDRESS, iif, oil)
+ result = verify_mroutes(tgen, dut, src_addr, GROUP_ADDRESS, iif, oil)
assert result is True, "Testcase {}:Failed \n Error: {}".format(tc_name, result)
step("clear BSM database before moving to next case")
pytest.skip(tgen.errors)
app_helper.stop_all_hosts()
- clear_ip_mroute(tgen)
+ clear_mroute(tgen)
reset_config_on_routers(tgen)
- clear_ip_pim_interface_traffic(tgen, topo)
+ clear_pim_interface_traffic(tgen, topo)
reset_config_on_routers(tgen)
iif = "l1-i1-eth0"
# Verify upstream rpf for 225.1.1.1 is chosen as rp1
step("Verify upstream rpf for 225.1.1.1 is chosen as rp1 in l1")
- result = verify_ip_pim_upstream_rpf(tgen, topo, dut, iif, GROUP_ADDRESS, rp1)
+ result = verify_pim_upstream_rpf(tgen, topo, dut, iif, GROUP_ADDRESS, rp1)
assert result is True, "Testcase {}:Failed \n Error: {}".format(tc_name, result)
# Send BSR packet from b1 with rp for 225.1.1.1/32 removed
# Verify upstream rpf for 225.1.1.1 is chosen as rp1
step("Verify upstream rpf for 225.1.1.1 is chosen as rp2 in l1")
- result = verify_ip_pim_upstream_rpf(tgen, topo, dut, iif, GROUP_ADDRESS, rp2)
+ result = verify_pim_upstream_rpf(tgen, topo, dut, iif, GROUP_ADDRESS, rp2)
assert result is True, "Testcase {}:Failed \n Error: {}".format(tc_name, result)
# Verify IIF/OIL in pim state
pytest.skip(tgen.errors)
app_helper.stop_all_hosts()
- clear_ip_mroute(tgen)
+ clear_mroute(tgen)
reset_config_on_routers(tgen)
- clear_ip_pim_interface_traffic(tgen, topo)
+ clear_pim_interface_traffic(tgen, topo)
reset_config_on_routers(tgen)
write_test_header(tc_name)
app_helper.stop_all_hosts()
- clear_ip_mroute(tgen)
+ clear_mroute(tgen)
reset_config_on_routers(tgen)
- clear_ip_pim_interface_traffic(tgen, topo)
+ clear_pim_interface_traffic(tgen, topo)
# Don"t run this test if we have any failure.
if tgen.routers_have_failure():
pytest.skip(tgen.errors)
app_helper.stop_all_hosts()
- clear_ip_mroute(tgen)
+ clear_mroute(tgen)
reset_config_on_routers(tgen)
- clear_ip_pim_interface_traffic(tgen, topo)
+ clear_pim_interface_traffic(tgen, topo)
reset_config_on_routers(tgen)
pytest.skip(tgen.errors)
app_helper.stop_all_hosts()
- clear_ip_mroute(tgen)
+ clear_mroute(tgen)
reset_config_on_routers(tgen)
- clear_ip_pim_interface_traffic(tgen, topo)
+ clear_pim_interface_traffic(tgen, topo)
reset_config_on_routers(tgen)
pytest.skip(tgen.errors)
app_helper.stop_all_hosts()
- clear_ip_mroute(tgen)
+ clear_mroute(tgen)
reset_config_on_routers(tgen)
- clear_ip_pim_interface_traffic(tgen, topo)
+ clear_pim_interface_traffic(tgen, topo)
step("pre-configure BSM packet")
result = pre_config_to_bsm(
from lib.pim import (
create_pim_config,
create_igmp_config,
- verify_ip_mroutes,
- clear_ip_mroute,
- clear_ip_pim_interface_traffic,
+ verify_mroutes,
+ clear_mroute,
+ clear_pim_interface_traffic,
verify_pim_config,
verify_upstream_iif,
verify_multicast_traffic,
verify_multicast_flag_state,
verify_igmp_groups,
- McastTesterHelper
+ McastTesterHelper,
)
from lib.topolog import logger
from lib.topojson import build_config_from_json
# Creating configuration from JSON
app_helper.stop_all_hosts()
- clear_ip_mroute(tgen)
+ clear_mroute(tgen)
check_router_status(tgen)
reset_config_on_routers(tgen)
- clear_ip_pim_interface_traffic(tgen, topo)
+ clear_pim_interface_traffic(tgen, topo)
# Don"t run this test if we have any failure.
if tgen.routers_have_failure():
]
for data in input_dict_r2:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
)
for data in input_dict_r2:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
)
assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
data["oil"],
expected=False,
)
- assert result is not True, (
- "Testcase {} : Failed \n "
- "mroutes are still present \n Error: {}".format(tc_name, result)
+ assert (
+ result is not True
+ ), "Testcase {} : Failed \n " "mroutes are still present \n Error: {}".format(
+ tc_name, result
)
step("Configure local join on R2 for group range (227.1.1.1)")
]
for data in input_dict_r1:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
data["oil"],
expected=False,
)
- assert result is not True, (
- "Testcase {} : Failed \n "
- "Mroutes are still present \n Error: {}".format(tc_name, result)
+ assert (
+ result is not True
+ ), "Testcase {} : Failed \n " "mroutes are still present \n Error: {}".format(
+ tc_name, result
)
step("Remove local join from DR and Non DR node")
)
for data in input_dict_r1:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
data["oil"],
expected=False,
)
- assert result is not True, (
- "Testcase {} : Failed \n "
- "Mroutes are still present \n Error: {}".format(tc_name, result)
+ assert (
+ result is not True
+ ), "Testcase {} : Failed \n " "mroutes are still present \n Error: {}".format(
+ tc_name, result
)
for data in input_dict_r2:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
data["oil"],
expected=False,
)
- assert result is not True, (
- "Testcase {} : Failed \n "
- "Mroutes are still present \n Error: {}".format(tc_name, result)
+ assert (
+ result is not True
+ ), "Testcase {} : Failed \n " "mroutes are still present \n Error: {}".format(
+ tc_name, result
)
write_test_footer(tc_name)
from lib.pim import (
create_pim_config,
create_igmp_config,
- verify_ip_mroutes,
- clear_ip_mroute,
- clear_ip_pim_interface_traffic,
+ verify_mroutes,
+ clear_mroute,
+ clear_pim_interface_traffic,
verify_pim_config,
verify_upstream_iif,
verify_multicast_flag_state,
)
logger.info("=" * 40)
+
#####################################################
#
# Local APIs
# Creating configuration from JSON
app_helper.stop_all_hosts()
- clear_ip_mroute(tgen)
+ clear_mroute(tgen)
check_router_status(tgen)
reset_config_on_routers(tgen)
- clear_ip_pim_interface_traffic(tgen, topo)
+ clear_pim_interface_traffic(tgen, topo)
# Don"t run this test if we have any failure.
if tgen.routers_have_failure():
]
for data in input_dict_r1_r2:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
step("R1 has mroute with none OIL and upstream with Not Join")
for data in input_dict_r1_r2:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
for data in input_dict_r1_r2:
if data["dut"] == "r1":
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
)
for data in input_dict_r1_r2:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
data["oil"],
expected=False,
)
- assert result is not True, (
- "Testcase {} : Failed \n "
- "mroutes are still present \n Error: {}".format(tc_name, result)
+ assert (
+ result is not True
+ ), "Testcase {} : Failed \n " "mroutes are still present \n Error: {}".format(
+ tc_name, result
)
step("start FRR for all the nodes")
assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
for data in input_dict_r1_r2:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
from lib.pim import (
create_pim_config,
create_igmp_config,
- verify_ip_mroutes,
- clear_ip_mroute,
- clear_ip_pim_interface_traffic,
+ verify_mroutes,
+ clear_mroute,
+ clear_pim_interface_traffic,
verify_pim_config,
verify_upstream_iif,
verify_multicast_traffic,
# Creating configuration from JSON
app_helper.stop_all_hosts()
- clear_ip_mroute(tgen)
+ clear_mroute(tgen)
check_router_status(tgen)
reset_config_on_routers(tgen)
- clear_ip_pim_interface_traffic(tgen, topo)
+ clear_pim_interface_traffic(tgen, topo)
# Don"t run this test if we have any failure.
if tgen.routers_have_failure():
]
for data in input_dict_r1:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
]
for data in input_dict_r5:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
)
for data in input_dict_r5:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
create_pim_config,
create_igmp_config,
verify_igmp_groups,
- verify_ip_mroutes,
+ verify_mroutes,
get_pim_interface_traffic,
verify_upstream_iif,
- verify_ip_pim_join,
- clear_ip_mroute,
- clear_ip_pim_interface_traffic,
+ verify_pim_join,
+ clear_mroute,
+ clear_pim_interface_traffic,
verify_igmp_config,
McastTesterHelper,
)
]
for data in input_dict:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen, data["dut"], data["src_address"], IGMP_JOIN, data["iif"], data["oil"]
)
assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
step("l1 sent PIM (S,G) join to f1 , verify using 'show ip pim join'")
dut = "f1"
interface = intf_f1_r2
- result = verify_ip_pim_join(tgen, topo, dut, interface, IGMP_JOIN)
+ result = verify_pim_join(tgen, topo, dut, interface, IGMP_JOIN)
assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
write_test_footer(tc_name)
# Creating configuration from JSON
app_helper.stop_all_hosts()
- clear_ip_mroute(tgen)
+ clear_mroute(tgen)
reset_config_on_routers(tgen)
- clear_ip_pim_interface_traffic(tgen, topo)
+ clear_pim_interface_traffic(tgen, topo)
step("Configure RP on R2 (loopback interface) for the" " group range 225.0.0.0/8")
# previous 80 retries with 2s wait if we assume .5s per vtysh/show ip mroute runtime
# (41 * (2 + .5)) == 102.
for data in input_dict:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
step("l1 sent PIM (S,G) join to f1 , verify using 'show ip pim join'")
dut = "f1"
interface = "f1-r2-eth3"
- result = verify_ip_pim_join(tgen, topo, dut, interface, IGMP_JOIN)
+ result = verify_pim_join(tgen, topo, dut, interface, IGMP_JOIN)
assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
write_test_footer(tc_name)
# Creating configuration from JSON
app_helper.stop_all_hosts()
- clear_ip_mroute(tgen)
+ clear_mroute(tgen)
reset_config_on_routers(tgen)
- clear_ip_pim_interface_traffic(tgen, topo)
+ clear_pim_interface_traffic(tgen, topo)
step("Configure static RP on c1 for group (225.1.1.1-5)")
input_dict = {
]
for data in input_dict:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen, data["dut"], data["src_address"], IGMP_JOIN, data["iif"], data["oil"]
)
assert result is True, "Testcase{} : Failed Error: {}".format(tc_name, result)
step("Clear mroutes on l1")
- clear_ip_mroute(tgen, "l1")
+ clear_mroute(tgen, "l1")
step(
"After clear ip mroute (*,g) entries are re-populated again"
)
for data in input_dict:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen, data["dut"], data["src_address"], IGMP_JOIN, data["iif"], data["oil"]
)
assert result is True, "Testcase{} : Failed Error: {}".format(tc_name, result)
# Creating configuration from JSON
app_helper.stop_all_hosts()
- clear_ip_mroute(tgen)
+ clear_mroute(tgen)
reset_config_on_routers(tgen)
- clear_ip_pim_interface_traffic(tgen, topo)
+ clear_pim_interface_traffic(tgen, topo)
step("Configure RP on R2 (loopback interface) for the" " group range 225.0.0.0/8")
{"dut": "f1", "src_address": source, "iif": "f1-i2-eth1", "oil": "f1-i8-eth2"},
]
for data in input_dict:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen, data["dut"], data["src_address"], IGMP_JOIN, data["iif"], data["oil"]
)
assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
# Creating configuration from JSON
app_helper.stop_all_hosts()
- clear_ip_mroute(tgen)
+ clear_mroute(tgen)
reset_config_on_routers(tgen)
- clear_ip_pim_interface_traffic(tgen, topo)
+ clear_pim_interface_traffic(tgen, topo)
step("Configure static RP for (226.1.1.1-5) and (232.1.1.1-5)" " in c1")
},
]
for data in input_dict:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
},
]
for data in input_dict:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
},
]
for data in input_dict:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
},
]
for data in input_dict:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
# Creating configuration from JSON
app_helper.stop_all_hosts()
- clear_ip_mroute(tgen)
+ clear_mroute(tgen)
reset_config_on_routers(tgen)
- clear_ip_pim_interface_traffic(tgen, topo)
+ clear_pim_interface_traffic(tgen, topo)
step("Configure static RP for (226.1.1.1-5) in c2")
input_dict = {
{"dut": "c2", "src_address": source, "iif": "c2-f1-eth1", "oil": "c2-c1-eth0"},
]
for data in input_dict:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen, data["dut"], data["src_address"], IGMP_JOIN, data["iif"], data["oil"]
)
assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
" router 'show ip mroute'"
)
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen, "c1", "*", IGMP_JOIN, "c1-c2-eth1", "c1-l1-eth0", expected=False
)
assert (
# Creating configuration from JSON
app_helper.stop_all_hosts()
- clear_ip_mroute(tgen)
+ clear_mroute(tgen)
reset_config_on_routers(tgen)
- clear_ip_pim_interface_traffic(tgen, topo)
+ clear_pim_interface_traffic(tgen, topo)
step("Configure RP on FRR2 (loopback interface) for " "the group range 225.0.0.0/8")
{"dut": "f1", "src_address": source, "iif": "f1-i2-eth1", "oil": "f1-i8-eth2"},
]
for data in input_dict:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen, data["dut"], data["src_address"], IGMP_JOIN, data["iif"], data["oil"]
)
assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
sleep(20)
step("Clear the mroute on f1")
- clear_ip_mroute(tgen, "f1")
+ clear_mroute(tgen, "f1")
step(
"After Shut the RP interface and clear the mroute verify all "
" 'show ip mroute' "
)
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen, "f1", "*", IGMP_JOIN, "f1-r2-eth3", "f1-i8-eth2", expected=False
)
assert (
# Creating configuration from JSON
app_helper.stop_all_hosts()
- clear_ip_mroute(tgen)
+ clear_mroute(tgen)
reset_config_on_routers(tgen)
- clear_ip_pim_interface_traffic(tgen, topo)
+ clear_pim_interface_traffic(tgen, topo)
step("Enable IGMP on FRR1 interface and send IGMP join (225.1.1.1)")
result = app_helper.run_join("i1", IGMP_JOIN, "l1")
{"dut": "f1", "src_address": source, "iif": "f1-i2-eth1", "oil": "f1-r2-eth3"},
]
for data in input_dict_4:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen, data["dut"], data["src_address"], IGMP_JOIN, data["iif"], data["oil"]
)
assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
# Creating configuration from JSON
app_helper.stop_all_hosts()
- clear_ip_mroute(tgen)
+ clear_mroute(tgen)
reset_config_on_routers(tgen)
- clear_ip_pim_interface_traffic(tgen, topo)
+ clear_pim_interface_traffic(tgen, topo)
step("Enable IGMP on FRR1 interface and send IGMP join (225.1.1.1)")
result = app_helper.run_join("i1", IGMP_JOIN, "l1")
{"dut": "f1", "src_address": source, "iif": "f1-i2-eth1", "oil": "f1-r2-eth3"},
]
for data in input_dict_5:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen, data["dut"], data["src_address"], IGMP_JOIN, data["iif"], data["oil"]
)
assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
create_pim_config,
create_igmp_config,
verify_igmp_groups,
- verify_ip_mroutes,
+ verify_mroutes,
get_pim_interface_traffic,
verify_upstream_iif,
verify_pim_neighbors,
verify_pim_state,
- clear_ip_mroute,
- clear_ip_pim_interface_traffic,
+ clear_mroute,
+ clear_pim_interface_traffic,
McastTesterHelper,
)
from lib.topolog import logger
# Creating configuration from JSON
app_helper.stop_all_hosts()
- clear_ip_mroute(tgen)
+ clear_mroute(tgen)
reset_config_on_routers(tgen)
- clear_ip_pim_interface_traffic(tgen, topo)
+ clear_pim_interface_traffic(tgen, topo)
step("Configure static RP for (226.1.1.1-5) in c1")
step("Configure static RP for (232.1.1.1-5) in c2")
{"dut": "f1", "src_address": source, "iif": "f1-i2-eth1", "oil": "f1-i8-eth2"},
]
for data in input_dict:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
start_router_daemons(tgen, "f1", ["pimd"])
for data in input_dict:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
{"dut": "f1", "src_address": "*", "iif": "f1-c2-eth0", "oil": "f1-i8-eth2"}
]
for data in input_dict:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
{"dut": "f1", "src_address": source, "iif": "f1-i2-eth1", "oil": "none"}
]
for data in input_dict:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
# Creating configuration from JSON
app_helper.stop_all_hosts()
- clear_ip_mroute(tgen)
+ clear_mroute(tgen)
reset_config_on_routers(tgen)
- clear_ip_pim_interface_traffic(tgen, topo)
+ clear_pim_interface_traffic(tgen, topo)
step("Configure static RP for (226.1.1.1-5) in c1")
step("Configure static RP for (232.1.1.1-5) in c2")
{"dut": "f1", "src_address": source, "iif": "f1-i2-eth1", "oil": "f1-i8-eth2"},
]
for data in input_dict:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
start_router(tgen, "f1")
for data in input_dict:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
{"dut": "f1", "src_address": "*", "iif": "f1-c2-eth0", "oil": "f1-i8-eth2"}
]
for data in input_dict:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
{"dut": "f1", "src_address": source, "iif": "f1-i2-eth1", "oil": "none"}
]
for data in input_dict:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
# Creating configuration from JSON
app_helper.stop_all_hosts()
- clear_ip_mroute(tgen)
+ clear_mroute(tgen)
reset_config_on_routers(tgen)
- clear_ip_pim_interface_traffic(tgen, topo)
+ clear_pim_interface_traffic(tgen, topo)
step("Configure static RP for (226.1.1.1-5) and " "(232.1.1.1-5) in c2")
" 'show ip mroute' and mroute OIL is towards RP."
)
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
"f1",
"10.0.5.2",
)
assert result is True, "Testcase {} : " "Failed Error: {}".format(tc_name, result)
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen, "f1", "10.0.5.2", _IGMP_JOIN_RANGE, "f1-i2-eth1", "f1-r2-eth3"
)
assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
{"dut": "l1", "src_address": source, "iif": "l1-r2-eth4", "oil": "l1-i1-eth1"},
]
for data in input_dict:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
{"dut": "c2", "src_address": source, "iif": "c2-f1-eth1", "oil": "none"}
]
for data in input_dict:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
# Creating configuration from JSON
app_helper.stop_all_hosts()
- clear_ip_mroute(tgen)
+ clear_mroute(tgen)
reset_config_on_routers(tgen)
- clear_ip_pim_interface_traffic(tgen, topo)
+ clear_pim_interface_traffic(tgen, topo)
step("Configure static RP for (226.1.1.1-5) in c1")
step("Configure static RP for (232.1.1.1-5) in c2")
{"dut": "f1", "src_address": source, "iif": "f1-i2-eth1", "oil": "f1-i8-eth2"},
]
for data in input_dict:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
shutdown_bringup_interface(tgen, dut, intf, True)
for data in input_dict:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
)
for data in input_dict:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
intf = "f1-r2-eth3"
shutdown_bringup_interface(tgen, dut, intf, True)
- clear_ip_mroute(tgen, "l1")
- clear_ip_mroute(tgen, "l1")
+ clear_mroute(tgen, "l1")
+ clear_mroute(tgen, "l1")
step(
"After no shut, verify traffic resume to all the receivers"
)
for data in input_dict:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
shutdown_bringup_interface(tgen, dut, intf_l1_c1, True)
for data in input_dict:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
app_helper.stop_all_hosts()
for data in input_dict:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
# Creating configuration from JSON
app_helper.stop_all_hosts()
- clear_ip_mroute(tgen)
+ clear_mroute(tgen)
reset_config_on_routers(tgen)
- clear_ip_pim_interface_traffic(tgen, topo)
+ clear_pim_interface_traffic(tgen, topo)
step("Configure static RP on c1 for group range " "(226.1.1.1-5) and (232.1.1.1-5)")
{"dut": "c2", "src_address": source, "iif": "c2-f1-eth1", "oil": "c2-i5-eth2"},
]
for data in input_dict:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
# Creating configuration from JSON
app_helper.stop_all_hosts()
- clear_ip_mroute(tgen)
+ clear_mroute(tgen)
reset_config_on_routers(tgen)
- clear_ip_pim_interface_traffic(tgen, topo)
+ clear_pim_interface_traffic(tgen, topo)
step("Configure static RP for group range (226.1.1.1-5) and " "(232.1.1.1-5) on c1")
_GROUP_RANGE = GROUP_RANGE_2 + GROUP_RANGE_3
{"dut": "l1", "src_address": source, "iif": "l1-r2-eth4", "oil": "l1-i1-eth1"},
]
for data in input_dict:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
{"dut": "l1", "src_address": "*", "iif": "l1-c1-eth0", "oil": "l1-i1-eth1"},
]
for data in input_dict:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
]
for data in input_dict:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
# Creating configuration from JSON
app_helper.stop_all_hosts()
- clear_ip_mroute(tgen)
+ clear_mroute(tgen)
reset_config_on_routers(tgen)
- clear_ip_pim_interface_traffic(tgen, topo)
+ clear_pim_interface_traffic(tgen, topo)
step("Configure static RP for (226.1.1.1-5) in c1")
step("Configure static RP for (232.1.1.1-5) in c2")
{"dut": "f1", "src_address": source, "iif": "f1-r2-eth3", "oil": "f1-i8-eth2"},
]
for data in input_dict_all:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
{"dut": "l1", "src_address": source, "iif": "l1-r2-eth4", "oil": "l1-i1-eth1"}
]
for data in input_dict:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
{"dut": "f1", "src_address": source, "iif": "f1-r2-eth3", "oil": "f1-i8-eth2"}
]
for data in input_dict:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
)
for data in input_dict_all:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
# Creating configuration from JSON
app_helper.stop_all_hosts()
- clear_ip_mroute(tgen)
+ clear_mroute(tgen)
reset_config_on_routers(tgen)
- clear_ip_pim_interface_traffic(tgen, topo)
+ clear_pim_interface_traffic(tgen, topo)
step("Configure static RP for (226.1.1.1-5) in c1")
step("Configure static RP for (232.1.1.1-5) in c2")
},
]
for data in input_dict_all:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
" 'show ip mroute' "
)
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
"f1",
source_i2,
create_pim_config,
create_igmp_config,
verify_igmp_groups,
- verify_ip_mroutes,
- clear_ip_mroute,
- clear_ip_pim_interface_traffic,
+ verify_mroutes,
+ clear_mroute,
+ clear_pim_interface_traffic,
verify_igmp_config,
verify_pim_config,
verify_pim_interface,
# Creating configuration from JSON
app_helper.stop_all_hosts()
- clear_ip_mroute(tgen)
+ clear_mroute(tgen)
reset_config_on_routers(tgen)
- clear_ip_pim_interface_traffic(tgen, topo)
+ clear_pim_interface_traffic(tgen, topo)
check_router_status(tgen)
step("Enable the PIM on all the interfaces of FRR1, FRR2, FRR3")
step("Verify mroutes and iff upstream")
for data in input_dict_all:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
step("Verify mroutes and iff upstream")
for data in input_dict_l1:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
step("Verify mroutes and iff upstream")
for data in input_dict_f1:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
step("Verify mroutes and iff upstream")
for data in input_dict_l1:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
# Creating configuration from JSON
app_helper.stop_all_hosts()
- clear_ip_mroute(tgen)
+ clear_mroute(tgen)
reset_config_on_routers(tgen)
- clear_ip_pim_interface_traffic(tgen, topo)
+ clear_pim_interface_traffic(tgen, topo)
check_router_status(tgen)
step("Removing FRR3 to simulate topo " "FHR(FRR1)---LHR(FRR2)")
step("Verify mroutes and iff upstream")
for data in input_dict_all:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
]
for data in input_dict_r2:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
step("Verify mroutes and iff upstream")
for data in input_dict_l1_r2:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
]
for data in input_dict_l1:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
# Creating configuration from JSON
app_helper.stop_all_hosts()
- clear_ip_mroute(tgen)
+ clear_mroute(tgen)
reset_config_on_routers(tgen)
- clear_ip_pim_interface_traffic(tgen, topo)
+ clear_pim_interface_traffic(tgen, topo)
check_router_status(tgen)
step("Enable the PIM on all the interfaces of FRR1, R2 and FRR3" " routers")
)
for data in input_dict_all:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
)
intf_f1_r2 = topo["routers"]["f1"]["links"]["r2"]["interface"]
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen, "f1", source_i2, IGMP_JOIN_RANGE_1, intf_f1_i2, intf_f1_r2, expected=False
)
assert (
)
for data in input_dict_all:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
)
for data in input_dict_all:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
# Creating configuration from JSON
app_helper.stop_all_hosts()
- clear_ip_mroute(tgen)
+ clear_mroute(tgen)
reset_config_on_routers(tgen)
- clear_ip_pim_interface_traffic(tgen, topo)
+ clear_pim_interface_traffic(tgen, topo)
check_router_status(tgen)
step("Enable the PIM on all the interfaces of FRR1, R2 and FRR3" " routers")
]
for data in input_dict_all:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
)
for data in input_dict_all:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
)
for data in input_dict_all:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
# Creating configuration from JSON
app_helper.stop_all_hosts()
- clear_ip_mroute(tgen)
+ clear_mroute(tgen)
reset_config_on_routers(tgen)
- clear_ip_pim_interface_traffic(tgen, topo)
+ clear_pim_interface_traffic(tgen, topo)
check_router_status(tgen)
step("Enable PIM on all routers")
]
for data in input_dict_all:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
)
for data in input_dict_all:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
]
for data in input_dict_l1_f1:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
iif = topo["routers"]["l1"]["links"]["i6"]["interface"]
oil = topo["routers"]["l1"]["links"]["i1"]["interface"]
source = source_i6
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen, dut, source, IGMP_JOIN_RANGE_1, iif, oil, expected=False
)
assert (
# Creating configuration from JSON
app_helper.stop_all_hosts()
- clear_ip_mroute(tgen)
+ clear_mroute(tgen)
reset_config_on_routers(tgen)
- clear_ip_pim_interface_traffic(tgen, topo)
+ clear_pim_interface_traffic(tgen, topo)
check_router_status(tgen)
step("Enable PIM on all routers")
]
for data in input_dict_all:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
# Creating configuration from JSON
app_helper.stop_all_hosts()
- clear_ip_mroute(tgen)
+ clear_mroute(tgen)
reset_config_on_routers(tgen)
- clear_ip_pim_interface_traffic(tgen, topo)
+ clear_pim_interface_traffic(tgen, topo)
check_router_status(tgen)
step("Configure 'ip pim' on receiver interface on FRR1")
# Creating configuration from JSON
app_helper.stop_all_hosts()
- clear_ip_mroute(tgen)
+ clear_mroute(tgen)
reset_config_on_routers(tgen)
- clear_ip_pim_interface_traffic(tgen, topo)
+ clear_pim_interface_traffic(tgen, topo)
check_router_status(tgen)
step("Configure 'ip pim' on receiver interface on FRR1")
]
for data in input_dict_all:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
for data in input_dict_all:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
for data in input_dict_all:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
for data in input_dict_all:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
# Creating configuration from JSON
app_helper.stop_all_hosts()
- clear_ip_mroute(tgen)
+ clear_mroute(tgen)
reset_config_on_routers(tgen)
- clear_ip_pim_interface_traffic(tgen, topo)
+ clear_pim_interface_traffic(tgen, topo)
check_router_status(tgen)
step("Configure 'ip pim' on receiver interface on FRR1")
# Creating configuration from JSON
app_helper.stop_all_hosts()
- clear_ip_mroute(tgen)
+ clear_mroute(tgen)
reset_config_on_routers(tgen)
- clear_ip_pim_interface_traffic(tgen, topo)
+ clear_pim_interface_traffic(tgen, topo)
check_router_status(tgen)
step(
step("Verify mroutes and iff upstream")
for data in input_dict_all:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
]
for data in input_dict_starg:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
logger.info("Expected Behaviour: {}".format(result))
for data in input_dict_sg:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
for data in input_dict_starg:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
source_i2 = topo["routers"]["i6"]["links"]["l1"]["ipv4"].split("/")[0]
for data in input_dict_sg:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
# Creating configuration from JSON
app_helper.stop_all_hosts()
- clear_ip_mroute(tgen)
+ clear_mroute(tgen)
reset_config_on_routers(tgen)
- clear_ip_pim_interface_traffic(tgen, topo)
+ clear_pim_interface_traffic(tgen, topo)
check_router_status(tgen)
step(
step("Verify mroutes and iff upstream")
for data in input_dict_all:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
step("Verify mroute after Shut the link from LHR to RP from RP node")
for data in input_dict_starg:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
logger.info("Expected Behaviour: {}".format(result))
for data in input_dict_sg_i1:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
step("Verify mroute after No shut the link from LHR to RP from RP node")
for data in input_dict_starg:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
for data in input_dict_sg_i2:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
for data in input_dict_sg_i1:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
step("Verify mroute after Shut the link from FHR to RP from RP node")
for data in input_dict_starg:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
for data in input_dict_sg_i1:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
step("Verify mroute after Noshut the link from FHR to RP from RP node")
for data in input_dict_starg:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
for data in input_dict_sg_i2:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
for data in input_dict_sg_i1:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
step("Verify mroute after Shut the link from FHR to RP from FHR node")
for data in input_dict_starg:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
for data in input_dict_sg_i1:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
step("Verify mroute after No Shut the link from FHR to RP from FHR node")
for data in input_dict_starg:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
for data in input_dict_sg_i2:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
for data in input_dict_sg_i1:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
# Creating configuration from JSON
app_helper.stop_all_hosts()
- clear_ip_mroute(tgen)
+ clear_mroute(tgen)
reset_config_on_routers(tgen)
- clear_ip_pim_interface_traffic(tgen, topo)
+ clear_pim_interface_traffic(tgen, topo)
check_router_status(tgen)
step(
step("Verify mroutes and iff upstream")
for data in input_dict_all:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
# Creating configuration from JSON
app_helper.stop_all_hosts()
- clear_ip_mroute(tgen)
+ clear_mroute(tgen)
reset_config_on_routers(tgen)
- clear_ip_pim_interface_traffic(tgen, topo)
+ clear_pim_interface_traffic(tgen, topo)
check_router_status(tgen)
step(
]
for data in input_dict_all:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
]
for data in input_dict_r2:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
)
for data in input_dict_r2:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
]
for data in input_dict_l1:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
)
for data in input_dict_l1:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
]
for data in input_dict_r2_f1:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
)
for data in input_dict_all:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
iif = topo["routers"]["r2"]["links"]["f1"]["interface"]
oil = topo["routers"]["r2"]["links"]["i3"]["interface"]
- result = verify_ip_mroutes(tgen, dut, src_address, _IGMP_JOIN_RANGE, iif, oil)
+ result = verify_mroutes(tgen, dut, src_address, _IGMP_JOIN_RANGE, iif, oil)
assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
src_address = source_i1
iif = topo["routers"]["r2"]["links"]["l1"]["interface"]
oil = topo["routers"]["r2"]["links"]["i3"]["interface"]
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen, dut, src_address, _IGMP_JOIN_RANGE, iif, oil, expected=False
)
assert (
)
for data in input_dict_all:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
# Creating configuration from JSON
app_helper.stop_all_hosts()
- clear_ip_mroute(tgen)
+ clear_mroute(tgen)
reset_config_on_routers(tgen)
- clear_ip_pim_interface_traffic(tgen, topo)
+ clear_pim_interface_traffic(tgen, topo)
check_router_status(tgen)
step(
]
for data in input_dict_all:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
]
for data in input_dict_l1:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
)
for data in input_dict_l1:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
]
for data in input_dict_r2:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
)
for data in input_dict_r2:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
]
for data in input_dict_all_star:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
from lib.pim import (
create_pim_config,
create_igmp_config,
- verify_ip_mroutes,
- clear_ip_pim_interface_traffic,
+ verify_mroutes,
+ clear_pim_interface_traffic,
verify_upstream_iif,
- clear_ip_mroute,
+ clear_mroute,
verify_pim_rp_info,
get_pim_interface_traffic,
McastTesterHelper,
# Creating configuration from JSON
app_helper.stop_all_hosts()
- clear_ip_mroute(tgen)
+ clear_mroute(tgen)
reset_config_on_routers(tgen)
- clear_ip_pim_interface_traffic(tgen, topo)
+ clear_pim_interface_traffic(tgen, topo)
step(
"Remove c1-c2 connected link to simulate topo "
step("Verify mroutes and iff upstream")
for data in input_dict_sg:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
for data in input_dict_starg:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
step("Verify mroute not present after Delete of static routes on c1")
for data in input_dict_sg:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
)
for data in input_dict_starg:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
step("Verify (s,g) populated after adding default route ")
for data in input_dict_sg:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
step("Verify (*,g) populated after adding default route ")
for data in input_dict_starg:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
# Creating configuration from JSON
app_helper.stop_all_hosts()
- clear_ip_mroute(tgen)
+ clear_mroute(tgen)
reset_config_on_routers(tgen)
- clear_ip_pim_interface_traffic(tgen, topo)
+ clear_pim_interface_traffic(tgen, topo)
step(
"Remove c1-c2 connected link to simulate topo "
step("Verify mroutes and iff upstream")
for data in input_dict_sg:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
for data in input_dict_starg:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
for data in input_dict_starg:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
step("Verify (s,g) populated after adding default route ")
for data in input_dict_sg:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
step("Verify (*,g) populated after adding default route ")
for data in input_dict_starg:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
# Creating configuration from JSON
app_helper.stop_all_hosts()
- clear_ip_mroute(tgen)
+ clear_mroute(tgen)
reset_config_on_routers(tgen)
- clear_ip_pim_interface_traffic(tgen, topo)
+ clear_pim_interface_traffic(tgen, topo)
step(
"Remove c1-c2 connected link to simulate topo "
step("(*,G) and (S,G) created on f1 and node verify using 'show ip mroute'")
for data in input_dict_sg:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
for data in input_dict_starg:
- result = verify_ip_mroutes(
+ result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
verify_igmp_groups,
verify_upstream_iif,
verify_join_state_and_timer,
- verify_ip_mroutes,
+ verify_mroutes,
verify_pim_neighbors,
get_pim_interface_traffic,
verify_pim_rp_info,
verify_pim_state,
- clear_ip_pim_interface_traffic,
- clear_ip_igmp_interfaces,
- clear_ip_pim_interfaces,
- clear_ip_mroute,
- clear_ip_mroute_verify,
+ clear_pim_interface_traffic,
+ clear_igmp_interfaces,
+ clear_pim_interfaces,
+ clear_mroute,
+ clear_mroute_verify,
McastTesterHelper,
)
result = verify_join_state_and_timer(tgen, dut, iif, STAR, GROUP_ADDRESS)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("r1: Verify ip mroutes")
- result = verify_ip_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif)
+ result = verify_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("r1: Verify ip pim join")
)
step("r1: Verify ip mroutes")
- result = verify_ip_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif, expected=False)
+ result = verify_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif, expected=False)
assert (
result is not True
), "Testcase {} : Failed \n " "r1: mroutes are still present \n Error: {}".format(
step("Creating configuration from JSON")
reset_config_on_routers(tgen)
app_helper.stop_all_hosts()
- clear_ip_mroute(tgen)
- clear_ip_pim_interface_traffic(tgen, TOPO)
+ clear_mroute(tgen)
+ clear_pim_interface_traffic(tgen, TOPO)
dut = "r1"
intf = "r1-r3-eth2"
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("r1: Verify (*, G) ip mroutes")
- result = verify_ip_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif)
+ result = verify_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("r1: Verify (S, G) upstream IIF interface")
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("r1: Verify (S, G) ip mroutes")
- result = verify_ip_mroutes(tgen, dut, SOURCE_ADDRESS, GROUP_ADDRESS, iif, oif)
+ result = verify_mroutes(tgen, dut, SOURCE_ADDRESS, GROUP_ADDRESS, iif, oif)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("r2: Verify (*, G) upstream IIF interface")
step("r2: Verify (*, G) ip mroutes")
oif = "r2-r1-eth0"
- result = verify_ip_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif)
+ result = verify_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("r2: Verify (S, G) upstream IIF interface")
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("r2: Verify (S, G) ip mroutes")
- result = verify_ip_mroutes(tgen, dut, SOURCE_ADDRESS, GROUP_ADDRESS, iif, oif)
+ result = verify_mroutes(tgen, dut, SOURCE_ADDRESS, GROUP_ADDRESS, iif, oif)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("r3: Verify (S, G) upstream IIF interface")
step("r3: Verify (S, G) ip mroutes")
oif = "r3-r2-eth1"
- result = verify_ip_mroutes(tgen, dut, SOURCE_ADDRESS, GROUP_ADDRESS, iif, oif)
+ result = verify_mroutes(tgen, dut, SOURCE_ADDRESS, GROUP_ADDRESS, iif, oif)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
# Uncomment next line for debugging
step("Creating configuration from JSON")
reset_config_on_routers(tgen)
app_helper.stop_all_hosts()
- clear_ip_mroute(tgen)
- clear_ip_pim_interface_traffic(tgen, TOPO)
+ clear_mroute(tgen)
+ clear_pim_interface_traffic(tgen, TOPO)
dut = "r1"
intf = "r1-r3-eth2"
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("r1 :Verify ip mroutes")
- result = verify_ip_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif)
+ result = verify_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("r1: Make RP un-reachable")
assert result is True, "Testcase{} : Failed Error: {}".format(tc_name, result)
step("r1: (*, G) cleared from mroute table using show ip mroute")
- result = verify_ip_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif, expected=False)
+ result = verify_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif, expected=False)
assert result is not True, (
"Testcase {} : Failed \n "
"r1: (*, G) are not cleared from mroute table \n Error: {}".format(
step("Creating configuration from JSON")
reset_config_on_routers(tgen)
app_helper.stop_all_hosts()
- clear_ip_mroute(tgen)
- clear_ip_pim_interface_traffic(tgen, TOPO)
+ clear_mroute(tgen)
+ clear_pim_interface_traffic(tgen, TOPO)
step("Enable IGMP on R1 interface")
step("Configure r2 loopback interface as RP")
)
step("r1: Verify ip mroutes")
- result = verify_ip_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif, expected=False)
+ result = verify_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif, expected=False)
assert (
result is not True
), "Testcase {} : Failed \n " "r1: mroutes are still present\n Error: {}".format(
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("r1 : Verify ip mroutes")
- result = verify_ip_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif)
+ result = verify_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
logger.info("Expected behavior: %s", result)
step("Creating configuration from JSON")
reset_config_on_routers(tgen)
app_helper.stop_all_hosts()
- clear_ip_mroute(tgen)
- clear_ip_pim_interface_traffic(tgen, TOPO)
+ clear_mroute(tgen)
+ clear_pim_interface_traffic(tgen, TOPO)
step("Enable IGMP on r1 interface and send IGMP " "join (225.1.1.1) to r1")
step("Configure r2 loopback interface as RP")
)
step("r1 : Verify ip mroutes")
- result = verify_ip_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif, expected=False)
+ result = verify_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif, expected=False)
assert (
result is not True
), "Testcase {} : Failed \n " "r1: mroutes are still present\n Error: {}".format(
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("r1 : Verify ip mroutes")
- result = verify_ip_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif)
+ result = verify_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
logger.info("Expected behavior: %s", result)
step("Creating configuration from JSON")
reset_config_on_routers(tgen)
app_helper.stop_all_hosts()
- clear_ip_mroute(tgen)
- clear_ip_pim_interface_traffic(tgen, TOPO)
+ clear_mroute(tgen)
+ clear_pim_interface_traffic(tgen, TOPO)
step("Enable IGMP on r1 interface")
step("Configure RP on r2 (loopback interface) for the group range " "224.0.0.0/4")
step("r1 : Verify ip mroutes")
iif = "r1-r4-eth3"
- result = verify_ip_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif)
+ result = verify_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("r1 : Verify PIM state")
result = verify_join_state_and_timer(tgen, dut, iif, STAR, GROUP_ADDRESS)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
- clear_ip_pim_interface_traffic(tgen, TOPO)
+ clear_pim_interface_traffic(tgen, TOPO)
step("r1 : Verify joinTx, pruneTx count before RP gets deleted")
state_dict = {"r1": {"r1-r2-eth1": ["joinTx"], "r1-r4-eth3": ["pruneTx"]}}
"RP gets deleted"
)
iif = "r1-r2-eth1"
- result = verify_ip_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif)
+ result = verify_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
logger.info("Expected behavior: %s", result)
step("Creating configuration from JSON")
reset_config_on_routers(tgen)
app_helper.stop_all_hosts()
- clear_ip_mroute(tgen)
- clear_ip_pim_interface_traffic(tgen, TOPO)
+ clear_mroute(tgen)
+ clear_pim_interface_traffic(tgen, TOPO)
step("Enable IGMP on r1 interface")
step("Configure RP on r1 (loopback interface) for the group range" " 224.0.0.0/4")
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("r1: Verify (*, G) ip mroutes")
- result = verify_ip_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif)
+ result = verify_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("r1: Verify (S, G) upstream IIF interface")
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("r1: Verify (S, G) ip mroutes")
- result = verify_ip_mroutes(tgen, dut, SOURCE_ADDRESS, GROUP_ADDRESS, iif, oif)
+ result = verify_mroutes(tgen, dut, SOURCE_ADDRESS, GROUP_ADDRESS, iif, oif)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("r3: Verify (S, G) upstream IIF interface")
step("r3: Verify (S, G) ip mroutes")
oif = "r3-r1-eth0"
- result = verify_ip_mroutes(tgen, dut, SOURCE_ADDRESS, GROUP_ADDRESS, iif, oif)
+ result = verify_mroutes(tgen, dut, SOURCE_ADDRESS, GROUP_ADDRESS, iif, oif)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
# Uncomment next line for debugging
step("Creating configuration from JSON")
reset_config_on_routers(tgen)
app_helper.stop_all_hosts()
- clear_ip_mroute(tgen)
- clear_ip_pim_interface_traffic(tgen, TOPO)
+ clear_mroute(tgen)
+ clear_pim_interface_traffic(tgen, TOPO)
step("Enable IGMP on r1 interface")
step("Configure RP on r1 (loopback interface) for the group range" " 224.0.0.0/4")
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("r1: Verify (*, G) ip mroutes")
- result = verify_ip_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif)
+ result = verify_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("r1: Verify (S, G) upstream IIF interface")
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("r1: Verify (S, G) ip mroutes")
- result = verify_ip_mroutes(tgen, dut, SOURCE_ADDRESS, GROUP_ADDRESS, iif, oif)
+ result = verify_mroutes(tgen, dut, SOURCE_ADDRESS, GROUP_ADDRESS, iif, oif)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("r3: Verify (S, G) upstream IIF interface")
step("r3: Verify (S, G) ip mroutes")
oif = "r3-r1-eth0"
- result = verify_ip_mroutes(tgen, dut, SOURCE_ADDRESS, GROUP_ADDRESS, iif, oif)
+ result = verify_mroutes(tgen, dut, SOURCE_ADDRESS, GROUP_ADDRESS, iif, oif)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
# Uncomment next line for debugging
step("Creating configuration from JSON")
reset_config_on_routers(tgen)
app_helper.stop_all_hosts()
- clear_ip_mroute(tgen)
- clear_ip_pim_interface_traffic(tgen, TOPO)
+ clear_mroute(tgen)
+ clear_pim_interface_traffic(tgen, TOPO)
step("Enable IGMP on r1 interface")
step("Configure RP on r2 (loopback interface) for the group range" " 225.1.1.0/24")
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("r1: Verify (*, G) ip mroutes")
- result = verify_ip_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif)
+ result = verify_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("r1: Verify (S, G) upstream IIF interface")
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("r1: Verify (S, G) ip mroutes")
- result = verify_ip_mroutes(tgen, dut, SOURCE_ADDRESS, GROUP_ADDRESS, iif, oif)
+ result = verify_mroutes(tgen, dut, SOURCE_ADDRESS, GROUP_ADDRESS, iif, oif)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("r3: Verify (S, G) upstream IIF interface")
step("r3: Verify (S, G) ip mroutes")
oif = "r3-r1-eth0"
- result = verify_ip_mroutes(tgen, dut, SOURCE_ADDRESS, GROUP_ADDRESS, iif, oif)
+ result = verify_mroutes(tgen, dut, SOURCE_ADDRESS, GROUP_ADDRESS, iif, oif)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
# Uncomment next line for debugging
step("Creating configuration from JSON")
reset_config_on_routers(tgen)
app_helper.stop_all_hosts()
- clear_ip_mroute(tgen)
- clear_ip_pim_interface_traffic(tgen, TOPO)
+ clear_mroute(tgen)
+ clear_pim_interface_traffic(tgen, TOPO)
step("Enable IGMP on r1 interface")
step("Configure RP on r2 (loopback interface) for the group range" " 225.1.1.0/24")
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("r1: Verify (*, G) ip mroutes")
- result = verify_ip_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif)
+ result = verify_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("r1: Verify (S, G) upstream IIF interface")
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("r1: Verify (S, G) ip mroutes")
- result = verify_ip_mroutes(tgen, dut, SOURCE_ADDRESS, GROUP_ADDRESS, iif, oif)
+ result = verify_mroutes(tgen, dut, SOURCE_ADDRESS, GROUP_ADDRESS, iif, oif)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("r3: Verify (S, G) upstream IIF interface")
step("r3: Verify (S, G) ip mroutes")
oif = "r3-r1-eth0"
- result = verify_ip_mroutes(tgen, dut, SOURCE_ADDRESS, GROUP_ADDRESS, iif, oif)
+ result = verify_mroutes(tgen, dut, SOURCE_ADDRESS, GROUP_ADDRESS, iif, oif)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
# Uncomment next line for debugging
step("Creating configuration from JSON")
reset_config_on_routers(tgen)
app_helper.stop_all_hosts()
- clear_ip_mroute(tgen)
- clear_ip_pim_interface_traffic(tgen, TOPO)
+ clear_mroute(tgen)
+ clear_pim_interface_traffic(tgen, TOPO)
step("Enable IGMP on r1 interface and send IGMP join (225.1.1.1) to r1")
step("Configure RP on r2 (loopback interface) for the group range" " 224.0.0.0/4")
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("r1: Verify (*, G) ip mroutes")
- result = verify_ip_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif)
+ result = verify_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("r1: Verify (S, G) upstream IIF interface")
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("r1: Verify (S, G) ip mroutes")
- result = verify_ip_mroutes(tgen, dut, SOURCE_ADDRESS, GROUP_ADDRESS, iif, oif)
+ result = verify_mroutes(tgen, dut, SOURCE_ADDRESS, GROUP_ADDRESS, iif, oif)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("r2: Verify (*, G) upstream IIF interface")
step("r2: Verify (*, G) ip mroutes")
oif = "r2-r1-eth0"
- result = verify_ip_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif)
+ result = verify_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("r3: Verify (S, G) upstream IIF interface")
step("r3: Verify (S, G) ip mroutes")
oif = "r3-r1-eth0"
- result = verify_ip_mroutes(tgen, dut, SOURCE_ADDRESS, GROUP_ADDRESS, iif, oif)
+ result = verify_mroutes(tgen, dut, SOURCE_ADDRESS, GROUP_ADDRESS, iif, oif)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("r2: Verify (S, G) upstream IIF interface")
step("r2: Verify (S, G) ip mroutes")
oif = "none"
- result = verify_ip_mroutes(tgen, dut, SOURCE_ADDRESS, GROUP_ADDRESS, iif, oif)
+ result = verify_mroutes(tgen, dut, SOURCE_ADDRESS, GROUP_ADDRESS, iif, oif)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
# Uncomment next line for debugging
step("Creating configuration from JSON")
reset_config_on_routers(tgen)
app_helper.stop_all_hosts()
- clear_ip_mroute(tgen)
- clear_ip_pim_interface_traffic(tgen, TOPO)
+ clear_mroute(tgen)
+ clear_pim_interface_traffic(tgen, TOPO)
step("Enable IGMP on r1 interface")
step("Configure RP on r2 (loopback interface) for the group range" " 224.0.0.0/4")
step("r1: Verify (*, G) ip mroutes")
oif = "r1-r0-eth0"
- result = verify_ip_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif)
+ result = verify_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("r1: Verify IGMP groups timer restarted")
- result = clear_ip_igmp_interfaces(tgen, dut)
+ result = clear_igmp_interfaces(tgen, dut)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("r1: Verify PIM neighbor timer restarted")
- result = clear_ip_pim_interfaces(tgen, dut)
+ result = clear_pim_interfaces(tgen, dut)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("r1: Verify PIM mroute timer restarted")
- result = clear_ip_mroute_verify(tgen, dut)
+ result = clear_mroute_verify(tgen, dut)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
# Uncomment next line for debugging
step("Creating configuration from JSON")
reset_config_on_routers(tgen)
app_helper.stop_all_hosts()
- clear_ip_mroute(tgen)
- clear_ip_pim_interface_traffic(tgen, TOPO)
+ clear_mroute(tgen)
+ clear_pim_interface_traffic(tgen, TOPO)
step("Enable IGMP on r1 interface and send IGMP join (225.1.1.1) to R1")
step("Configure RP on r3 (loopback interface) for the group range" " 224.0.0.0/4")
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("r1: Verify (*, G) ip mroutes")
- result = verify_ip_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif)
+ result = verify_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("r1: Verify (S, G) upstream IIF interface")
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("r1: Verify (S, G) ip mroutes")
- result = verify_ip_mroutes(tgen, dut, SOURCE_ADDRESS, GROUP_ADDRESS, iif, oif)
+ result = verify_mroutes(tgen, dut, SOURCE_ADDRESS, GROUP_ADDRESS, iif, oif)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("r2: Verify (*, G) upstream IIF interface")
step("r2: Verify (*, G) ip mroutes")
oif = "r2-r1-eth0"
- result = verify_ip_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif)
+ result = verify_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("r3: Verify (S, G) upstream IIF interface")
step("r3: Verify (S, G) ip mroutes")
oif = "r3-r1-eth0"
- result = verify_ip_mroutes(tgen, dut, SOURCE_ADDRESS, GROUP_ADDRESS, iif, oif)
+ result = verify_mroutes(tgen, dut, SOURCE_ADDRESS, GROUP_ADDRESS, iif, oif)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
dut = "r1"
logger.info("waiting for 10 sec to make sure old mroute time is higher")
sleep(10)
# Why do we then wait 60 seconds below before checking the routes?
- uptime_before = verify_ip_mroutes(
+ uptime_before = verify_mroutes(
tgen, dut, STAR, GROUP_ADDRESS, iif, oil, return_uptime=True, mwait=60
)
assert isinstance(uptime_before, dict), "Testcase{} : Failed Error: {}".format(
sleep(5)
# Why do we then wait 10 seconds below before checking the routes?
- uptime_after = verify_ip_mroutes(
+ uptime_after = verify_mroutes(
tgen, dut, STAR, GROUP_ADDRESS, iif, oil, return_uptime=True, mwait=10
)
assert isinstance(uptime_after, dict), "Testcase{} : Failed Error: {}".format(
step("Creating configuration from JSON")
reset_config_on_routers(tgen)
app_helper.stop_all_hosts()
- clear_ip_mroute(tgen)
- clear_ip_pim_interface_traffic(tgen, TOPO)
+ clear_mroute(tgen)
+ clear_pim_interface_traffic(tgen, TOPO)
step("Enable IGMP on r1 interface and send IGMP join (225.1.1.1) to r1")
step("Configure RP on r2 (loopback interface) for the group range" "225.1.1.0/24")
step("r1: Verify (*, G) ip mroutes")
oif = "r1-r0-eth0"
- result = verify_ip_mroutes(tgen, dut, STAR, group_address_list, iif, oif)
+ result = verify_mroutes(tgen, dut, STAR, group_address_list, iif, oif)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("r1: Verify (S, G) upstream IIF interface")
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("r1: Verify (S, G) ip mroutes")
- result = verify_ip_mroutes(tgen, dut, SOURCE_ADDRESS, group_address_list, iif, oif)
+ result = verify_mroutes(tgen, dut, SOURCE_ADDRESS, group_address_list, iif, oif)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("r2: Verify (*, G) upstream IIF interface")
step("r2: Verify (*, G) ip mroutes")
oif = "r2-r1-eth0"
- result = verify_ip_mroutes(tgen, dut, STAR, group_address_list, iif, oif)
+ result = verify_mroutes(tgen, dut, STAR, group_address_list, iif, oif)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("r3: Verify (S, G) upstream IIF interface")
step("r3: Verify (S, G) ip mroutes")
oif = "r3-r1-eth0"
- result = verify_ip_mroutes(tgen, dut, SOURCE_ADDRESS, group_address_list, iif, oif)
+ result = verify_mroutes(tgen, dut, SOURCE_ADDRESS, group_address_list, iif, oif)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("r2: Verify (S, G) upstream IIF interface")
step("r2: Verify (S, G) ip mroutes")
oif = "none"
- result = verify_ip_mroutes(tgen, dut, SOURCE_ADDRESS, group_address_list, iif, oif)
+ result = verify_mroutes(tgen, dut, SOURCE_ADDRESS, group_address_list, iif, oif)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("r1: Delete RP configuration")
step("r1: Verify (*, G) ip mroutes")
oif = "r1-r0-eth0"
- result = verify_ip_mroutes(tgen, dut, STAR, group_address_list, iif, oif)
+ result = verify_mroutes(tgen, dut, STAR, group_address_list, iif, oif)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("r1: Verify (S, G) upstream IIF interface")
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("r1: Verify (S, G) ip mroutes")
- result = verify_ip_mroutes(tgen, dut, SOURCE_ADDRESS, group_address_list, iif, oif)
+ result = verify_mroutes(tgen, dut, SOURCE_ADDRESS, group_address_list, iif, oif)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("r2: Verify (*, G) upstream IIF interface")
step("r2: Verify (*, G) ip mroutes")
oif = "r2-r1-eth0"
- result = verify_ip_mroutes(tgen, dut, STAR, group_address_list, iif, oif)
+ result = verify_mroutes(tgen, dut, STAR, group_address_list, iif, oif)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("r2: Verify (S, G) upstream IIF interface")
step("r2: Verify (S, G) ip mroutes")
oif = "none"
- result = verify_ip_mroutes(tgen, dut, SOURCE_ADDRESS, group_address_list, iif, oif)
+ result = verify_mroutes(tgen, dut, SOURCE_ADDRESS, group_address_list, iif, oif)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("r3: Verify (S, G) upstream IIF interface")
step("r3: Verify (S, G) ip mroutes")
oif = "r3-r1-eth0"
- result = verify_ip_mroutes(tgen, dut, SOURCE_ADDRESS, group_address_list, iif, oif)
+ result = verify_mroutes(tgen, dut, SOURCE_ADDRESS, group_address_list, iif, oif)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
write_test_footer(tc_name)
step("Creating configuration from JSON")
reset_config_on_routers(tgen)
app_helper.stop_all_hosts()
- clear_ip_mroute(tgen)
- clear_ip_pim_interface_traffic(tgen, TOPO)
+ clear_mroute(tgen)
+ clear_pim_interface_traffic(tgen, TOPO)
step("Delete existing RP configuration")
input_dict = {
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("r1: Verify (*, G) ip mroutes")
- result = verify_ip_mroutes(tgen, dut, STAR, GROUP_ADDRESS_LIST_1, iif, oif)
+ result = verify_mroutes(tgen, dut, STAR, GROUP_ADDRESS_LIST_1, iif, oif)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("r1: Verify (S, G) upstream IIF interface")
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("r1: Verify (S, G) ip mroutes")
- result = verify_ip_mroutes(
- tgen, dut, SOURCE_ADDRESS, GROUP_ADDRESS_LIST_1, iif, oif
- )
+ result = verify_mroutes(tgen, dut, SOURCE_ADDRESS, GROUP_ADDRESS_LIST_1, iif, oif)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("r2: Verify (*, G) upstream IIF interface")
step("r2: Verify (*, G) ip mroutes")
oif = "r2-r1-eth0"
- result = verify_ip_mroutes(tgen, dut, STAR, GROUP_ADDRESS_LIST_1, iif, oif)
+ result = verify_mroutes(tgen, dut, STAR, GROUP_ADDRESS_LIST_1, iif, oif)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("r2: Verify (S, G) upstream IIF interface")
step("r2: Verify (S, G) ip mroutes")
oif = "none"
- result = verify_ip_mroutes(
- tgen, dut, SOURCE_ADDRESS, GROUP_ADDRESS_LIST_1, iif, oif
- )
+ result = verify_mroutes(tgen, dut, SOURCE_ADDRESS, GROUP_ADDRESS_LIST_1, iif, oif)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("r3: Verify (S, G) upstream IIF interface")
step("r3: Verify (S, G) ip mroutes")
oif = "r3-r1-eth0"
- result = verify_ip_mroutes(
- tgen, dut, SOURCE_ADDRESS, GROUP_ADDRESS_LIST_1, iif, oif
- )
+ result = verify_mroutes(tgen, dut, SOURCE_ADDRESS, GROUP_ADDRESS_LIST_1, iif, oif)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("r1: Verify (*, G) upstream IIF interface")
step("r1: Verify (*, G) ip mroutes")
oif = "r1-r0-eth0"
- result = verify_ip_mroutes(tgen, dut, STAR, GROUP_ADDRESS_LIST_2, iif, oif)
+ result = verify_mroutes(tgen, dut, STAR, GROUP_ADDRESS_LIST_2, iif, oif)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("r1: Verify (S, G) upstream IIF interface")
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("r1: Verify (S, G) ip mroutes")
- result = verify_ip_mroutes(
- tgen, dut, SOURCE_ADDRESS, GROUP_ADDRESS_LIST_2, iif, oif
- )
+ result = verify_mroutes(tgen, dut, SOURCE_ADDRESS, GROUP_ADDRESS_LIST_2, iif, oif)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("r4: Verify (*, G) upstream IIF interface")
step("r4: Verify (*, G) ip mroutes")
oif = "r4-r1-eth0"
- result = verify_ip_mroutes(tgen, dut, STAR, GROUP_ADDRESS_LIST_2, iif, oif)
+ result = verify_mroutes(tgen, dut, STAR, GROUP_ADDRESS_LIST_2, iif, oif)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("r4: Verify (S, G) upstream IIF interface")
step("r4: Verify (S, G) ip mroutes")
oif = "none"
- result = verify_ip_mroutes(
- tgen, dut, SOURCE_ADDRESS, GROUP_ADDRESS_LIST_2, iif, oif
- )
+ result = verify_mroutes(tgen, dut, SOURCE_ADDRESS, GROUP_ADDRESS_LIST_2, iif, oif)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("r3: Verify (S, G) upstream IIF interface")
step("r3: Verify (S, G) ip mroutes")
oif = "r3-r1-eth0"
- result = verify_ip_mroutes(
- tgen, dut, SOURCE_ADDRESS, GROUP_ADDRESS_LIST_2, iif, oif
- )
+ result = verify_mroutes(tgen, dut, SOURCE_ADDRESS, GROUP_ADDRESS_LIST_2, iif, oif)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("Delete RP configuration")
step("r1: Verify (*, G) ip mroutes")
oif = "r1-r0-eth0"
- result = verify_ip_mroutes(tgen, dut, STAR, GROUP_ADDRESS_LIST_1, iif, oif)
+ result = verify_mroutes(tgen, dut, STAR, GROUP_ADDRESS_LIST_1, iif, oif)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("r1: Verify (S, G) upstream IIF interface")
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("r1: Verify (S, G) ip mroutes")
- result = verify_ip_mroutes(
- tgen, dut, SOURCE_ADDRESS, GROUP_ADDRESS_LIST_1, iif, oif
- )
+ result = verify_mroutes(tgen, dut, SOURCE_ADDRESS, GROUP_ADDRESS_LIST_1, iif, oif)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("r2: Verify (*, G) upstream IIF interface")
step("r2: Verify (*, G) ip mroutes")
oif = "r2-r1-eth0"
- result = verify_ip_mroutes(tgen, dut, STAR, GROUP_ADDRESS_LIST_1, iif, oif)
+ result = verify_mroutes(tgen, dut, STAR, GROUP_ADDRESS_LIST_1, iif, oif)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("r2: Verify (S, G) upstream IIF interface")
step("r2: Verify (S, G) ip mroutes")
oif = "none"
- result = verify_ip_mroutes(
- tgen, dut, SOURCE_ADDRESS, GROUP_ADDRESS_LIST_1, iif, oif
- )
+ result = verify_mroutes(tgen, dut, SOURCE_ADDRESS, GROUP_ADDRESS_LIST_1, iif, oif)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("r3: Verify (S, G) upstream IIF interface")
step("r3: Verify (S, G) ip mroutes")
oif = "r3-r1-eth0"
- result = verify_ip_mroutes(
- tgen, dut, SOURCE_ADDRESS, GROUP_ADDRESS_LIST_1, iif, oif
- )
+ result = verify_mroutes(tgen, dut, SOURCE_ADDRESS, GROUP_ADDRESS_LIST_1, iif, oif)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("r1: Verify (*, G) upstream IIF interface")
step("r1: Verify (*, G) ip mroutes")
oif = "r1-r0-eth0"
- result = verify_ip_mroutes(tgen, dut, STAR, GROUP_ADDRESS_LIST_2, iif, oif)
+ result = verify_mroutes(tgen, dut, STAR, GROUP_ADDRESS_LIST_2, iif, oif)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("r1: Verify (S, G) upstream IIF interface")
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("r1: Verify (S, G) ip mroutes")
- result = verify_ip_mroutes(
- tgen, dut, SOURCE_ADDRESS, GROUP_ADDRESS_LIST_2, iif, oif
- )
+ result = verify_mroutes(tgen, dut, SOURCE_ADDRESS, GROUP_ADDRESS_LIST_2, iif, oif)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("r4: Verify (*, G) upstream IIF interface")
step("r4: Verify (*, G) ip mroutes")
oif = "r4-r1-eth0"
- result = verify_ip_mroutes(tgen, dut, STAR, GROUP_ADDRESS_LIST_2, iif, oif)
+ result = verify_mroutes(tgen, dut, STAR, GROUP_ADDRESS_LIST_2, iif, oif)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("r4: Verify (S, G) upstream IIF interface")
step("r4: Verify (S, G) ip mroutes")
oif = "none"
- result = verify_ip_mroutes(
- tgen, dut, SOURCE_ADDRESS, GROUP_ADDRESS_LIST_2, iif, oif
- )
+ result = verify_mroutes(tgen, dut, SOURCE_ADDRESS, GROUP_ADDRESS_LIST_2, iif, oif)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("r3: Verify (S, G) upstream IIF interface")
step("r3: Verify (S, G) ip mroutes")
oif = "r3-r1-eth0"
- result = verify_ip_mroutes(
- tgen, dut, SOURCE_ADDRESS, GROUP_ADDRESS_LIST_2, iif, oif
- )
+ result = verify_mroutes(tgen, dut, SOURCE_ADDRESS, GROUP_ADDRESS_LIST_2, iif, oif)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
write_test_footer(tc_name)
step("Creating configuration from JSON")
reset_config_on_routers(tgen)
app_helper.stop_all_hosts()
- clear_ip_mroute(tgen)
- clear_ip_pim_interface_traffic(tgen, TOPO)
+ clear_mroute(tgen)
+ clear_pim_interface_traffic(tgen, TOPO)
# Steps to execute
step("Enable IGMP on r1 interface")
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("r1: Verify (*, G) ip mroutes")
- result = verify_ip_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif)
+ result = verify_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("r2: Verify (*, G) ip mroutes")
dut = "r2"
iif = "lo"
oif = "r2-r1-eth0"
- result = verify_ip_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif)
+ result = verify_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("r1: Shut the interface r1-r2-eth1 from R1 to R2")
dut = "r1"
iif = "r1-r3-eth2"
oif = "r1-r0-eth0"
- result = verify_ip_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif)
+ result = verify_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("r2: Verify (*, G) ip mroutes")
dut = "r2"
iif = "lo"
oif = "r2-r3-eth1"
- result = verify_ip_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif)
+ result = verify_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("r3: Verify (*, G) ip mroutes")
dut = "r3"
iif = "r3-r2-eth1"
oif = "r3-r1-eth0"
- result = verify_ip_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif)
+ result = verify_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("r3: Shut the link from R1 to R3 from R3 node")
dut = "r1"
iif = "r1-r3-eth2"
oif = "r1-r0-eth0"
- result = verify_ip_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif, expected=False)
+ result = verify_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif, expected=False)
assert result is not True, (
"Testcase {} : Failed \n "
"r1: (*,G) mroutes are not cleared after shut of R1 to R3 link\n Error: {}".format(
dut = "r2"
iif = "lo"
oif = "r2-r3-eth1"
- result = verify_ip_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif, expected=False)
+ result = verify_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif, expected=False)
assert result is not True, (
"Testcase {} : Failed \n "
"r2: (*,G) mroutes are not cleared after shut of R1 to R3 link\n Error: {}".format(
dut = "r3"
iif = "r3-r2-eth1"
oif = "r3-r1-eth0"
- result = verify_ip_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif, expected=False)
+ result = verify_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif, expected=False)
assert result is not True, (
"Testcase {} : Failed \n "
"r3: (*,G) mroutes are not cleared after shut of R1 to R3 link\n Error: {}".format(
dut = "r1"
iif = "r1-r3-eth2"
oif = "r1-r0-eth0"
- result = verify_ip_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif)
+ result = verify_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("r2: Verify (*, G) ip mroutes")
dut = "r2"
iif = "lo"
oif = "r2-r3-eth1"
- result = verify_ip_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif)
+ result = verify_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("r3: Verify (*, G) ip mroutes")
dut = "r3"
iif = "r3-r2-eth1"
oif = "r3-r1-eth0"
- result = verify_ip_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif)
+ result = verify_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("r1: No shutdown the link from R1 to R2 from R1 node")
dut = "r1"
iif = "r1-r2-eth1"
oif = "r1-r0-eth0"
- result = verify_ip_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif)
+ result = verify_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("r2: Verify (*, G) ip mroutes")
dut = "r2"
iif = "lo"
oif = "r2-r1-eth0"
- result = verify_ip_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif)
+ result = verify_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
write_test_footer(tc_name)
step("Creating configuration from JSON")
reset_config_on_routers(tgen)
app_helper.stop_all_hosts()
- clear_ip_mroute(tgen)
- clear_ip_pim_interface_traffic(tgen, TOPO)
+ clear_mroute(tgen)
+ clear_pim_interface_traffic(tgen, TOPO)
step("Enable IGMP on r1 interface")
step("Configure RP on r2 (loopback interface) for the group range" " 224.0.0.0/4")
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("r1: Verify (*, G) ip mroutes created")
- result = verify_ip_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif)
+ result = verify_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("r2: Verify (*, G) ip mroutes created")
dut = "r2"
iif = "lo"
oif = "r2-r1-eth0"
- result = verify_ip_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif)
+ result = verify_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("r1: Delete RP configuration")
dut = "r1"
iif = "r1-r2-eth1"
oif = "r1-r0-eth0"
- result = verify_ip_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif, expected=False)
+ result = verify_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif, expected=False)
assert result is not True, (
"Testcase {} : Failed \n "
"r1: (*,G) mroutes are not cleared after shut of R1 to R0 link\n Error: {}".format(
dut = "r2"
iif = "lo"
oif = "r2-r1-eth0"
- result = verify_ip_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif, expected=False)
+ result = verify_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif, expected=False)
assert result is not True, (
"Testcase {} : Failed \n "
"r2: (*,G) mroutes are not cleared after shut of R1 to R0 link\n Error: {}".format(
step("Creating configuration from JSON")
reset_config_on_routers(tgen)
app_helper.stop_all_hosts()
- clear_ip_mroute(tgen)
- clear_ip_pim_interface_traffic(tgen, TOPO)
+ clear_mroute(tgen)
+ clear_pim_interface_traffic(tgen, TOPO)
step("Enable IGMP on r1 interface")
step("Configure RP on r2 (lo) for the group range" " 224.0.0.0/4")
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("r1: Verify (*, G) ip mroutes created")
- result = verify_ip_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif)
+ result = verify_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("r2: Verify (*, G) ip mroutes created")
dut = "r2"
iif = "lo"
oif = "r2-r1-eth0"
- result = verify_ip_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif)
+ result = verify_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("r2: Delete RP configuration")
dut = "r1"
iif = "r1-r2-eth1"
oif = "r1-r0-eth0"
- result = verify_ip_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif, expected=False)
+ result = verify_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif, expected=False)
assert result is not True, (
"Testcase {} : Failed \n "
"r1: (*,G) mroutes are not cleared after shut of R1 to R2 and R3 link\n Error: {}".format(
dut = "r2"
iif = "lo"
oif = "r2-r1-eth0"
- result = verify_ip_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif, expected=False)
+ result = verify_mroutes(tgen, dut, STAR, GROUP_ADDRESS, iif, oif, expected=False)
assert result is not True, (
"Testcase {} : Failed \n "
"r2: (*,G) mroutes are not cleared after shut of R1 to R2 and R3 link\n Error: {}".format(
projects / mirror_frr.git / commitdiff