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

Merge pull request #13726 from LabNConsulting/chopps/fixpim6topotest

Browse Source 
tests: fixing pim6 topotest bugs
This commit is contained in:
Jafar Al-Gharaibeh 2023-06-08 08:47:15 -05:00 committed by GitHub
commit 87d347e66d
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
7 changed files with 228 additions and 384 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

42
tests/topotests/lib/mcast-tester.py
View File

@ -11,6 +11,7 @@ for the multicast group we subscribed to.
import argparse
import json
import ipaddress
import os
import socket
import struct
@ -35,13 +36,16 @@ def interface_name_to_index(name):
def multicast_join(sock, ifindex, group, port):
"Joins a multicast group."
mreq = struct.pack(
"=4sLL", socket.inet_aton(args.group), socket.INADDR_ANY, ifindex
)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.bind((group, port))
sock.setsockopt(socket.IPPROTO_IP, socket.IP_ADD_MEMBERSHIP, mreq)
if ip_version == 4:
mreq = group.packed + struct.pack("@II", socket.INADDR_ANY, ifindex)
opt = socket.IP_ADD_MEMBERSHIP
else:
mreq = group.packed + struct.pack("@I", ifindex)
opt = socket.IPV6_JOIN_GROUP
sock.bind((str(group), port))
sock.setsockopt(ip_proto, opt, mreq)
#
@ -50,15 +54,14 @@ def multicast_join(sock, ifindex, group, port):
parser = argparse.ArgumentParser(description="Multicast RX utility")
parser.add_argument("group", help="Multicast IP")
parser.add_argument("interface", help="Interface name")
parser.add_argument("--port", type=int, default=1000, help="port to send to")
parser.add_argument("--ttl", type=int, default=16, help="TTL/hops for sending packets")
parser.add_argument("--socket", help="Point to topotest UNIX socket")
parser.add_argument(
"--send", help="Transmit instead of join with interval", type=float, default=0
)
args = parser.parse_args()
ttl = 16
port = 1000
# Get interface index/validate.
ifindex = interface_name_to_index(args.interface)
if ifindex is None:
@ -85,7 +88,12 @@ else:
# Set topotest socket non blocking so we can multiplex the main loop.
toposock.setblocking(False)
msock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
args.group = ipaddress.ip_address(args.group)
ip_version = args.group.version
ip_family = socket.AF_INET if ip_version == 4 else socket.AF_INET6
ip_proto = socket.IPPROTO_IP if ip_version == 4 else socket.IPPROTO_IPV6
msock = socket.socket(ip_family, socket.SOCK_DGRAM, socket.IPPROTO_UDP)
if args.send > 0:
# Prepare multicast bit in that interface.
msock.setsockopt(
@ -93,12 +101,18 @@ if args.send > 0:
25,
struct.pack("%ds" % len(args.interface), args.interface.encode("utf-8")),
)
# Set packets TTL.
msock.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_TTL, struct.pack("b", ttl))
# Set packets TTL/hops.
ttlopt = socket.IP_MULTICAST_TTL if ip_version == 4 else socket.IPV6_MULTICAST_HOPS
if ip_version == 4:
msock.setsockopt(ip_proto, ttlopt, struct.pack("B", args.ttl))
else:
msock.setsockopt(ip_proto, ttlopt, struct.pack("I", args.ttl))
# Block to ensure packet send.
msock.setblocking(True)
else:
multicast_join(msock, ifindex, args.group, port)
multicast_join(msock, ifindex, args.group, args.port)
def should_exit():
@ -120,7 +134,7 @@ def should_exit():
counter = 0
while not should_exit():
if args.send > 0:
msock.sendto(b"test %d" % counter, (args.group, port))
msock.sendto(b"test %d" % counter, (str(args.group), args.port))
counter += 1
time.sleep(args.send)

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

@ -1,35 +1,35 @@
# -*- coding: utf-8 eval: (blacken-mode 1) -*-
# SPDX-License-Identifier: ISC
# Copyright (c) 2019 by VMware, Inc. ("VMware")
# Used Copyright (c) 2018 by Network Device Education Foundation, Inc.
# ("NetDEF") in this file.
import datetime
import functools
import os
import re
import sys
import traceback
import functools
from copy import deepcopy
from time import sleep
from lib import topotest
# Import common_config to use commomnly used APIs
from lib.common_config import (
create_common_configurations,
HostApplicationHelper,
InvalidCLIError,
create_common_configuration,
InvalidCLIError,
create_common_configurations,
get_frr_ipv6_linklocal,
retry,
run_frr_cmd,
validate_ip_address,
get_frr_ipv6_linklocal,
)
from lib.micronet import get_exec_path
from lib.topolog import logger
from lib.topotest import frr_unicode
from lib import topotest
####
CWD = os.path.dirname(os.path.realpath(__file__))

72
tests/topotests/multicast_mld_join_topo1/test_multicast_mld_local_join.py
View File

