tests/topotests/multicast_pim6_static_rp_topo1/test_multicast_pim6_static_rp2.py

   1 #!/usr/bin/env python
   2
   3 #
   4 # Copyright (c) 2022 by VMware, Inc. ("VMware")
   5 # Used Copyright (c) 2018 by Network Device Education Foundation,
   6 # Inc. ("NetDEF") in this file.
   7 #
   8 # Permission to use, copy, modify, and/or distribute this software
   9 # for any purpose with or without fee is hereby granted, provided
  10 # that the above copyright notice and this permission notice appear
  11 # in all copies.
  12 #
  13 # THE SOFTWARE IS PROVIDED "AS IS" AND VMWARE DISCLAIMS ALL WARRANTIES
  14 # WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
  15 # MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL VMWARE BE LIABLE FOR
  16 # ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY
  17 # DAMAGES WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS,
  18 # WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS
  19 # ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR PERFORMANCE
  20 # OF THIS SOFTWARE.
  21 #
  22
  23 """
  24 Following tests are covered to test Multicast basic functionality:
  25
  26 Topology:
  27
  28                  _______r2_____
  29                 |             |
  30       iperf     |             |     iperf
  31         r0-----r1-------------r3-----r5
  32                 |             |
  33                 |_____________|
  34                         r4
  35
  36 Test steps
  37 - Create topology (setup module)
  38 - Bring up topology
  39
  40 1. Configure multiple  groups (10 grps) with same RP address
  41 2. Verify IIF and OIL in updated in mroute when upstream interface
  42    configure as RP
  43 3. Verify RP info and (*,G) mroute after deleting the RP and shut /
  44     no shut the RPF interface.
  45 """
  46
  47 import os
  48 import sys
  49 import json
  50 import time
  51 import pytest
  52
  53 # Save the Current Working Directory to find configuration files.
  54 CWD = os.path.dirname(os.path.realpath(__file__))
  55 sys.path.append(os.path.join(CWD, "../"))
  56 sys.path.append(os.path.join(CWD, "../lib/"))
  57
  58 # Required to instantiate the topology builder class.
  59
  60 # pylint: disable=C0413
  61 # Import topogen and topotest helpers
  62 from lib.topogen import Topogen, get_topogen
  63
  64 from lib.common_config import (
  65     start_topology,
  66     write_test_header,
  67     write_test_footer,
  68     reset_config_on_routers,
  69     step,
  70     shutdown_bringup_interface,
  71     kill_router_daemons,
  72     start_router_daemons,
  73     create_static_routes,
  74     check_router_status,
  75     socat_send_mld_join,
  76     socat_send_pim6_traffic,
  77     kill_socat,
  78 )
  79 from lib.pim import (
  80     create_pim_config,
  81     verify_upstream_iif,
  82     verify_join_state_and_timer,
  83     verify_mroutes,
  84     verify_pim_neighbors,
  85     verify_pim_interface_traffic,
  86     verify_pim_rp_info,
  87     verify_pim_state,
  88     clear_pim6_interface_traffic,
  89     clear_pim6_mroute,
  90     verify_pim6_neighbors,
  91     get_pim6_interface_traffic,
  92     clear_pim6_interfaces,
  93     verify_mld_groups,
  94 )
  95 from lib.topolog import logger
  96 from lib.topojson import build_topo_from_json, build_config_from_json
  97
  98 # Global variables
  99 GROUP_RANGE_1 = "ff08::/64"
 100 GROUP_ADDRESS_1 = "ff08::1"
 101 GROUP_RANGE_3 = "ffaa::/64"
 102 GROUP_ADDRESS_3 = "ffaa::1"
 103 GROUP_RANGE_LIST_1 = [
 104     "ffaa::1/128",
 105     "ffaa::2/128",
 106     "ffaa::3/128",
 107     "ffaa::4/128",
 108     "ffaa::5/128",
 109 ]
 110 GROUP_RANGE_LIST_2 = [
 111     "ffaa::6/128",
 112     "ffaa::7/128",
 113     "ffaa::8/128",
 114     "ffaa::9/128",
 115     "ffaa::10/128",
 116 ]
 117 GROUP_ADDRESS_LIST_1 = ["ffaa::1", "ffaa::2", "ffaa::3", "ffaa::4", "ffaa::5"]
 118 GROUP_ADDRESS_LIST_2 = ["ffaa::6", "ffaa::7", "ffaa::8", "ffaa::9", "ffaa::10"]
 119 STAR = "*"
 120 SOURCE = "Static"
 121 ASSERT_MSG = "Testcase {} : Failed Error: {}"
 122
 123 pytestmark = [pytest.mark.pim6d]
 124
 125
 126 def build_topo(tgen):
 127     """Build function"""
 128
 129     # Building topology from json file
 130     build_topo_from_json(tgen, TOPO)
 131
 132
 133 def setup_module(mod):
 134     """
 135     Sets up the pytest environment
 136
 137     * `mod`: module name
 138     """
 139
 140     testsuite_run_time = time.asctime(time.localtime(time.time()))
 141     logger.info("Testsuite start time: %s", testsuite_run_time)
 142     logger.info("=" * 40)
 143
 144     topology = """
 145
 146                  _______r2_____
 147                 |             |
 148       iperf     |             |     iperf
 149         r0-----r1-------------r3-----r5
 150                 |             |
 151                 |_____________|
 152                         r4
 153
 154     """
 155     logger.info("Master Topology: \n %s", topology)
 156
 157     logger.info("Running setup_module to create topology")
 158
 159     # This function initiates the topology build with Topogen...
 160     json_file = "{}/multicast_pim6_static_rp.json".format(CWD)
 161     tgen = Topogen(json_file, mod.__name__)
 162     global TOPO
 163     TOPO = tgen.json_topo
 164
 165     # ... and here it calls Mininet initialization functions.
 166
 167     # Starting topology, create tmp files which are loaded to routers
 168     #  to start daemons and then start routers
 169     start_topology(tgen)
 170
 171     # Don"t run this test if we have any failure.
 172     if tgen.routers_have_failure():
 173         pytest.skip(tgen.errors)
 174
 175     # Creating configuration from JSON
 176     build_config_from_json(tgen, TOPO)
 177
 178     # Verify PIM6 neighbors
 179     result = verify_pim6_neighbors(tgen, TOPO)
 180     assert result is True, "setup_module :Failed \n Error:" " {}".format(result)
 181
 182     logger.info("Running setup_module() done")
 183
 184
 185 def teardown_module():
 186     """Teardown the pytest environment"""
 187
 188     logger.info("Running teardown_module to delete topology")
 189     tgen = get_topogen()
 190
 191     # Stop toplogy and Remove tmp files
 192     tgen.stop_topology()
 193
 194     logger.info("Testsuite end time: %s", time.asctime(time.localtime(time.time())))
 195     logger.info("=" * 40)
 196
 197
 198 #####################################################
 199 #
 200 #   Local API
 201 #
 202 #####################################################
 203
 204
 205 def verify_state_incremented(state_before, state_after):
 206     """
 207     API to compare interface traffic state incrementing
 208
 209     Parameters
 210     ----------
 211     * `state_before` : State dictionary for any particular instance
 212     * `state_after` : State dictionary for any particular instance
 213     """
 214
 215     for router, state_data in state_before.items():
 216         for state, value in state_data.items():
 217             if state_before[router][state] >= state_after[router][state]:
 218                 errormsg = (
 219                     "[DUT: %s]: state %s value has not"
 220                     " incremented, Initial value: %s, "
 221                     "Current value: %s [FAILED!!]"
 222                     % (
 223                         router,
 224                         state,
 225                         state_before[router][state],
 226                         state_after[router][state],
 227                     )
 228                 )
 229                 return errormsg
 230
 231             logger.info(
 232                 "[DUT: %s]: State %s value is "
 233                 "incremented, Initial value: %s, Current value: %s"
 234                 " [PASSED!!]",
 235                 router,
 236                 state,
 237                 state_before[router][state],
 238                 state_after[router][state],
 239             )
 240
 241     return True
 242
 243
 244 #####################################################
 245 #
 246 #   Testcases
 247 #
 248 #####################################################
 249
 250
 251 def test_pim6_multiple_groups_same_RP_address_p2(request):
 252     """
 253     Configure multiple  groups (10 grps) with same RP address
 254
 255     Topology used:
 256                 ________r2_____
 257                 |             |
 258       iperf     |             |     iperf
 259         r0-----r1-------------r3-----r5
 260
 261     r1 : LHR
 262     r2 : RP
 263     r3 : FHR
 264     """
 265
 266     tgen = get_topogen()
 267     tc_name = request.node.name
 268     write_test_header(tc_name)
 269
 270     # Don"t run this test if we have any failure.
 271     if tgen.routers_have_failure():
 272         pytest.skip(tgen.errors)
 273
 274     step("Creating configuration from JSON")
 275     kill_socat(tgen)
 276     clear_pim6_mroute(tgen)
 277     clear_pim6_interface_traffic(tgen, TOPO)
 278     reset_config_on_routers(tgen)
 279
 280     step("Enable MLD on r1 interface")
 281     step("Enable the PIM6 on all the interfaces of r1, r2, r3 and r4 routers")
 282     step("r2: Configure r2 as RP")
 283     input_dict = {
 284         "r2": {
 285             "pim6": {
 286                 "rp": [
 287                     {
 288                         "rp_addr": TOPO["routers"]["r2"]["links"]["lo"]["ipv6"].split(
 289                             "/"
 290                         )[0],
 291                         "group_addr_range": GROUP_RANGE_3,
 292                     }
 293                 ]
 294             }
 295         }
 296     }
 297
 298     result = create_pim_config(tgen, TOPO, input_dict)
 299     assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
 300
 301     step("r2: verify rp-info")
 302     dut = "r2"
 303     oif = "lo"
 304     rp_address = TOPO["routers"]["r2"]["links"]["lo"]["ipv6"].split("/")[0]
 305     result = verify_pim_rp_info(tgen, TOPO, dut, GROUP_RANGE_3, oif, rp_address, SOURCE)
 306     assert result is True, ASSERT_MSG.format(tc_name, result)
 307
 308     group_address_list = GROUP_ADDRESS_LIST_1 + GROUP_ADDRESS_LIST_2
 309     step("r0: Send MLD join for 10 groups")
 310     intf = TOPO["routers"]["r0"]["links"]["r1"]["interface"]
 311     intf_ip = TOPO["routers"]["r0"]["links"]["r1"]["ipv6"].split("/")[0]
 312     result = socat_send_mld_join(
 313         tgen, "r0", "UDP6-RECV", group_address_list, intf, intf_ip
 314     )
 315     assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 316
 317     step("r1: Verify MLD groups")
 318     dut = "r1"
 319     intf_r1_r0 = TOPO["routers"]["r1"]["links"]["r0"]["interface"]
 320     result = verify_mld_groups(tgen, dut, intf_r1_r0, group_address_list)
 321     assert result is True, ASSERT_MSG.format(tc_name, result)
 322
 323     step("r5: Send multicast traffic for group {}".format(group_address_list))
 324     intf = TOPO["routers"]["r5"]["links"]["r3"]["interface"]
 325     SOURCE_ADDRESS = TOPO["routers"]["r5"]["links"]["r3"]["ipv6"].split("/")[0]
 326     result = socat_send_pim6_traffic(tgen, "r5", "UDP6-SEND", group_address_list, intf)
 327     assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 328
 329     step("r1: Verify (*, G) upstream IIF interface")
 330     dut = "r1"
 331     iif = TOPO["routers"]["r1"]["links"]["r2"]["interface"]
 332     result = verify_upstream_iif(tgen, dut, iif, STAR, group_address_list)
 333     assert result is True, ASSERT_MSG.format(tc_name, result)
 334
 335     step("r1: Verify (*, G) upstream join state and join timer")
 336     result = verify_join_state_and_timer(
 337         tgen, dut, iif, STAR, group_address_list, addr_type="ipv6"
 338     )
 339     assert result is True, ASSERT_MSG.format(tc_name, result)
 340
 341     step("r1: Verify (*, G) ip mroutes")
 342     oif = TOPO["routers"]["r1"]["links"]["r0"]["interface"]
 343     result = verify_mroutes(tgen, dut, STAR, group_address_list, iif, oif)
 344     assert result is True, ASSERT_MSG.format(tc_name, result)
 345
 346     step("r1: Verify (S, G) ip mroutes")
 347     iif = TOPO["routers"]["r1"]["links"]["r3"]["interface"]
 348     result = verify_mroutes(tgen, dut, SOURCE_ADDRESS, group_address_list, iif, oif)
 349     assert result is True, ASSERT_MSG.format(tc_name, result)
 350
 351     step("r1: Verify (S, G) upstream IIF interface")
 352     result = verify_upstream_iif(tgen, dut, iif, SOURCE_ADDRESS, group_address_list)
 353     assert result is True, ASSERT_MSG.format(tc_name, result)
 354
 355     step("r1: Verify (S, G) upstream join state and join timer")
 356     result = verify_join_state_and_timer(
 357         tgen, dut, iif, SOURCE_ADDRESS, group_address_list, addr_type="ipv6"
 358     )
 359     assert result is True, ASSERT_MSG.format(tc_name, result)
 360
 361     step("r2: Verify (*, G) upstream IIF interface")
 362     dut = "r2"
 363     iif = "lo"
 364     result = verify_upstream_iif(tgen, dut, iif, STAR, group_address_list)
 365     assert result is True, ASSERT_MSG.format(tc_name, result)
 366
 367     step("r2: Verify (*, G) upstream join state and join timer")
 368     result = verify_join_state_and_timer(
 369         tgen, dut, iif, STAR, group_address_list, addr_type="ipv6"
 370     )
 371     assert result is True, ASSERT_MSG.format(tc_name, result)
 372
 373     step("r2: Verify (*, G) ip mroutes")
 374     oif = TOPO["routers"]["r2"]["links"]["r1"]["interface"]
 375     result = verify_mroutes(tgen, dut, STAR, group_address_list, iif, oif)
 376     assert result is True, ASSERT_MSG.format(tc_name, result)
 377
 378     step("r3: Verify (S, G) upstream IIF interface")
 379     dut = "r3"
 380     iif = TOPO["routers"]["r3"]["links"]["r5"]["interface"]
 381     result = verify_upstream_iif(tgen, dut, iif, SOURCE_ADDRESS, group_address_list)
 382     assert result is True, ASSERT_MSG.format(tc_name, result)
 383
 384     step("r3: Verify (S, G) upstream join state and join timer")
 385     result = verify_join_state_and_timer(
 386         tgen,
 387         dut,
 388         iif,
 389         SOURCE_ADDRESS,
 390         group_address_list,
 391         addr_type="ipv6",
 392         expected=False,
 393     )
 394     assert result is not True, (
 395         "Testcase {} : Failed \n "
 396         "r3: (S,G) upstream state is joined and join timer is running\n Error: {}".format(
 397             tc_name, result
 398         )
 399     )
 400
 401     step("r3: Verify (S, G) ip mroutes")
 402     oif = TOPO["routers"]["r3"]["links"]["r1"]["interface"]
 403     result = verify_mroutes(tgen, dut, SOURCE_ADDRESS, group_address_list, iif, oif)
 404     assert result is True, ASSERT_MSG.format(tc_name, result)
 405
 406     step("r2: Verify (S, G) upstream IIF interface")
 407     dut = "r2"
 408     iif = TOPO["routers"]["r2"]["links"]["r3"]["interface"]
 409     result = verify_upstream_iif(
 410         tgen, dut, iif, SOURCE_ADDRESS, group_address_list, joinState="NotJoined"
 411     )
 412     assert result is True, ASSERT_MSG.format(tc_name, result)
 413
 414     step("r2: Verify (S, G) upstream join state and join timer")
 415     result = verify_join_state_and_timer(
 416         tgen,
 417         dut,
 418         iif,
 419         SOURCE_ADDRESS,
 420         group_address_list,
 421         addr_type="ipv6",
 422         expected=False,
 423     )
 424     assert result is not True, (
 425         "Testcase {} : Failed \n "
 426         "r2: (S,G) upstream state is joined and join timer is running\n Error: {}".format(
 427             tc_name, result
 428         )
 429     )
 430
 431     step("r2: Verify (S, G) ip mroutes")
 432     oif = "none"
 433     result = verify_mroutes(tgen, dut, SOURCE_ADDRESS, group_address_list, iif, oif)
 434     assert result is True, ASSERT_MSG.format(tc_name, result)
 435
 436     step("r1: Delete RP configuration")
 437     input_dict = {
 438         "r1": {
 439             "pim6": {
 440                 "rp": [
 441                     {
 442                         "rp_addr": TOPO["routers"]["r2"]["links"]["lo"]["ipv6"].split(
 443                             "/"
 444                         )[0],
 445                         "group_addr_range": GROUP_RANGE_3,
 446                         "delete": True,
 447                     }
 448                 ]
 449             }
 450         }
 451     }
 452
 453     result = create_pim_config(tgen, TOPO, input_dict)
 454     assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
 455
 456     step("r1: Shut the interface r1-r2-eth1 from R1 to R2")
 457     dut = "r1"
 458     intf = TOPO["routers"]["r1"]["links"]["r2"]["interface"]
 459     shutdown_bringup_interface(tgen, dut, intf, False)
 460
 461     step("r1: No Shut the interface r1-r2-eth1 from R1 to R2")
 462     shutdown_bringup_interface(tgen, dut, intf, True)
 463
 464     step("r1: Configure RP")
 465     input_dict = {
 466         "r1": {
 467             "pim6": {
 468                 "rp": [
 469                     {
 470                         "rp_addr": TOPO["routers"]["r2"]["links"]["lo"]["ipv6"].split(
 471                             "/"
 472                         )[0],
 473                         "group_addr_range": GROUP_RANGE_3,
 474                     }
 475                 ]
 476             }
 477         }
 478     }
 479
 480     result = create_pim_config(tgen, TOPO, input_dict)
 481     assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
 482
 483     step("r1: Shut the interface r1-r0-eth0 from R1 to R2")
 484     intf = TOPO["routers"]["r1"]["links"]["r0"]["interface"]
 485     shutdown_bringup_interface(tgen, dut, intf, False)
 486
 487     step("r1: No Shut the interface r1-r0-eth0 from R1 to R2")
 488     shutdown_bringup_interface(tgen, dut, intf, True)
 489
 490     step("r1: Verify (*, G) upstream IIF interface")
 491     dut = "r1"
 492     iif = TOPO["routers"]["r1"]["links"]["r2"]["interface"]
 493     result = verify_upstream_iif(tgen, dut, iif, STAR, group_address_list)
 494     assert result is True, ASSERT_MSG.format(tc_name, result)
 495
 496     step("r1: Verify (*, G) upstream join state and join timer")
 497     result = verify_join_state_and_timer(
 498         tgen, dut, iif, STAR, group_address_list, addr_type="ipv6"
 499     )
 500     assert result is True, ASSERT_MSG.format(tc_name, result)
 501
 502     step("r1: Verify (*, G) ip mroutes")
 503     oif = TOPO["routers"]["r1"]["links"]["r0"]["interface"]
 504     result = verify_mroutes(tgen, dut, STAR, group_address_list, iif, oif)
 505     assert result is True, ASSERT_MSG.format(tc_name, result)
 506
 507     step("r1: Verify (S, G) upstream IIF interface")
 508     iif = TOPO["routers"]["r1"]["links"]["r3"]["interface"]
 509     result = verify_upstream_iif(tgen, dut, iif, SOURCE_ADDRESS, group_address_list)
 510     assert result is True, ASSERT_MSG.format(tc_name, result)
 511
 512     step("r1: Verify (S, G) upstream join state and join timer")
 513     result = verify_join_state_and_timer(
 514         tgen, dut, iif, SOURCE_ADDRESS, group_address_list, addr_type="ipv6"
 515     )
 516     assert result is True, ASSERT_MSG.format(tc_name, result)
 517
 518     step("r1: Verify (S, G) ip mroutes")
 519     result = verify_mroutes(tgen, dut, SOURCE_ADDRESS, group_address_list, iif, oif)
 520     assert result is True, ASSERT_MSG.format(tc_name, result)
 521
 522     step("r2: Verify (*, G) upstream IIF interface")
 523     dut = "r2"
 524     iif = "lo"
 525     result = verify_upstream_iif(tgen, dut, iif, STAR, group_address_list)
 526     assert result is True, ASSERT_MSG.format(tc_name, result)
 527
 528     step("r2: Verify (*, G) upstream join state and join timer")
 529     result = verify_join_state_and_timer(
 530         tgen, dut, iif, STAR, group_address_list, addr_type="ipv6"
 531     )
 532     assert result is True, ASSERT_MSG.format(tc_name, result)
 533
 534     step("r2: Verify (*, G) ip mroutes")
 535     oif = TOPO["routers"]["r2"]["links"]["r1"]["interface"]
 536     result = verify_mroutes(tgen, dut, STAR, group_address_list, iif, oif)
 537     assert result is True, ASSERT_MSG.format(tc_name, result)
 538
 539     step("r3: Verify (S, G) upstream IIF interface")
 540     dut = "r3"
 541     iif = TOPO["routers"]["r3"]["links"]["r5"]["interface"]
 542     result = verify_upstream_iif(tgen, dut, iif, SOURCE_ADDRESS, group_address_list)
 543     assert result is True, ASSERT_MSG.format(tc_name, result)
 544
 545     step("r3: Verify (S, G) upstream join state and join timer")
 546     result = verify_join_state_and_timer(
 547         tgen,
 548         dut,
 549         iif,
 550         SOURCE_ADDRESS,
 551         group_address_list,
 552         addr_type="ipv6",
 553         expected=False,
 554     )
 555     assert result is not True, (
 556         "Testcase {} : Failed \n "
 557         "r3: (S,G) upstream state is joined and join timer is running\n Error: {}".format(
 558             tc_name, result
 559         )
 560     )
 561
 562     step("r3: Verify (S, G) ip mroutes")
 563     oif = TOPO["routers"]["r3"]["links"]["r1"]["interface"]
 564     result = verify_mroutes(tgen, dut, SOURCE_ADDRESS, group_address_list, iif, oif)
 565     assert result is True, ASSERT_MSG.format(tc_name, result)
 566
 567     write_test_footer(tc_name)
 568
 569
 570 def test_pim6_multiple_groups_different_RP_address_p2(request):
 571     """
 572     Verify IIF and OIL in updated in mroute when upstream interface
 573     configure as RP
 574
 575     Topology used:
 576                 ________r2_____
 577                 |             |
 578       iperf     |             |     iperf
 579         r0-----r1-------------r3-----r5
 580                 |             |
 581                 |_____________|
 582                         r4
 583     r1 : LHR
 584     r2 & r4 : RP
 585     r3 : FHR
 586     """
 587
 588     tgen = get_topogen()
 589     tc_name = request.node.name
 590     write_test_header(tc_name)
 591
 592     # Don"t run this test if we have any failure.
 593     if tgen.routers_have_failure():
 594         pytest.skip(tgen.errors)
 595
 596     step("Creating configuration from JSON")
 597     kill_socat(tgen)
 598     clear_pim6_mroute(tgen)
 599     clear_pim6_interface_traffic(tgen, TOPO)
 600     reset_config_on_routers(tgen)
 601
 602     step("Enable MLD on r1 interface")
 603     step("Enable the PIM6 on all the interfaces of r1, r2, r3 and r4 routers")
 604     step("r2: Configure r2 as RP")
 605     step("r4: Configure r4 as RP")
 606     input_dict = {
 607         "r2": {
 608             "pim6": {
 609                 "rp": [
 610                     {
 611                         "rp_addr": TOPO["routers"]["r2"]["links"]["lo"]["ipv6"].split(
 612                             "/"
 613                         )[0],
 614                         "group_addr_range": GROUP_RANGE_LIST_1,
 615                     }
 616                 ]
 617             }
 618         },
 619         "r4": {
 620             "pim6": {
 621                 "rp": [
 622                     {
 623                         "rp_addr": TOPO["routers"]["r4"]["links"]["lo"]["ipv6"].split(
 624                             "/"
 625                         )[0],
 626                         "group_addr_range": GROUP_RANGE_LIST_2,
 627                     }
 628                 ]
 629             }
 630         },
 631     }
 632
 633     result = create_pim_config(tgen, TOPO, input_dict)
 634     assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
 635
 636     step("r2: Verify RP info")
 637     dut = "r2"
 638     oif = "lo"
 639     rp_address = TOPO["routers"]["r2"]["links"]["lo"]["ipv6"].split("/")[0]
 640     result = verify_pim_rp_info(
 641         tgen, TOPO, dut, GROUP_RANGE_LIST_1, oif, rp_address, SOURCE
 642     )
 643     assert result is True, ASSERT_MSG.format(tc_name, result)
 644
 645     step("r4: Verify RP info")
 646     dut = "r4"
 647     rp_address = TOPO["routers"]["r4"]["links"]["lo"]["ipv6"].split("/")[0]
 648     result = verify_pim_rp_info(
 649         tgen, TOPO, dut, GROUP_RANGE_LIST_2, oif, rp_address, SOURCE
 650     )
 651     assert result is True, ASSERT_MSG.format(tc_name, result)
 652
 653     group_address_list = GROUP_ADDRESS_LIST_1 + GROUP_ADDRESS_LIST_2
 654     step("r0: Send MLD join for 10 groups")
 655     intf = TOPO["routers"]["r0"]["links"]["r1"]["interface"]
 656     intf_ip = TOPO["routers"]["r0"]["links"]["r1"]["ipv6"].split("/")[0]
 657     result = socat_send_mld_join(
 658         tgen, "r0", "UDP6-RECV", group_address_list, intf, intf_ip
 659     )
 660     assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 661
 662     step("r1: Verify MLD groups")
 663     dut = "r1"
 664     intf_r1_r0 = TOPO["routers"]["r1"]["links"]["r0"]["interface"]
 665     result = verify_mld_groups(tgen, dut, intf_r1_r0, group_address_list)
 666     assert result is True, ASSERT_MSG.format(tc_name, result)
 667
 668     step("r5: Send multicast traffic for group {}".format(group_address_list))
 669     intf = TOPO["routers"]["r5"]["links"]["r3"]["interface"]
 670     SOURCE_ADDRESS = TOPO["routers"]["r5"]["links"]["r3"]["ipv6"].split("/")[0]
 671     result = socat_send_pim6_traffic(tgen, "r5", "UDP6-SEND", group_address_list, intf)
 672     assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 673
 674     step("r1: Verify (*, G) upstream IIF interface")
 675     dut = "r1"
 676     iif = TOPO["routers"]["r1"]["links"]["r2"]["interface"]
 677     result = verify_upstream_iif(tgen, dut, iif, STAR, group_address_list)
 678     assert result is True, ASSERT_MSG.format(tc_name, result)
 679
 680     step("r1: Verify (*, G) upstream join state and join timer")
 681     result = verify_join_state_and_timer(
 682         tgen, dut, iif, STAR, group_address_list, addr_type="ipv6"
 683     )
 684     assert result is True, ASSERT_MSG.format(tc_name, result)
 685
 686     step("r1: Verify (*, G) ip mroutes")
 687     oif = TOPO["routers"]["r1"]["links"]["r0"]["interface"]
 688     result = verify_mroutes(tgen, dut, STAR, GROUP_ADDRESS_LIST_1, iif, oif)
 689     assert result is True, ASSERT_MSG.format(tc_name, result)
 690
 691     step("r1: Verify (S, G) upstream IIF interface")
 692     iif = TOPO["routers"]["r1"]["links"]["r3"]["interface"]
 693     result = verify_upstream_iif(tgen, dut, iif, SOURCE_ADDRESS, GROUP_ADDRESS_LIST_1)
 694     assert result is True, ASSERT_MSG.format(tc_name, result)
 695
 696     step("r1: Verify (S, G) upstream join state and join timer")
 697     result = verify_join_state_and_timer(
 698         tgen, dut, iif, SOURCE_ADDRESS, GROUP_ADDRESS_LIST_1, addr_type="ipv6"
 699     )
 700     assert result is True, ASSERT_MSG.format(tc_name, result)
 701
 702     step("r1: Verify (S, G) ip mroutes")
 703     result = verify_mroutes(tgen, dut, SOURCE_ADDRESS, GROUP_ADDRESS_LIST_1, iif, oif)
 704     assert result is True, ASSERT_MSG.format(tc_name, result)
 705
 706     step("r2: Verify (*, G) upstream IIF interface")
 707     dut = "r2"
 708     iif = "lo"
 709     result = verify_upstream_iif(tgen, dut, iif, STAR, GROUP_ADDRESS_LIST_1)
 710     assert result is True, ASSERT_MSG.format(tc_name, result)
 711
 712     step("r2: Verify (*, G) upstream join state and join timer")
 713     result = verify_join_state_and_timer(
 714         tgen, dut, iif, STAR, GROUP_ADDRESS_LIST_1, addr_type="ipv6"
 715     )
 716     assert result is True, ASSERT_MSG.format(tc_name, result)
 717
 718     step("r2: Verify (*, G) ip mroutes")
 719     oif = TOPO["routers"]["r2"]["links"]["r1"]["interface"]
 720     result = verify_mroutes(tgen, dut, STAR, GROUP_ADDRESS_LIST_1, iif, oif)
 721     assert result is True, ASSERT_MSG.format(tc_name, result)
 722
 723     step("r2: Verify (S, G) upstream IIF interface")
 724     iif = TOPO["routers"]["r2"]["links"]["r3"]["interface"]
 725     result = verify_upstream_iif(
 726         tgen, dut, iif, SOURCE_ADDRESS, GROUP_ADDRESS_LIST_1, joinState="NotJoined"
 727     )
 728     assert result is True, ASSERT_MSG.format(tc_name, result)
 729
 730     step("r2: Verify (S, G) upstream join state and join timer")
 731     result = verify_join_state_and_timer(
 732         tgen,
 733         dut,
 734         iif,
 735         SOURCE_ADDRESS,
 736         GROUP_ADDRESS_LIST_1,
 737         addr_type="ipv6",
 738         expected=False,
 739     )
 740     assert result is not True, (
 741         "Testcase {} : Failed \n "
 742         "r2: (S,G) upstream state is joined and join timer is running\n Error: {}".format(
 743             tc_name, result
 744         )
 745     )
 746
 747     step("r2: Verify (S, G) ip mroutes")
 748     oif = "none"
 749     result = verify_mroutes(tgen, dut, SOURCE_ADDRESS, GROUP_ADDRESS_LIST_1, iif, oif)
 750     assert result is True, ASSERT_MSG.format(tc_name, result)
 751
 752     step("r3: Verify (S, G) upstream IIF interface")
 753     dut = "r3"
 754     iif = TOPO["routers"]["r3"]["links"]["r5"]["interface"]
 755     result = verify_upstream_iif(tgen, dut, iif, SOURCE_ADDRESS, GROUP_ADDRESS_LIST_1)
 756     assert result is True, ASSERT_MSG.format(tc_name, result)
 757
 758     step("r3: Verify (S, G) upstream join state and join timer")
 759     result = verify_join_state_and_timer(
 760         tgen,
 761         dut,
 762         iif,
 763         SOURCE_ADDRESS,
 764         GROUP_ADDRESS_LIST_1,
 765         addr_type="ipv6",
 766         expected=False,
 767     )
 768     assert result is not True, (
 769         "Testcase {} : Failed \n "
 770         "r3: (S,G) upstream state is joined and join timer is running\n Error: {}".format(
 771             tc_name, result
 772         )
 773     )
 774
 775     step("r3: Verify (S, G) ip mroutes")
 776     oif = TOPO["routers"]["r3"]["links"]["r1"]["interface"]
 777     result = verify_mroutes(tgen, dut, SOURCE_ADDRESS, GROUP_ADDRESS_LIST_1, iif, oif)
 778     assert result is True, ASSERT_MSG.format(tc_name, result)
 779
 780     step("r1: Verify (*, G) upstream IIF interface")
 781     dut = "r1"
 782     iif = TOPO["routers"]["r1"]["links"]["r4"]["interface"]
 783     result = verify_upstream_iif(tgen, dut, iif, STAR, GROUP_ADDRESS_LIST_2)
 784     assert result is True, ASSERT_MSG.format(tc_name, result)
 785
 786     step("r1: Verify (*, G) upstream join state and join timer")
 787     result = verify_join_state_and_timer(
 788         tgen, dut, iif, STAR, GROUP_ADDRESS_LIST_2, addr_type="ipv6"
 789     )
 790     assert result is True, ASSERT_MSG.format(tc_name, result)
 791
 792     step("r1: Verify (*, G) ip mroutes")
 793     oif = TOPO["routers"]["r1"]["links"]["r0"]["interface"]
 794     result = verify_mroutes(tgen, dut, STAR, GROUP_ADDRESS_LIST_2, iif, oif)
 795     assert result is True, ASSERT_MSG.format(tc_name, result)
 796
 797     step("r1: Verify (S, G) upstream IIF interface")
 798     iif = TOPO["routers"]["r1"]["links"]["r3"]["interface"]
 799     result = verify_upstream_iif(tgen, dut, iif, SOURCE_ADDRESS, GROUP_ADDRESS_LIST_2)
 800     assert result is True, ASSERT_MSG.format(tc_name, result)
 801
 802     step("r1: Verify (S, G) upstream join state and join timer")
 803     result = verify_join_state_and_timer(
 804         tgen, dut, iif, SOURCE_ADDRESS, GROUP_ADDRESS_LIST_2, addr_type="ipv6"
 805     )
 806     assert result is True, ASSERT_MSG.format(tc_name, result)
 807
 808     step("r1: Verify (S, G) ip mroutes")
 809     result = verify_mroutes(tgen, dut, SOURCE_ADDRESS, GROUP_ADDRESS_LIST_2, iif, oif)
 810     assert result is True, ASSERT_MSG.format(tc_name, result)
 811
 812     step("r4: Verify (*, G) upstream IIF interface")
 813     dut = "r4"
 814     iif = "lo"
 815     result = verify_upstream_iif(tgen, dut, iif, STAR, GROUP_ADDRESS_LIST_2)
 816     assert result is True, ASSERT_MSG.format(tc_name, result)
 817
 818     step("r4: Verify (*, G) upstream join state and join timer")
 819     result = verify_join_state_and_timer(
 820         tgen, dut, iif, STAR, GROUP_ADDRESS_LIST_2, addr_type="ipv6"
 821     )
 822     assert result is True, ASSERT_MSG.format(tc_name, result)
 823
 824     step("r4: Verify (*, G) ip mroutes")
 825     oif = TOPO["routers"]["r4"]["links"]["r1"]["interface"]
 826     result = verify_mroutes(tgen, dut, STAR, GROUP_ADDRESS_LIST_2, iif, oif)
 827     assert result is True, ASSERT_MSG.format(tc_name, result)
 828
 829     step("r4: Verify (S, G) upstream IIF interface")
 830     iif = TOPO["routers"]["r4"]["links"]["r3"]["interface"]
 831     result = verify_upstream_iif(
 832         tgen, dut, iif, SOURCE_ADDRESS, GROUP_ADDRESS_LIST_2, joinState="NotJoined"
 833     )
 834     assert result is True, ASSERT_MSG.format(tc_name, result)
 835
 836     step("r4: Verify (S, G) upstream join state and join timer")
 837     result = verify_join_state_and_timer(
 838         tgen,
 839         dut,
 840         iif,
 841         SOURCE_ADDRESS,
 842         GROUP_ADDRESS_LIST_2,
 843         addr_type="ipv6",
 844         expected=False,
 845     )
 846     assert result is not True, (
 847         "Testcase {} : Failed \n "
 848         "r4: (S,G) upstream state is joined and join timer is running\n Error: {}".format(
 849             tc_name, result
 850         )
 851     )
 852
 853     step("r4: Verify (S, G) ip mroutes")
 854     oif = "none"
 855     result = verify_mroutes(tgen, dut, SOURCE_ADDRESS, GROUP_ADDRESS_LIST_2, iif, oif)
 856     assert result is True, ASSERT_MSG.format(tc_name, result)
 857
 858     step("r3: Verify (S, G) upstream IIF interface")
 859     dut = "r3"
 860     iif = TOPO["routers"]["r3"]["links"]["r5"]["interface"]
 861     result = verify_upstream_iif(tgen, dut, iif, SOURCE_ADDRESS, GROUP_ADDRESS_LIST_2)
 862     assert result is True, ASSERT_MSG.format(tc_name, result)
 863
 864     step("r3: Verify (S, G) upstream join state and join timer")
 865     result = verify_join_state_and_timer(
 866         tgen,
 867         dut,
 868         iif,
 869         SOURCE_ADDRESS,
 870         GROUP_ADDRESS_LIST_2,
 871         addr_type="ipv6",
 872         expected=False,
 873     )
 874     assert result is not True, "Testcase {} :Failed \n Error: {}".format(
 875         tc_name, result
 876     )
 877
 878     step("r3: Verify (S, G) ip mroutes")
 879     oif = TOPO["routers"]["r3"]["links"]["r1"]["interface"]
 880     result = verify_mroutes(tgen, dut, SOURCE_ADDRESS, GROUP_ADDRESS_LIST_2, iif, oif)
 881     assert result is True, ASSERT_MSG.format(tc_name, result)
 882
 883     step("Delete RP configuration")
 884     input_dict = {
 885         "r2": {
 886             "pim6": {
 887                 "rp": [
 888                     {
 889                         "rp_addr": TOPO["routers"]["r2"]["links"]["lo"]["ipv6"].split(
 890                             "/"
 891                         )[0],
 892                         "group_addr_range": GROUP_RANGE_LIST_1,
 893                         "delete": True,
 894                     }
 895                 ]
 896             }
 897         },
 898         "r4": {
 899             "pim6": {
 900                 "rp": [
 901                     {
 902                         "rp_addr": TOPO["routers"]["r4"]["links"]["lo"]["ipv6"].split(
 903                             "/"
 904                         )[0],
 905                         "group_addr_range": GROUP_RANGE_LIST_2,
 906                         "delete": True,
 907                     }
 908                 ]
 909             }
 910         },
 911     }
 912
 913     result = create_pim_config(tgen, TOPO, input_dict)
 914     assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
 915
 916     step("r1, r2, r3, r4: Re-configure RP")
 917     input_dict = {
 918         "r2": {
 919             "pim6": {
 920                 "rp": [
 921                     {
 922                         "rp_addr": TOPO["routers"]["r2"]["links"]["lo"]["ipv6"].split(
 923                             "/"
 924                         )[0],
 925                         "group_addr_range": GROUP_RANGE_LIST_1,
 926                     }
 927                 ]
 928             }
 929         },
 930         "r4": {
 931             "pim6": {
 932                 "rp": [
 933                     {
 934                         "rp_addr": TOPO["routers"]["r4"]["links"]["lo"]["ipv6"].split(
 935                             "/"
 936                         )[0],
 937                         "group_addr_range": GROUP_RANGE_LIST_2,
 938                     }
 939                 ]
 940             }
 941         },
 942     }
 943
 944     result = create_pim_config(tgen, TOPO, input_dict)
 945     assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
 946
 947     step("r1: Shut/No Shut the interfacesfrom R1 to R2, R4 and R0")
 948     dut = "r1"
 949     intf1 = TOPO["routers"]["r1"]["links"]["r2"]["interface"]
 950     intf2 = TOPO["routers"]["r1"]["links"]["r4"]["interface"]
 951     intf3 = TOPO["routers"]["r1"]["links"]["r0"]["interface"]
 952     for intf in [intf1, intf2, intf3]:
 953         shutdown_bringup_interface(tgen, dut, intf, False)
 954         shutdown_bringup_interface(tgen, dut, intf, True)
 955
 956     step("r1: Verify (*, G) upstream IIF interface")
 957     dut = "r1"
 958     iif = TOPO["routers"]["r1"]["links"]["r2"]["interface"]
 959     result = verify_upstream_iif(tgen, dut, iif, STAR, GROUP_ADDRESS_LIST_1)
 960     assert result is True, ASSERT_MSG.format(tc_name, result)
 961
 962     step("r1: Verify (*, G) upstream join state and join timer")
 963     result = verify_join_state_and_timer(
 964         tgen, dut, iif, STAR, GROUP_ADDRESS_LIST_1, addr_type="ipv6"
 965     )
 966     assert result is True, ASSERT_MSG.format(tc_name, result)
 967
 968     step("r1: Verify (*, G) ip mroutes")
 969     oif = TOPO["routers"]["r1"]["links"]["r0"]["interface"]
 970     result = verify_mroutes(tgen, dut, STAR, GROUP_ADDRESS_LIST_1, iif, oif)
 971     assert result is True, ASSERT_MSG.format(tc_name, result)
 972
 973     step("r1: Verify (S, G) upstream IIF interface")
 974     iif = TOPO["routers"]["r1"]["links"]["r3"]["interface"]
 975     result = verify_upstream_iif(tgen, dut, iif, SOURCE_ADDRESS, GROUP_ADDRESS_LIST_1)
 976     assert result is True, ASSERT_MSG.format(tc_name, result)
 977
 978     step("r1: Verify (S, G) upstream join state and join timer")
 979     result = verify_join_state_and_timer(
 980         tgen, dut, iif, SOURCE_ADDRESS, GROUP_ADDRESS_LIST_1, addr_type="ipv6"
 981     )
 982     assert result is True, ASSERT_MSG.format(tc_name, result)
 983
 984     step("r1: Verify (S, G) ip mroutes")
 985     result = verify_mroutes(tgen, dut, SOURCE_ADDRESS, GROUP_ADDRESS_LIST_1, iif, oif)
 986     assert result is True, ASSERT_MSG.format(tc_name, result)
 987
 988     step("r2: Verify (*, G) upstream IIF interface")
 989     dut = "r2"
 990     iif = "lo"
 991     result = verify_upstream_iif(tgen, dut, iif, STAR, GROUP_ADDRESS_LIST_1)
 992     assert result is True, ASSERT_MSG.format(tc_name, result)
 993
 994     step("r2: Verify (*, G) upstream join state and join timer")
 995     result = verify_join_state_and_timer(
 996         tgen, dut, iif, STAR, GROUP_ADDRESS_LIST_1, addr_type="ipv6"
 997     )
 998     assert result is True, ASSERT_MSG.format(tc_name, result)
 999
