tests/topotests/multicast_pim6_static_rp_topo1/test_multicast_pim6_static_rp1.py

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