@ -1,4 +1,4 @@
#!/usr/bin/env python
# -*- coding: utf-8 eval: (blacken-mode 1) -*-
# SPDX-License-Identifier: ISC
#
# Copyright (c) 2023 by VMware, Inc. ("VMware")
@ -20,52 +20,31 @@ Following tests are covered:
5. Verify static MLD groups after removing and adding MLD config
"""
import os
import sys
import time
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 re import search as re_search
from re import findall as findall
from lib.common_config import (
start_topology,
write_test_header,
write_test_footer,
step,
kill_router_daemons,
start_router_daemons,
reset_config_on_routers,
do_countdown,
apply_raw_config,
socat_send_pim6_traffic,
start_topology,
step,
write_test_footer,
write_test_header,
)
from lib.pim import (
create_pim_config,
verify_mroutes,
verify_upstream_iif,
verify_mld_groups,
clear_pim6_mroute,
McastTesterHelper,
verify_pim_neighbors,
create_mld_config,
verify_mld_groups,
create_pim_config,
verify_local_mld_groups,
verify_mld_groups,
verify_mroutes,
verify_pim_neighbors,
verify_pim_rp_info,
verify_upstream_iif,
)
from lib.topolog import logger
from lib.topogen import Topogen, get_topogen
from lib.topojson import build_config_from_json
from lib.topolog import logger
r1_r2_links = []
r1_r3_links = []
@ -131,7 +110,7 @@ def setup_module(mod):
logger.info("Running setup_module to create topology")
# This function initiates the topology build with Topogen...
json_file = "{}/multicast_mld_local_join.json".format(CWD)
json_file = "multicast_mld_local_join.json"
tgen = Topogen(json_file, mod.__name__)
global topo
topo = tgen.json_topo
@ -151,6 +130,9 @@ def setup_module(mod):
result = verify_pim_neighbors(tgen, topo)
assert result is True, " Verify PIM neighbor: Failed Error: {}".format(result)
global app_helper
app_helper = McastTesterHelper(tgen)
logger.info("Running setup_module() done")
@ -161,6 +143,8 @@ def teardown_module():
tgen = get_topogen()
app_helper.cleanup()
# Stop toplogy and Remove tmp files
tgen.stop_topology()
@ -265,6 +249,8 @@ def test_mroute_with_mld_local_joins_p0(request):
reset_config_on_routers(tgen)
app_helper.stop_all_hosts()
step("Enable the PIM on all the interfaces of R1, R2, R3, R4")
step("configure BGP on R1, R2, R3, R4 and enable redistribute static/connected")
step("Enable the MLD on R11 interfac of R1 and configure local mld groups")
@ -330,9 +316,7 @@ def test_mroute_with_mld_local_joins_p0(request):
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("Send traffic from R4 to all the groups ( ffaa::1 to ffaa::5)")
intf_ip = topo["routers"]["i4"]["links"]["r4"]["ipv6"].split("/")[0]
intf = topo["routers"]["i4"]["links"]["r4"]["interface"]
result = socat_send_pim6_traffic(tgen, "i4", "UDP6-SEND", MLD_JOIN_RANGE_1, intf)
result = app_helper.run_traffic("i4", MLD_JOIN_RANGE_1, "r4")
assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
step(
@ -458,6 +442,8 @@ def test_remove_add_mld_local_joins_p1(request):
reset_config_on_routers(tgen)
app_helper.stop_all_hosts()
step("Enable the PIM on all the interfaces of R1, R2, R3, R4")
step("configure BGP on R1, R2, R3, R4 and enable redistribute static/connected")
step("Enable the MLD on R11 interfac of R1 and configure local mld groups")
@ -517,9 +503,7 @@ def test_remove_add_mld_local_joins_p1(request):
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("Send traffic from R4 to all the groups ( ffaa::1 to ffaa::5)")
intf_ip = topo["routers"]["i4"]["links"]["r4"]["ipv6"].split("/")[0]
intf = topo["routers"]["i4"]["links"]["r4"]["interface"]
result = socat_send_pim6_traffic(tgen, "i4", "UDP6-SEND", MLD_JOIN_RANGE_1, intf)
result = app_helper.run_traffic("i4", MLD_JOIN_RANGE_1, "r4")
assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
step(
@ -710,6 +694,8 @@ def test_remove_add_mld_config_with_local_joins_p1(request):
reset_config_on_routers(tgen)
app_helper.stop_all_hosts()
step("Enable the PIM on all the interfaces of R1, R2, R3, R4")
step("configure BGP on R1, R2, R3, R4 and enable redistribute static/connected")
step("Enable the MLD on R11 interfac of R1 and configure local mld groups")
@ -759,9 +745,7 @@ def test_remove_add_mld_config_with_local_joins_p1(request):
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("Send traffic from R4 to all the groups ( ffaa::1 to ffaa::5)")
intf_ip = topo["routers"]["i4"]["links"]["r4"]["ipv6"].split("/")[0]
intf = topo["routers"]["i4"]["links"]["r4"]["interface"]
result = socat_send_pim6_traffic(tgen, "i4", "UDP6-SEND", MLD_JOIN_RANGE_1, intf)
result = app_helper.run_traffic("i4", MLD_JOIN_RANGE_1, "r4")
assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
step(

172
tests/topotests/multicast_pim6_sm_topo1/test_multicast_pim6_sm1.py
View File

@ -1,4 +1,4 @@
#!/usr/bin/env python
# -*- coding: utf-8 eval: (blacken-mode 1) -*-
# SPDX-License-Identifier: ISC
#
@ -30,61 +30,40 @@ should get update accordingly
data traffic
"""
import os
import sys
import json
import time
import datetime
import sys
import time
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,
get_frr_ipv6_linklocal,
required_linux_kernel_version,
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,
kill_socat,
start_topology,
step,
write_test_footer,
write_test_header,
)
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,
clear_pim6_mroute,
create_mld_config,
create_pim_config,
verify_mld_config,
verify_mld_groups,
verify_mroute_summary,
verify_mroutes,
verify_pim_interface_traffic,
verify_pim_join,
verify_pim_nexthop,
verify_pim_state,
verify_sg_traffic,
verify_upstream_iif,
)
from lib.topolog import logger
from lib.topogen import Topogen, get_topogen
from lib.topojson import build_config_from_json
from lib.topolog import logger
# Global variables
GROUP_RANGE = "ff00::/8"
@ -141,8 +120,7 @@ def setup_module(mod):
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)
json_file = "multicast_pim6_sm_topo1.json"
tgen = Topogen(json_file, mod.__name__)
global topo
topo = tgen.json_topo
@ -159,6 +137,9 @@ def setup_module(mod):
# Creating configuration from JSON
build_config_from_json(tgen, tgen.json_topo)
global app_helper
app_helper = McastTesterHelper(tgen)
logger.info("Running setup_module() done")
@ -169,8 +150,7 @@ def teardown_module():
tgen = get_topogen()
# Clean up socat
kill_socat(tgen)
app_helper.cleanup()
# Stop toplogy and Remove tmp files
tgen.stop_topology()
@ -296,6 +276,8 @@ def test_multicast_data_traffic_static_RP_send_traffic_then_join_p0(request):
# Creating configuration from JSON
reset_config_on_routers(tgen)
app_helper.stop_all_hosts()
# Don"t run this test if we have any failure.
if tgen.routers_have_failure():
pytest.skip(tgen.errors)
@ -334,9 +316,7 @@ def test_multicast_data_traffic_static_RP_send_traffic_then_join_p0(request):
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)
result = app_helper.run_traffic("i2", MLD_JOIN_RANGE_1, "r3")
assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
source = topo["routers"]["i2"]["links"]["r3"]["ipv6"].split("/")[0]
@ -375,11 +355,7 @@ def test_multicast_data_traffic_static_RP_send_traffic_then_join_p0(request):
)
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
)
result = app_helper.run_join("i1", MLD_JOIN_RANGE_1, "r1")
assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
step(
@ -532,11 +508,7 @@ def test_verify_mroute_when_receiver_is_outside_frr_p0(request):
assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
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, intf, intf_ip
)
result = app_helper.run_join("i1", _MLD_JOIN_RANGE, "r1")
assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
step("verify MLD joins received on r1")
@ -546,9 +518,7 @@ def test_verify_mroute_when_receiver_is_outside_frr_p0(request):
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, intf)
result = app_helper.run_traffic("i2", _MLD_JOIN_RANGE, "r3")
assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
step(
@ -561,11 +531,7 @@ def test_verify_mroute_when_receiver_is_outside_frr_p0(request):
result = create_mld_config(tgen, topo, input_dict)
assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
i5_r5 = topo["routers"]["i5"]["links"]["r5"]["interface"]
intf_ip = topo["routers"]["i5"]["links"]["r5"]["ipv6"].split("/")[0]
result = socat_send_mld_join(
tgen, "i5", "UDP6-RECV", _MLD_JOIN_RANGE, i5_r5, intf_ip
)
result = app_helper.run_join("i5", _MLD_JOIN_RANGE, "r5")
assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
step("FRR1 has 10 (*.G) and 10 (S,G) verify using 'show ipv6 mroute'")
@ -682,6 +648,8 @@ def test_verify_mroute_when_frr_is_transit_router_p2(request):
# Creating configuration from JSON
reset_config_on_routers(tgen)
app_helper.stop_all_hosts()
# Don"t run this test if we have any failure.
if tgen.routers_have_failure():
pytest.skip(tgen.errors)
@ -708,11 +676,7 @@ def test_verify_mroute_when_frr_is_transit_router_p2(request):
step("Enable mld on FRR1 interface and send mld join ")
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
)
result = app_helper.run_join("i1", MLD_JOIN_RANGE_1, "r1")
assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
step("verify mld groups received on R1")
@ -722,9 +686,7 @@ def test_verify_mroute_when_frr_is_transit_router_p2(request):
assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
step("Send multicast traffic from FRR3 to ffaa::1-5 receivers")
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)
result = app_helper.run_traffic("i2", MLD_JOIN_RANGE_1, "r3")
assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
step("shut the direct link to R1 ")
@ -841,6 +803,8 @@ def test_verify_mroute_when_RP_unreachable_p1(request):
# Creating configuration from JSON
reset_config_on_routers(tgen)
app_helper.stop_all_hosts()
# Don"t run this test if we have any failure.
if tgen.routers_have_failure():
pytest.skip(tgen.errors)
@ -868,17 +832,11 @@ def test_verify_mroute_when_RP_unreachable_p1(request):
step("Enable mld on FRR1 interface and send mld join ffaa::1-5")
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
)
result = app_helper.run_join("i1", MLD_JOIN_RANGE_1, "r1")
assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
step("Send multicast traffic from FRR3 to ffaa::1-5 receivers")
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)
result = app_helper.run_traffic("i2", MLD_JOIN_RANGE_1, "r3")
assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
step("Configure one MLD interface on FRR3 node and send MLD" " join (ffcc::1)")
@ -888,11 +846,7 @@ def test_verify_mroute_when_RP_unreachable_p1(request):
assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
step("send mld join (ffaa::1-5) to R1")
intf = topo["routers"]["i8"]["links"]["r3"]["interface"]
intf_ip = topo["routers"]["i8"]["links"]["r3"]["ipv6"].split("/")[0]
result = socat_send_mld_join(
tgen, "i8", "UDP6-RECV", MLD_JOIN_RANGE_1, intf, intf_ip
)
result = app_helper.run_join("i8", MLD_JOIN_RANGE_1, "r3")
assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
step("verify MLD groups received ")
@ -975,16 +929,14 @@ def test_modify_mld_query_timer_p0(request):
# Creating configuration from JSON
reset_config_on_routers(tgen)
app_helper.stop_all_hosts()
# Don"t run this test if we have any failure.
if tgen.routers_have_failure():
pytest.skip(tgen.errors)
step("send mld join (ffaa::1-5) to R1")
intf = topo["routers"]["i8"]["links"]["r3"]["interface"]
intf_ip = topo["routers"]["i8"]["links"]["r3"]["ipv6"].split("/")[0]
result = socat_send_mld_join(
tgen, "i8", "UDP6-RECV", MLD_JOIN_RANGE_1, intf, intf_ip
)
result = app_helper.run_join("i8", MLD_JOIN_RANGE_1, "r3")
assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
step("Enable MLD on receiver interface")
@ -1023,9 +975,7 @@ def test_modify_mld_query_timer_p0(request):
assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
step("Send multicast traffic from FRR3 to ffaa::1-5 receivers")
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)
result = app_helper.run_traffic("i2", MLD_JOIN_RANGE_1, "r3")
assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
step(
@ -1158,17 +1108,15 @@ def test_modify_mld_max_query_response_timer_p0(request):
# Creating configuration from JSON
reset_config_on_routers(tgen)
app_helper.stop_all_hosts()
# Don"t run this test if we have any failure.
if tgen.routers_have_failure():
pytest.skip(tgen.errors)
step("Enable mld on FRR1 interface and send MLD join")
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
)
result = app_helper.run_join("i1", MLD_JOIN_RANGE_1, "r1")
assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
r1_i1 = topo["routers"]["r1"]["links"]["i1"]["interface"]
@ -1214,9 +1162,7 @@ def test_modify_mld_max_query_response_timer_p0(request):
assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
step("Send multicast traffic from FRR3 to ffaa::1-5 receivers")
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)
result = app_helper.run_traffic("i2", MLD_JOIN_RANGE_1, "r3")
assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
step(
@ -1431,6 +1377,8 @@ def test_verify_impact_on_multicast_traffic_when_RP_removed_p0(request):
# Creating configuration from JSON
reset_config_on_routers(tgen)
app_helper.stop_all_hosts()
# Don"t run this test if we have any failure.
if tgen.routers_have_failure():
pytest.skip(tgen.errors)
@ -1438,9 +1386,7 @@ def test_verify_impact_on_multicast_traffic_when_RP_removed_p0(request):
step("send multicast traffic for group range ffaa::1-5")
step("Send multicast traffic from FRR3 to ffaa::1-5 receivers")
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)
result = app_helper.run_traffic("i2", MLD_JOIN_RANGE_1, "r3")
assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
step("Configure static RP for group (ffaa::1) on r5")
@ -1464,11 +1410,7 @@ def test_verify_impact_on_multicast_traffic_when_RP_removed_p0(request):
step("Enable mld on FRR1 interface and send MLD join")
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
)
result = app_helper.run_join("i1", MLD_JOIN_RANGE_1, "r1")
assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
step(

102
tests/topotests/multicast_pim6_sm_topo1/test_multicast_pim6_sm2.py
View File

@ -1,4 +1,4 @@
#!/usr/bin/env python
# -*- coding: utf-8 eval: (blacken-mode 1) -*-
# SPDX-License-Identifier: ISC
#
@ -21,61 +21,31 @@ PIM nbr and mroute from FRR node
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,
required_linux_kernel_version,
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,
kill_socat,
start_topology,
step,
write_test_footer,
write_test_header,
)
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,
clear_pim6_mroute,
create_pim_config,
verify_mroutes,
verify_pim_interface_traffic,
verify_sg_traffic,
verify_mld_config,
verify_upstream_iif,
)
from lib.topolog import logger
from lib.topogen import Topogen, get_topogen
from lib.topojson import build_config_from_json
from lib.topolog import logger
# Global variables
GROUP_RANGE = "ff00::/8"
@ -114,6 +84,16 @@ ASSERT_MSG = "Testcase {} : Failed Error: {}"
pytestmark = [pytest.mark.pim6d]
@pytest.fixture(scope="function")
def app_helper():
# helper = McastTesterHelper(get_topogen())
# yield helepr
# helper.cleanup()
# Even better use contextmanager functionality:
with McastTesterHelper(get_topogen()) as ah:
yield ah
def setup_module(mod):
"""
Sets up the pytest environment
@ -132,8 +112,7 @@ def setup_module(mod):
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)
json_file = "multicast_pim6_sm_topo1.json"
tgen = Topogen(json_file, mod.__name__)
global topo
topo = tgen.json_topo
@ -160,9 +139,6 @@ def teardown_module():
tgen = get_topogen()
# Clean up socat
kill_socat(tgen)
# Stop toplogy and Remove tmp files
tgen.stop_topology()
@ -225,7 +201,7 @@ def verify_state_incremented(state_before, state_after):
#####################################################
def test_clear_mroute_and_verify_multicast_data_p0(request):
def test_clear_mroute_and_verify_multicast_data_p0(request, app_helper):
"""
Verify (*,G) and (S,G) entry populated again after clear the
PIM nbr and mroute from FRR node
@ -237,6 +213,8 @@ def test_clear_mroute_and_verify_multicast_data_p0(request):
# Creating configuration from JSON
reset_config_on_routers(tgen)
app_helper.stop_all_hosts()
# Don"t run this test if we have any failure.
if tgen.routers_have_failure():
pytest.skip(tgen.errors)
@ -266,18 +244,12 @@ def test_clear_mroute_and_verify_multicast_data_p0(request):
)
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
)
result = app_helper.run_join("i1", MLD_JOIN_RANGE_1, "r1")
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)
result = app_helper.run_traffic("i2", MLD_JOIN_RANGE_1, "r3")
assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
step("Clear the mroute on r1, wait for 5 sec")
@ -457,7 +429,9 @@ def test_clear_mroute_and_verify_multicast_data_p0(request):
write_test_footer(tc_name)
def test_verify_SPT_switchover_when_RPT_and_SPT_path_is_different_p0(request):
def test_verify_SPT_switchover_when_RPT_and_SPT_path_is_different_p0(
request, app_helper
):
"""
Verify SPT switchover working when RPT and SPT path is
different
@ -498,11 +472,7 @@ def test_verify_SPT_switchover_when_RPT_and_SPT_path_is_different_p0(request):
assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
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
)
result = app_helper.run_join("i1", _MLD_JOIN_RANGE, "r1")
assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
step("registerRx and registerStopTx value before traffic sent")
@ -518,9 +488,7 @@ def test_verify_SPT_switchover_when_RPT_and_SPT_path_is_different_p0(request):
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)
result = app_helper.run_traffic("i2", _MLD_JOIN_RANGE, "r3")
assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
step(

121
tests/topotests/multicast_pim6_static_rp_topo1/test_multicast_pim6_static_rp1.py
View File

@ -1,4 +1,4 @@
#!/usr/bin/env python
# -*- coding: utf-8 eval: (blacken-mode 1) -*-
# SPDX-License-Identifier: ISC
#
@ -41,57 +41,36 @@ Test steps
8. Verify PIM6 join send towards the higher preferred RP
9. Verify PIM6 prune send towards the lower preferred RP
"""
import os
import sys
import json
import time
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,
reset_config_on_routers,
step,
shutdown_bringup_interface,
kill_router_daemons,
start_router_daemons,
create_static_routes,
check_router_status,
socat_send_mld_join,
socat_send_pim6_traffic,
kill_socat,
reset_config_on_routers,
shutdown_bringup_interface,
start_topology,
step,
write_test_footer,
write_test_header,
)
from lib.pim import (
McastTesterHelper,
clear_pim6_interface_traffic,
create_pim_config,
verify_upstream_iif,
get_pim6_interface_traffic,
verify_join_state_and_timer,
verify_mld_groups,
verify_mroutes,
verify_pim_neighbors,
verify_pim6_neighbors,
verify_pim_interface_traffic,
verify_pim_rp_info,
verify_pim_state,
clear_pim6_interface_traffic,
clear_pim6_mroute,
verify_pim6_neighbors,
get_pim6_interface_traffic,
clear_pim6_interfaces,
verify_mld_groups,
verify_upstream_iif,
)
from lib.topogen import Topogen, get_topogen
from lib.topojson import build_config_from_json, build_topo_from_json
from lib.topolog import logger
from lib.topojson import build_topo_from_json, build_config_from_json
# Global variables
GROUP_RANGE_1 = "ff08::/64"
@ -141,7 +120,7 @@ def setup_module(mod):
logger.info("Running setup_module to create topology")
# This function initiates the topology build with Topogen...
json_file = "{}/multicast_pim6_static_rp.json".format(CWD)
json_file = "multicast_pim6_static_rp.json"
tgen = Topogen(json_file, mod.__name__)
global TOPO
TOPO = tgen.json_topo
@ -163,6 +142,9 @@ def setup_module(mod):
result = verify_pim6_neighbors(tgen, TOPO)
assert result is True, "setup_module :Failed \n Error:" " {}".format(result)
global app_helper
app_helper = McastTesterHelper(tgen)
logger.info("Running setup_module() done")
@ -172,8 +154,7 @@ def teardown_module():
logger.info("Running teardown_module to delete topology")
tgen = get_topogen()
# Clean up socat
kill_socat(tgen)
app_helper.cleanup()
# Stop toplogy and Remove tmp files
tgen.stop_topology()
@ -260,6 +241,8 @@ def test_pim6_add_delete_static_RP_p0(request):
step("Creating configuration from JSON")
reset_config_on_routers(tgen)
app_helper.stop_all_hosts()
step("Shut link b/w R1 and R3 and R1 and R4 as per testcase topology")
intf_r1_r3 = TOPO["routers"]["r1"]["links"]["r3"]["interface"]
intf_r1_r4 = TOPO["routers"]["r1"]["links"]["r4"]["interface"]
@ -313,11 +296,7 @@ def test_pim6_add_delete_static_RP_p0(request):
)
step("send mld join {} to R1".format(GROUP_ADDRESS_1))
intf = TOPO["routers"]["r0"]["links"]["r1"]["interface"]
intf_ip = TOPO["routers"]["r0"]["links"]["r1"]["ipv6"].split("/")[0]
result = socat_send_mld_join(
tgen, "r0", "UDP6-RECV", GROUP_ADDRESS_1, intf, intf_ip
)
result = app_helper.run_join("r0", GROUP_ADDRESS_1, "r1")
assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
step("r1: Verify MLD groups")
@ -457,6 +436,8 @@ def test_pim6_SPT_RPT_path_same_p1(request):
step("Creating configuration from JSON")
reset_config_on_routers(tgen)
app_helper.stop_all_hosts()
step("Shut link b/w R1->R3, R1->R4 and R3->R1, R3->R4 as per " "testcase topology")
intf_r1_r3 = TOPO["routers"]["r1"]["links"]["r3"]["interface"]
intf_r1_r4 = TOPO["routers"]["r1"]["links"]["r4"]["interface"]
@ -494,11 +475,7 @@ def test_pim6_SPT_RPT_path_same_p1(request):
step(
"Enable MLD on r1 interface and send MLD join {} to R1".format(GROUP_ADDRESS_1)
)
intf = TOPO["routers"]["r0"]["links"]["r1"]["interface"]
intf_ip = TOPO["routers"]["r0"]["links"]["r1"]["ipv6"].split("/")[0]
result = socat_send_mld_join(
tgen, "r0", "UDP6-RECV", GROUP_ADDRESS_1, intf, intf_ip
)
result = app_helper.run_join("r0", GROUP_ADDRESS_1, "r1")
assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
step("r1: Verify MLD groups")
@ -508,9 +485,8 @@ def test_pim6_SPT_RPT_path_same_p1(request):
assert result is True, ASSERT_MSG.format(tc_name, result)
step("Send multicast traffic from R5")
intf = TOPO["routers"]["r5"]["links"]["r3"]["interface"]
SOURCE_ADDRESS = TOPO["routers"]["r5"]["links"]["r3"]["ipv6"].split("/")[0]
result = socat_send_pim6_traffic(tgen, "r5", "UDP6-SEND", GROUP_ADDRESS_1, intf)
result = app_helper.run_traffic("r5", GROUP_ADDRESS_1, "r3")
assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
step("r2: Verify RP info")
@ -630,6 +606,8 @@ def test_pim6_RP_configured_as_LHR_p1(request):
step("Creating configuration from JSON")
reset_config_on_routers(tgen)
app_helper.stop_all_hosts()
step("Enable MLD on r1 interface")
step("Enable the PIM6 on all the interfaces of r1, r2, r3 and r4 routers")
@ -665,11 +643,7 @@ def test_pim6_RP_configured_as_LHR_p1(request):
assert result is True, ASSERT_MSG.format(tc_name, result)
step("send mld join {} to R1".format(GROUP_ADDRESS_1))
intf = TOPO["routers"]["r0"]["links"]["r1"]["interface"]
intf_ip = TOPO["routers"]["r0"]["links"]["r1"]["ipv6"].split("/")[0]
result = socat_send_mld_join(
tgen, "r0", "UDP6-RECV", GROUP_ADDRESS_1, intf, intf_ip
)
result = app_helper.run_join("r0", GROUP_ADDRESS_1, "r1")
assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
step("r1: Verify MLD groups")
@ -679,9 +653,8 @@ def test_pim6_RP_configured_as_LHR_p1(request):
assert result is True, ASSERT_MSG.format(tc_name, result)
step("r5: Send multicast traffic for group {}".format(GROUP_ADDRESS_1))
intf = TOPO["routers"]["r5"]["links"]["r3"]["interface"]
SOURCE_ADDRESS = TOPO["routers"]["r5"]["links"]["r3"]["ipv6"].split("/")[0]
result = socat_send_pim6_traffic(tgen, "r5", "UDP6-SEND", GROUP_ADDRESS_1, intf)
result = app_helper.run_traffic("r5", GROUP_ADDRESS_1, "r3")
assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
step("r1: Verify (*, G) upstream IIF interface")
@ -762,6 +735,8 @@ def test_pim6_RP_configured_as_FHR_p1(request):
step("Creating configuration from JSON")
reset_config_on_routers(tgen)
app_helper.stop_all_hosts()
step("Enable MLD on r1 interface")
step("Enable the PIM6 on all the interfaces of r1, r2, r3 and r4 routers")
step("r3: Configure r3(FHR) as RP")
@ -792,11 +767,7 @@ def test_pim6_RP_configured_as_FHR_p1(request):
assert result is True, ASSERT_MSG.format(tc_name, result)
step("send mld join {} to R1".format(GROUP_ADDRESS_1))
intf = TOPO["routers"]["r0"]["links"]["r1"]["interface"]
intf_ip = TOPO["routers"]["r0"]["links"]["r1"]["ipv6"].split("/")[0]
result = socat_send_mld_join(
tgen, "r0", "UDP6-RECV", GROUP_ADDRESS_1, intf, intf_ip
)
result = app_helper.run_join("r0", GROUP_ADDRESS_1, "r1")
assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
step("r1: Verify MLD groups")
@ -806,9 +777,8 @@ def test_pim6_RP_configured_as_FHR_p1(request):
assert result is True, ASSERT_MSG.format(tc_name, result)
step("r5: Send multicast traffic for group {}".format(GROUP_ADDRESS_1))
intf = TOPO["routers"]["r5"]["links"]["r3"]["interface"]
SOURCE_ADDRESS = TOPO["routers"]["r5"]["links"]["r3"]["ipv6"].split("/")[0]
result = socat_send_pim6_traffic(tgen, "r5", "UDP6-SEND", GROUP_ADDRESS_1, intf)
result = app_helper.run_traffic("r5", GROUP_ADDRESS_1, "r3")
assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
step("r1: Verify (*, G) upstream IIF interface")
@ -890,6 +860,8 @@ def test_pim6_SPT_RPT_path_different_p1(request):
step("Creating configuration from JSON")
reset_config_on_routers(tgen)
app_helper.stop_all_hosts()
step("Enable MLD on r1 interface")
step("Enable the PIM6 on all the interfaces of r1, r2, r3 and r4 routers")
step("r2: Configure r2 as RP")
@ -921,11 +893,7 @@ def test_pim6_SPT_RPT_path_different_p1(request):
assert result is True, ASSERT_MSG.format(tc_name, result)
step("send mld join {} to R1".format(GROUP_ADDRESS_1))
intf = TOPO["routers"]["r0"]["links"]["r1"]["interface"]
intf_ip = TOPO["routers"]["r0"]["links"]["r1"]["ipv6"].split("/")[0]
result = socat_send_mld_join(
tgen, "r0", "UDP6-RECV", GROUP_ADDRESS_1, intf, intf_ip
)
result = app_helper.run_join("r0", GROUP_ADDRESS_1, "r1")
assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
step("r1: Verify MLD groups")
@ -935,9 +903,8 @@ def test_pim6_SPT_RPT_path_different_p1(request):
assert result is True, ASSERT_MSG.format(tc_name, result)
step("r5: Send multicast traffic for group {}".format(GROUP_ADDRESS_1))
intf = TOPO["routers"]["r5"]["links"]["r3"]["interface"]
SOURCE_ADDRESS = TOPO["routers"]["r5"]["links"]["r3"]["ipv6"].split("/")[0]
result = socat_send_pim6_traffic(tgen, "r5", "UDP6-SEND", GROUP_ADDRESS_1, intf)
result = app_helper.run_traffic("r5", GROUP_ADDRESS_1, "r3")
assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
step("r1: Verify (*, G) upstream IIF interface")
@ -1060,6 +1027,8 @@ def test_pim6_send_join_on_higher_preffered_rp_p1(request):
step("Creating configuration from JSON")
reset_config_on_routers(tgen)
app_helper.stop_all_hosts()
step("Enable MLD on r1 interface")
step("Enable the PIM66 on all the interfaces of r1, r2, r3 and r4 routers")
step(
@ -1109,11 +1078,7 @@ def test_pim6_send_join_on_higher_preffered_rp_p1(request):
)
step("r0: send mld join {} to R1".format(GROUP_ADDRESS_3))
intf = TOPO["routers"]["r0"]["links"]["r1"]["interface"]
intf_ip = TOPO["routers"]["r0"]["links"]["r1"]["ipv6"].split("/")[0]
result = socat_send_mld_join(
tgen, "r0", "UDP6-RECV", GROUP_ADDRESS_3, intf, intf_ip
)
result = app_helper.run_join("r0", GROUP_ADDRESS_3, "r1")
assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
step("r1: Verify MLD groups")

