#!/usr/bin/env python
+# SPDX-License-Identifier: ISC
#
# Copyright (c) 2020 by VMware, Inc. ("VMware")
# Used Copyright (c) 2018 by Network Device Education Foundation,
# Inc. ("NetDEF") in this file.
#
-# Permission to use, copy, modify, and/or distribute this software
-# for any purpose with or without fee is hereby granted, provided
-# that the above copyright notice and this permission notice appear
-# in all copies.
-#
-# THE SOFTWARE IS PROVIDED "AS IS" AND VMWARE DISCLAIMS ALL WARRANTIES
-# WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
-# MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL VMWARE BE LIABLE FOR
-# ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY
-# DAMAGES WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS,
-# WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS
-# ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR PERFORMANCE
-# OF THIS SOFTWARE.
-#
"""
Following tests are covered to test multicast pim sm:
import os
import sys
-import json
import time
-import datetime
-from time import sleep
import pytest
pytestmark = pytest.mark.pimd
# pylint: disable=C0413
# Import topogen and topotest helpers
from lib.topogen import Topogen, get_topogen
-from mininet.topo import Topo
from lib.common_config import (
start_topology,
write_test_header,
write_test_footer,
step,
- iperfSendIGMPJoin,
- addKernelRoute,
reset_config_on_routers,
- iperfSendTraffic,
- kill_iperf,
shutdown_bringup_interface,
kill_router_daemons,
start_router,
create_pim_config,
create_igmp_config,
verify_igmp_groups,
- verify_ip_mroutes,
- verify_pim_interface_traffic,
+ verify_mroutes,
+ get_pim_interface_traffic,
verify_upstream_iif,
verify_pim_neighbors,
verify_pim_state,
- verify_ip_pim_join,
- clear_ip_mroute,
- clear_ip_pim_interface_traffic,
- verify_igmp_config,
+ clear_mroute,
+ clear_pim_interface_traffic,
+ McastTesterHelper,
)
from lib.topolog import logger
-from lib.topojson import build_topo_from_json, build_config_from_json
-
-pytestmark = [pytest.mark.pimd]
+from lib.topojson import build_config_from_json
-# Reading the data from JSON File for topology creation
-jsonFile = "{}/multicast_pim_sm_topo2.json".format(CWD)
-try:
- with open(jsonFile, "r") as topoJson:
- topo = json.load(topoJson)
-except IOError:
- assert False, "Could not read file {}".format(jsonFile)
+pytestmark = [pytest.mark.pimd]
TOPOLOGY = """
IGMP_JOIN_RANGE_3 = ["227.1.1.1", "227.1.1.2", "227.1.1.3", "227.1.1.4", "227.1.1.5"]
-class CreateTopo(Topo):
- """
- Test BasicTopo - topology 1
-
- * `Topo`: Topology object
- """
-
- def build(self, *_args, **_opts):
- """Build function"""
- tgen = get_topogen(self)
-
- # Building topology from json file
- build_topo_from_json(tgen, topo)
-
-
def setup_module(mod):
"""
Sets up the pytest environment
# Required linux kernel version for this suite to run.
result = required_linux_kernel_version("4.19")
if result is not True:
- pytest.skip("Kernel requirements are not met")
+ pytest.skip("Kernel version should be >= 4.19")
testsuite_run_time = time.asctime(time.localtime(time.time()))
logger.info("Testsuite start time: {}".format(testsuite_run_time))
logger.info("Running setup_module to create topology")
- tgen = Topogen(CreateTopo, mod.__name__)
+ json_file = "{}/multicast_pim_sm_topo2.json".format(CWD)
+ tgen = Topogen(json_file, mod.__name__)
+ global topo
+ topo = tgen.json_topo
# ... and here it calls Mininet initialization functions.
- # get list of daemons needs to be started for this suite.
- daemons = topo_daemons(tgen, topo)
-
# Starting topology, create tmp files which are loaded to routers
- # to start deamons and then start routers
- start_topology(tgen, daemons)
+ # to start daemons and then start routers
+ start_topology(tgen)
# Don"t run this test if we have any failure.
if tgen.routers_have_failure():
# Creating configuration from JSON
build_config_from_json(tgen, topo)
+ # XXX Replace this using "with McastTesterHelper()... " in each test if possible.
+ global app_helper
+ app_helper = McastTesterHelper(tgen)
+
logger.info("Running setup_module() done")
tgen = get_topogen()
+ app_helper.cleanup()
+
# Stop toplogy and Remove tmp files
tgen.stop_topology()
#####################################################
-def config_to_send_igmp_join_and_traffic(
- tgen, topo, tc_name, iperf, iperf_intf, GROUP_RANGE, join=False, traffic=False
-):
- """
- API to do pre-configuration to send IGMP join and multicast
- traffic
-
- parameters:
- -----------
- * `tgen`: topogen object
- * `topo`: input json data
- * `tc_name`: caller test case name
- * `iperf`: router running iperf
- * `iperf_intf`: interface name router running iperf
- * `GROUP_RANGE`: group range
- * `join`: IGMP join, default False
- * `traffic`: multicast traffic, default False
- """
-
- if join:
- # Add route to kernal
- result = addKernelRoute(tgen, iperf, iperf_intf, GROUP_RANGE)
- assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
-
- if traffic:
- # Add route to kernal
- result = addKernelRoute(tgen, iperf, iperf_intf, GROUP_RANGE)
- assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
-
- router_list = tgen.routers()
- for router in router_list.keys():
- if router == iperf:
- continue
-
- rnode = router_list[router]
- rnode.run("echo 2 > /proc/sys/net/ipv4/conf/all/rp_filter")
-
- return True
-
-
def verify_state_incremented(state_before, state_after):
"""
API to compare interface traffic state incrementing
tc_name = request.node.name
write_test_header(tc_name)
- # Creating configuration from JSON
- kill_iperf(tgen)
- clear_ip_mroute(tgen)
- reset_config_on_routers(tgen)
- clear_ip_pim_interface_traffic(tgen, topo)
-
# Don"t run this test if we have any failure.
if tgen.routers_have_failure():
pytest.skip(tgen.errors)
+ # Creating configuration from JSON
+ app_helper.stop_all_hosts()
+ clear_mroute(tgen)
+ reset_config_on_routers(tgen)
+ 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")
)
input_dict = {
- "f1": {"igmp": {"interfaces": {"f1-i8-eth2": {"igmp": {"version": "2"}}}}}
+ "f1": {
+ "igmp": {
+ "interfaces": {
+ "f1-i8-eth2": {
+ "igmp": {"version": "2", "query": {"query-interval": 15}}
+ }
+ }
+ }
+ }
}
result = create_igmp_config(tgen, topo, input_dict)
assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
input_join = {"i1": "i1-l1-eth0", "i8": "i8-f1-eth0"}
for recvr, recvr_intf in input_join.items():
- result = config_to_send_igmp_join_and_traffic(
- tgen, topo, tc_name, recvr, recvr_intf, _GROUP_RANGE, join=True
- )
- assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
-
- result = iperfSendIGMPJoin(tgen, recvr, _IGMP_JOIN_RANGE, join_interval=1)
+ result = app_helper.run_join(recvr, _IGMP_JOIN_RANGE, join_intf=recvr_intf)
assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
step(
input_src = {"i2": "i2-f1-eth0", "i5": "i5-c2-eth0"}
for src, src_intf in input_src.items():
- result = config_to_send_igmp_join_and_traffic(
- tgen, topo, tc_name, src, src_intf, _GROUP_RANGE, traffic=True
- )
- assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
-
- result = iperfSendTraffic(tgen, src, _IGMP_JOIN_RANGE, 32, 2500)
+ result = app_helper.run_traffic(src, _IGMP_JOIN_RANGE, bind_intf=src_intf)
assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
# Verifying mroutes before PIMd restart, fetching uptime
{"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"],
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 "
+ "Expected: [{}]: mroute (S, G) should not be present in mroute table \n "
+ "Found: {}".format(tc_name, data["dut"], result)
)
- logger.info("Expected Behavior: {}".format(result))
write_test_footer(tc_name)
tc_name = request.node.name
write_test_header(tc_name)
- # Creating configuration from JSON
- kill_iperf(tgen)
- clear_ip_mroute(tgen)
- reset_config_on_routers(tgen)
- clear_ip_pim_interface_traffic(tgen, topo)
-
# Don"t run this test if we have any failure.
if tgen.routers_have_failure():
pytest.skip(tgen.errors)
+ # Creating configuration from JSON
+ app_helper.stop_all_hosts()
+ clear_mroute(tgen)
+ reset_config_on_routers(tgen)
+ 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")
)
input_dict = {
- "f1": {"igmp": {"interfaces": {"f1-i8-eth2": {"igmp": {"version": "2"}}}}}
+ "f1": {
+ "igmp": {
+ "interfaces": {
+ "f1-i8-eth2": {
+ "igmp": {"version": "2", "query": {"query-interval": 15}}
+ }
+ }
+ }
+ }
}
result = create_igmp_config(tgen, topo, input_dict)
assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
input_join = {"i1": "i1-l1-eth0", "i8": "i8-f1-eth0"}
for recvr, recvr_intf in input_join.items():
- result = config_to_send_igmp_join_and_traffic(
- tgen, topo, tc_name, recvr, recvr_intf, _GROUP_RANGE, join=True
- )
- assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
-
- result = iperfSendIGMPJoin(tgen, recvr, _IGMP_JOIN_RANGE, join_interval=1)
+ result = app_helper.run_join(recvr, _IGMP_JOIN_RANGE, join_intf=recvr_intf)
assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
step(
input_src = {"i2": "i2-f1-eth0", "i5": "i5-c2-eth0"}
for src, src_intf in input_src.items():
- result = config_to_send_igmp_join_and_traffic(
- tgen, topo, tc_name, src, src_intf, _GROUP_RANGE, traffic=True
- )
- assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
-
- result = iperfSendTraffic(tgen, src, _IGMP_JOIN_RANGE, 32, 2500)
+ result = app_helper.run_traffic(src, _IGMP_JOIN_RANGE, bind_intf=src_intf)
assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
step("Verifying mroutes before FRR restart, fetching uptime")
{"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"],
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 "
+ "Expected: [{}]: mroute (S, G) should not be present in mroute table \n "
+ "Found: {}".format(tc_name, data["dut"], result)
)
- logger.info("Expected Behavior: {}".format(result))
write_test_footer(tc_name)
tc_name = request.node.name
write_test_header(tc_name)
- # Creating configuration from JSON
- kill_iperf(tgen)
- clear_ip_mroute(tgen)
- reset_config_on_routers(tgen)
- clear_ip_pim_interface_traffic(tgen, topo)
-
# Don"t run this test if we have any failure.
if tgen.routers_have_failure():
pytest.skip(tgen.errors)
+ # Creating configuration from JSON
+ app_helper.stop_all_hosts()
+ clear_mroute(tgen)
+ reset_config_on_routers(tgen)
+ clear_pim_interface_traffic(tgen, topo)
+
step("Configure static RP for (226.1.1.1-5) and " "(232.1.1.1-5) in c2")
_GROUP_RANGE = GROUP_RANGE_2 + GROUP_RANGE_3
"(226.1.1.1-5) and (232.1.1.1-5)"
)
- result = config_to_send_igmp_join_and_traffic(
- tgen, topo, tc_name, "i1", "i1-l1-eth0", _GROUP_RANGE, join=True
- )
- assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
-
- result = iperfSendIGMPJoin(tgen, "i1", _IGMP_JOIN_RANGE, join_interval=1)
+ result = app_helper.run_join("i1", _IGMP_JOIN_RANGE, "l1")
assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
step("Send multicast traffic from FRR3 to '226.1.1.1-5'" ", '232.1.1.1-5' receiver")
- result = config_to_send_igmp_join_and_traffic(
- tgen, topo, tc_name, "i2", "i2-f1-eth0", _GROUP_RANGE, traffic=True
- )
- assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
-
step("registerRx and registerStopTx value before traffic sent")
state_dict = {"c2": {"c2-f1-eth1": ["registerRx", "registerStopTx"]}}
- state_before = verify_pim_interface_traffic(tgen, state_dict)
+ state_before = get_pim_interface_traffic(tgen, state_dict)
assert isinstance(
state_before, dict
), "Testcase {} : Failed \n state_before is not dictionary \nError: {}".format(
tc_name, result
)
- result = iperfSendTraffic(tgen, "i2", _IGMP_JOIN_RANGE, 32, 2500)
+ result = app_helper.run_traffic("i2", _IGMP_JOIN_RANGE, "f1")
assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
step(
" '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"],
step("Stop the traffic to all the receivers")
- kill_iperf(tgen, "i2", "remove_traffic")
+ app_helper.stop_host("i2")
step(
"Null register packet being send periodically from FRR3 to RP, "
{"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"],
assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
step("registerRx and registerStopTx value after traffic sent")
- state_after = verify_pim_interface_traffic(tgen, state_dict)
+ state_after = get_pim_interface_traffic(tgen, state_dict)
assert isinstance(
state_after, dict
), "Testcase {} : Failed \n state_before is not dictionary \nError: {}".format(
tc_name = request.node.name
write_test_header(tc_name)
- # Creating configuration from JSON
- kill_iperf(tgen)
- clear_ip_mroute(tgen)
- reset_config_on_routers(tgen)
- clear_ip_pim_interface_traffic(tgen, topo)
-
# Don"t run this test if we have any failure.
if tgen.routers_have_failure():
pytest.skip(tgen.errors)
+ # Creating configuration from JSON
+ app_helper.stop_all_hosts()
+ clear_mroute(tgen)
+ reset_config_on_routers(tgen)
+ 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")
)
input_dict = {
- "f1": {"igmp": {"interfaces": {"f1-i8-eth2": {"igmp": {"version": "2"}}}}}
+ "f1": {
+ "igmp": {
+ "interfaces": {
+ "f1-i8-eth2": {
+ "igmp": {"version": "2", "query": {"query-interval": 15}}
+ }
+ }
+ }
+ }
}
result = create_igmp_config(tgen, topo, input_dict)
assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
input_join = {"i1": "i1-l1-eth0", "i8": "i8-f1-eth0"}
for recvr, recvr_intf in input_join.items():
- result = config_to_send_igmp_join_and_traffic(
- tgen, topo, tc_name, recvr, recvr_intf, _GROUP_RANGE, join=True
- )
- assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
-
- result = iperfSendIGMPJoin(tgen, recvr, _IGMP_JOIN_RANGE, join_interval=1)
+ result = app_helper.run_join(recvr, _IGMP_JOIN_RANGE, join_intf=recvr_intf)
assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
step(
input_src = {"i2": "i2-f1-eth0", "i5": "i5-c2-eth0"}
for src, src_intf in input_src.items():
- result = config_to_send_igmp_join_and_traffic(
- tgen, topo, tc_name, src, src_intf, _GROUP_RANGE, traffic=True
- )
- assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
-
- result = iperfSendTraffic(tgen, src, _IGMP_JOIN_RANGE, 32, 2500)
+ result = app_helper.run_traffic(src, _IGMP_JOIN_RANGE, bind_intf=src_intf)
assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
step(
{"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"],
intf_l1_c1 = "l1-c1-eth0"
shutdown_bringup_interface(tgen, dut, intf_l1_c1, False)
- done_flag = False
- for retry in range(1, 11):
- result = verify_upstream_iif(
- tgen, "l1", "Unknown", source, IGMP_JOIN_RANGE_2, expected=False
- )
- if result is not True:
- done_flag = True
- else:
- continue
-
- if done_flag:
- logger.info("Expected Behavior: {}".format(result))
- break
-
- assert done_flag is True, (
- "Testcase {} : Failed Error: \n "
- "mroutes are still present, after waiting for 10 mins".format(tc_name)
+ result = verify_upstream_iif(
+ tgen, "l1", "Unknown", source, IGMP_JOIN_RANGE_2, expected=False
+ )
+ assert result is not True, (
+ "Testcase {} : Failed \n "
+ "Expected: [{}]: Upstream IIF should be unknown \n "
+ "Found: {}".format(tc_name, "l1", result)
)
step("No shut the Source interface just after the upstream is expired" " from FRR1")
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"],
assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
step("Stop the traffic to all the receivers")
- kill_iperf(tgen)
+ app_helper.stop_all_hosts()
for data in input_dict:
- result = verify_ip_mroutes(
+ 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 "
+ "Expected: [{}]: mroute (S, G) should not be present in mroute table \n "
+ "Found: {}".format(tc_name, data["dut"], result)
)
- logger.info("Expected Behavior: {}".format(result))
write_test_footer(tc_name)
tc_name = request.node.name
write_test_header(tc_name)
- # Creating configuration from JSON
- kill_iperf(tgen)
- clear_ip_mroute(tgen)
- reset_config_on_routers(tgen)
- clear_ip_pim_interface_traffic(tgen, topo)
-
# Don"t run this test if we have any failure.
if tgen.routers_have_failure():
pytest.skip(tgen.errors)
+ # Creating configuration from JSON
+ app_helper.stop_all_hosts()
+ clear_mroute(tgen)
+ reset_config_on_routers(tgen)
+ 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)")
_GROUP_RANGE = GROUP_RANGE_2 + GROUP_RANGE_3
"Enable IGMP on FRR1 interface and send IGMP join"
" (226.1.1.1-5) and (232.1.1.1-5)"
)
- result = config_to_send_igmp_join_and_traffic(
- tgen, topo, tc_name, "i1", "i1-l1-eth0", _GROUP_RANGE, join=True
- )
- assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
-
- result = iperfSendIGMPJoin(tgen, "i1", _IGMP_JOIN_RANGE, join_interval=1)
+ result = app_helper.run_join("i1", _IGMP_JOIN_RANGE, "l1")
assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
step(
"Send multicast traffic from FRR3 to all the receivers "
"(226.1.1.1-5) and (232.1.1.1-5)"
)
- result = config_to_send_igmp_join_and_traffic(
- tgen, topo, tc_name, "i2", "i2-f1-eth0", _GROUP_RANGE, traffic=True
- )
- assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
-
- result = iperfSendTraffic(tgen, "i2", _IGMP_JOIN_RANGE, 32, 2500)
+ result = app_helper.run_traffic("i2", _IGMP_JOIN_RANGE, "f1")
assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
step(
" join (226.1.1.1-5) and (232.1.1.1-5)"
)
input_dict = {
- "c2": {"igmp": {"interfaces": {"c2-i5-eth2": {"igmp": {"version": "2"}}}}}
+ "c2": {
+ "igmp": {
+ "interfaces": {
+ "c2-i5-eth2": {
+ "igmp": {"version": "2", "query": {"query-interval": 15}}
+ }
+ }
+ }
+ }
}
result = create_igmp_config(tgen, topo, input_dict)
assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
- result = config_to_send_igmp_join_and_traffic(
- tgen, topo, tc_name, "i5", "i5-c2-eth0", _GROUP_RANGE, join=True
- )
- assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
-
- result = iperfSendIGMPJoin(tgen, "i5", _IGMP_JOIN_RANGE, join_interval=1)
+ result = app_helper.run_join("i5", _IGMP_JOIN_RANGE, "c2")
assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
step("FRR1 has 10 (*.G) and 10 (S,G) verify using 'show ip mroute count'")
{"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"],
tc_name = request.node.name
write_test_header(tc_name)
- # Creating configuration from JSON
- kill_iperf(tgen)
- clear_ip_mroute(tgen)
- reset_config_on_routers(tgen)
- clear_ip_pim_interface_traffic(tgen, topo)
-
# Don"t run this test if we have any failure.
if tgen.routers_have_failure():
pytest.skip(tgen.errors)
+ # Creating configuration from JSON
+ app_helper.stop_all_hosts()
+ clear_mroute(tgen)
+ reset_config_on_routers(tgen)
+ 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
_IGMP_JOIN_RANGE = IGMP_JOIN_RANGE_2 + IGMP_JOIN_RANGE_3
"(226.1.1.1-5) and (232.1.1.1-5)"
)
- result = config_to_send_igmp_join_and_traffic(
- tgen, topo, tc_name, "i1", "i1-l1-eth0", _GROUP_RANGE, join=True
- )
- assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
-
- result = config_to_send_igmp_join_and_traffic(
- tgen, topo, tc_name, "i2", "i2-f1-eth0", _GROUP_RANGE, traffic=True
- )
- assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
-
step("Send IGMP join (226.1.1.1-5, 232.1.1.1-5) to LHR(l1)")
- result = iperfSendIGMPJoin(tgen, "i1", _IGMP_JOIN_RANGE, join_interval=1)
+ result = app_helper.run_join("i1", _IGMP_JOIN_RANGE, "l1")
assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
step("Send multicast traffic from FRR3 to '226.1.1.1-5'" ", '232.1.1.1-5' receiver")
- result = iperfSendTraffic(tgen, "i2", _IGMP_JOIN_RANGE, 32, 2500)
+ result = app_helper.run_traffic("i2", _IGMP_JOIN_RANGE, "f1")
assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
step(
step("Configure one IGMP interface on f1 node and send IGMP" " join (225.1.1.1)")
input_dict = {
- "f1": {"igmp": {"interfaces": {"f1-i8-eth2": {"igmp": {"version": "2"}}}}}
+ "f1": {
+ "igmp": {
+ "interfaces": {
+ "f1-i8-eth2": {
+ "igmp": {"version": "2", "query": {"query-interval": 15}}
+ }
+ }
+ }
+ }
}
result = create_igmp_config(tgen, topo, input_dict)
assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
- result = config_to_send_igmp_join_and_traffic(
- tgen, topo, tc_name, "i8", "i8-f1-eth0", _GROUP_RANGE, join=True
- )
- assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
-
- result = iperfSendIGMPJoin(tgen, "i8", _IGMP_JOIN_RANGE, join_interval=1)
+ result = app_helper.run_join("i8", _IGMP_JOIN_RANGE, "f1")
assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
step(
"l1 and f1 has 10 IGMP groups (226.1.1.1-5, 232.1.1.1-5),"
{"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"],
# Stop the multicast traffic
step("Stop the traffic to all the receivers")
- kill_iperf(tgen)
+ app_helper.stop_all_hosts()
step(
"After traffic stopped , verify (*,G) entries are not flushed"
{"dut": "f1", "src_address": "*", "iif": "f1-c2-eth0", "oil": "f1-i8-eth2"},
{"dut": "l1", "src_address": "*", "iif": "l1-c1-eth0", "oil": "l1-i1-eth1"},
]
-
- done_flag = False
- for retry in range(1, 11):
- for data in input_dict:
- result = verify_ip_mroutes(
- tgen,
- data["dut"],
- data["src_address"],
- _IGMP_JOIN_RANGE,
- data["iif"],
- data["oil"],
- )
-
- if result is True:
- done_flag = True
- else:
- continue
-
- if done_flag:
- break
-
- assert done_flag is True, (
- "Testcase {} : Failed Error: \n "
- "mroutes are still present, after waiting for 10 mins".format(tc_name)
- )
+ for data in input_dict:
+ result = verify_mroutes(
+ tgen,
+ data["dut"],
+ data["src_address"],
+ _IGMP_JOIN_RANGE,
+ data["iif"],
+ data["oil"],
+ )
+ assert (
+ result is True
+ ), "Testcase {} : Failed Error mroutes were flushed.".format(tc_name)
step(
"After traffic stopped , verify (S,G) entries are flushed out"
{"dut": "f1", "src_address": source, "iif": "i2-f1-eth0", "oil": "f1-r2-eth3"},
]
- done_flag = False
- for retry in range(1, 11):
- for data in input_dict:
- result = verify_ip_mroutes(
- tgen,
- data["dut"],
- data["src_address"],
- _IGMP_JOIN_RANGE,
- data["iif"],
- data["oil"],
- expected=False,
- )
- if result is not True:
- done_flag = True
- else:
- continue
-
- if done_flag:
- logger.info("Expected Behavior: {}".format(result))
- break
-
- assert done_flag is True, (
- "Testcase {} : Failed Error: \n "
- "mroutes are still present, after waiting for 10 mins".format(tc_name)
- )
+ for data in input_dict:
+ result = verify_mroutes(
+ tgen,
+ data["dut"],
+ data["src_address"],
+ _IGMP_JOIN_RANGE,
+ data["iif"],
+ data["oil"],
+ expected=False,
+ )
+ assert result is not True, (
+ "Testcase {} : Failed \n "
+ "Expected: [{}]: mroute (S, G) should not be present in mroute table \n "
+ "Found: {}".format(tc_name, data["dut"], result)
+ )
write_test_footer(tc_name)
tc_name = request.node.name
write_test_header(tc_name)
- # Creating configuration from JSON
- kill_iperf(tgen)
- clear_ip_mroute(tgen)
- reset_config_on_routers(tgen)
- clear_ip_pim_interface_traffic(tgen, topo)
-
# Don"t run this test if we have any failure.
if tgen.routers_have_failure():
pytest.skip(tgen.errors)
+ # Creating configuration from JSON
+ app_helper.stop_all_hosts()
+ clear_mroute(tgen)
+ reset_config_on_routers(tgen)
+ 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")
"f1": {
"igmp": {
"interfaces": {
- "f1-i8-eth2": {"igmp": {"version": "2"}},
- "f1-i2-eth1": {"igmp": {"version": "2"}},
+ "f1-i8-eth2": {
+ "igmp": {"version": "2", "query": {"query-interval": 15}}
+ },
+ "f1-i2-eth1": {
+ "igmp": {"version": "2", "query": {"query-interval": 15}}
+ },
+ }
+ }
+ },
+ "l1": {
+ "igmp": {
+ "interfaces": {
+ "l1-i6-eth2": {
+ "igmp": {"version": "2", "query": {"query-interval": 15}}
+ }
}
}
},
- "l1": {"igmp": {"interfaces": {"l1-i6-eth2": {"igmp": {"version": "2"}}}}},
}
result = create_igmp_config(tgen, topo, input_dict)
assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
}
for recvr, recvr_intf in input_join.items():
- result = config_to_send_igmp_join_and_traffic(
- tgen, topo, tc_name, recvr, recvr_intf, _GROUP_RANGE, join=True
- )
+ result = app_helper.run_join(recvr, _IGMP_JOIN_RANGE, join_intf=recvr_intf)
assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
- result = iperfSendIGMPJoin(tgen, recvr, _IGMP_JOIN_RANGE, join_interval=1)
- assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
step("Configure one source in FRR2 , one in c1")
step(
"Send multicast traffic from both the sources to all the"
input_src = {"i3": "i3-r2-eth0"}
for src, src_intf in input_src.items():
- result = config_to_send_igmp_join_and_traffic(
- tgen, topo, tc_name, src, src_intf, _GROUP_RANGE, traffic=True
- )
- assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
-
- result = iperfSendTraffic(tgen, src, _IGMP_JOIN_RANGE, 32, 2500)
+ result = app_helper.run_traffic(src, _IGMP_JOIN_RANGE, bind_intf=src_intf)
assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
step(
"After all the IGMP groups received with correct port using"
source = topo["routers"]["i3"]["links"]["r2"]["ipv4"].split("/")[0]
input_dict_all = [
- {"dut": "l1", "src_address": source, "iif": ["l1-r2-eth4", "l1-c1-eth0"],
- "oil": ["l1-i1-eth1", "l1-i6-eth2"]},
+ {
+ "dut": "l1",
+ "src_address": source,
+ "iif": ["l1-r2-eth4", "l1-c1-eth0"],
+ "oil": ["l1-i1-eth1", "l1-i6-eth2"],
+ },
{"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"],
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 "
+ "Expected: [{}]: mroute (S, G) should not be present in mroute table \n "
+ "Found: {}".format(tc_name, data["dut"], result)
)
- logger.info("Expected Behavior: {}".format(result))
step(
"No traffic impact observed on other receivers verify using"
{"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"],
tc_name = request.node.name
write_test_header(tc_name)
- # Creating configuration from JSON
- kill_iperf(tgen)
- clear_ip_mroute(tgen)
- reset_config_on_routers(tgen)
- clear_ip_pim_interface_traffic(tgen, topo)
-
# Don"t run this test if we have any failure.
if tgen.routers_have_failure():
pytest.skip(tgen.errors)
+ # Creating configuration from JSON
+ app_helper.stop_all_hosts()
+ clear_mroute(tgen)
+ reset_config_on_routers(tgen)
+ 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")
)
input_dict = {
- "f1": {"igmp": {"interfaces": {"f1-i8-eth2": {"igmp": {"version": "2"}}}}}
+ "f1": {
+ "igmp": {
+ "interfaces": {
+ "f1-i8-eth2": {
+ "igmp": {"version": "2", "query": {"query-interval": 15}}
+ }
+ }
+ }
+ }
}
result = create_igmp_config(tgen, topo, input_dict)
assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
input_join = {"i1": "i1-l1-eth0", "i8": "i8-f1-eth0"}
for recvr, recvr_intf in input_join.items():
- result = config_to_send_igmp_join_and_traffic(
- tgen, topo, tc_name, recvr, recvr_intf, _GROUP_RANGE, join=True
- )
- assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
-
- result = iperfSendIGMPJoin(tgen, recvr, _IGMP_JOIN_RANGE, join_interval=1)
+ result = app_helper.run_join(recvr, _IGMP_JOIN_RANGE, join_intf=recvr_intf)
assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
step("Configure 1 source in FRR1 , 1 in FRR3")
input_src = {"i6": "i6-l1-eth0", "i2": "i2-f1-eth0"}
for src, src_intf in input_src.items():
- result = config_to_send_igmp_join_and_traffic(
- tgen, topo, tc_name, src, src_intf, _GROUP_RANGE, traffic=True
- )
- assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
-
- result = iperfSendTraffic(tgen, src, _IGMP_JOIN_RANGE, 32, 2500)
+ result = app_helper.run_traffic(src, _IGMP_JOIN_RANGE, bind_intf=src_intf)
assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
step(
},
]
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,
"f1-i8-eth2",
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 "
+ "Expected: [{}]: mroute (S, G) should not be present in mroute table \n "
+ "Found: {}".format(tc_name, data["dut"], result)
)
- logger.info("Expected Behavior: {}".format(result))
result = verify_upstream_iif(
tgen, "f1", "Unknown", "10.0.5.2", _IGMP_JOIN_RANGE, joinState="NotJoined"
projects / mirror_frr.git / blobdiff