diff --git a/tests/topotests/lib/mcast-tester.py b/tests/topotests/lib/mcast-tester.py index 8a8251010c..5efbecd5e5 100755 --- a/tests/topotests/lib/mcast-tester.py +++ b/tests/topotests/lib/mcast-tester.py @@ -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) diff --git a/tests/topotests/lib/pim.py b/tests/topotests/lib/pim.py index e26bdb3af3..f69718a5bd 100644 --- a/tests/topotests/lib/pim.py +++ b/tests/topotests/lib/pim.py @@ -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__)) diff --git a/tests/topotests/multicast_mld_join_topo1/test_multicast_mld_local_join.py b/tests/topotests/multicast_mld_join_topo1/test_multicast_mld_local_join.py index 2c4fb4e998..826d6e2941 100644 --- a/tests/topotests/multicast_mld_join_topo1/test_multicast_mld_local_join.py +++ b/tests/topotests/multicast_mld_join_topo1/test_multicast_mld_local_join.py @@ -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( diff --git a/tests/topotests/multicast_pim6_sm_topo1/test_multicast_pim6_sm1.py b/tests/topotests/multicast_pim6_sm_topo1/test_multicast_pim6_sm1.py index 87b04b41be..aff623705c 100644 --- a/tests/topotests/multicast_pim6_sm_topo1/test_multicast_pim6_sm1.py +++ b/tests/topotests/multicast_pim6_sm_topo1/test_multicast_pim6_sm1.py @@ -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( diff --git a/tests/topotests/multicast_pim6_sm_topo1/test_multicast_pim6_sm2.py b/tests/topotests/multicast_pim6_sm_topo1/test_multicast_pim6_sm2.py index 788a839918..767264a7c0 100644 --- a/tests/topotests/multicast_pim6_sm_topo1/test_multicast_pim6_sm2.py +++ b/tests/topotests/multicast_pim6_sm_topo1/test_multicast_pim6_sm2.py @@ -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( diff --git a/tests/topotests/multicast_pim6_static_rp_topo1/test_multicast_pim6_static_rp1.py b/tests/topotests/multicast_pim6_static_rp_topo1/test_multicast_pim6_static_rp1.py index 977cd477c8..23326337d6 100755 --- a/tests/topotests/multicast_pim6_static_rp_topo1/test_multicast_pim6_static_rp1.py +++ b/tests/topotests/multicast_pim6_static_rp_topo1/test_multicast_pim6_static_rp1.py @@ -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") diff --git a/tests/topotests/multicast_pim6_static_rp_topo1/test_multicast_pim6_static_rp2.py b/tests/topotests/multicast_pim6_static_rp_topo1/test_multicast_pim6_static_rp2.py index a61164baa2..39497e91ed 100755 --- a/tests/topotests/multicast_pim6_static_rp_topo1/test_multicast_pim6_static_rp2.py +++ b/tests/topotests/multicast_pim6_static_rp_topo1/test_multicast_pim6_static_rp2.py @@ -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")