proxmox-mirrors/mirror_frr
1
0
Fork 0
mirror of https://git.proxmox.com/git/mirror_frr synced 2025-08-16 00:25:01 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

Merge pull request #12310 from kuldeepkash/pim_v6

Browse Source 
tests: [PIMv6] Adding multicast pim6 SM testsuite
This commit is contained in:
Donatas Abraitis 2023-03-15 08:55:24 +02:00 committed by GitHub
commit 52523db51c
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 2675 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

166
tests/topotests/lib/pim.py
View File

@ -4524,12 +4524,11 @@ def verify_mld_config(tgen, input_dict, stats_return=False, expected=True):
for dut in input_dict.keys():
rnode = tgen.routers()[dut]
for interface, data in input_dict[dut]["igmp"]["interfaces"].items():
for interface, data in input_dict[dut]["mld"]["interfaces"].items():
statistics = False
report = False
if "statistics" in input_dict[dut]["igmp"]["interfaces"][interface]["igmp"]:
if "statistics" in input_dict[dut]["mld"]["interfaces"][interface]["mld"]:
statistics = True
cmd = "show ipv6 mld statistics"
else:
@ -4556,6 +4555,8 @@ def verify_mld_config(tgen, input_dict, stats_return=False, expected=True):
rnode, "{} interface {} json".format(cmd, interface), isjson=True
)
show_ipv6_mld_intf_json = show_ipv6_mld_intf_json["default"]
if not report:
if interface not in show_ipv6_mld_intf_json:
errormsg = (
@ -4636,7 +4637,7 @@ def verify_mld_config(tgen, input_dict, stats_return=False, expected=True):
for query, value in data["mld"]["query"].items():
if query == "query-interval":
# Verifying IGMP interface query interval timer
if intf_detail_json["timerQueryInterval"] != value:
if intf_detail_json["timerQueryIntervalMsec"] != value * 1000:
errormsg = (
"[DUT %s]: MLD interface: %s "
" query-interval verification "
@ -4646,7 +4647,7 @@ def verify_mld_config(tgen, input_dict, stats_return=False, expected=True):
dut,
interface,
value,
intf_detail_json["timerQueryInterval"],
intf_detail_json["timerQueryIntervalMsec"],
)
)
return errormsg
@ -4655,13 +4656,13 @@ def verify_mld_config(tgen, input_dict, stats_return=False, expected=True):
"[DUT %s]: MLD interface: %s " "query-interval is %s",
dut,
interface,
value,
value * 1000,
)
if query == "query-max-response-time":
# Verifying IGMP interface query max response timer
if (
intf_detail_json["timerQueryResponseIntervalMsec"]
intf_detail_json["timerQueryResponseTimerMsec"]
!= value * 100
):
errormsg = (
@ -4672,8 +4673,8 @@ def verify_mld_config(tgen, input_dict, stats_return=False, expected=True):
% (
dut,
interface,
value * 1000,
intf_detail_json["timerQueryResponseIntervalMsec"],
value * 100,
intf_detail_json["timerQueryResponseTimerMsec"],
)
)
return errormsg
@ -4714,8 +4715,8 @@ def verify_mld_config(tgen, input_dict, stats_return=False, expected=True):
if query == "last-member-query-interval":
# Verifying IGMP interface last member query interval
if (
intf_detail_json["timerLastMemberQueryMsec"]
!= value * 100 * intf_detail_json["lastMemberQueryCount"]
intf_detail_json["timerLastMemberQueryIntervalMsec"]
!= value * 100
):
errormsg = (
"[DUT %s]: MLD interface: %s "
@ -4725,8 +4726,10 @@ def verify_mld_config(tgen, input_dict, stats_return=False, expected=True):
% (
dut,
interface,
value * 1000,
intf_detail_json["timerLastMemberQueryMsec"],
value * 100,
intf_detail_json[
"timerLastMemberQueryIntervalMsec"
],
)
)
return errormsg
@ -4736,7 +4739,7 @@ def verify_mld_config(tgen, input_dict, stats_return=False, expected=True):
"last-member-query-interval is %s ms",
dut,
interface,
value * intf_detail_json["lastMemberQueryCount"] * 100,
value * 100,
)
if "version" in data["mld"]:
@ -4761,7 +4764,7 @@ def verify_mld_config(tgen, input_dict, stats_return=False, expected=True):
@retry(retry_timeout=60, diag_pct=0)
def verify_pim_nexthop(tgen, topo, dut, nexthop, addr_type):
def verify_pim_nexthop(tgen, topo, dut, nexthop, addr_type="ipv4"):
"""
Verify all PIM nexthop details using "show ip/ipv6 pim neighbor" cli
@ -4895,6 +4898,7 @@ def verify_mroute_summary(
return True
@retry(retry_timeout=60, diag_pct=0)
def verify_sg_traffic(tgen, dut, groups, src, addr_type="ipv4"):
"""
Verify multicast traffic by running
@ -4907,7 +4911,7 @@ def verify_sg_traffic(tgen, dut, groups, src, addr_type="ipv4"):
Usage
-----
result = verify_sg_traffic(tgen, "r1", igmp_groups, srcaddress)
result = verify_sg_traffic(tgen, "r1", igmp_groups/mld_groups, srcaddress)
Returns
-------
@ -4973,7 +4977,7 @@ def verify_sg_traffic(tgen, dut, groups, src, addr_type="ipv4"):
after_traffic[grp] = show_mroute_sg_traffic_json[grp][src]["packets"]
for grp in groups:
if after_traffic[grp] < before_traffic[grp]:
if after_traffic[grp] <= before_traffic[grp]:
errormsg = (
"[DUT %s]: Verifying igmp group %s source %s not increamenting traffic"
" [FAILED]!! " % (dut, grp, src)
@ -4988,6 +4992,134 @@ def verify_sg_traffic(tgen, dut, groups, src, addr_type="ipv4"):
return result
@retry(retry_timeout=60, diag_pct=0)
def verify_pim6_config(tgen, input_dict, expected=True):
"""
Verify pim interface details, verifying following configs:
drPriority
helloPeriod
helloReceived
helloSend
drAddress
Parameters
----------
* `tgen`: topogen object
* `input_dict` : Input dict data, required to verify
timer
* `expected` : expected results from API, by-default True
Usage
-----
input_dict ={
"l1": {
"mld": {
"interfaces": {
"l1-i1-eth1": {
"pim6": {
"drPriority" : 10,
"helloPeriod" : 5
}
}
}
}
}
}
}
result = verify_pim6_config(tgen, input_dict)
Returns
-------
errormsg(str) or True
"""
logger.debug("Entering lib API: {}".format(sys._getframe().f_code.co_name))
for dut in input_dict.keys():
rnode = tgen.routers()[dut]
for interface, data in input_dict[dut]["pim6"]["interfaces"].items():
logger.info(
"[DUT: %s]: Verifying PIM6 interface %s detail:", dut, interface
)
show_ipv6_pim_intf_json = run_frr_cmd(
rnode, "show ipv6 pim interface {} json".format(interface), isjson=True
)
if interface not in show_ipv6_pim_intf_json:
errormsg = (
"[DUT %s]: PIM6 interface: %s "
" is not present in CLI output "
"[FAILED]!! " % (dut, interface)
)
return errormsg
intf_detail_json = show_ipv6_pim_intf_json[interface]
for config, value in data.items():
if config == "helloPeriod":
# Verifying PIM interface helloPeriod
if intf_detail_json["helloPeriod"] != value:
errormsg = (
"[DUT %s]: PIM6 interface: %s "
" helloPeriod verification "
"[FAILED]!! Expected : %s,"
" Found : %s"
% (dut, interface, value, intf_detail_json["helloPeriod"])
)
return errormsg
logger.info(
"[DUT %s]: PIM6 interface: %s " "helloPeriod is %s",
dut,
interface,
value,
)
if config == "drPriority":
# Verifying PIM interface drPriority
if intf_detail_json["drPriority"] != value:
errormsg = (
"[DUT %s]: PIM6 interface: %s "
" drPriority verification "
"[FAILED]!! Expected : %s,"
" Found : %s"
% (dut, interface, value, intf_detail_json["drPriority"])
)
return errormsg
logger.info(
"[DUT %s]: PIM6 interface: %s " "drPriority is %s",
dut,
interface,
value,
)
if config == "drAddress":
# Verifying PIM interface drAddress
if intf_detail_json["drAddress"] != value:
errormsg = (
"[DUT %s]: PIM6 interface: %s "
" drAddress verification "
"[FAILED]!! Expected : %s,"
" Found : %s"
% (dut, interface, value, intf_detail_json["drAddress"])
)
return errormsg
logger.info(
"[DUT %s]: PIM6 interface: %s " "drAddress is %s",
dut,
interface,
value,
)
logger.debug("Exiting lib API: {}".format(sys._getframe().f_code.co_name))
return True
# def cleanup(self):
# super(McastTesterHelper, self).cleanup()

300
tests/topotests/multicast_pim6_sm_topo1/multicast_pim6_sm_topo1.json Normal file
View File

@ -0,0 +1,300 @@
{
"address_types": ["ipv6"],
"ipv6base": "fd00::",
"ipv6mask": 64,
"link_ip_start": {
"ipv6": "fd00::",
"v6mask": 64
},
"lo_prefix": {
"ipv6": "2001:db8:f::",
"v6mask": 128
},
"routers": {
"r1": {
"links": {
"lo": {"ipv6": "auto", "type": "loopback", "pim6": "enable"},
"r2": {"ipv6": "auto", "pim6": "enable"},
"r4": {"ipv6": "auto", "pim6": "enable"},
"i1": {"ipv6": "auto", "pim6": "enable"},
"i6": {"ipv6": "auto", "pim6": "enable"},
"i7": {"ipv6": "auto", "pim6": "enable"}
},
"bgp": {
"local_as": "100",
"router_id": "192.168.1.1",
"address_family": {
"ipv4": {
"unicast": {
"neighbor": {
"r2": {
"dest_link": {
"r1": {}
}
},
"r4": {
"dest_link": {
"r1": {}
}
}
}
}
},
"ipv6": {
"unicast": {
"redistribute": [
{"redist_type": "static"},
{"redist_type": "connected"}
],
"neighbor": {
"r2": {
"dest_link": {
"r1": {}
}
},
"r4": {
"dest_link": {
"r1": {}
}
}
}
}
}
}
},
"mld": {
"interfaces": {
"r1-i1-eth0" :{
"mld":{
"version": 1
}
}
}
}
},
"r2": {
"links": {
"lo": {"ipv6": "auto", "type": "loopback", "pim6": "enable"},
"r1": {"ipv6": "auto", "pim6": "enable"},
"r3": {"ipv6": "auto", "pim6": "enable"},
"i3": {"ipv6": "auto", "pim6": "enable"}
},
"bgp": {
"local_as": "200",
"router_id": "192.168.1.2",
"address_family": {
"ipv4": {
"unicast": {
"neighbor": {
"r1": {
"dest_link": {
"r2": {}
}
},
"r3": {
"dest_link": {
"r2": {}
}
}
}
}
},
"ipv6": {
"unicast": {
"redistribute": [
{"redist_type": "static"},
{"redist_type": "connected"}
],
"neighbor": {
"r1": {
"dest_link": {
"r2": {}
}
},
"r3": {
"dest_link": {
"r2": {}
}
}
}
}
}
}
}
},
"r3": {
"links": {
"lo": {"ipv6": "auto", "type": "loopback", "pim6": "enable"},
"r2": {"ipv6": "auto", "pim6": "enable"},
"r5": {"ipv6": "auto", "pim6": "enable"},
"i2": {"ipv6": "auto", "pim6": "enable"},
"i8": {"ipv6": "auto", "pim6": "enable"}
},
"bgp": {
"local_as": "300",
"router_id": "192.168.1.3",
"address_family": {
"ipv4": {
"unicast": {
}
},
"ipv6": {
"unicast": {
"redistribute": [
{"redist_type": "static"},
{"redist_type": "connected"}
],
"neighbor": {
"r2": {
"dest_link": {
"r3": {}
}
},
"r5": {
"dest_link": {
"r3": {}
}
}
}
}
}
}
}
},
"r4": {
"links": {
"lo": {"ipv6": "auto", "type": "loopback", "pim6": "enable"},
"r1": {"ipv6": "auto", "pim6": "enable"},
"r5": {"ipv6": "auto", "pim6": "enable"},
"i4": {"ipv6": "auto", "pim6": "enable"}
},
"bgp": {
"local_as": "400",
"router_id": "192.168.1.4",
"address_family": {
"ipv4": {
"unicast": {
"neighbor": {
"r1": {
"dest_link": {
"r4": {}
}
},
"r5": {
"dest_link": {
"r4": {}
}
}
}
}
},
"ipv6": {
"unicast": {
"redistribute": [
{"redist_type": "static"},
{"redist_type": "connected"}
],
"neighbor": {
"r1": {
"dest_link": {
"r4": {}
}
},
"r5": {
"dest_link": {
"r4": {}
}
}
}
}
}
}
}
},
"r5": {
"links": {
"lo": {"ipv6": "auto", "type": "loopback", "pim6": "enable"},
"r3": {"ipv6": "auto", "pim6": "enable"},
"r4": {"ipv6": "auto", "pim6": "enable"},
"i5": {"ipv6": "auto", "pim6": "enable"}
},
"bgp": {
"local_as": "500",
"router_id": "192.168.1.5",
"address_family": {
"ipv4": {
"unicast": {
}
},
"ipv6": {
"unicast": {
"redistribute": [
{"redist_type": "static"},
{"redist_type": "connected"}
],
"neighbor": {
"r3": {
"dest_link": {
"r5": {}
}
},
"r4": {
"dest_link": {
"r5": {}
}
}
}
}
}
}
}
},
"i1": {
"links": {
"r1": {"ipv6": "auto"}
}
},
"i2": {
"links": {
"r3": {"ipv6": "auto"}
}
},
"i3": {
"links": {
"r2": {"ipv6": "auto"}
}
},
"i4": {
"links": {
"r4": {"ipv6": "auto"}
}
},
"i5": {
"links": {
"r5": {"ipv6": "auto"}
}
},
"i6": {
"links": {
"r1": {"ipv6": "auto"}
}
},
"i7": {
"links": {
"r1": {"ipv6": "auto"}
}
},
"i8": {
"links": {
"r3": {"ipv6": "auto"}
}
}
}
}

1618
tests/topotests/multicast_pim6_sm_topo1/test_multicast_pim6_sm1.py Normal file
View File

File diff suppressed because it is too large Load Diff

608
tests/topotests/multicast_pim6_sm_topo1/test_multicast_pim6_sm2.py Normal file
View File

@ -0,0 +1,608 @@
#!/usr/bin/env python
# SPDX-License-Identifier: ISC
#
# Copyright (c) 2022 by VMware, Inc. ("VMware")
# Used Copyright (c) 2018 by Network Device Education Foundation,
# Inc. ("NetDEF") in this file.
#
"""
Following tests are covered to test multicast pim6 sm:
Test steps
- Create topology (setup module)
- Bring up topology
Following tests are covered:
1. Verify (*,G) and (S,G) entry populated again after clear the
PIM nbr and mroute from FRR node
2. Verify SPT switchover working when RPT and SPT path is
different
"""
import os
import sys
import json
import time
import datetime
import pytest
# Save the Current Working Directory to find configuration files.
CWD = os.path.dirname(os.path.realpath(__file__))
sys.path.append(os.path.join(CWD, "../"))
sys.path.append(os.path.join(CWD, "../lib/"))
# Required to instantiate the topology builder class.
# pylint: disable=C0413
# Import topogen and topotest helpers
from lib.topogen import Topogen, get_topogen
from lib.common_config import (
start_topology,
write_test_header,
write_test_footer,
step,
reset_config_on_routers,
shutdown_bringup_interface,
start_router,
stop_router,
create_static_routes,
required_linux_kernel_version,
socat_send_mld_join,
socat_send_pim6_traffic,
get_frr_ipv6_linklocal,
)
from lib.bgp import create_router_bgp
from lib.pim import (
create_pim_config,
create_mld_config,
verify_mld_groups,
verify_mroutes,
clear_pim6_interface_traffic,
verify_upstream_iif,
clear_pim6_mroute,
verify_pim_interface_traffic,
verify_pim_state,
McastTesterHelper,
verify_pim_join,
verify_mroute_summary,
verify_pim_nexthop,
verify_sg_traffic,
verify_mld_config,
)
from lib.topolog import logger
from lib.topojson import build_config_from_json
# Global variables
GROUP_RANGE = "ff00::/8"
GROUP_RANGE_1 = [
"ffaa::1/128",
"ffaa::2/128",
"ffaa::3/128",
"ffaa::4/128",
"ffaa::5/128",
]
MLD_JOIN_RANGE_1 = ["ffaa::1", "ffaa::2", "ffaa::3", "ffaa::4", "ffaa::5"]
GROUP_RANGE_2 = [
"ffbb::1/128",
"ffbb::2/128",
"ffbb::3/128",
"ffbb::4/128",
"ffbb::5/128",
]
MLD_JOIN_RANGE_2 = ["ffbb::1", "ffbb::2", "ffbb::3", "ffbb::4", "ffbb::5"]
GROUP_RANGE_3 = [
"ffcc::1/128",
"ffcc::2/128",
"ffcc::3/128",
"ffcc::4/128",
"ffcc::5/128",
]
MLD_JOIN_RANGE_3 = ["ffcc::1", "ffcc::2", "ffcc::3", "ffcc::4", "ffcc::5"]
HELLO_TIMER = 1
HOLD_TIMER = 3
PREFERRED_NEXT_HOP = "link_local"
ASSERT_MSG = "Testcase {} : Failed Error: {}"
pytestmark = [pytest.mark.pim6d]
def setup_module(mod):
"""
Sets up the pytest environment
* `mod`: module name
"""
# 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")
testsuite_run_time = time.asctime(time.localtime(time.time()))
logger.info("Testsuite start time: {}".format(testsuite_run_time))
logger.info("=" * 40)
logger.info("Running setup_module to create topology")
testdir = os.path.dirname(os.path.realpath(__file__))
json_file = "{}/multicast_pim6_sm_topo1.json".format(testdir)
tgen = Topogen(json_file, mod.__name__)
global topo
topo = tgen.json_topo
# ... and here it calls Mininet initialization functions.
# Starting topology, create tmp files which are loaded to routers
# to start deamons and then start routers
start_topology(tgen)
# Don"t run this test if we have any failure.
if tgen.routers_have_failure():
pytest.skip(tgen.errors)
# Creating configuration from JSON
build_config_from_json(tgen, tgen.json_topo)
# XXX Replace this using "with McastTesterHelper()... " in each test if possible.
global app_helper
app_helper = McastTesterHelper(tgen)
logger.info("Running setup_module() done")
def teardown_module():
"""Teardown the pytest environment"""
logger.info("Running teardown_module to delete topology")
tgen = get_topogen()
app_helper.cleanup()
# Stop toplogy and Remove tmp files
tgen.stop_topology()
logger.info(
"Testsuite end time: {}".format(time.asctime(time.localtime(time.time())))
)
logger.info("=" * 40)
#####################################################
#
# Local APIs
#
#####################################################
def verify_state_incremented(state_before, state_after):
"""
API to compare interface traffic state incrementing
Parameters
----------
* `state_before` : State dictionary for any particular instance
* `state_after` : State dictionary for any particular instance
"""
for router, state_data in state_before.items():
for state, value in state_data.items():
if state_before[router][state] >= state_after[router][state]:
errormsg = (
"[DUT: %s]: state %s value has not"
" incremented, Initial value: %s, "
"Current value: %s [FAILED!!]"
% (
router,
state,
state_before[router][state],
state_after[router][state],
)
)
return errormsg
logger.info(
"[DUT: %s]: State %s value is "
"incremented, Initial value: %s, Current value: %s"
" [PASSED!!]",
router,
state,
state_before[router][state],
state_after[router][state],
)
return True
#####################################################
#
# Testcases
#
#####################################################
def test_clear_mroute_and_verify_multicast_data_p0(request):
"""
Verify (*,G) and (S,G) entry populated again after clear the
PIM nbr and mroute from FRR node
"""
tgen = get_topogen()
tc_name = request.node.name
write_test_header(tc_name)
# Creating configuration from JSON
reset_config_on_routers(tgen)
# Don"t run this test if we have any failure.
if tgen.routers_have_failure():
pytest.skip(tgen.errors)
step("Configure static RP on r4 for group (ffcc::1-5)")
input_dict = {
"r4": {
"pim6": {
"rp": [
{
"rp_addr": topo["routers"]["r4"]["links"]["lo"]["ipv6"].split(
"/"
)[0],
"group_addr_range": GROUP_RANGE_1,
}
]
}
}
}
result = create_pim_config(tgen, topo, input_dict)
assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
step(
"Enable mld on FRR1 interface and send mld join ffaa::1 "
"to ffaa::5 from different interfaces"
)
step("send mld join (ffaa::1-5) to R1")
intf = topo["routers"]["i1"]["links"]["r1"]["interface"]
intf_ip = topo["routers"]["i1"]["links"]["r1"]["ipv6"].split("/")[0]
result = socat_send_mld_join(
tgen, "i1", "UDP6-RECV", MLD_JOIN_RANGE_1, intf, intf_ip
)
assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
step("Send multicast traffic from FRR3 to all the receivers" "ffaa::1-5")
intf_ip = topo["routers"]["i2"]["links"]["r3"]["ipv6"].split("/")[0]
intf = topo["routers"]["i2"]["links"]["r3"]["interface"]
result = socat_send_pim6_traffic(tgen, "i2", "UDP6-SEND", MLD_JOIN_RANGE_1, intf)
assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
step("Clear the mroute on r1, wait for 5 sec")
result = clear_pim6_mroute(tgen, "r1")
assert result is True, "Testcase{}: Failed Error: {}".format(tc_name, result)
step(
"After clear ip mroute (*,g) entries are re-populated again"
" with same OIL and IIF, verify using 'show ipv6 mroute' and "
" 'show ipv6 pim upstream' "
)
source = topo["routers"]["i2"]["links"]["r3"]["ipv6"].split("/")[0]
input_dict = [
{
"dut": "r1",
"src_address": "*",
"iif": topo["routers"]["r1"]["links"]["r4"]["interface"],
"oil": topo["routers"]["r1"]["links"]["i1"]["interface"],
}
]
for data in input_dict:
result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
MLD_JOIN_RANGE_1,
data["iif"],
data["oil"],
)
assert result is True, "Testcase{} : Failed Error: {}".format(tc_name, result)
step(
"Verify 'show ipv6 pim upstream' showing correct OIL and IIF"
" on all the nodes"
)
for data in input_dict:
result = verify_upstream_iif(
tgen, data["dut"], data["iif"], data["src_address"], MLD_JOIN_RANGE_1
)
assert result is True, "Testcase{} : Failed Error: {}".format(tc_name, result)
step("Clear the mroute on r3, wait for 5 sec")
result = clear_pim6_mroute(tgen, "r3")
assert result is True, "Testcase{}: Failed Error: {}".format(tc_name, result)
step(
"Verify 'show ipv6 mroute' showing correct RPF and OIF"
" interface for (*,G) and (S,G) entries on all the nodes"
)
input_dict = [
{
"dut": "r3",
"src_address": source,
"iif": topo["routers"]["r3"]["links"]["i2"]["interface"],
"oil": topo["routers"]["r3"]["links"]["r2"]["interface"],
}
]
for data in input_dict:
result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
MLD_JOIN_RANGE_1,
data["iif"],
data["oil"],
)
assert result is True, "Testcase{} : Failed Error: {}".format(tc_name, result)
step(
"Verify 'show ipv6 pim upstream' showing correct OIL and IIF"
" on all the nodes"
)
for data in input_dict:
result = verify_upstream_iif(
tgen, data["dut"], data["iif"], data["src_address"], MLD_JOIN_RANGE_1
)
assert result is True, "Testcase{} : Failed Error: {}".format(tc_name, result)
step("Clear the mroute on r2, wait for 5 sec")
result = clear_pim6_mroute(tgen, "r2")
assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
step(
"Verify 'show ipv6 mroute' showing correct RPF and OIF"
" interface for (*,G) and (S,G) entries on all the nodes"
)
input_dict = [
{
"dut": "r2",
"src_address": source,
"iif": topo["routers"]["r2"]["links"]["r3"]["interface"],
"oil": topo["routers"]["r2"]["links"]["r1"]["interface"],
}
]
for data in input_dict:
result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
MLD_JOIN_RANGE_1,
data["iif"],
data["oil"],
)
assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
step(
"Verify 'show ipv6 pim upstream' showing correct OIL and IIF"
" on all the nodes"
)
for data in input_dict:
result = verify_upstream_iif(
tgen, data["dut"], data["iif"], data["src_address"], MLD_JOIN_RANGE_1
)
assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
step("Clear the mroute on r1, r3, wait for 5 sec")
result = clear_pim6_mroute(tgen, "r1")
assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
result = clear_pim6_mroute(tgen, "r3")
assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
step(
"Verify 'show ipv6 mroute' showing correct RPF and OIF"
" interface for (*,G) and (S,G) entries on all the nodes"
)
input_dict = [
{
"dut": "r1",
"src_address": "*",
"iif": topo["routers"]["r1"]["links"]["r4"]["interface"],
"oil": topo["routers"]["r1"]["links"]["i1"]["interface"],
},
{
"dut": "r3",
"src_address": source,
"iif": topo["routers"]["r3"]["links"]["i2"]["interface"],
"oil": topo["routers"]["r3"]["links"]["r2"]["interface"],
},
]
for data in input_dict:
result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
MLD_JOIN_RANGE_1,
data["iif"],
data["oil"],
)
assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
step(
"Verify 'show ipv6 pim upstream' showing correct OIL and IIF"
" on all the nodes"
)
for data in input_dict:
result = verify_upstream_iif(
tgen, data["dut"], data["iif"], data["src_address"], MLD_JOIN_RANGE_1
)
assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
step(
"multicast traffic is resume for all the receivers using "
" 'show ip multicast' "
)
result = verify_sg_traffic(tgen, "r1", MLD_JOIN_RANGE_1, source, "ipv6")
assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
write_test_footer(tc_name)
def test_verify_SPT_switchover_when_RPT_and_SPT_path_is_different_p0(request):
"""
Verify SPT switchover working when RPT and SPT path is
different
"""
tgen = get_topogen()
tc_name = request.node.name
write_test_header(tc_name)
# Creating configuration from JSON
reset_config_on_routers(tgen)
# Don"t run this test if we have any failure.
if tgen.routers_have_failure():
pytest.skip(tgen.errors)
step("Configure static RP for (ffcc::1-5) and " "(ffbb::1-5) in r5")
_GROUP_RANGE = GROUP_RANGE_2 + GROUP_RANGE_3
_MLD_JOIN_RANGE = MLD_JOIN_RANGE_2 + MLD_JOIN_RANGE_3
input_dict = {
"r5": {
"pim6": {
"rp": [
{
"rp_addr": topo["routers"]["r5"]["links"]["lo"]["ipv6"].split(
"/"
)[0],
"group_addr_range": _GROUP_RANGE,
}
]
}
}
}
result = create_pim_config(tgen, topo, input_dict)
assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
step("send mld join (ffbb::1-5, ffcc::1-5) to R1")
intf = topo["routers"]["i1"]["links"]["r1"]["interface"]
intf_ip = topo["routers"]["i1"]["links"]["r1"]["ipv6"].split("/")[0]
result = socat_send_mld_join(
tgen, "i1", "UDP6-RECV", _MLD_JOIN_RANGE, intf, intf_ip
)
assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
step("registerRx and registerStopTx value before traffic sent")
intf_r5 = topo["routers"]["r5"]["links"]["r3"]["interface"]
state_dict = {"r5": {intf_r5: ["registerRx", "registerStopTx"]}}
state_before = verify_pim_interface_traffic(tgen, state_dict, addr_type="ipv6")
assert isinstance(
state_before, dict
), "Testcase {} : Failed \n state_before is not dictionary \n " "Error: {}".format(
tc_name, result
)
step(
"Send multicast traffic from FRR3 to all the receivers" "ffbb::1-5 , ffcc::1-5"
)
intf_ip = topo["routers"]["i2"]["links"]["r3"]["ipv6"].split("/")[0]
intf = topo["routers"]["i2"]["links"]["r3"]["interface"]
result = socat_send_pim6_traffic(tgen, "i2", "UDP6-SEND", _MLD_JOIN_RANGE, intf)
assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
step(
"Verify in FRR3 sending initial packet to RP using"
" 'show ipv6 mroute' and mroute OIL is towards RP."
)
source = topo["routers"]["i2"]["links"]["r3"]["ipv6"].split("/")[0]
r3_i2 = topo["routers"]["r3"]["links"]["i2"]["interface"]
r3_r5 = topo["routers"]["r3"]["links"]["r5"]["interface"]
r3_r2 = topo["routers"]["r3"]["links"]["r2"]["interface"]
r1_r2 = topo["routers"]["r1"]["links"]["r2"]["interface"]
r1_i1 = topo["routers"]["r1"]["links"]["i1"]["interface"]
result = verify_mroutes(tgen, "r3", source, _MLD_JOIN_RANGE, r3_i2, [r3_r5, r3_r2])
assert result is True, "Testcase {} : " "Failed Error: {}".format(tc_name, result)
result = verify_mroutes(tgen, "r3", source, _MLD_JOIN_RANGE, r3_i2, r3_r2)
assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
step(
" After spt switchover traffic is flowing between"
" (LHR(FRR1)-FHR(FRR3)) and (S,G) OIL is updated toward FRR1"
" 'show mroute' and 'show pim upstream'"
)
input_dict = [
{"dut": "r3", "src_address": source, "iif": r3_i2, "oil": r3_r2},
{"dut": "r1", "src_address": source, "iif": r1_r2, "oil": r1_i1},
]
for data in input_dict:
result = verify_mroutes(
tgen,
data["dut"],
data["src_address"],
_MLD_JOIN_RANGE,
data["iif"],
data["oil"],
)
assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
for data in input_dict:
result = verify_upstream_iif(
tgen, data["dut"], data["iif"], data["src_address"], _MLD_JOIN_RANGE
)
assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
step("Stop the traffic to all the receivers")
dut = "i2"
intf = topo["routers"]["i2"]["links"]["r3"]["interface"]
shutdown_bringup_interface(tgen, dut, intf, False)
step(
"Null register packet being send periodically from FRR3 to RP, "
"verify using show ipv6 mroute on RP, have (S, G) entries null OIL"
" 'show ipv6 mroute' and verify show ip pim interface traffic"
"(In RP Register msg should be received and Register stop should"
" be transmitted)"
)
result = verify_upstream_iif(
tgen, "r3", "Unknown", source, _MLD_JOIN_RANGE, joinState="NotJoined"
)
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, addr_type="ipv6")
assert isinstance(
state_after, dict
), "Testcase {} : Failed \n state_before is not dictionary \n " "Error: {}".format(
tc_name, result
)
result = verify_state_incremented(state_before, state_after)
assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
write_test_footer(tc_name)
if __name__ == "__main__":
args = ["-s"] + sys.argv[1:]
sys.exit(pytest.main(args))