1000     step("r2: Verify (*, G) ip mroutes")
1001     oif = TOPO["routers"]["r2"]["links"]["r1"]["interface"]
1002     result = verify_mroutes(tgen, dut, STAR, GROUP_ADDRESS_LIST_1, iif, oif)
1003     assert result is True, ASSERT_MSG.format(tc_name, result)
1004
1005     step("r2: Verify (S, G) upstream IIF interface")
1006     iif = TOPO["routers"]["r2"]["links"]["r3"]["interface"]
1007     result = verify_upstream_iif(
1008         tgen, dut, iif, SOURCE_ADDRESS, GROUP_ADDRESS_LIST_1, joinState="NotJoined"
1009     )
1010     assert result is True, ASSERT_MSG.format(tc_name, result)
1011
1012     step("r2: Verify (S, G) upstream join state and join timer")
1013     result = verify_join_state_and_timer(
1014         tgen,
1015         dut,
1016         iif,
1017         SOURCE_ADDRESS,
1018         GROUP_ADDRESS_LIST_1,
1019         addr_type="ipv6",
1020         expected=False,
1021     )
1022     assert result is not True, (
1023         "Testcase {} : Failed \n "
1024         "r2: (S,G) upstream state is joined and join timer is running\n Error: {}".format(
1025             tc_name, result
1026         )
1027     )
1028
1029     step("r2: Verify (S, G) ip mroutes")
1030     oif = "none"
1031     result = verify_mroutes(tgen, dut, SOURCE_ADDRESS, GROUP_ADDRESS_LIST_1, iif, oif)
1032     assert result is True, ASSERT_MSG.format(tc_name, result)
1033
1034     step("r3: Verify (S, G) upstream IIF interface")
1035     dut = "r3"
1036     iif = TOPO["routers"]["r3"]["links"]["r5"]["interface"]
1037     result = verify_upstream_iif(tgen, dut, iif, SOURCE_ADDRESS, GROUP_ADDRESS_LIST_1)
1038     assert result is True, ASSERT_MSG.format(tc_name, result)
1039
1040     step("r3: Verify (S, G) upstream join state and join timer")
1041     result = verify_join_state_and_timer(
1042         tgen,
1043         dut,
1044         iif,
1045         SOURCE_ADDRESS,
1046         GROUP_ADDRESS_LIST_1,
1047         addr_type="ipv6",
1048         expected=False,
1049     )
1050     assert result is not True, (
1051         "Testcase {} : Failed \n "
1052         "r3: (S,G) upstream state is joined and join timer is running\n Error: {}".format(
1053             tc_name, result
1054         )
1055     )
1056
1057     step("r3: Verify (S, G) ip mroutes")
1058     oif = TOPO["routers"]["r3"]["links"]["r1"]["interface"]
1059     result = verify_mroutes(tgen, dut, SOURCE_ADDRESS, GROUP_ADDRESS_LIST_1, iif, oif)
1060     assert result is True, ASSERT_MSG.format(tc_name, result)
1061
1062     step("r1: Verify (*, G) upstream IIF interface")
1063     dut = "r1"
1064     iif = TOPO["routers"]["r1"]["links"]["r4"]["interface"]
1065     result = verify_upstream_iif(tgen, dut, iif, STAR, GROUP_ADDRESS_LIST_2)
1066     assert result is True, ASSERT_MSG.format(tc_name, result)
1067
1068     step("r1: Verify (*, G) upstream join state and join timer")
1069     result = verify_join_state_and_timer(
1070         tgen, dut, iif, STAR, GROUP_ADDRESS_LIST_2, addr_type="ipv6"
1071     )
1072     assert result is True, ASSERT_MSG.format(tc_name, result)
1073
1074     step("r1: Verify (*, G) ip mroutes")
1075     oif = TOPO["routers"]["r1"]["links"]["r0"]["interface"]
1076     result = verify_mroutes(tgen, dut, STAR, GROUP_ADDRESS_LIST_2, iif, oif)
1077     assert result is True, ASSERT_MSG.format(tc_name, result)
1078
1079     step("r1: Verify (S, G) upstream IIF interface")
1080     iif = TOPO["routers"]["r1"]["links"]["r3"]["interface"]
1081     result = verify_upstream_iif(tgen, dut, iif, SOURCE_ADDRESS, GROUP_ADDRESS_LIST_2)
1082     assert result is True, ASSERT_MSG.format(tc_name, result)
1083
1084     step("r1: Verify (S, G) upstream join state and join timer")
1085     result = verify_join_state_and_timer(
1086         tgen, dut, iif, SOURCE_ADDRESS, GROUP_ADDRESS_LIST_2, addr_type="ipv6"
1087     )
1088     assert result is True, ASSERT_MSG.format(tc_name, result)
1089
1090     step("r1: Verify (S, G) ip mroutes")
1091     result = verify_mroutes(tgen, dut, SOURCE_ADDRESS, GROUP_ADDRESS_LIST_2, iif, oif)
1092     assert result is True, ASSERT_MSG.format(tc_name, result)
1093
1094     step("r4: Verify (*, G) upstream IIF interface")
1095     dut = "r4"
1096     iif = "lo"
1097     result = verify_upstream_iif(tgen, dut, iif, STAR, GROUP_ADDRESS_LIST_2)
1098     assert result is True, ASSERT_MSG.format(tc_name, result)
1099
1100     step("r4: Verify (*, G) upstream join state and join timer")
1101     result = verify_join_state_and_timer(
1102         tgen, dut, iif, STAR, GROUP_ADDRESS_LIST_2, addr_type="ipv6"
1103     )
1104     assert result is True, ASSERT_MSG.format(tc_name, result)
1105
1106     step("r4: Verify (*, G) ip mroutes")
1107     oif = TOPO["routers"]["r4"]["links"]["r1"]["interface"]
1108     result = verify_mroutes(tgen, dut, STAR, GROUP_ADDRESS_LIST_2, iif, oif)
1109     assert result is True, ASSERT_MSG.format(tc_name, result)
1110
1111     step("r4: Verify (S, G) upstream IIF interface")
1112     iif = TOPO["routers"]["r4"]["links"]["r3"]["interface"]
1113     result = verify_upstream_iif(
1114         tgen, dut, iif, SOURCE_ADDRESS, GROUP_ADDRESS_LIST_2, joinState="NotJoined"
1115     )
1116     assert result is True, ASSERT_MSG.format(tc_name, result)
1117
1118     step("r4: Verify (S, G) upstream join state and join timer")
1119     result = verify_join_state_and_timer(
1120         tgen,
1121         dut,
1122         iif,
1123         SOURCE_ADDRESS,
1124         GROUP_ADDRESS_LIST_2,
1125         addr_type="ipv6",
1126         expected=False,
1127     )
1128     assert result is not True, (
1129         "Testcase {} : Failed \n "
1130         "r4: (S,G) upstream state is joined and join timer is running\n Error: {}".format(
1131             tc_name, result
1132         )
1133     )
1134
1135     step("r4: Verify (S, G) ip mroutes")
1136     oif = "none"
1137     result = verify_mroutes(tgen, dut, SOURCE_ADDRESS, GROUP_ADDRESS_LIST_2, iif, oif)
1138     assert result is True, ASSERT_MSG.format(tc_name, result)
1139
1140     step("r3: Verify (S, G) upstream IIF interface")
1141     dut = "r3"
1142     iif = TOPO["routers"]["r3"]["links"]["r5"]["interface"]
1143     result = verify_upstream_iif(tgen, dut, iif, SOURCE_ADDRESS, GROUP_ADDRESS_LIST_2)
1144     assert result is True, ASSERT_MSG.format(tc_name, result)
1145
1146     step("r3: Verify (S, G) upstream join state and join timer")
1147     result = verify_join_state_and_timer(
1148         tgen,
1149         dut,
1150         iif,
1151         SOURCE_ADDRESS,
1152         GROUP_ADDRESS_LIST_2,
1153         addr_type="ipv6",
1154         expected=False,
1155     )
1156     assert result is not True, (
1157         "Testcase {} : Failed \n "
1158         "r3: (S,G) upstream state is joined and join timer is running\n Error: {}".format(
1159             tc_name, result
1160         )
1161     )
1162
1163     step("r3: Verify (S, G) ip mroutes")
1164     oif = TOPO["routers"]["r3"]["links"]["r1"]["interface"]
1165     result = verify_mroutes(tgen, dut, SOURCE_ADDRESS, GROUP_ADDRESS_LIST_2, iif, oif)
1166     assert result is True, ASSERT_MSG.format(tc_name, result)
1167
1168     write_test_footer(tc_name)
1169
1170
1171 def test_pim6_delete_RP_shut_noshut_upstream_interface_p1(request):
1172     """
1173     Verify RP info and (*,G) mroute after deleting the RP and shut /
1174     no shut the RPF interface.
1175
1176     Topology used:
1177                 ________r2_____
1178                 |             |
1179       iperf     |             |
1180         r0-----r1-------------r3
1181     """
1182
1183     tgen = get_topogen()
1184     tc_name = request.node.name
1185     write_test_header(tc_name)
1186
1187     # Don"t run this test if we have any failure.
1188     if tgen.routers_have_failure():
1189         pytest.skip(tgen.errors)
1190
1191     step("Creating configuration from JSON")
1192     kill_socat(tgen)
1193     clear_pim6_mroute(tgen)
1194     clear_pim6_interface_traffic(tgen, TOPO)
1195     reset_config_on_routers(tgen)
1196
1197     step("Enable MLD on r1 interface")
1198     step("Enable the PIM6 on all the interfaces of r1, r2, r3 and r4 routers")
1199     step("r2: Configure r2 as RP")
1200     input_dict = {
1201         "r2": {
1202             "pim6": {
1203                 "rp": [
1204                     {
1205                         "rp_addr": TOPO["routers"]["r2"]["links"]["lo"]["ipv6"].split(
1206                             "/"
1207                         )[0],
1208                         "group_addr_range": GROUP_RANGE_1,
1209                     }
1210                 ]
1211             }
1212         }
1213     }
1214
1215     result = create_pim_config(tgen, TOPO, input_dict)
1216     assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
1217
1218     step("r2: verify rp-info")
1219     dut = "r2"
1220     oif = "lo"
1221     rp_address = TOPO["routers"]["r2"]["links"]["lo"]["ipv6"].split("/")[0]
1222     result = verify_pim_rp_info(
1223         tgen, TOPO, dut, GROUP_ADDRESS_1, oif, rp_address, SOURCE
1224     )
1225     assert result is True, ASSERT_MSG.format(tc_name, result)
1226
1227     step("r0: Send MLD join")
1228     intf = TOPO["routers"]["r0"]["links"]["r1"]["interface"]
1229     intf_ip = TOPO["routers"]["r0"]["links"]["r1"]["ipv6"].split("/")[0]
1230     result = socat_send_mld_join(
1231         tgen, "r0", "UDP6-RECV", GROUP_ADDRESS_1, intf, intf_ip
1232     )
1233     assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
1234
1235     step("r1: Verify MLD groups")
1236     dut = "r1"
1237     intf_r1_r0 = TOPO["routers"]["r1"]["links"]["r0"]["interface"]
1238     result = verify_mld_groups(tgen, dut, intf_r1_r0, GROUP_ADDRESS_1)
1239     assert result is True, ASSERT_MSG.format(tc_name, result)
1240
1241     step("r1: Verify (*, G) ip mroutes created")
1242     iif = TOPO["routers"]["r1"]["links"]["r2"]["interface"]
1243     oif = TOPO["routers"]["r1"]["links"]["r0"]["interface"]
1244     result = verify_mroutes(tgen, dut, STAR, GROUP_ADDRESS_1, iif, oif)
1245     assert result is True, ASSERT_MSG.format(tc_name, result)
1246
1247     step("r2: Verify (*, G) ip mroutes created")
1248     dut = "r2"
1249     iif = "lo"
1250     oif = TOPO["routers"]["r2"]["links"]["r1"]["interface"]
1251     result = verify_mroutes(tgen, dut, STAR, GROUP_ADDRESS_1, iif, oif)
1252     assert result is True, ASSERT_MSG.format(tc_name, result)
1253
1254     step("r1: Delete RP configuration")
1255     input_dict = {
1256         "r1": {
1257             "pim6": {
1258                 "rp": [
1259                     {
1260                         "rp_addr": TOPO["routers"]["r2"]["links"]["lo"]["ipv6"].split(
1261                             "/"
1262                         )[0],
1263                         "group_addr_range": GROUP_RANGE_1,
1264                         "delete": True,
1265                     }
1266                 ]
1267             }
1268         }
1269     }
1270
1271     result = create_pim_config(tgen, TOPO, input_dict)
1272     assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
1273
1274     step("r1: Shut/No Shut the interface r1-r2-eth1/r1-r0-eth0 from R1 to R2")
1275     dut = "r1"
1276     intf1 = TOPO["routers"]["r1"]["links"]["r2"]["interface"]
1277     intf2 = TOPO["routers"]["r1"]["links"]["r0"]["interface"]
1278     for intf in [intf1, intf2]:
1279         shutdown_bringup_interface(tgen, dut, intf, False)
1280         shutdown_bringup_interface(tgen, dut, intf, True)
1281
1282     step("r2: Shut the RP interface lo")
1283     dut = "r2"
1284     intf = "lo"
1285     shutdown_bringup_interface(tgen, dut, intf, False)
1286
1287     step("r1: Shut the interface r1-r2-eth1/r1-r3-eth2 towards RP")
1288     intf3 = TOPO["routers"]["r1"]["links"]["r3"]["interface"]
1289     for intf in [intf1, intf3]:
1290         shutdown_bringup_interface(tgen, dut, intf, False)
1291
1292     step("r1: Verify (*, G) ip mroutes cleared")
1293     dut = "r1"
1294     iif = TOPO["routers"]["r1"]["links"]["r2"]["interface"]
1295     oif = TOPO["routers"]["r1"]["links"]["r0"]["interface"]
1296     result = verify_mroutes(tgen, dut, STAR, GROUP_ADDRESS_1, iif, oif, expected=False)
1297     assert result is not True, (
1298         "Testcase {} : Failed \n "
1299         "r1: (*,G) mroutes are not cleared after shut of R1 to R0 link\n Error: {}".format(
1300             tc_name, result
1301         )
1302     )
1303
1304     step("r2: Verify (*, G) ip mroutes cleared")
1305     dut = "r2"
1306     iif = "lo"
1307     oif = TOPO["routers"]["r2"]["links"]["r1"]["interface"]
1308     result = verify_mroutes(tgen, dut, STAR, GROUP_ADDRESS_1, iif, oif, expected=False)
1309     assert result is not True, (
1310         "Testcase {} : Failed \n "
1311         "r2: (*,G) mroutes are not cleared after shut of R1 to R0 link\n Error: {}".format(
1312             tc_name, result
1313         )
1314     )
1315
1316     write_test_footer(tc_name)
1317
1318
1319 if __name__ == "__main__":
1320     args = ["-s"] + sys.argv[1:]
1321     sys.exit(pytest.main(args))