From 71231d304f257aed9a086fcf6c2146052574b2f4 Mon Sep 17 00:00:00 2001 From: Christian Hopps Date: Thu, 8 Jun 2023 02:11:50 -0400 Subject: [PATCH 1/3] tests: mcast-tester.py handles IPv6 Signed-off-by: Christian Hopps --- tests/topotests/lib/mcast-tester.py | 42 +++++++++++++++++++---------- 1 file changed, 28 insertions(+), 14 deletions(-) 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) From efedb1897616c6de76aaaff7a51f443dbbe93e31 Mon Sep 17 00:00:00 2001 From: Christian Hopps Date: Thu, 8 Jun 2023 02:42:32 -0400 Subject: [PATCH 2/3] tests: fixing pim6 topotest bugs - Remove use of bespoke socat - Use ipv6 support in mcast-tester.py - do not run processes in the background behind munet/micronet's back with `&` (ever) -- use popen or the helper class Signed-off-by: Christian Hopps --- tests/topotests/lib/pim.py | 12 +- .../test_multicast_mld_local_join.py | 72 +++----- .../test_multicast_pim6_sm1.py | 172 ++++++------------ .../test_multicast_pim6_sm2.py | 91 +++------ .../test_multicast_pim6_static_rp1.py | 121 +++++------- .../test_multicast_pim6_static_rp2.py | 91 ++++----- 6 files changed, 192 insertions(+), 367 deletions(-) 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..b049a1a57d 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" @@ -132,8 +102,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 @@ -150,6 +119,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") @@ -160,8 +132,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() @@ -237,6 +208,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 +239,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") @@ -470,6 +437,8 @@ def test_verify_SPT_switchover_when_RPT_and_SPT_path_is_different_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) @@ -498,11 +467,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 +483,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") From b28bc2561e7231e410e304f7ec23f9d795e3e479 Mon Sep 17 00:00:00 2001 From: Christian Hopps Date: Thu, 8 Jun 2023 04:12:26 -0400 Subject: [PATCH 3/3] tests: convert old pim test to more cleanly use pytest fixture This is a good way to run a per-test background helper process. Here the helper object is created before the test function requesting it (through param name match), and then cleaned up after the test function exits (pass or failed). A context manager is used to further guarantee the cleanup is done. Signed-off-by: Christian Hopps --- .../test_multicast_pim6_sm2.py | 23 +++++++++++-------- 1 file changed, 14 insertions(+), 9 deletions(-) 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 b049a1a57d..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 @@ -84,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 @@ -119,9 +129,6 @@ 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") @@ -132,8 +139,6 @@ def teardown_module(): tgen = get_topogen() - app_helper.cleanup() - # Stop toplogy and Remove tmp files tgen.stop_topology() @@ -196,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 @@ -424,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 @@ -437,8 +444,6 @@ def test_verify_SPT_switchover_when_RPT_and_SPT_path_is_different_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)