tests/topotests/bgp_ecmp_topo2/test_ebgp_ecmp_topo2.py

   1 #!/usr/bin/env python
   2
   3 #
   4 # Copyright (c) 2019 by VMware, Inc. ("VMware")
   5 # Used Copyright (c) 2018 by Network Device Education Foundation, Inc.
   6 # ("NetDEF") in this file.
   7 #
   8 # Permission to use, copy, modify, and/or distribute this software
   9 # for any purpose with or without fee is hereby granted, provided
  10 # that the above copyright notice and this permission notice appear
  11 # in all copies.
  12 #
  13 # THE SOFTWARE IS PROVIDED "AS IS" AND VMWARE DISCLAIMS ALL WARRANTIES
  14 # WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
  15 # MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL VMWARE BE LIABLE FOR
  16 # ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY
  17 # DAMAGES WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS,
  18 # WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS
  19 # ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR PERFORMANCE
  20 # OF THIS SOFTWARE.
  21 #
  22
  23
  24 """
  25 Following tests are covered to test ecmp functionality on EBGP.
  26 1. Verify routes installed as per maximum-paths configuration (8/16/32)
  27 2. Disable/Shut selected paths nexthops and verify other next are installed in
  28    the RIB of DUT. Enable interfaces and verify RIB count.
  29 3. Verify BGP table and RIB in DUT after clear BGP routes and neighbors.
  30 4. Verify routes are cleared from BGP and RIB table of DUT when
  31    redistribute static configuration is removed.
  32 5. Shut BGP neighbors one by one and verify BGP and routing table updated
  33    accordingly in DUT
  34 6. Delete static routes and verify routers are cleared from BGP table and RIB
  35    of DUT.
  36 7. Verify routes are cleared from BGP and RIB table of DUT when advertise
  37    network configuration is removed.
  38 """
  39 import os
  40 import sys
  41 import time
  42 import pytest
  43
  44 # Save the Current Working Directory to find configuration files.
  45 CWD = os.path.dirname(os.path.realpath(__file__))
  46 sys.path.append(os.path.join(CWD, "../"))
  47 sys.path.append(os.path.join(CWD, "../../"))
  48
  49 # pylint: disable=C0413
  50 # Import topogen and topotest helpers
  51 from lib.topogen import Topogen, get_topogen
  52
  53 from lib.common_config import (
  54     start_topology,
  55     write_test_header,
  56     write_test_footer,
  57     verify_rib,
  58     create_static_routes,
  59     check_address_types,
  60     interface_status,
  61     reset_config_on_routers,
  62     required_linux_kernel_version,
  63 )
  64 from lib.topolog import logger
  65 from lib.bgp import verify_bgp_convergence, create_router_bgp, clear_bgp
  66 from lib.topojson import build_config_from_json
  67
  68
  69 pytestmark = [pytest.mark.bgpd, pytest.mark.staticd]
  70
  71
  72 # Global variables
  73 NEXT_HOPS = {"ipv4": [], "ipv6": []}
  74 INTF_LIST_R3 = []
  75 INTF_LIST_R2 = []
  76 NETWORK = {"ipv4": "11.0.20.1/32", "ipv6": "1::/64"}
  77 NEXT_HOP_IP = {"ipv4": "10.0.0.1", "ipv6": "fd00::1"}
  78 BGP_CONVERGENCE = False
  79
  80
  81 def setup_module(mod):
  82     """
  83     Sets up the pytest environment.
  84
  85     * `mod`: module name
  86     """
  87     global NEXT_HOPS, INTF_LIST_R3, INTF_LIST_R2, TEST_STATIC
  88     global ADDR_TYPES
  89
  90     # Required linux kernel version for this suite to run.
  91     result = required_linux_kernel_version("4.15")
  92     if result is not True:
  93         pytest.skip("Kernel requirements are not met, kernel version should be >=4.15")
  94
  95     testsuite_run_time = time.asctime(time.localtime(time.time()))
  96     logger.info("Testsuite start time: {}".format(testsuite_run_time))
  97     logger.info("=" * 40)
  98
  99     logger.info("Running setup_module to create topology")
 100
 101     # This function initiates the topology build with Topogen...
 102     json_file = "{}/ebgp_ecmp_topo2.json".format(CWD)
 103     tgen = Topogen(json_file, mod.__name__)
 104     global topo
 105     topo = tgen.json_topo
 106
 107     # Starting topology, create tmp files which are loaded to routers
 108     #  to start daemons and then start routers
 109     start_topology(tgen)
 110
 111     # Creating configuration from JSON
 112     build_config_from_json(tgen, topo)
 113
 114     # Don't run this test if we have any failure.
 115     if tgen.routers_have_failure():
 116         pytest.skip(tgen.errors)
 117
 118     # tgen.mininet_cli()
 119     # Api call verify whether BGP is converged
 120     ADDR_TYPES = check_address_types()
 121
 122     BGP_CONVERGENCE = verify_bgp_convergence(tgen, topo)
 123     assert BGP_CONVERGENCE is True, "setup_module :Failed \n Error:" " {}".format(
 124         BGP_CONVERGENCE
 125     )
 126
 127     link_data = [
 128         val for links, val in topo["routers"]["r2"]["links"].items() if "r3" in links
 129     ]
 130     for adt in ADDR_TYPES:
 131         NEXT_HOPS[adt] = [val[adt].split("/")[0] for val in link_data]
 132         if adt == "ipv4":
 133             NEXT_HOPS[adt] = sorted(NEXT_HOPS[adt], key=lambda x: int(x.split(".")[2]))
 134         elif adt == "ipv6":
 135             NEXT_HOPS[adt] = sorted(
 136                 NEXT_HOPS[adt], key=lambda x: int(x.split(":")[-3], 16)
 137             )
 138
 139     INTF_LIST_R2 = [val["interface"].split("/")[0] for val in link_data]
 140     INTF_LIST_R2 = sorted(INTF_LIST_R2, key=lambda x: int(x.split("eth")[1]))
 141
 142     link_data = [
 143         val for links, val in topo["routers"]["r3"]["links"].items() if "r2" in links
 144     ]
 145     INTF_LIST_R3 = [val["interface"].split("/")[0] for val in link_data]
 146     INTF_LIST_R3 = sorted(INTF_LIST_R3, key=lambda x: int(x.split("eth")[1]))
 147
 148     # STATIC_ROUTE = True
 149     logger.info("Running setup_module() done")
 150
 151
 152 def teardown_module():
 153     """
 154     Teardown the pytest environment.
 155
 156     * `mod`: module name
 157     """
 158
 159     logger.info("Running teardown_module to delete topology")
 160
 161     tgen = get_topogen()
 162
 163     # Stop toplogy and Remove tmp files
 164     tgen.stop_topology()
 165
 166
 167 def static_or_nw(tgen, topo, tc_name, test_type, dut):
 168
 169     if test_type == "redist_static":
 170         input_dict_static = {
 171             dut: {
 172                 "static_routes": [
 173                     {"network": NETWORK["ipv4"], "next_hop": NEXT_HOP_IP["ipv4"]},
 174                     {"network": NETWORK["ipv6"], "next_hop": NEXT_HOP_IP["ipv6"]},
 175                 ]
 176             }
 177         }
 178         logger.info("Configuring static route on router %s", dut)
 179         result = create_static_routes(tgen, input_dict_static)
 180         assert result is True, "Testcase {} : Failed \n Error: {}".format(
 181             tc_name, result
 182         )
 183
 184         input_dict_2 = {
 185             dut: {
 186                 "bgp": {
 187                     "address_family": {
 188                         "ipv4": {
 189                             "unicast": {"redistribute": [{"redist_type": "static"}]}
 190                         },
 191                         "ipv6": {
 192                             "unicast": {"redistribute": [{"redist_type": "static"}]}
 193                         },
 194                     }
 195                 }
 196             }
 197         }
 198
 199         logger.info("Configuring redistribute static route on router %s", dut)
 200         result = create_router_bgp(tgen, topo, input_dict_2)
 201         assert result is True, "Testcase {} : Failed \n Error: {}".format(
 202             tc_name, result
 203         )
 204
 205     elif test_type == "advertise_nw":
 206         input_dict_nw = {
 207             dut: {
 208                 "bgp": {
 209                     "address_family": {
 210                         "ipv4": {
 211                             "unicast": {
 212                                 "advertise_networks": [{"network": NETWORK["ipv4"]}]
 213                             }
 214                         },
 215                         "ipv6": {
 216                             "unicast": {
 217                                 "advertise_networks": [{"network": NETWORK["ipv6"]}]
 218                             }
 219                         },
 220                     }
 221                 }
 222             }
 223         }
 224
 225         logger.info(
 226             "Advertising networks %s %s from router %s",
 227             NETWORK["ipv4"],
 228             NETWORK["ipv6"],
 229             dut,
 230         )
 231         result = create_router_bgp(tgen, topo, input_dict_nw)
 232         assert result is True, "Testcase {} : Failed \n Error: {}".format(
 233             tc_name, result
 234         )
 235
 236
 237 @pytest.mark.parametrize("ecmp_num", ["8", "16", "32"])
 238 @pytest.mark.parametrize("test_type", ["redist_static", "advertise_nw"])
 239 def test_modify_ecmp_max_paths(request, ecmp_num, test_type):
 240     """
 241     Verify routes installed as per maximum-paths
 242     configuration (8/16/32).
 243     """
 244
 245     tc_name = request.node.name
 246     write_test_header(tc_name)
 247     tgen = get_topogen()
 248
 249     reset_config_on_routers(tgen)
 250
 251     static_or_nw(tgen, topo, tc_name, test_type, "r2")
 252
 253     input_dict = {
 254         "r3": {
 255             "bgp": {
 256                 "address_family": {
 257                     "ipv4": {
 258                         "unicast": {
 259                             "maximum_paths": {
 260                                 "ebgp": ecmp_num,
 261                             }
 262                         }
 263                     },
 264                     "ipv6": {
 265                         "unicast": {
 266                             "maximum_paths": {
 267                                 "ebgp": ecmp_num,
 268                             }
 269                         }
 270                     },
 271                 }
 272             }
 273         }
 274     }
 275
 276     logger.info("Configuring bgp maximum-paths %s on router r3", ecmp_num)
 277     result = create_router_bgp(tgen, topo, input_dict)
 278     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 279
 280     # Verifying RIB routes
 281     dut = "r3"
 282     protocol = "bgp"
 283
 284     for addr_type in ADDR_TYPES:
 285         input_dict_1 = {"r3": {"static_routes": [{"network": NETWORK[addr_type]}]}}
 286
 287         logger.info("Verifying %s routes on r3", addr_type)
 288
 289         # Only test the count of nexthops; the actual nexthop addresses
 290         # can vary and are not deterministic.
 291         #
 292         result = verify_rib(
 293             tgen,
 294             addr_type,
 295             dut,
 296             input_dict_1,
 297             next_hop=NEXT_HOPS[addr_type][: int(ecmp_num)],
 298             protocol=protocol,
 299             count_only=True,
 300         )
 301
 302         assert result is True, "Testcase {} : Failed \n Error: {}".format(
 303             tc_name, result
 304         )
 305
 306     write_test_footer(tc_name)
 307
 308
 309 @pytest.mark.parametrize("ecmp_num", ["8", "16", "32"])
 310 @pytest.mark.parametrize("test_type", ["redist_static", "advertise_nw"])
 311 def test_ecmp_after_clear_bgp(request, ecmp_num, test_type):
 312     """Verify BGP table and RIB in DUT after clear BGP routes and neighbors"""
 313
 314     tc_name = request.node.name
 315     write_test_header(tc_name)
 316     tgen = get_topogen()
 317
 318     reset_config_on_routers(tgen)
 319
 320     # Verifying RIB routes
 321     dut = "r3"
 322     protocol = "bgp"
 323
 324     static_or_nw(tgen, topo, tc_name, test_type, "r2")
 325     for addr_type in ADDR_TYPES:
 326         input_dict_1 = {"r3": {"static_routes": [{"network": NETWORK[addr_type]}]}}
 327
 328         logger.info("Verifying %s routes on r3", addr_type)
 329         result = verify_rib(
 330             tgen,
 331             addr_type,
 332             dut,
 333             input_dict_1,
 334             next_hop=NEXT_HOPS[addr_type][: int(ecmp_num)],
 335             protocol=protocol,
 336         )
 337         assert result is True, "Testcase {} : Failed \n Error: {}".format(
 338             tc_name, result
 339         )
 340
 341     # Clear BGP
 342     for addr_type in ADDR_TYPES:
 343         clear_bgp(tgen, addr_type, dut)
 344
 345     # Verify BGP convergence
 346     result = verify_bgp_convergence(tgen, topo)
 347     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 348
 349     for addr_type in ADDR_TYPES:
 350         input_dict_1 = {"r3": {"static_routes": [{"network": NETWORK[addr_type]}]}}
 351         logger.info("Verifying %s routes on r3", addr_type)
 352         result = verify_rib(
 353             tgen,
 354             addr_type,
 355             dut,
 356             input_dict_1,
 357             next_hop=NEXT_HOPS[addr_type][: int(ecmp_num)],
 358             protocol=protocol,
 359         )
 360         assert result is True, "Testcase {} : Failed \n Error: {}".format(
 361             tc_name, result
 362         )
 363
 364     write_test_footer(tc_name)
 365
 366
 367 def test_ecmp_remove_redistribute_static(request):
 368     """Verify routes are cleared from BGP and RIB table of DUT when
 369     redistribute static configuration is removed."""
 370
 371     tc_name = request.node.name
 372     write_test_header(tc_name)
 373     tgen = get_topogen()
 374
 375     reset_config_on_routers(tgen)
 376     static_or_nw(tgen, topo, tc_name, "redist_static", "r2")
 377     for addr_type in ADDR_TYPES:
 378
 379         # Verifying RIB routes
 380         dut = "r3"
 381         protocol = "bgp"
 382         input_dict_1 = {"r3": {"static_routes": [{"network": NETWORK[addr_type]}]}}
 383
 384         logger.info("Verifying %s routes on r3", addr_type)
 385         result = verify_rib(
 386             tgen,
 387             addr_type,
 388             dut,
 389             input_dict_1,
 390             next_hop=NEXT_HOPS[addr_type],
 391             protocol=protocol,
 392         )
 393         assert result is True, "Testcase {} : Failed \n Error: {}".format(
 394             tc_name, result
 395         )
 396
 397     input_dict_2 = {
 398         "r2": {
 399             "bgp": {
 400                 "address_family": {
 401                     "ipv4": {
 402                         "unicast": {
 403                             "redistribute": [{"redist_type": "static", "delete": True}]
 404                         }
 405                     },
 406                     "ipv6": {
 407                         "unicast": {
 408                             "redistribute": [{"redist_type": "static", "delete": True}]
 409                         }
 410                     },
 411                 }
 412             }
 413         }
 414     }
 415
 416     logger.info("Remove redistribute static")
 417     result = create_router_bgp(tgen, topo, input_dict_2)
 418     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 419
 420     for addr_type in ADDR_TYPES:
 421
 422         # Verifying RIB routes
 423         dut = "r3"
 424         protocol = "bgp"
 425         input_dict_1 = {"r3": {"static_routes": [{"network": NETWORK[addr_type]}]}}
 426
 427         logger.info("Verifying %s routes on r3 are deleted", addr_type)
 428         result = verify_rib(
 429             tgen,
 430             addr_type,
 431             dut,
 432             input_dict_1,
 433             next_hop=[],
 434             protocol=protocol,
 435             expected=False,
 436         )
 437         assert (
 438             result is not True
 439         ), "Testcase {} : Failed \n Expected: Routes still present in {} RIB. Found: {}".format(
 440             tc_name, dut, result
 441         )
 442
 443     logger.info("Enable redistribute static")
 444     input_dict_2 = {
 445         "r2": {
 446             "bgp": {
 447                 "address_family": {
 448                     "ipv4": {"unicast": {"redistribute": [{"redist_type": "static"}]}},
 449                     "ipv6": {"unicast": {"redistribute": [{"redist_type": "static"}]}},
 450                 }
 451             }
 452         }
 453     }
 454     result = create_router_bgp(tgen, topo, input_dict_2)
 455     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 456
 457     for addr_type in ADDR_TYPES:
 458         # Verifying RIB routes
 459         dut = "r3"
 460         protocol = "bgp"
 461         input_dict_1 = {"r3": {"static_routes": [{"network": NETWORK[addr_type]}]}}
 462         logger.info("Verifying %s routes on r3", addr_type)
 463         result = verify_rib(
 464             tgen,
 465             addr_type,
 466             dut,
 467             input_dict_1,
 468             next_hop=NEXT_HOPS[addr_type],
 469             protocol=protocol,
 470         )
 471         assert result is True, "Testcase {} : Failed \n Error: {}".format(
 472             tc_name, result
 473         )
 474
 475     write_test_footer(tc_name)
 476
 477
 478 @pytest.mark.parametrize("test_type", ["redist_static", "advertise_nw"])
 479 def test_ecmp_shut_bgp_neighbor(request, test_type):
 480     """Shut BGP neighbors one by one and verify BGP and routing table updated
 481     accordingly in DUT"""
 482
 483     tc_name = request.node.name
 484     write_test_header(tc_name)
 485     tgen = get_topogen()
 486
 487     logger.info(INTF_LIST_R2)
 488     # Verifying RIB routes
 489     dut = "r3"
 490     protocol = "bgp"
 491
 492     reset_config_on_routers(tgen)
 493     static_or_nw(tgen, topo, tc_name, test_type, "r2")
 494
 495     for addr_type in ADDR_TYPES:
 496         input_dict = {"r3": {"static_routes": [{"network": NETWORK[addr_type]}]}}
 497
 498         logger.info("Verifying %s routes on r3", addr_type)
 499         result = verify_rib(
 500             tgen,
 501             addr_type,
 502             dut,
 503             input_dict,
 504             next_hop=NEXT_HOPS[addr_type],
 505             protocol=protocol,
 506         )
 507         assert result is True, "Testcase {} : Failed \n Error: {}".format(
 508             tc_name, result
 509         )
 510
 511     for intf_num in range(len(INTF_LIST_R2) + 1, 16):
 512         intf_val = INTF_LIST_R2[intf_num : intf_num + 16]
 513
 514         input_dict_1 = {"r2": {"interface_list": [intf_val], "status": "down"}}
 515         logger.info("Shutting down neighbor interface {} on r2".format(intf_val))
 516         result = interface_status(tgen, topo, input_dict_1)
 517         assert result is True, "Testcase {} : Failed \n Error: {}".format(
 518             tc_name, result
 519         )
 520
 521         for addr_type in ADDR_TYPES:
 522             if intf_num + 16 < 32:
 523                 check_hops = NEXT_HOPS[addr_type]
 524             else:
 525                 check_hops = []
 526
 527             input_dict = {"r3": {"static_routes": [{"network": NETWORK[addr_type]}]}}
 528             logger.info("Verifying %s routes on r3", addr_type)
 529             result = verify_rib(
 530                 tgen, addr_type, dut, input_dict, next_hop=check_hops, protocol=protocol
 531             )
 532             assert result is True, "Testcase {} : Failed \n Error: {}".format(
 533                 tc_name, result
 534             )
 535
 536     input_dict_1 = {"r2": {"interface_list": INTF_LIST_R2, "status": "up"}}
 537
 538     logger.info("Enabling all neighbor interface {} on r2")
 539     result = interface_status(tgen, topo, input_dict_1)
 540     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 541
 542     static_or_nw(tgen, topo, tc_name, test_type, "r2")
 543     for addr_type in ADDR_TYPES:
 544         input_dict = {"r3": {"static_routes": [{"network": NETWORK[addr_type]}]}}
 545
 546         logger.info("Verifying %s routes on r3", addr_type)
 547         result = verify_rib(
 548             tgen,
 549             addr_type,
 550             dut,
 551             input_dict,
 552             next_hop=NEXT_HOPS[addr_type],
 553             protocol=protocol,
 554         )
 555         assert result is True, "Testcase {} : Failed \n Error: {}".format(
 556             tc_name, result
 557         )
 558
 559     write_test_footer(tc_name)
 560
 561
 562 def test_ecmp_remove_static_route(request):
 563     """
 564     Delete static routes and verify routers are cleared from BGP table,
 565     and RIB of DUT.
 566     """
 567
 568     tc_name = request.node.name
 569     write_test_header(tc_name)
 570     tgen = get_topogen()
 571
 572     # Verifying RIB routes
 573     dut = "r3"
 574     protocol = "bgp"
 575
 576     reset_config_on_routers(tgen)
 577
 578     static_or_nw(tgen, topo, tc_name, "redist_static", "r2")
 579     for addr_type in ADDR_TYPES:
 580         input_dict_1 = {"r3": {"static_routes": [{"network": NETWORK[addr_type]}]}}
 581
 582         logger.info("Verifying %s routes on r3", addr_type)
 583         result = verify_rib(
 584             tgen,
 585             addr_type,
 586             dut,
 587             input_dict_1,
 588             next_hop=NEXT_HOPS[addr_type],
 589             protocol=protocol,
 590         )
 591         assert result is True, "Testcase {} : Failed \n Error: {}".format(
 592             tc_name, result
 593         )
 594
 595     for addr_type in ADDR_TYPES:
 596         input_dict_2 = {
 597             "r2": {
 598                 "static_routes": [
 599                     {
 600                         "network": NETWORK[addr_type],
 601                         "next_hop": NEXT_HOP_IP[addr_type],
 602                         "delete": True,
 603                     }
 604                 ]
 605             }
 606         }
 607
 608         logger.info("Remove static routes")
 609         result = create_static_routes(tgen, input_dict_2)
 610         assert result is True, "Testcase {} : Failed \n Error: {}".format(
 611             tc_name, result
 612         )
 613
 614         logger.info("Verifying %s routes on r3 are removed", addr_type)
 615         result = verify_rib(
 616             tgen,
 617             addr_type,
 618             dut,
 619             input_dict_2,
 620             next_hop=[],
 621             protocol=protocol,
 622             expected=False,
 623         )
 624         assert (
 625             result is not True
 626         ), "Testcase {} : Failed \n  Expected: Routes still present in {} RIB. Found: {}".format(
 627             tc_name, dut, result
 628         )
 629
 630     for addr_type in ADDR_TYPES:
 631         # Enable static routes
 632         input_dict_4 = {
 633             "r2": {
 634                 "static_routes": [
 635                     {"network": NETWORK[addr_type], "next_hop": NEXT_HOP_IP[addr_type]}
 636                 ]
 637             }
 638         }
 639
 640         logger.info("Enable static route")
 641         result = create_static_routes(tgen, input_dict_4)
 642         assert result is True, "Testcase {} : Failed \n Error: {}".format(
 643             tc_name, result
 644         )
 645
 646         logger.info("Verifying %s routes on r3", addr_type)
 647         result = verify_rib(
 648             tgen,
 649             addr_type,
 650             dut,
 651             input_dict_4,
 652             next_hop=NEXT_HOPS[addr_type],
 653             protocol=protocol,
 654         )
 655         assert result is True, "Testcase {} : Failed \n Error: {}".format(
 656             tc_name, result
 657         )
 658
 659
 660 def test_ecmp_remove_nw_advertise(request):
 661     """
 662     Verify routes are cleared from BGP and RIB table of DUT,
 663     when advertise network configuration is removed
 664     """
 665
 666     tc_name = request.node.name
 667     write_test_header(tc_name)
 668     tgen = get_topogen()
 669
 670     # Verifying RIB routes
 671     dut = "r3"
 672     protocol = "bgp"
 673
 674     reset_config_on_routers(tgen)
 675     static_or_nw(tgen, topo, tc_name, "advertise_nw", "r2")
 676     for addr_type in ADDR_TYPES:
 677         input_dict = {"r3": {"static_routes": [{"network": NETWORK[addr_type]}]}}
 678
 679         logger.info("Verifying %s routes on r3", addr_type)
 680         result = verify_rib(
 681             tgen,
 682             addr_type,
 683             dut,
 684             input_dict,
 685             next_hop=NEXT_HOPS[addr_type],
 686             protocol=protocol,
 687         )
 688         assert result is True, "Testcase {} : Failed \n Error: {}".format(
 689             tc_name, result
 690         )
 691
 692     input_dict_3 = {
 693         "r2": {
 694             "bgp": {
 695                 "address_family": {
 696                     "ipv4": {
 697                         "unicast": {
 698                             "advertise_networks": [
 699                                 {"network": NETWORK["ipv4"], "delete": True}
 700                             ]
 701                         }
 702                     },
 703                     "ipv6": {
 704                         "unicast": {
 705                             "advertise_networks": [
 706                                 {"network": NETWORK["ipv6"], "delete": True}
 707                             ]
 708                         }
 709                     },
 710                 }
 711             }
 712         }
 713     }
 714
 715     logger.info("Withdraw advertised networks")
 716     result = create_router_bgp(tgen, topo, input_dict_3)
 717     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 718
 719     for addr_type in ADDR_TYPES:
 720         input_dict = {"r3": {"static_routes": [{"network": NETWORK[addr_type]}]}}
 721
 722         logger.info("Verifying %s routes on r3", addr_type)
 723         result = verify_rib(
 724             tgen,
 725             addr_type,
 726             dut,
 727             input_dict,
 728             next_hop=[],
 729             protocol=protocol,
 730             expected=False,
 731         )
 732         assert (
 733             result is not True
 734         ), "Testcase {} : Failed \n  Expected: Routes still present in {} RIB. Found: {}".format(
 735             tc_name, dut, result
 736         )
 737
 738     static_or_nw(tgen, topo, tc_name, "advertise_nw", "r2")
 739     for addr_type in ADDR_TYPES:
 740         input_dict = {"r3": {"static_routes": [{"network": NETWORK[addr_type]}]}}
 741         logger.info("Verifying %s routes on r3", addr_type)
 742         result = verify_rib(
 743             tgen,
 744             addr_type,
 745             dut,
 746             input_dict,
 747             next_hop=NEXT_HOPS[addr_type],
 748             protocol=protocol,
 749         )
 750         assert result is True, "Testcase {} : Failed \n Error: {}".format(
 751             tc_name, result
 752         )
 753
 754     write_test_footer(tc_name)
 755
 756
 757 if __name__ == "__main__":
 758     args = ["-s"] + sys.argv[1:]
 759     sys.exit(pytest.main(args))