91
tests/topotests/multicast_pim6_static_rp_topo1/test_multicast_pim6_static_rp2.py
View File

@ -1,4 +1,4 @@
#!/usr/bin/env python
# -*- coding: utf-8 eval: (blacken-mode 1) -*-
# SPDX-License-Identifier: ISC
#
@ -33,55 +33,31 @@ Test steps
import os
import sys
import json
import time
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,
reset_config_on_routers,
step,
shutdown_bringup_interface,
kill_router_daemons,
start_router_daemons,
create_static_routes,
check_router_status,
socat_send_mld_join,
socat_send_pim6_traffic,
kill_socat,
create_debug_log_config,
reset_config_on_routers,
shutdown_bringup_interface,
start_topology,
step,
write_test_footer,
write_test_header,
)
from lib.pim import (
McastTesterHelper,
create_pim_config,
verify_upstream_iif,
verify_join_state_and_timer,
verify_mroutes,
verify_pim_neighbors,
verify_pim_interface_traffic,
verify_pim_rp_info,
verify_pim_state,
clear_pim6_interface_traffic,
clear_pim6_mroute,
verify_pim6_neighbors,
get_pim6_interface_traffic,
clear_pim6_interfaces,
verify_mld_groups,
verify_mroutes,
verify_pim6_neighbors,
verify_pim_rp_info,
verify_upstream_iif,
)
from lib.topogen import Topogen, get_topogen
from lib.topojson import build_config_from_json, build_topo_from_json
from lib.topolog import logger
from lib.topojson import build_topo_from_json, build_config_from_json
# Global variables
GROUP_RANGE_1 = "ff08::/64"
@ -145,7 +121,7 @@ def setup_module(mod):
logger.info("Running setup_module to create topology")
# This function initiates the topology build with Topogen...
json_file = "{}/multicast_pim6_static_rp.json".format(CWD)
json_file = "multicast_pim6_static_rp.json"
tgen = Topogen(json_file, mod.__name__)
global TOPO
TOPO = tgen.json_topo
@ -167,6 +143,9 @@ def setup_module(mod):
result = verify_pim6_neighbors(tgen, TOPO)
assert result is True, "setup_module :Failed \n Error:" " {}".format(result)
global app_helper
app_helper = McastTesterHelper(tgen)
logger.info("Running setup_module() done")
@ -176,8 +155,7 @@ def teardown_module():
logger.info("Running teardown_module to delete topology")
tgen = get_topogen()
# Clean up socat
kill_socat(tgen)
app_helper.cleanup()
# Stop toplogy and Remove tmp files
tgen.stop_topology()
@ -265,6 +243,8 @@ def test_pim6_multiple_groups_same_RP_address_p2(request):
step("Creating configuration from JSON")
reset_config_on_routers(tgen)
app_helper.stop_all_hosts()
input_dict = {
"r1": {"debug": {"log_file": "r1_debug.log", "enable": ["pim6d"]}},
"r2": {"debug": {"log_file": "r2_debug.log", "enable": ["pim6d"]}},
@ -305,10 +285,7 @@ def test_pim6_multiple_groups_same_RP_address_p2(request):
group_address_list = GROUP_ADDRESS_LIST_1 + GROUP_ADDRESS_LIST_2
step("r0: Send MLD join for 10 groups")
intf = TOPO["routers"]["r0"]["links"]["r1"]["interface"]
intf_ip = TOPO["routers"]["r0"]["links"]["r1"]["ipv6"].split("/")[0]
result = socat_send_mld_join(
tgen, "r0", "UDP6-RECV", group_address_list, intf, intf_ip
)
result = app_helper.run_join("r0", group_address_list, "r1")
assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
step("r1: Verify MLD groups")
@ -318,9 +295,8 @@ def test_pim6_multiple_groups_same_RP_address_p2(request):
assert result is True, ASSERT_MSG.format(tc_name, result)
step("r5: Send multicast traffic for group {}".format(group_address_list))
intf = TOPO["routers"]["r5"]["links"]["r3"]["interface"]
SOURCE_ADDRESS = TOPO["routers"]["r5"]["links"]["r3"]["ipv6"].split("/")[0]
result = socat_send_pim6_traffic(tgen, "r5", "UDP6-SEND", group_address_list, intf)
result = app_helper.run_traffic("r5", group_address_list, "r3")
assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
step("r1: Verify (*, G) upstream IIF interface")
@ -593,6 +569,8 @@ def test_pim6_multiple_groups_different_RP_address_p2(request):
step("Creating configuration from JSON")
reset_config_on_routers(tgen)
app_helper.stop_all_hosts()
step("Enable MLD on r1 interface")
step("Enable the PIM6 on all the interfaces of r1, r2, r3 and r4 routers")
step("r2: Configure r2 as RP")
@ -646,11 +624,7 @@ def test_pim6_multiple_groups_different_RP_address_p2(request):
group_address_list = GROUP_ADDRESS_LIST_1 + GROUP_ADDRESS_LIST_2
step("r0: Send MLD join for 10 groups")
intf = TOPO["routers"]["r0"]["links"]["r1"]["interface"]
intf_ip = TOPO["routers"]["r0"]["links"]["r1"]["ipv6"].split("/")[0]
result = socat_send_mld_join(
tgen, "r0", "UDP6-RECV", group_address_list, intf, intf_ip
)
result = app_helper.run_join("r0", group_address_list, "r1")
assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
step("r1: Verify MLD groups")
@ -660,9 +634,8 @@ def test_pim6_multiple_groups_different_RP_address_p2(request):
assert result is True, ASSERT_MSG.format(tc_name, result)
step("r5: Send multicast traffic for group {}".format(group_address_list))
intf = TOPO["routers"]["r5"]["links"]["r3"]["interface"]
SOURCE_ADDRESS = TOPO["routers"]["r5"]["links"]["r3"]["ipv6"].split("/")[0]
result = socat_send_pim6_traffic(tgen, "r5", "UDP6-SEND", group_address_list, intf)
result = app_helper.run_traffic("r5", group_address_list, "r3")
assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
step("r1: Verify (*, G) upstream IIF interface")
@ -1189,6 +1162,8 @@ def test_pim6_delete_RP_shut_noshut_upstream_interface_p1(request):
step("Creating configuration from JSON")
reset_config_on_routers(tgen)
app_helper.stop_all_hosts()
step("Enable MLD on r1 interface")
step("Enable the PIM6 on all the interfaces of r1, r2, r3 and r4 routers")
step("r2: Configure r2 as RP")
@ -1220,11 +1195,7 @@ def test_pim6_delete_RP_shut_noshut_upstream_interface_p1(request):
assert result is True, ASSERT_MSG.format(tc_name, result)
step("r0: Send MLD join")
intf = TOPO["routers"]["r0"]["links"]["r1"]["interface"]
intf_ip = TOPO["routers"]["r0"]["links"]["r1"]["ipv6"].split("/")[0]
result = socat_send_mld_join(
tgen, "r0", "UDP6-RECV", GROUP_ADDRESS_1, intf, intf_ip
)
result = app_helper.run_join("r0", GROUP_ADDRESS_1, "r1")
assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
step("r1: Verify MLD groups")