tests: PIM AutoRP candidate RP tests

Signed-off-by: Nathan Bahr <[email protected]>

Author: nabahr

tests/topotests/lib/pim.py

@@ -1779,16 +1779,15 @@ def verify_pim_rp_info_is_empty(tgen, dut, af="ipv4"):
     logger.info("[DUT: %s]: Verifying %s rp info", dut, ip_cmd)
     cmd = "show {} pim rp-info json".format(ip_cmd)
     show_ip_rp_info_json = run_frr_cmd(rnode, cmd, isjson=True)
-    
+
     if show_ip_rp_info_json:
-        errormsg = (
-            "[DUT %s]: Verifying empty rp-info [FAILED]!!" % (dut)
-        )
+        errormsg = "[DUT %s]: Verifying empty rp-info [FAILED]!!" % (dut)
         return errormsg
 
     logger.debug("Exiting lib API: {}".format(sys._getframe().f_code.co_name))
     return True
 
+
 @retry(retry_timeout=60, diag_pct=0)
 def verify_pim_state(
     tgen,
@@ -2794,6 +2793,7 @@ def scapy_send_bsr_raw_packet(tgen, topo, senderRouter, receiverRouter, packet=N
     logger.debug("Exiting lib API: scapy_send_bsr_raw_packet")
     return True
 
+
 def scapy_send_autorp_raw_packet(tgen, senderRouter, senderInterface, packet=None):
     """
     Using scapy Raw() method to send AutoRP raw packet from one FRR
@@ -2835,6 +2835,7 @@ def scapy_send_autorp_raw_packet(tgen, senderRouter, senderInterface, packet=Non
     logger.debug("Exiting lib API: scapy_send_autorp_raw_packet")
     return True
 
+
 def find_rp_from_bsrp_info(tgen, dut, bsr, grp=None):
     """
     Find which RP is having lowest prioriy and returns rp IP
@@ -4344,6 +4345,7 @@ def verify_local_igmp_groups(tgen, dut, interface, group_addresses):
     logger.debug("Exiting lib API: {}".format(sys._getframe().f_code.co_name))
     return True
 
+
 @retry(retry_timeout=62)
 def verify_static_groups(tgen, dut, interface, group_addresses):
     """
@@ -4377,7 +4379,9 @@ def verify_static_groups(tgen, dut, interface, group_addresses):
     rnode = tgen.routers()[dut]
 
     logger.info("[DUT: %s]: Verifying static groups received:", dut)
-    show_static_group_json = run_frr_cmd(rnode, "show ip igmp static-group json", isjson=True)
+    show_static_group_json = run_frr_cmd(
+        rnode, "show ip igmp static-group json", isjson=True
+    )
 
     if type(group_addresses) is not list:
         group_addresses = [group_addresses]
@@ -4414,6 +4418,7 @@ def verify_static_groups(tgen, dut, interface, group_addresses):
     logger.debug("Exiting lib API: {}".format(sys._getframe().f_code.co_name))
     return True
 
+
 def verify_pim_interface_traffic(tgen, input_dict, return_stats=True, addr_type="ipv4"):
     """
     Verify ip pim interface traffic by running

tests/topotests/multicast_pim_autorp/__init__.py renamed to tests/topotests/pim_autorp/__init__.py

@@ -5,7 +5,7 @@ log file /tmp/r1-frr.log
 debug pim autorp
 !
 interface r1-eth0
- ip address 10.10.76.2/24
+ ip address 10.10.76.1/24
  ip igmp
  ip pim
 !

tests/topotests/multicast_pim_autorp/r1/frr.conf renamed to tests/topotests/pim_autorp/r1/frr.conf

@@ -5,7 +5,7 @@ log file /tmp/r2-frr.log
 debug pim autorp
 !
 interface r2-eth0
- ip address 10.10.76.1/24
+ ip address 10.10.76.2/24
  ip igmp
  ip pim
 !

tests/topotests/multicast_pim_autorp/r2/frr.conf renamed to tests/topotests/pim_autorp/r2/frr.conf

@@ -2,7 +2,7 @@
 # SPDX-License-Identifier: ISC
 
 #
-# test_multicast_pim_autorp.py
+# test_pim_autorp.py
 #
 # Copyright (c) 2024 ATCorp
 # Nathan Bahr
@@ -19,8 +19,10 @@
 from lib.pim import scapy_send_autorp_raw_packet, verify_pim_rp_info, verify_pim_rp_info_is_empty
 from lib.common_config import step, write_test_header
 
+from time import sleep
+
 """
-test_multicast_pim_autorp.py: Test general PIM AutoRP functionality
+test_pim_autorp.py: Test general PIM AutoRP functionality
 """
 
 TOPOLOGY = """
@@ -52,7 +54,7 @@ def build_topo(tgen):
     switch = tgen.add_switch("s1-2")
     switch.add_link(tgen.gears["r1"])
     switch.add_link(tgen.gears["r2"])
-    
+
 def setup_module(mod):
     logger.info("PIM AutoRP basic functionality:\n {}".format(TOPOLOGY))
 
@@ -97,20 +99,20 @@ def test_pim_autorp_discovery_single_rp(request):
     step("Start with no RP configuration")
     result = verify_pim_rp_info_is_empty(tgen, "r1")
     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
-    
-    step("Send AutoRP packet from r2 to r1")
+
+    step("Send AutoRP packet from r1 to r2")
     # 1 RP(s), hold time 5 secs, 10.10.76.1, group(s) 224.0.0.0/4
     data = "01005e00012800127f55cfb1080045c00030700c000008110abe0a0a4c01e000012801f001f0001c798b12010005000000000a0a4c0103010004e0000000"
-    scapy_send_autorp_raw_packet(tgen, "r2", "r2-eth0", data)
-    
+    scapy_send_autorp_raw_packet(tgen, "r1", "r1-eth0", data)
+
     step("Verify rp-info from AutoRP packet")
-    result = verify_pim_rp_info(tgen, None, "r1", "224.0.0.0/4", "r1-eth0", "10.10.76.1", "AutoRP", False, "ipv4", True)
+    result = verify_pim_rp_info(tgen, None, "r2", "224.0.0.0/4", "r2-eth0", "10.10.76.1", "AutoRP", False, "ipv4", True)
     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
-    
+
     step("Verify AutoRP configuration times out")
-    result = verify_pim_rp_info_is_empty(tgen, "r1")
+    result = verify_pim_rp_info_is_empty(tgen, "r2")
     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
-    
+
 def test_pim_autorp_discovery_multiple_rp(request):
     "Test PIM AutoRP Discovery with multiple RP's"
     tgen = get_topogen()
@@ -121,20 +123,20 @@ def test_pim_autorp_discovery_multiple_rp(request):
         pytest.skip("skipped because of router(s) failure")
 
     step("Start with no RP configuration")
-    result = verify_pim_rp_info_is_empty(tgen, "r1")
+    result = verify_pim_rp_info_is_empty(tgen, "r2")
     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
-    
-    step("Send AutoRP packet from r2 to r1")
+
+    step("Send AutoRP packet from r1 to r2")
     # 2 RP(s), hold time 5 secs, 10.10.76.1, group(s) 224.0.0.0/8, 10.10.76.3, group(s) 225.0.0.0/8
     data = "01005e00012800127f55cfb1080045c0003c700c000008110ab20a0a4c01e000012801f001f000283f5712020005000000000a0a4c0103010008e00000000a0a4c0303010008e1000000"
-    scapy_send_autorp_raw_packet(tgen, "r2", "r2-eth0", data)
-    
+    scapy_send_autorp_raw_packet(tgen, "r1", "r1-eth0", data)
+
     step("Verify rp-info from AutoRP packet")
-    result = verify_pim_rp_info(tgen, None, "r1", "224.0.0.0/8", "r1-eth0", "10.10.76.1", "AutoRP", False, "ipv4", True)
+    result = verify_pim_rp_info(tgen, None, "r2", "224.0.0.0/8", "r2-eth0", "10.10.76.1", "AutoRP", False, "ipv4", True)
     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
-    result = verify_pim_rp_info(tgen, None, "r1", "225.0.0.0/8", "r1-eth0", "10.10.76.3", "AutoRP", False, "ipv4", True)
+    result = verify_pim_rp_info(tgen, None, "r2", "225.0.0.0/8", "r2-eth0", "10.10.76.3", "AutoRP", False, "ipv4", True)
     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
-    
+
 
 def test_pim_autorp_discovery_static(request):
     "Test PIM AutoRP Discovery with Static RP"
@@ -146,26 +148,51 @@ def test_pim_autorp_discovery_static(request):
         pytest.skip("skipped because of router(s) failure")
 
     step("Start with no RP configuration")
-    result = verify_pim_rp_info_is_empty(tgen, "r1")
+    result = verify_pim_rp_info_is_empty(tgen, "r2")
     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
-    
-    step("Add static RP configuration to r1")
-    rnode = tgen.routers()["r1"]
+
+    step("Add static RP configuration to r2")
+    rnode = tgen.routers()["r2"]
     rnode.cmd("vtysh -c 'conf t' -c 'router pim' -c 'rp 10.10.76.3 224.0.0.0/4'")
-    
-    step("Verify static rp-info from r1")
-    result = verify_pim_rp_info(tgen, None, "r1", "224.0.0.0/4", "r1-eth0", "10.10.76.3", "Static", False, "ipv4", True)
+
+    step("Verify static rp-info from r2")
+    result = verify_pim_rp_info(tgen, None, "r2", "224.0.0.0/4", "r2-eth0", "10.10.76.3", "Static", False, "ipv4", True)
     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
-    
-    step("Send AutoRP packet from r2 to r1")
+
+    step("Send AutoRP packet from r1 to r2")
     # 1 RP(s), hold time 5 secs, 10.10.76.1, group(s) 224.0.0.0/4
     data = "01005e00012800127f55cfb1080045c00030700c000008110abe0a0a4c01e000012801f001f0001c798b12010005000000000a0a4c0103010004e0000000"
-    scapy_send_autorp_raw_packet(tgen, "r2", "r2-eth0", data)
-    
+    scapy_send_autorp_raw_packet(tgen, "r1", "r1-eth0", data)
+
     step("Verify rp-info from AutoRP packet")
-    result = verify_pim_rp_info(tgen, None, "r1", "224.0.0.0/4", "r1-eth0", "10.10.76.1", "AutoRP", False, "ipv4", True)
+    result = verify_pim_rp_info(tgen, None, "r2", "224.0.0.0/4", "r2-eth0", "10.10.76.1", "AutoRP", False, "ipv4", True)
     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 
+
+def test_pim_autorp_announce_group(request):
+    "Test PIM AutoRP Announcement with a single group"
+    tgen = get_topogen()
+    tc_name = request.node.name
+    write_test_header(tc_name)
+
+    if tgen.routers_have_failure():
+        pytest.skip("skipped because of router(s) failure")
+
+    step("Add candidate RP configuration to r1")
+    rnode = tgen.routers()["r1"]
+    rnode.cmd("vtysh -c 'conf t' -c 'router pim' -c 'send-rp-announce 10.10.76.1 224.0.0.0/4'")
+    step("Verify Announcement sent data")
+    # TODO: Verify AutoRP mapping agent receives candidate RP announcement
+    # Mapping agent is not yet implemented
+    #sleep(10)
+    step("Change AutoRP Announcement packet parameters")
+    rnode.cmd("vtysh -c 'conf t' -c 'router pim' -c 'send-rp-announce scope 8 interval 10 holdtime 60'")
+    step("Verify Announcement sent data")
+    # TODO: Verify AutoRP mapping agent receives updated candidate RP announcement
+    # Mapping agent is not yet implemented
+    #sleep(10)
+
+
 def test_memory_leak():
     "Run the memory leak test and report results."
     tgen = get_topogen()