tests/topotests/multicast_pim_sm_topo3/test_multicast_pim_sm_topo4.py

   1 #!/usr/bin/env python
   2 # SPDX-License-Identifier: ISC
   3
   4 #
   5 # Copyright (c) 2020 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 pim sm:
  12
  13 Test steps
  14 - Create topology (setup module)
  15 - Bring up topology
  16
  17 Following tests are covered:
  18
  19 1. TC:48 Verify mroute after configuring black-hole route for RP and source
  20 2. TC:49 Verify mroute when RP is reachable using default route
  21 3. TC:50 Verify mroute when LHR,FHR,RP and transit routers reachable
  22     using default routes
  23 4. TC:52 Verify PIM nbr after changing interface ip
  24 5. TC:53 Verify IGMP interface updated with correct detail after changing interface config
  25 6. TC:54 Verify received and transmit hello stats are getting cleared after PIM nbr reset
  26
  27
  28 """
  29
  30 import os
  31 import sys
  32 import time
  33 import pytest
  34
  35 pytestmark = pytest.mark.pimd
  36
  37 # Save the Current Working Directory to find configuration files.
  38 CWD = os.path.dirname(os.path.realpath(__file__))
  39 sys.path.append(os.path.join(CWD, "../"))
  40 sys.path.append(os.path.join(CWD, "../lib/"))
  41
  42 # Required to instantiate the topology builder class.
  43
  44 # pylint: disable=C0413
  45 # Import topogen and topotest helpers
  46 from lib.topogen import Topogen, get_topogen
  47
  48 from lib.common_config import (
  49     start_topology,
  50     write_test_header,
  51     write_test_footer,
  52     step,
  53     reset_config_on_routers,
  54     shutdown_bringup_interface,
  55     apply_raw_config,
  56     create_static_routes,
  57     required_linux_kernel_version,
  58     topo_daemons,
  59 )
  60 from lib.pim import (
  61     create_pim_config,
  62     create_igmp_config,
  63     verify_mroutes,
  64     clear_pim_interface_traffic,
  65     verify_upstream_iif,
  66     clear_mroute,
  67     verify_pim_rp_info,
  68     get_pim_interface_traffic,
  69     McastTesterHelper,
  70 )
  71 from lib.topolog import logger
  72 from lib.topojson import build_config_from_json
  73 from time import sleep
  74
  75
  76 TOPOLOGY = """
  77
  78
  79     i4-----c1-------------c2---i5
  80             |              |
  81             |              |
  82     i1-----l1------r2-----f1---i2
  83        |    |      |       |
  84        |    |      |       |
  85       i7    i6     i3     i8
  86
  87     Description:
  88     i1, i2, i3. i4, i5, i6, i7, i8 - FRR running iperf to send IGMP
  89                                      join and traffic
  90     l1 - LHR
  91     f1 - FHR
  92     r2 - FRR router
  93     c1 - FRR router
  94     c2 - FRR router
  95 """
  96
  97 # Global variables
  98
  99 GROUP_RANGE = "224.0.0.0/4"
 100 IGMP_GROUP = "225.1.1.1/32"
 101 IGMP_JOIN = "225.1.1.1"
 102 GROUP_RANGE_1 = [
 103     "225.1.1.1/32",
 104     "225.1.1.2/32",
 105     "225.1.1.3/32",
 106     "225.1.1.4/32",
 107     "225.1.1.5/32",
 108 ]
 109 IGMP_JOIN_RANGE_1 = ["225.1.1.1", "225.1.1.2", "225.1.1.3", "225.1.1.4", "225.1.1.5"]
 110 NEW_ADDRESS_1 = "192.168.20.1"
 111 NEW_ADDRESS_2 = "192.168.20.2"
 112 NEW_ADDRESS_1_SUBNET = "192.168.20.1/24"
 113 NEW_ADDRESS_2_SUBNET = "192.168.20.2/24"
 114
 115
 116 def setup_module(mod):
 117     """
 118     Sets up the pytest environment
 119
 120     * `mod`: module name
 121     """
 122
 123     # Required linux kernel version for this suite to run.
 124     result = required_linux_kernel_version("4.19")
 125     if result is not True:
 126         pytest.skip("Kernel version should be >= 4.19")
 127
 128     testsuite_run_time = time.asctime(time.localtime(time.time()))
 129     logger.info("Testsuite start time: {}".format(testsuite_run_time))
 130     logger.info("=" * 40)
 131     logger.info("Master Topology: \n {}".format(TOPOLOGY))
 132
 133     logger.info("Running setup_module to create topology")
 134
 135     json_file = "{}/multicast_pim_sm_topo4.json".format(CWD)
 136     tgen = Topogen(json_file, mod.__name__)
 137     global topo
 138     topo = tgen.json_topo
 139     # ... and here it calls Mininet initialization functions.
 140
 141     # Starting topology, create tmp files which are loaded to routers
 142     #  to start daemons and then start routers
 143     start_topology(tgen)
 144
 145     # Don"t run this test if we have any failure.
 146     if tgen.routers_have_failure():
 147         pytest.skip(tgen.errors)
 148
 149     # Creating configuration from JSON
 150     build_config_from_json(tgen, topo)
 151
 152     # XXX Replace this using "with McastTesterHelper()... " in each test if possible.
 153     global app_helper
 154     app_helper = McastTesterHelper(tgen)
 155
 156     logger.info("Running setup_module() done")
 157
 158
 159 def teardown_module():
 160     """Teardown the pytest environment"""
 161
 162     logger.info("Running teardown_module to delete topology")
 163
 164     tgen = get_topogen()
 165
 166     app_helper.cleanup()
 167
 168     # Stop toplogy and Remove tmp files
 169     tgen.stop_topology()
 170
 171     logger.info(
 172         "Testsuite end time: {}".format(time.asctime(time.localtime(time.time())))
 173     )
 174     logger.info("=" * 40)
 175
 176
 177 #####################################################
 178 #
 179 #   Testcases
 180 #
 181 #####################################################
 182
 183
 184 def reset_stats(stats):
 185     """
 186     API to reset the stats
 187
 188     Parameters
 189     ----------
 190     * `stats` : State dictionary holding helloRx and helloTx values
 191     """
 192
 193     for router, state_data in stats.items():
 194         for state, value in state_data.items():
 195             stats[router][state] = 0
 196             logger.info(
 197                 "[DUT: %s]: stats %s value has reset" " reset, Current value: %s",
 198                 router,
 199                 state,
 200                 stats[router][state],
 201             )
 202
 203     return True
 204
 205
 206 def verify_state_incremented(state_before, state_after):
 207     """
 208     API to compare interface traffic state incrementing
 209
 210     Parameters
 211     ----------
 212     * `state_before` : State dictionary for any particular instance
 213     * `state_after` : State dictionary for any particular instance
 214     """
 215
 216     for router, state_data in state_before.items():
 217         for state, value in state_data.items():
 218             if state_before[router][state] >= state_after[router][state]:
 219                 errormsg = (
 220                     "[DUT: %s]: state %s value has not"
 221                     " incremented, Initial value: %s, "
 222                     "Current value: %s [FAILED!!]"
 223                     % (
 224                         router,
 225                         state,
 226                         state_before[router][state],
 227                         state_after[router][state],
 228                     )
 229                 )
 230                 return errormsg
 231
 232             logger.info(
 233                 "[DUT: %s]: State %s value is "
 234                 "incremented, Initial value: %s, Current value: %s"
 235                 " [PASSED!!]",
 236                 router,
 237                 state,
 238                 state_before[router][state],
 239                 state_after[router][state],
 240             )
 241
 242     return True
 243
 244
 245 def test_mroute_when_RP_reachable_default_route_p2(request):
 246     """
 247     TC_49 Verify mroute when and source RP is reachable using default route
 248     """
 249
 250     tgen = get_topogen()
 251     tc_name = request.node.name
 252     write_test_header(tc_name)
 253
 254     # Don"t run this test if we have any failure.
 255     if tgen.routers_have_failure():
 256         pytest.skip(tgen.errors)
 257
 258     # Creating configuration from JSON
 259     app_helper.stop_all_hosts()
 260     clear_mroute(tgen)
 261     reset_config_on_routers(tgen)
 262     clear_pim_interface_traffic(tgen, topo)
 263
 264     step(
 265         "Remove c1-c2 connected link to simulate topo "
 266         "c1(FHR)---l1(RP)----r2---f1-----c2(LHR)"
 267     )
 268
 269     intf_c1_c2 = topo["routers"]["c1"]["links"]["c2"]["interface"]
 270     intf_c2_c1 = topo["routers"]["c2"]["links"]["c1"]["interface"]
 271     shutdown_bringup_interface(tgen, "c1", intf_c1_c2, False)
 272     shutdown_bringup_interface(tgen, "c2", intf_c2_c1, False)
 273
 274     step("Enable the PIM on all the interfaces of FRR1, FRR2, FRR3")
 275     step(
 276         "Enable IGMP of FRR1 interface and send IGMP joins "
 277         " from FRR1 node for group range (225.1.1.1-5)"
 278     )
 279
 280     intf_c2_i5 = topo["routers"]["c2"]["links"]["i5"]["interface"]
 281     input_dict = {
 282         "c2": {"igmp": {"interfaces": {intf_c2_i5: {"igmp": {"version": "2"}}}}}
 283     }
 284     result = create_igmp_config(tgen, topo, input_dict)
 285     assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 286
 287     input_join = {"i5": topo["routers"]["i5"]["links"]["c2"]["interface"]}
 288
 289     for recvr, recvr_intf in input_join.items():
 290         result = app_helper.run_join(recvr, IGMP_JOIN_RANGE_1, join_intf=recvr_intf)
 291         assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 292
 293     step("Configure static RP for (225.1.1.1-5) as R2")
 294
 295     input_dict = {
 296         "l1": {
 297             "pim": {
 298                 "rp": [
 299                     {
 300                         "rp_addr": topo["routers"]["l1"]["links"]["lo"]["ipv4"].split(
 301                             "/"
 302                         )[0],
 303                         "group_addr_range": GROUP_RANGE,
 304                     }
 305                 ]
 306             }
 307         }
 308     }
 309
 310     result = create_pim_config(tgen, topo, input_dict)
 311     assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
 312
 313     step("Send traffic from C1 to all the groups ( 225.1.1.1 to 225.1.1.5)")
 314
 315     input_src = {"i4": topo["routers"]["i4"]["links"]["c1"]["interface"]}
 316
 317     for src, src_intf in input_src.items():
 318         result = app_helper.run_traffic(src, IGMP_JOIN_RANGE_1, bind_intf=src_intf)
 319         assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
 320
 321     source_i4 = topo["routers"]["i4"]["links"]["c1"]["ipv4"].split("/")[0]
 322
 323     input_dict_starg = [
 324         {
 325             "dut": "c2",
 326             "src_address": "*",
 327             "iif": topo["routers"]["c2"]["links"]["f1"]["interface"],
 328             "oil": topo["routers"]["c2"]["links"]["i5"]["interface"],
 329         }
 330     ]
 331
 332     input_dict_sg = [
 333         {
 334             "dut": "c2",
 335             "src_address": source_i4,
 336             "iif": topo["routers"]["c2"]["links"]["f1"]["interface"],
 337             "oil": topo["routers"]["c2"]["links"]["i5"]["interface"],
 338         }
 339     ]
 340
 341     step("Verify mroutes and iff upstream")
 342
 343     for data in input_dict_sg:
 344         result = verify_mroutes(
 345             tgen,
 346             data["dut"],
 347             data["src_address"],
 348             IGMP_JOIN_RANGE_1,
 349             data["iif"],
 350             data["oil"],
 351         )
 352         assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
 353
 354         result = verify_upstream_iif(
 355             tgen, data["dut"], data["iif"], data["src_address"], IGMP_JOIN_RANGE_1
 356         )
 357         assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
 358
 359     for data in input_dict_starg:
 360         result = verify_mroutes(
 361             tgen,
 362             data["dut"],
 363             data["src_address"],
 364             IGMP_JOIN_RANGE_1,
 365             data["iif"],
 366             data["oil"],
 367         )
 368         assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
 369
 370         result = verify_upstream_iif(
 371             tgen, data["dut"], data["iif"], data["src_address"], IGMP_JOIN_RANGE_1
 372         )
 373         assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
 374
 375     step("Delete static routes on c2")
 376     input_dict = {
 377         "c2": {
 378             "static_routes": [
 379                 {
 380                     "network": ["1.0.4.11/32", "10.0.2.1/24", "10.0.1.2/24"],
 381                     "next_hop": "10.0.3.2",
 382                     "delete": True,
 383                 }
 384             ]
 385         }
 386     }
 387
 388     result = create_static_routes(tgen, input_dict)
 389     assert result is True, "Testcase {} :Failed \n Error {}".format(tc_name, result)
 390
 391     step("Verify RP info unknown after removing static route from c2 ")
 392     dut = "c2"
 393     rp_address = topo["routers"]["l1"]["links"]["lo"]["ipv4"].split("/")[0]
 394     SOURCE = "Static"
 395     result = verify_pim_rp_info(
 396         tgen, topo, dut, GROUP_RANGE_1, "Unknown", rp_address, SOURCE
 397     )
 398     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 399
 400     step("Verify mroute not present after Delete of static routes on c1")
 401
 402     for data in input_dict_sg:
 403         result = verify_mroutes(
 404             tgen,
 405             data["dut"],
 406             data["src_address"],
 407             IGMP_JOIN_RANGE_1,
 408             data["iif"],
 409             data["oil"],
 410             expected=False,
 411         )
 412         assert result is not True, (
 413             "Testcase {} : Failed \n "
 414             "Expected: [{}]: mroute (S, G) should not be present in mroute table \n "
 415             "Found: {}".format(tc_name, data["dut"], result)
 416         )
 417
 418         result = verify_upstream_iif(
 419             tgen,
 420             data["dut"],
 421             data["iif"],
 422             data["src_address"],
 423             IGMP_JOIN_RANGE_1,
 424             expected=False,
 425         )
 426         assert result is not True, (
 427             "Testcase {} : Failed \n "
 428             "Expected: [{}]: Upstream IIF interface {} should not be present\n "
 429             "Found: {}".format(tc_name, data["dut"], data["iif"], result)
 430         )
 431
 432     for data in input_dict_starg:
 433         result = verify_mroutes(
 434             tgen,
 435             data["dut"],
 436             data["src_address"],
 437             IGMP_JOIN_RANGE_1,
 438             data["iif"],
 439             data["oil"],
 440             expected=False,
 441         )
 442         assert result is not True, (
 443             "Testcase {} : Failed \n "
 444             "Expected: [{}]: mroute (S, G) should not be present in mroute table \n "
 445             "Found: {}".format(tc_name, data["dut"], result)
 446         )
 447
 448         result = verify_upstream_iif(
 449             tgen,
 450             data["dut"],
 451             data["iif"],
 452             data["src_address"],
 453             IGMP_JOIN_RANGE_1,
 454             expected=False,
 455         )
 456         assert result is not True, (
 457             "Testcase {} : Failed \n "
 458             "Expected: [{}]: Upstream IIF interface {} should not be present\n "
 459             "Found: {}".format(tc_name, data["dut"], data["iif"], result)
 460         )
 461
 462     step("Configure default routes on c2")
 463
 464     intf_f1_c2 = topo["routers"]["f1"]["links"]["c2"]["ipv4"].split("/")[0]
 465
 466     input_dict = {
 467         "c2": {"static_routes": [{"network": "0.0.0.0/0", "next_hop": intf_f1_c2}]}
 468     }
 469     result = create_static_routes(tgen, input_dict)
 470     assert result is True, "Testcase {} :Failed \n Error {}".format(tc_name, result)
 471
 472     step("applying ip nht config  on c2")
 473
 474     raw_config = {"c2": {"raw_config": ["ip nht resolve-via-default"]}}
 475
 476     result = apply_raw_config(tgen, raw_config)
 477     assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
 478
 479     step("Verify RP info is NOT unknown after removing static route from c2 ")
 480     result = verify_pim_rp_info(
 481         tgen, topo, dut, GROUP_RANGE_1, "Unknown", rp_address, SOURCE, expected=False
 482     )
 483     assert result is not True, (
 484         "Testcase {} : Failed \n "
 485         "Expected: [{}]: RP Info should not be Unknown after removing static"
 486         " route from c2 \n"
 487         "Found: {}".format(tc_name, data["dut"], result)
 488     )
 489
 490     step("Verify (s,g) populated after adding default route ")
 491
 492     for data in input_dict_sg:
 493         result = verify_mroutes(
 494             tgen,
 495             data["dut"],
 496             data["src_address"],
 497             IGMP_JOIN_RANGE_1,
 498             data["iif"],
 499             data["oil"],
 500         )
 501         assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
 502
 503         result = verify_upstream_iif(
 504             tgen, data["dut"], data["iif"], data["src_address"], IGMP_JOIN_RANGE_1
 505         )
 506         assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
 507
 508     step("Verify (*,g) populated after adding default route ")
 509
 510     for data in input_dict_starg:
 511         result = verify_mroutes(
 512             tgen,
 513             data["dut"],
 514             data["src_address"],
 515             IGMP_JOIN_RANGE_1,
 516             data["iif"],
 517             data["oil"],
 518         )
 519         assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
 520
 521         result = verify_upstream_iif(
 522             tgen, data["dut"], data["iif"], data["src_address"], IGMP_JOIN_RANGE_1
 523         )
 524         assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
 525
 526     write_test_footer(tc_name)
 527
 528
 529 def test_mroute_with_RP_default_route_all_nodes_p2(request):
 530     """
 531     TC_50 Verify mroute when LHR,FHR,RP and transit routers reachable
 532     using default routes
 533     """
 534
 535     tgen = get_topogen()
 536     tc_name = request.node.name
 537     write_test_header(tc_name)
 538
 539     # Don"t run this test if we have any failure.
 540     if tgen.routers_have_failure():
 541         pytest.skip(tgen.errors)
 542
 543     # Creating configuration from JSON
 544     app_helper.stop_all_hosts()
 545     clear_mroute(tgen)
 546     reset_config_on_routers(tgen)
 547     clear_pim_interface_traffic(tgen, topo)
 548
 549     step(
 550         "Remove c1-c2 connected link to simulate topo "
 551         "c1(LHR)---l1(RP)----r2---f1-----c2(FHR)"
 552     )
 553
 554     intf_c1_c2 = topo["routers"]["c1"]["links"]["c2"]["interface"]
 555     intf_c2_c1 = topo["routers"]["c2"]["links"]["c1"]["interface"]
 556     shutdown_bringup_interface(tgen, "c1", intf_c1_c2, False)
 557     shutdown_bringup_interface(tgen, "c2", intf_c2_c1, False)
 558
 559     step("Enable the PIM on all the interfaces of FRR1, FRR2, FRR3")
 560     step(
 561         "Enable IGMP of FRR1 interface and send IGMP joins "
 562         " from FRR1 node for group range (225.1.1.1-5)"
 563     )
 564
 565     intf_c1_i4 = topo["routers"]["c1"]["links"]["i4"]["interface"]
 566     input_dict = {
 567         "c1": {"igmp": {"interfaces": {intf_c1_i4: {"igmp": {"version": "2"}}}}}
 568     }
 569     result = create_igmp_config(tgen, topo, input_dict)
 570     assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 571
 572     input_join = {"i4": topo["routers"]["i4"]["links"]["c1"]["interface"]}
 573
 574     for recvr, recvr_intf in input_join.items():
 575         result = app_helper.run_join(recvr, IGMP_JOIN_RANGE_1, join_intf=recvr_intf)
 576         assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 577
 578     step("Configure static RP for (225.1.1.1-5) as R2")
 579
 580     input_dict = {
 581         "l1": {
 582             "pim": {
 583                 "rp": [
 584                     {
 585                         "rp_addr": topo["routers"]["l1"]["links"]["lo"]["ipv4"].split(
 586                             "/"
 587                         )[0],
 588                         "group_addr_range": GROUP_RANGE,
 589                     }
 590                 ]
 591             }
 592         }
 593     }
 594
 595     result = create_pim_config(tgen, topo, input_dict)
 596     assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
 597
 598     step("Send traffic from C2 to all the groups ( 225.1.1.1 to 225.1.1.5)")
 599
 600     input_src = {"i5": topo["routers"]["i5"]["links"]["c2"]["interface"]}
 601
 602     for src, src_intf in input_src.items():
 603         result = app_helper.run_traffic(src, IGMP_JOIN_RANGE_1, bind_intf=src_intf)
 604         assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
 605
 606     source_i5 = topo["routers"]["i5"]["links"]["c2"]["ipv4"].split("/")[0]
 607
 608     input_dict_starg = [
 609         {
 610             "dut": "c1",
 611             "src_address": "*",
 612             "iif": topo["routers"]["c1"]["links"]["l1"]["interface"],
 613             "oil": topo["routers"]["c1"]["links"]["i4"]["interface"],
 614         }
 615     ]
 616
 617     input_dict_sg = [
 618         {
 619             "dut": "c1",
 620             "src_address": source_i5,
 621             "iif": topo["routers"]["c1"]["links"]["l1"]["interface"],
 622             "oil": topo["routers"]["c1"]["links"]["i4"]["interface"],
 623         }
 624     ]
 625
 626     step("Verify mroutes and iff upstream")
 627
 628     for data in input_dict_sg:
 629         result = verify_mroutes(
 630             tgen,
 631             data["dut"],
 632             data["src_address"],
 633             IGMP_JOIN_RANGE_1,
 634             data["iif"],
 635             data["oil"],
 636         )
 637         assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
 638
 639         result = verify_upstream_iif(
 640             tgen, data["dut"], data["iif"], data["src_address"], IGMP_JOIN_RANGE_1
 641         )
 642         assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
 643
 644     for data in input_dict_starg:
 645         result = verify_mroutes(
 646             tgen,
 647             data["dut"],
 648             data["src_address"],
 649             IGMP_JOIN_RANGE_1,
 650             data["iif"],
 651             data["oil"],
 652         )
 653         assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
 654
 655         result = verify_upstream_iif(
 656             tgen, data["dut"], data["iif"], data["src_address"], IGMP_JOIN_RANGE_1
 657         )
 658         assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
 659
 660     step("Delete static routes RP on all the nodes")
 661     input_dict = {
 662         "c2": {
 663             "static_routes": [
 664                 {"network": ["1.0.4.11/32"], "next_hop": "10.0.3.2", "delete": True}
 665             ]
 666         },
 667         "c1": {
 668             "static_routes": [
 669                 {"network": ["1.0.4.11/32"], "next_hop": "10.0.2.2", "delete": True}
 670             ]
 671         },
 672         "r2": {
 673             "static_routes": [
 674                 {"network": ["1.0.4.11/32"], "next_hop": "10.0.12.1", "delete": True}
 675             ]
 676         },
 677         "f1": {
 678             "static_routes": [
 679                 {"network": ["1.0.4.11/32"], "next_hop": "10.0.7.2", "delete": True}
 680             ]
 681         },
 682     }
 683
 684     result = create_static_routes(tgen, input_dict)
 685     assert result is True, "Testcase {} :Failed \n Error {}".format(tc_name, result)
 686
 687     step("Verify RP info unknown after removing static route from c2 ")
 688     dut = "c2"
 689     rp_address = topo["routers"]["l1"]["links"]["lo"]["ipv4"].split("/")[0]
 690     SOURCE = "Static"
 691     result = verify_pim_rp_info(
 692         tgen, topo, dut, GROUP_RANGE_1, "Unknown", rp_address, SOURCE
 693     )
 694     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 695
 696     for data in input_dict_starg:
 697         result = verify_mroutes(
 698             tgen,
 699             data["dut"],
 700             data["src_address"],
 701             IGMP_JOIN_RANGE_1,
 702             data["iif"],
 703             data["oil"],
 704             expected=False,
 705         )
 706         assert result is not True, (
 707             "Testcase {} : Failed \n "
 708             "Expected: [{}]: mroute (S, G) should not be present in mroute table \n "
 709             "Found: {}".format(tc_name, data["dut"], result)
 710         )
 711
 712         result = verify_upstream_iif(
 713             tgen,
 714             data["dut"],
 715             data["iif"],
 716             data["src_address"],
 717             IGMP_JOIN_RANGE_1,
 718             expected=False,
 719         )
 720         assert result is not True, (
 721             "Testcase {} : Failed \n "
 722             "Expected: [{}]: Upstream IIF interface {} should not be present\n "
 723             "Found: {}".format(tc_name, data["dut"], data["iif"], result)
 724         )
 725
 726     step("Configure default routes on all the nodes")
 727
 728     intf_f1_c2 = topo["routers"]["f1"]["links"]["c2"]["ipv4"].split("/")[0]
 729     intf_l1_c1 = topo["routers"]["l1"]["links"]["c1"]["ipv4"].split("/")[0]
 730     intf_l1_r2 = topo["routers"]["l1"]["links"]["r2"]["ipv4"].split("/")[0]
 731     intf_r2_f1 = topo["routers"]["r2"]["links"]["f1"]["ipv4"].split("/")[0]
 732
 733     input_dict = {
 734         "c1": {"static_routes": [{"network": "0.0.0.0/0", "next_hop": intf_l1_c1}]},
 735         "c2": {"static_routes": [{"network": "0.0.0.0/0", "next_hop": intf_f1_c2}]},
 736         "r2": {"static_routes": [{"network": "0.0.0.0/0", "next_hop": intf_l1_r2}]},
 737         "f1": {"static_routes": [{"network": "0.0.0.0/0", "next_hop": intf_r2_f1}]},
 738     }
 739     result = create_static_routes(tgen, input_dict)
 740     assert result is True, "Testcase {} :Failed \n Error {}".format(tc_name, result)
 741
 742     step("applying ip nht config  on c2")
 743
 744     raw_config = {
 745         "c1": {"raw_config": ["ip nht resolve-via-default"]},
 746         "c2": {"raw_config": ["ip nht resolve-via-default"]},
 747         "r2": {"raw_config": ["ip nht resolve-via-default"]},
 748         "f1": {"raw_config": ["ip nht resolve-via-default"]},
 749         "l1": {"raw_config": ["ip nht resolve-via-default"]},
 750     }
 751
 752     result = apply_raw_config(tgen, raw_config)
 753     assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
 754
 755     step("Verify RP info Not unknown after removing static route from c2 ")
 756     dut = "c2"
 757     step("Verify RP info is NOT unknown after removing static route from c2 ")
 758     result = verify_pim_rp_info(
 759         tgen, topo, dut, GROUP_RANGE_1, "Unknown", rp_address, SOURCE, expected=False
 760     )
 761     assert result is not True, (
 762         "Testcase {} : Failed \n "
 763         "Expected: [{}]: RP Info should not be Unknown after removing static"
 764         " route from c2 \n"
 765         "Found: {}".format(tc_name, data["dut"], result)
 766     )
 767
 768     step("Verify (s,g) populated after adding default route ")
 769
 770     for data in input_dict_sg:
 771         result = verify_mroutes(
 772             tgen,
 773             data["dut"],
 774             data["src_address"],
 775             IGMP_JOIN_RANGE_1,
 776             data["iif"],
 777             data["oil"],
 778         )
 779         assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
 780
 781         result = verify_upstream_iif(
 782             tgen, data["dut"], data["iif"], data["src_address"], IGMP_JOIN_RANGE_1
 783         )
 784         assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
 785
 786     step("Verify (*,g) populated after adding default route ")
 787
 788     for data in input_dict_starg:
 789         result = verify_mroutes(
 790             tgen,
 791             data["dut"],
 792             data["src_address"],
 793             IGMP_JOIN_RANGE_1,
 794             data["iif"],
 795             data["oil"],
 796         )
 797         assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
 798
 799         result = verify_upstream_iif(
 800             tgen, data["dut"], data["iif"], data["src_address"], IGMP_JOIN_RANGE_1
 801         )
 802         assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
 803
 804     write_test_footer(tc_name)
 805
 806
 807 def test_PIM_hello_tx_rx_p1(request):
 808     """
 809     TC_54 Verify received and transmit hello stats
 810         are getting cleared after PIM nbr reset
 811     """
 812
 813     tgen = get_topogen()
 814     tc_name = request.node.name
 815     write_test_header(tc_name)
 816
 817     # Don"t run this test if we have any failure.
 818     if tgen.routers_have_failure():
 819         pytest.skip(tgen.errors)
 820
 821     # Creating configuration from JSON
 822     app_helper.stop_all_hosts()
 823     clear_mroute(tgen)
 824     reset_config_on_routers(tgen)
 825     clear_pim_interface_traffic(tgen, topo)
 826
 827     step(
 828         "Remove c1-c2 connected link to simulate topo "
 829         "c1(LHR)---l1(RP)----r2---f1-----c2(FHR)"
 830     )
 831
 832     intf_c1_c2 = topo["routers"]["c1"]["links"]["c2"]["interface"]
 833     intf_c2_c1 = topo["routers"]["c2"]["links"]["c1"]["interface"]
 834     shutdown_bringup_interface(tgen, "c1", intf_c1_c2, False)
 835     shutdown_bringup_interface(tgen, "c2", intf_c2_c1, False)
 836
 837     step("Enable the PIM on all the interfaces of FRR1, FRR2, FRR3")
 838     step(
 839         "Enable IGMP of FRR1 interface and send IGMP joins "
 840         " from FRR1 node for group range (225.1.1.1-5)"
 841     )
 842
 843     intf_c1_i4 = topo["routers"]["c1"]["links"]["i4"]["interface"]
 844     input_dict = {
 845         "c1": {"igmp": {"interfaces": {intf_c1_i4: {"igmp": {"version": "2"}}}}}
 846     }
 847     result = create_igmp_config(tgen, topo, input_dict)
 848     assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 849
 850     input_join = {"i4": topo["routers"]["i4"]["links"]["c1"]["interface"]}
 851
 852     for recvr, recvr_intf in input_join.items():
 853         result = app_helper.run_join(recvr, IGMP_JOIN_RANGE_1, join_intf=recvr_intf)
 854         assert result is True, "Testcase {}: Failed Error: {}".format(tc_name, result)
 855
 856     step("Configure static RP for (225.1.1.1-5) as R2")
 857
 858     input_dict = {
 859         "l1": {
 860             "pim": {
 861                 "rp": [
 862                     {
 863                         "rp_addr": topo["routers"]["l1"]["links"]["lo"]["ipv4"].split(
 864                             "/"
 865                         )[0],
 866                         "group_addr_range": GROUP_RANGE,
 867                     }
 868                 ]
 869             }
 870         }
 871     }
 872
 873     result = create_pim_config(tgen, topo, input_dict)
 874     assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
 875
 876     step("Send Mcast traffic from C2 to all the groups ( 225.1.1.1 to 225.1.1.5)")
 877
 878     input_src = {"i5": topo["routers"]["i5"]["links"]["c2"]["interface"]}
 879
 880     for src, src_intf in input_src.items():
 881         result = app_helper.run_traffic(src, IGMP_JOIN_RANGE_1, bind_intf=src_intf)
 882         assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
 883
 884     source_i5 = topo["routers"]["i5"]["links"]["c2"]["ipv4"].split("/")[0]
 885
 886     input_dict_starg = [
 887         {
 888             "dut": "c1",
 889             "src_address": "*",
 890             "iif": topo["routers"]["c1"]["links"]["l1"]["interface"],
 891             "oil": topo["routers"]["c1"]["links"]["i4"]["interface"],
 892         }
 893     ]
 894
 895     input_dict_sg = [
 896         {
 897             "dut": "c1",
 898             "src_address": source_i5,
 899             "iif": topo["routers"]["c1"]["links"]["l1"]["interface"],
 900             "oil": topo["routers"]["c1"]["links"]["i4"]["interface"],
 901         }
 902     ]
 903
 904     step("(*,G) and (S,G) created on f1 and node verify using 'show ip mroute'")
 905     for data in input_dict_sg:
 906         result = verify_mroutes(
 907             tgen,
 908             data["dut"],
 909             data["src_address"],
 910             IGMP_JOIN_RANGE_1,
 911             data["iif"],
 912             data["oil"],
 913         )
 914         assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
 915
 916     for data in input_dict_starg:
 917         result = verify_mroutes(
 918             tgen,
 919             data["dut"],
 920             data["src_address"],
 921             IGMP_JOIN_RANGE_1,
 922             data["iif"],
 923             data["oil"],
 924         )
 925         assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
 926
 927     intf_l1_c1 = topo["routers"]["l1"]["links"]["c1"]["interface"]
 928     intf_c1_l1 = topo["routers"]["c1"]["links"]["l1"]["interface"]
 929
 930     state_dict = {
 931         "c1": {
 932             intf_c1_l1: ["helloTx", "helloRx"],
 933         }
 934     }
 935
 936     c1_state_before = get_pim_interface_traffic(tgen, state_dict)
 937     assert isinstance(
 938         c1_state_before, dict
 939     ), "Testcase{} : Failed \n state_before is not dictionary \n Error: {}".format(
 940         tc_name, result
 941     )
 942
 943     step("Flap PIM nbr while doing interface c1-l1 interface shut from f1 side")
 944     shutdown_bringup_interface(tgen, "c1", intf_c1_l1, False)
 945
 946     """ Resetting the stats here since shutdown resets the stats.
 947     """
 948     reset_stats(c1_state_before)
 949     shutdown_bringup_interface(tgen, "c1", intf_c1_l1, True)
 950
 951     step("verify stats after no shutdown on c1 and that they are incremented")
 952
 953     count = 0
 954     done = False
 955     while not done and count <= 7:
 956         c1_state_after = get_pim_interface_traffic(tgen, state_dict)
 957         assert isinstance(
 958             c1_state_after, dict
 959         ), "Testcase{} : Failed \n state_before is not dictionary \n Error: {}".format(
 960             tc_name, result
 961         )
 962
 963         result = verify_state_incremented(c1_state_before, c1_state_after)
 964         if result is not True:
 965             sleep(5)
 966             count += 1
 967         else:
 968             done = True
 969
 970     assert (
 971         result is True
 972     ), "Testcase{} : Failed Error: {}" "stats is not incremented".format(
 973         tc_name, result
 974     )
 975
 976     step("verify before stats on l1")
 977     l1_state_dict = {
 978         "l1": {
 979             intf_l1_c1: ["helloTx", "helloRx"],
 980         }
 981     }
 982
 983     l1_state_before = get_pim_interface_traffic(tgen, l1_state_dict)
 984     assert isinstance(
 985         l1_state_before, dict
 986     ), "Testcase{} : Failed \n state_before is not dictionary \n Error: {}".format(
 987         tc_name, result
 988     )
 989
 990     step("Flap PIM nbr while doing interface r2-c1 shut from r2 side")
 991     shutdown_bringup_interface(tgen, "l1", intf_l1_c1, False)
 992
 993     step(
 994         "After shut the interface from r2 side , verify r2 side rx and tx of hello"
 995         "counters are resetted show ip pim interface traffic"
 996     )
 997     shutdown_bringup_interface(tgen, "l1", intf_l1_c1, True)
 998
 999     step("verify stats after on l1 are incremented")
1000     count = 0
1001     done = False
1002     while not done and count <= 7:
1003         l1_state_after = get_pim_interface_traffic(tgen, l1_state_dict)
1004         assert isinstance(
1005             l1_state_after, dict
1006         ), "Testcase{} : Failed \n state_before is not dictionary \n Error: {}".format(
1007             tc_name, result
1008         )
1009
1010         result = verify_state_incremented(l1_state_before, l1_state_after)
1011         if result is True:
1012             sleep(5)
1013             count += 1
1014         else:
1015             done = True
1016
1017     assert (
1018         result is not True
1019     ), "Testcase{} : Failed Error: {}" "stats incremented".format(tc_name, result)
1020
1021     step("Reinit the dict")
1022     c1_state_before = {}
1023     l1_state_before = {}
1024     c1_state_after = {}
1025     l1_state_after = {}
1026
1027     step("verify before stats on C1")
1028     state_dict = {
1029         "c1": {
1030             intf_c1_l1: ["helloTx", "helloRx"],
1031         }
1032     }
1033
1034     c1_state_before = get_pim_interface_traffic(tgen, state_dict)
1035     assert isinstance(
1036         c1_state_before, dict
1037     ), "Testcase{} : Failed \n state_before is not dictionary \n Error: {}".format(
1038         tc_name, result
1039     )
1040
1041     step("Flap c1-r2 pim nbr while changing ip address from c1 side")
1042     c1_l1_ip_subnet = topo["routers"]["c1"]["links"]["l1"]["ipv4"]
1043
1044     raw_config = {
1045         "c1": {
1046             "raw_config": [
1047                 "interface {}".format(intf_c1_l1),
1048                 "no ip address {}".format(c1_l1_ip_subnet),
1049                 "ip address {}".format(NEW_ADDRESS_2_SUBNET),
1050             ]
1051         }
1052     }
1053
1054     result = apply_raw_config(tgen, raw_config)
1055     assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
1056
1057     step("verify stats after on c1 are incremented")
1058     count = 0
1059     done = False
1060     while not done and count <= 7:
1061         c1_state_after = get_pim_interface_traffic(tgen, state_dict)
1062         assert isinstance(
1063             c1_state_after, dict
1064         ), "Testcase{} : Failed \n state_before is not dictionary \n Error: {}".format(
1065             tc_name, result
1066         )
1067
1068         result = verify_state_incremented(c1_state_before, c1_state_after)
1069         if result is not True:
1070             sleep(5)
1071             count += 1
1072         else:
1073             done = True
1074
1075     assert result is True, "Testcase{} : Failed Error: {}" "stats incremented".format(
1076         tc_name, result
1077     )
1078
1079     write_test_footer(tc_name)
1080
1081
1082 if __name__ == "__main__":
1083     args = ["-s"] + sys.argv[1:]
1084     sys.exit(pytest.main(args))