tests/topotests/bgp_vrf_dynamic_route_leak_topo4/test_bgp_vrf_dynamic_route_leak_topo4-3.py

   1 #!/usr/bin/env python
   2
   3 #
   4 # Copyright (c) 2021 by VMware, Inc. ("VMware")
   5 # Used Copyright (c) 2018 by Network Device Education Foundation,
   6 # Inc. ("NetDEF") in this file.
   7 #
   8 # Permission to use, copy, modify, and/or distribute this software
   9 # for any purpose with or without fee is hereby granted, provided
  10 # that the above copyright notice and this permission notice appear
  11 # in all copies.
  12 #
  13 # THE SOFTWARE IS PROVIDED "AS IS" AND VMWARE DISCLAIMS ALL WARRANTIES
  14 # WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
  15 # MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL VMWARE BE LIABLE FOR
  16 # ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY
  17 # DAMAGES WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS,
  18 # WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS
  19 # ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR PERFORMANCE
  20 # OF THIS SOFTWARE.
  21 #
  22
  23 """
  24 Following tests are covered to test BGP Multi-VRF Dynamic Route Leaking:
  25 1. Verify recursive import among Tenant VRFs.
  26 2. Verify that dynamic import works fine between two different Tenant VRFs.
  27     When next-hop IPs are same across all VRFs.
  28     When next-hop IPs are different across all VRFs.
  29 3. Verify that with multiple tenant VRFs, dynamic import works fine between
  30     Tenant VRFs to default VRF.
  31     When next-hop IPs and prefixes are same across all VRFs.
  32     When next-hop IPs and prefixes are different across all VRFs.
  33 """
  34
  35 import os
  36 import sys
  37 import time
  38 import pytest
  39 import platform
  40 from time import sleep
  41
  42 # Save the Current Working Directory to find configuration files.
  43 CWD = os.path.dirname(os.path.realpath(__file__))
  44 sys.path.append(os.path.join(CWD, "../"))
  45 sys.path.append(os.path.join(CWD, "../lib/"))
  46
  47 # Required to instantiate the topology builder class.
  48
  49 # pylint: disable=C0413
  50 # Import topogen and topotest helpers
  51 from lib.topogen import Topogen, get_topogen
  52 from lib.topotest import version_cmp
  53
  54 from lib.common_config import (
  55     start_topology,
  56     write_test_header,
  57     check_address_types,
  58     write_test_footer,
  59     reset_config_on_routers,
  60     verify_rib,
  61     step,
  62     create_route_maps,
  63     create_static_routes,
  64     create_prefix_lists,
  65     create_bgp_community_lists,
  66     get_frr_ipv6_linklocal,
  67 )
  68
  69 from lib.topolog import logger
  70 from lib.bgp import (
  71     verify_bgp_convergence,
  72     create_router_bgp,
  73     verify_bgp_community,
  74     verify_bgp_rib,
  75 )
  76 from lib.topojson import build_config_from_json
  77
  78 pytestmark = [pytest.mark.bgpd, pytest.mark.staticd]
  79
  80 # Global variables
  81 NETWORK1_1 = {"ipv4": "11.11.11.1/32", "ipv6": "11:11::1/128"}
  82 NETWORK1_2 = {"ipv4": "11.11.11.11/32", "ipv6": "11:11::11/128"}
  83 NETWORK1_3 = {"ipv4": "10.10.10.1/32", "ipv6": "10:10::1/128"}
  84 NETWORK1_4 = {"ipv4": "10.10.10.100/32", "ipv6": "10:10::100/128"}
  85 NETWORK1_5 = {"ipv4": "110.110.110.1/32", "ipv6": "110:110::1/128"}
  86 NETWORK1_6 = {"ipv4": "110.110.110.100/32", "ipv6": "110:110::100/128"}
  87
  88 NETWORK2_1 = {"ipv4": "22.22.22.2/32", "ipv6": "22:22::2/128"}
  89 NETWORK2_2 = {"ipv4": "22.22.22.22/32", "ipv6": "22:22::22/128"}
  90 NETWORK2_3 = {"ipv4": "20.20.20.20/32", "ipv6": "20:20::20/128"}
  91 NETWORK2_4 = {"ipv4": "20.20.20.200/32", "ipv6": "20:20::200/128"}
  92 NETWORK2_5 = {"ipv4": "220.220.220.20/32", "ipv6": "220:220::20/128"}
  93 NETWORK2_6 = {"ipv4": "220.220.220.200/32", "ipv6": "220:220::200/128"}
  94
  95 NETWORK3_1 = {"ipv4": "30.30.30.3/32", "ipv6": "30:30::3/128"}
  96 NETWORK3_2 = {"ipv4": "30.30.30.30/32", "ipv6": "30:30::30/128"}
  97
  98 PREFIX_LIST = {
  99     "ipv4": ["11.11.11.1", "22.22.22.2", "22.22.22.22"],
 100     "ipv6": ["11:11::1", "22:22::2", "22:22::22"],
 101 }
 102 PREFERRED_NEXT_HOP = "global"
 103 VRF_LIST = ["RED", "BLUE", "GREEN"]
 104 COMM_VAL_1 = "100:100"
 105 COMM_VAL_2 = "500:500"
 106 COMM_VAL_3 = "600:600"
 107
 108
 109 def setup_module(mod):
 110     """
 111     Sets up the pytest environment
 112
 113     * `mod`: module name
 114     """
 115
 116     testsuite_run_time = time.asctime(time.localtime(time.time()))
 117     logger.info("Testsuite start time: {}".format(testsuite_run_time))
 118     logger.info("=" * 40)
 119
 120     logger.info("Running setup_module to create topology")
 121
 122     # This function initiates the topology build with Topogen...
 123     json_file = "{}/bgp_vrf_dynamic_route_leak_topo4.json".format(CWD)
 124     tgen = Topogen(json_file, mod.__name__)
 125     global topo
 126     topo = tgen.json_topo
 127     # ... and here it calls Mininet initialization functions.
 128
 129     # Starting topology, create tmp files which are loaded to routers
 130     #  to start daemons and then start routers
 131     start_topology(tgen)
 132
 133     # Run these tests for kernel version 4.19 or above
 134     if version_cmp(platform.release(), "4.19") < 0:
 135         error_msg = (
 136             "BGP vrf dynamic route leak tests will not run "
 137             '(have kernel "{}", but it requires >= 4.19)'.format(platform.release())
 138         )
 139         pytest.skip(error_msg)
 140
 141     # Creating configuration from JSON
 142     build_config_from_json(tgen, topo)
 143
 144     global BGP_CONVERGENCE
 145     global ADDR_TYPES
 146     ADDR_TYPES = check_address_types()
 147
 148     BGP_CONVERGENCE = verify_bgp_convergence(tgen, topo)
 149     assert BGP_CONVERGENCE is True, "setup_module : Failed \n Error: {}".format(
 150         BGP_CONVERGENCE
 151     )
 152
 153     logger.info("Running setup_module() done")
 154
 155
 156 def teardown_module():
 157     """Teardown the pytest environment"""
 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     logger.info(
 167         "Testsuite end time: {}".format(time.asctime(time.localtime(time.time())))
 168     )
 169     logger.info("=" * 40)
 170
 171
 172 #####################################################
 173 #
 174 #   Testcases
 175 #
 176 #####################################################
 177
 178
 179 def test_dynamic_import_routes_between_tenant_to_default_vrf_p0(request):
 180     """
 181     Verify that with multiple tenant VRFs, dynamic import works fine between
 182     Tenant VRFs to default VRF.
 183
 184     When next-hop IPs and prefixes are same across all VRFs.
 185     When next-hop IPs and prefixes are different across all VRFs.
 186     """
 187
 188     tgen = get_topogen()
 189     tc_name = request.node.name
 190     write_test_header(tc_name)
 191     reset_config_on_routers(tgen)
 192     if tgen.routers_have_failure():
 193         pytest.skip(tgen.errors)
 194
 195     step(
 196         "Configure static routes on R3 for each vrf and redistribute in "
 197         "respective BGP instance"
 198     )
 199
 200     for vrf_name, network in zip(VRF_LIST, [NETWORK1_1, NETWORK2_1, NETWORK3_1]):
 201         step("Configure static route for VRF : {}".format(vrf_name))
 202         for addr_type in ADDR_TYPES:
 203             static_routes = {
 204                 "r3": {
 205                     "static_routes": [
 206                         {
 207                             "network": [network[addr_type]],
 208                             "next_hop": "blackhole",
 209                             "vrf": vrf_name,
 210                         }
 211                     ]
 212                 }
 213             }
 214
 215             result = create_static_routes(tgen, static_routes)
 216             assert result is True, "Testcase {} :Failed \n Error: {}".format(
 217                 tc_name, result
 218             )
 219
 220         step("Redistribute static route on BGP VRF : {}".format(vrf_name))
 221         temp = {}
 222         for addr_type in ADDR_TYPES:
 223             temp.update(
 224                 {addr_type: {"unicast": {"redistribute": [{"redist_type": "static"}]}}}
 225             )
 226
 227         redist_dict = {
 228             "r3": {"bgp": [{"vrf": vrf_name, "local_as": 3, "address_family": temp}]}
 229         }
 230
 231         result = create_router_bgp(tgen, topo, redist_dict)
 232         assert result is True, "Testcase {} :Failed \n Error: {}".format(
 233             tc_name, result
 234         )
 235
 236     for vrf_name, network in zip(VRF_LIST, [NETWORK1_1, NETWORK2_1, NETWORK3_1]):
 237         step(
 238             "Verify that R3 has installed redistributed routes in respective "
 239             "vrfs: {}".format(vrf_name)
 240         )
 241         for addr_type in ADDR_TYPES:
 242             static_routes = {
 243                 "r3": {
 244                     "static_routes": [
 245                         {
 246                             "network": [network[addr_type]],
 247                             "next_hop": "blackhole",
 248                             "vrf": vrf_name,
 249                         }
 250                     ]
 251                 }
 252             }
 253
 254             result = verify_rib(tgen, addr_type, "r3", static_routes)
 255             assert result is True, "Testcase {} : Failed \n Error {}".format(
 256                 tc_name, result
 257             )
 258
 259     step("Import all tenant vrfs(GREEN+BLUE+RED) in default vrf on R3")
 260
 261     for vrf_name in ["RED", "BLUE", "GREEN"]:
 262         temp = {}
 263         for addr_type in ADDR_TYPES:
 264             temp.update({addr_type: {"unicast": {"import": {"vrf": vrf_name}}}})
 265
 266         redist_dict = {"r3": {"bgp": [{"local_as": 3, "address_family": temp}]}}
 267
 268         result = create_router_bgp(tgen, topo, redist_dict)
 269         assert result is True, "Testcase {} :Failed \n Error: {}".format(
 270             tc_name, result
 271         )
 272
 273     step(
 274         "Verify on R3 that it installs all the routes(imported from tenant "
 275         "VRFs) in default vrf. Additionally, verify that R1 & R2 also "
 276         "receive these routes from R3 and install in default vrf, next-hop "
 277         "pointing to R3"
 278     )
 279
 280     for vrf_name, network in zip(VRF_LIST, [NETWORK1_1, NETWORK2_1, NETWORK3_1]):
 281         for addr_type in ADDR_TYPES:
 282             static_routes = {
 283                 "r3": {
 284                     "static_routes": [
 285                         {
 286                             "network": [network[addr_type]],
 287                             "next_hop": "blackhole",
 288                         }
 289                     ]
 290                 }
 291             }
 292
 293             for dut in ["r2", "r1"]:
 294                 next_hop_val = topo["routers"]["r3"]["links"]["{}-link4".format(dut)][
 295                     addr_type
 296                 ].split("/")[0]
 297
 298                 result = verify_bgp_rib(
 299                     tgen, addr_type, dut, static_routes, next_hop=next_hop_val
 300                 )
 301                 assert result is True, "Testcase {} : Failed \n Error {}".format(
 302                     tc_name, result
 303                 )
 304
 305                 result = verify_rib(
 306                     tgen, addr_type, dut, static_routes, next_hop=next_hop_val
 307                 )
 308                 assert result is True, "Testcase {} : Failed \n Error {}".format(
 309                     tc_name, result
 310                 )
 311
 312             result = verify_bgp_rib(tgen, addr_type, "r3", static_routes)
 313             assert result is True, "Testcase {} : Failed \n Error {}".format(
 314                 tc_name, result
 315             )
 316
 317             result = verify_rib(tgen, addr_type, "r3", static_routes)
 318             assert result is True, "Testcase {} : Failed \n Error {}".format(
 319                 tc_name, result
 320             )
 321
 322     for action, value, status in zip(
 323         ["Remove", "Add"], [True, False], ["withdraw", "re-install"]
 324     ):
 325         step(
 326             "{} import vrf GREEN/BLUE/RED/all command from default vrf "
 327             "instance on R3".format(action)
 328         )
 329         for vrf_name in ["RED", "BLUE", "GREEN"]:
 330             temp = {}
 331             for addr_type in ADDR_TYPES:
 332                 temp.update(
 333                     {
 334                         addr_type: {
 335                             "unicast": {"import": {"vrf": vrf_name, "delete": value}}
 336                         }
 337                     }
 338                 )
 339
 340             import_dict = {"r3": {"bgp": [{"local_as": 3, "address_family": temp}]}}
 341
 342             result = create_router_bgp(tgen, topo, import_dict)
 343             assert result is True, "Testcase {} :Failed \n Error: {}".format(
 344                 tc_name, result
 345             )
 346
 347         step(
 348             "Verify that R1,R2 & R3 {} imported routes from GREEN/BLUE/RED/all"
 349             " in default vrf's RIB".format(status)
 350         )
 351         for dut in ["r1", "r2", "r3"]:
 352             for addr_type in ADDR_TYPES:
 353                 static_routes = {
 354                     dut: {
 355                         "static_routes": [
 356                             {
 357                                 "network": [
 358                                     NETWORK1_1[addr_type],
 359                                     NETWORK2_1[addr_type],
 360                                     NETWORK3_1[addr_type],
 361                                 ],
 362                                 "next_hop": "blackhole",
 363                             }
 364                         ]
 365                     }
 366                 }
 367
 368                 if value:
 369                     result = verify_bgp_rib(
 370                         tgen, addr_type, dut, static_routes, expected=False
 371                     )
 372                     assert (
 373                         result is not True
 374                     ), "Testcase {} : Failed \nError {}\n" "Routes {} still in BGP table".format(
 375                         tc_name,
 376                         result,
 377                         static_routes[dut]["static_routes"][0]["network"],
 378                     )
 379
 380                     result = verify_rib(
 381                         tgen, addr_type, dut, static_routes, expected=False
 382                     )
 383                     assert (
 384                         result is not True
 385                     ), "Testcase {} : Failed \nError {}\n" "Routes {} still in BGP table".format(
 386                         tc_name,
 387                         result,
 388                         static_routes[dut]["static_routes"][0]["network"],
 389                     )
 390                 else:
 391                     result = verify_bgp_rib(tgen, addr_type, dut, static_routes)
 392                     assert result is True, "Testcase {} : Failed \n Error {}".format(
 393                         tc_name, result
 394                     )
 395
 396                     result = verify_rib(tgen, addr_type, dut, static_routes)
 397                     assert result is True, "Testcase {} : Failed \n Error {}".format(
 398                         tc_name, result
 399                     )
 400
 401     for action, value in zip(["Shut", "No shut"], [True, False]):
 402         step(
 403             "{} the neighborship between R1-R3 and R1-R2 for vrf RED, GREEN "
 404             "and BLUE".format(action)
 405         )
 406         bgp_disable = {"r3": {"bgp": []}}
 407         for vrf_name in ["RED", "GREEN", "BLUE"]:
 408             temp = {}
 409             for addr_type in ADDR_TYPES:
 410                 temp.update(
 411                     {
 412                         addr_type: {
 413                             "unicast": {
 414                                 "neighbor": {
 415                                     "r1": {
 416                                         "dest_link": {"r3-link4": {"shutdown": value}}
 417                                     },
 418                                     "r2": {
 419                                         "dest_link": {"r3-link4": {"shutdown": value}}
 420                                     },
 421                                 }
 422                             }
 423                         }
 424                     }
 425                 )
 426
 427             bgp_disable["r3"]["bgp"].append(
 428                 {"vrf": vrf_name, "local_as": 3, "address_family": temp}
 429             )
 430         result = create_router_bgp(tgen, topo, bgp_disable)
 431         assert result is True, "Testcase {} :Failed \n Error: {}".format(
 432             tc_name, result
 433         )
 434
 435         step(
 436             "Verify that when peering is shutdown for tenant vrfs, it "
 437             "doesn't impact the RIB/FIB of default vrf on router R1 and R2"
 438         )
 439         for dut in ["r1", "r2"]:
 440             step("Verify RIB/FIB for default vrf on {}".format(dut))
 441             for addr_type in ADDR_TYPES:
 442                 static_routes = {
 443                     dut: {
 444                         "static_routes": [
 445                             {
 446                                 "network": [
 447                                     NETWORK1_1[addr_type],
 448                                     NETWORK2_1[addr_type],
 449                                     NETWORK3_1[addr_type],
 450                                 ],
 451                                 "next_hop": "blackhole",
 452                             }
 453                         ]
 454                     }
 455                 }
 456                 result = verify_bgp_rib(tgen, addr_type, dut, static_routes)
 457                 assert result is True, "Testcase {} : Failed \n Error {}".format(
 458                     tc_name, result
 459                 )
 460
 461                 result = verify_rib(tgen, addr_type, dut, static_routes)
 462                 assert result is True, "Testcase {} : Failed \n Error {}".format(
 463                     tc_name, result
 464                 )
 465
 466     for action, value, status in zip(
 467         ["Shut", "No shut"], [True, False], ["Withdraw", "Reinstall"]
 468     ):
 469         step(
 470             "{} the neighborship between R1-R3 and R2-R3 for default vrf".format(action)
 471         )
 472         temp = {}
 473         for addr_type in ADDR_TYPES:
 474             temp.update(
 475                 {
 476                     addr_type: {
 477                         "unicast": {
 478                             "neighbor": {
 479                                 "r1": {"dest_link": {"r3-link4": {"shutdown": value}}},
 480                                 "r2": {"dest_link": {"r3-link4": {"shutdown": value}}},
 481                             }
 482                         }
 483                     }
 484                 }
 485             )
 486
 487         bgp_disable = {"r3": {"bgp": [{"local_as": 3, "address_family": temp}]}}
 488         result = create_router_bgp(tgen, topo, bgp_disable)
 489         assert result is True, "Testcase {} :Failed \n Error: {}".format(
 490             tc_name, result
 491         )
 492
 493         step(
 494             "Verify that R1 and R2 {} all the routes from default vrf's RIB"
 495             " and FIB".format(status)
 496         )
 497         for dut in ["r1", "r2"]:
 498             step("Verify RIB/FIB for default vrf on {}".format(dut))
 499             for addr_type in ADDR_TYPES:
 500                 static_routes = {
 501                     dut: {
 502                         "static_routes": [
 503                             {
 504                                 "network": [
 505                                     NETWORK1_1[addr_type],
 506                                     NETWORK2_1[addr_type],
 507                                     NETWORK3_1[addr_type],
 508                                 ],
 509                                 "next_hop": "blackhole",
 510                             }
 511                         ]
 512                     }
 513                 }
 514
 515                 if value:
 516                     result = verify_bgp_rib(
 517                         tgen, addr_type, dut, static_routes, expected=False
 518                     )
 519                     assert (
 520                         result is not True
 521                     ), "Testcase {} : Failed \nError {}\n" "Routes {} still in BGP table".format(
 522                         tc_name,
 523                         result,
 524                         static_routes[dut]["static_routes"][0]["network"],
 525                     )
 526
 527                     result = verify_rib(
 528                         tgen, addr_type, dut, static_routes, expected=False
 529                     )
 530                     assert (
 531                         result is not True
 532                     ), "Testcase {} : Failed Error {}" "Routes {} still in Route table".format(
 533                         tc_name,
 534                         result,
 535                         static_routes[dut]["static_routes"][0]["network"],
 536                     )
 537                 else:
 538                     result = verify_bgp_rib(tgen, addr_type, dut, static_routes)
 539                     assert result is True, "Testcase {} : Failed \n Error {}".format(
 540                         tc_name, result
 541                     )
 542
 543                     result = verify_rib(tgen, addr_type, dut, static_routes)
 544                     assert result is True, "Testcase {} : Failed \n Error {}".format(
 545                         tc_name, result
 546                     )
 547
 548     step("Remove import command from router R3 and configure the same on R2")
 549     temp = {}
 550     for vrf_name in VRF_LIST:
 551         for addr_type in ADDR_TYPES:
 552             temp.update(
 553                 {addr_type: {"unicast": {"import": {"vrf": vrf_name, "delete": True}}}}
 554             )
 555
 556         import_dict = {"r3": {"bgp": [{"local_as": 3, "address_family": temp}]}}
 557
 558         result = create_router_bgp(tgen, topo, import_dict)
 559         assert result is True, "Testcase {} :Failed \n Error: {}".format(
 560             tc_name, result
 561         )
 562
 563     step(
 564         "Verify that once import commands are removed from R3, all imported "
 565         "routes are withdrawn from RIB/FIB of default vrf on R1/R2/R3"
 566     )
 567
 568     for dut in ["r1", "r2", "r3"]:
 569         step("Verify RIB/FIB for default vrf on {}".format(dut))
 570         for addr_type in ADDR_TYPES:
 571             static_routes = {
 572                 dut: {
 573                     "static_routes": [
 574                         {
 575                             "network": [
 576                                 NETWORK1_1[addr_type],
 577                                 NETWORK2_1[addr_type],
 578                                 NETWORK3_1[addr_type],
 579                             ],
 580                             "next_hop": "blackhole",
 581                         }
 582                     ]
 583                 }
 584             }
 585             result = verify_bgp_rib(tgen, addr_type, dut, static_routes, expected=False)
 586             assert result is not True, (
 587                 "Testcase {} : Failed \nError {}\n"
 588                 "Routes {} still in BGP table".format(
 589                     tc_name, result, static_routes[dut]["static_routes"][0]["network"]
 590                 )
 591             )
 592
 593             result = verify_rib(tgen, addr_type, dut, static_routes, expected=False)
 594             assert result is not True, "Testcase {} : Failed \n Error {}".format(
 595                 tc_name, result
 596             )
 597
 598     for vrf_name, network in zip(VRF_LIST, [NETWORK1_1, NETWORK2_1, NETWORK3_1]):
 599         step("Configure static route for VRF : {} on r2".format(vrf_name))
 600         for addr_type in ADDR_TYPES:
 601             static_routes = {
 602                 "r2": {
 603                     "static_routes": [
 604                         {
 605                             "network": [network[addr_type]],
 606                             "next_hop": "blackhole",
 607                             "vrf": vrf_name,
 608                         }
 609                     ]
 610                 }
 611             }
 612
 613             result = create_static_routes(tgen, static_routes)
 614             assert result is True, "Testcase {} :Failed \n Error: {}".format(
 615                 tc_name, result
 616             )
 617
 618         step("Redistribute static route on BGP VRF : {}".format(vrf_name))
 619         temp = {}
 620         for addr_type in ADDR_TYPES:
 621             temp.update(
 622                 {addr_type: {"unicast": {"redistribute": [{"redist_type": "static"}]}}}
 623             )
 624
 625         redist_dict = {
 626             "r2": {"bgp": [{"vrf": vrf_name, "local_as": 2, "address_family": temp}]}
 627         }
 628
 629         result = create_router_bgp(tgen, topo, redist_dict)
 630         assert result is True, "Testcase {} :Failed \n Error: {}".format(
 631             tc_name, result
 632         )
 633
 634         step("Remove redistribute static route on BGP VRF : {} on r3".format(vrf_name))
 635         temp = {}
 636         for addr_type in ADDR_TYPES:
 637             temp.update(
 638                 {
 639                     addr_type: {
 640                         "unicast": {
 641                             "redistribute": [{"redist_type": "static", "delete": True}]
 642                         }
 643                     }
 644                 }
 645             )
 646
 647         redist_dict = {
 648             "r3": {"bgp": [{"vrf": vrf_name, "local_as": 3, "address_family": temp}]}
 649         }
 650
 651         result = create_router_bgp(tgen, topo, redist_dict)
 652         assert result is True, "Testcase {} :Failed \n Error: {}".format(
 653             tc_name, result
 654         )
 655
 656     for vrf_name in ["RED", "BLUE", "GREEN"]:
 657         temp = {}
 658         for addr_type in ADDR_TYPES:
 659             temp.update({addr_type: {"unicast": {"import": {"vrf": vrf_name}}}})
 660
 661         import_dict = {"r2": {"bgp": [{"local_as": 2, "address_family": temp}]}}
 662
 663         result = create_router_bgp(tgen, topo, import_dict)
 664         assert result is True, "Testcase {} :Failed \n Error: {}".format(
 665             tc_name, result
 666         )
 667
 668     step(
 669         "Verify after import commands are re-configured on R2's vrf RED, all "
 670         "those routes are installed again in default vrf of R1,R2,R3"
 671     )
 672     for dut in ["r1", "r2", "r3"]:
 673         step("Verify RIB/FIB for vrf RED on {}".format(dut))
 674         for addr_type in ADDR_TYPES:
 675             static_routes = {
 676                 dut: {
 677                     "static_routes": [
 678                         {
 679                             "network": [
 680                                 NETWORK1_1[addr_type],
 681                                 NETWORK2_1[addr_type],
 682                                 NETWORK3_1[addr_type],
 683                             ],
 684                             "next_hop": "blackhole",
 685                         }
 686                     ]
 687                 }
 688             }
 689             result = verify_bgp_rib(tgen, addr_type, dut, static_routes)
 690             assert result is True, "Testcase {} : Failed \n Error {}".format(
 691                 tc_name, result
 692             )
 693
 694             result = verify_rib(tgen, addr_type, dut, static_routes)
 695             assert result is True, "Testcase {} : Failed \n Error {}".format(
 696                 tc_name, result
 697             )
 698
 699     step("Remove import vrf RED/GREEN/BLUE/all one by one from default vrf" " on R2")
 700     for vrf_name in ["RED", "BLUE", "GREEN"]:
 701         temp = {}
 702         for addr_type in ADDR_TYPES:
 703             temp.update(
 704                 {addr_type: {"unicast": {"import": {"vrf": vrf_name, "delete": True}}}}
 705             )
 706
 707         import_dict = {"r2": {"bgp": [{"local_as": 2, "address_family": temp}]}}
 708
 709         result = create_router_bgp(tgen, topo, import_dict)
 710         assert result is True, "Testcase {} :Failed \n Error: {}".format(
 711             tc_name, result
 712         )
 713
 714     step(
 715         "Verify that R1, R2 and R3 withdraws imported routes from default "
 716         "vrf's RIB and FIB "
 717     )
 718     for dut in ["r1", "r2", "r3"]:
 719         for addr_type in ADDR_TYPES:
 720             static_routes = {
 721                 dut: {
 722                     "static_routes": [
 723                         {
 724                             "network": [
 725                                 NETWORK1_1[addr_type],
 726                                 NETWORK2_1[addr_type],
 727                                 NETWORK3_1[addr_type],
 728                             ],
 729                             "next_hop": "blackhole",
 730                         }
 731                     ]
 732                 }
 733             }
 734             result = verify_bgp_rib(tgen, addr_type, dut, static_routes, expected=False)
 735             assert result is not True, (
 736                 "Testcase {} : Failed \nError {}\n"
 737                 "Routes {} still in BGP table".format(
 738                     tc_name, result, static_routes[dut]["static_routes"][0]["network"]
 739                 )
 740             )
 741
 742             result = verify_rib(tgen, addr_type, dut, static_routes, expected=False)
 743             assert result is not True, "Testcase {} : Failed \n Error {}".format(
 744                 tc_name, result
 745             )
 746
 747     step("Add import vrf RED/GREEN/BLUE/all one by one from default vrf on R2")
 748     for vrf_name in ["RED", "BLUE", "GREEN"]:
 749         temp = {}
 750         for addr_type in ADDR_TYPES:
 751             temp.update({addr_type: {"unicast": {"import": {"vrf": vrf_name}}}})
 752
 753         import_dict = {"r2": {"bgp": [{"local_as": 2, "address_family": temp}]}}
 754
 755         result = create_router_bgp(tgen, topo, import_dict)
 756         assert result is True, "Testcase {} :Failed \n Error: {}".format(
 757             tc_name, result
 758         )
 759
 760     for dut in ["r1", "r2", "r3"]:
 761         step("Verify that {} reinstall imported routes from vrf RED's RIB".format(dut))
 762         for addr_type in ADDR_TYPES:
 763             static_routes = {
 764                 dut: {
 765                     "static_routes": [
 766                         {
 767                             "network": [
 768                                 NETWORK1_1[addr_type],
 769                                 NETWORK2_1[addr_type],
 770                                 NETWORK3_1[addr_type],
 771                             ],
 772                             "next_hop": "blackhole",
 773                         }
 774                     ]
 775                 }
 776             }
 777             result = verify_bgp_rib(tgen, addr_type, dut, static_routes)
 778             assert result is True, "Testcase {} : Failed \n Error {}".format(
 779                 tc_name, result
 780             )
 781
 782             result = verify_rib(tgen, addr_type, dut, static_routes)
 783             assert result is True, "Testcase {} : Failed \n Error {}".format(
 784                 tc_name, result
 785             )
 786
 787     for action, value in zip(["Shut", "No shut"], [True, False]):
 788         step(
 789             "{} the neighborship between R2-R3 for vrf GREEN, BLUE and RED".format(
 790                 action
 791             )
 792         )
 793         bgp_disable = {"r2": {"bgp": []}}
 794         for vrf_name in ["GREEN", "BLUE", "RED"]:
 795             temp = {}
 796             for addr_type in ADDR_TYPES:
 797                 temp.update(
 798                     {
 799                         addr_type: {
 800                             "unicast": {
 801                                 "neighbor": {
 802                                     "r3": {
 803                                         "dest_link": {"r2-link4": {"shutdown": value}}
 804                                     }
 805                                 }
 806                             }
 807                         }
 808                     }
 809                 )
 810
 811             bgp_disable["r2"]["bgp"].append(
 812                 {"vrf": vrf_name, "local_as": 2, "address_family": temp}
 813             )
 814         result = create_router_bgp(tgen, topo, bgp_disable)
 815         assert result is True, "Testcase {} :Failed \n Error: {}".format(
 816             tc_name, result
 817         )
 818
 819         step("Verify RIB/FIB of vrf RED will be unchanged on all 3 routers")
 820         for dut in ["r1", "r2", "r3"]:
 821             step("Verify RIB/FIB for vrf RED on {}".format(dut))
 822             for addr_type in ADDR_TYPES:
 823                 static_routes = {
 824                     dut: {
 825                         "static_routes": [
 826                             {
 827                                 "network": [
 828                                     NETWORK1_1[addr_type],
 829                                     NETWORK2_1[addr_type],
 830                                     NETWORK3_1[addr_type],
 831                                 ],
 832                                 "next_hop": "blackhole",
 833                             }
 834                         ]
 835                     }
 836                 }
 837
 838                 result = verify_bgp_rib(tgen, addr_type, dut, static_routes)
 839                 assert result is True, "Testcase {} : Failed \n Error {}".format(
 840                     tc_name, result
 841                 )
 842
 843                 result = verify_rib(tgen, addr_type, dut, static_routes)
 844                 assert result is True, "Testcase {} : Failed \n Error {}".format(
 845                     tc_name, result
 846                 )
 847
 848     for action, value, status in zip(
 849         ["Shut", "No shut"], [True, False], ["Withdraw", "Reinstall"]
 850     ):
 851         step("{} the neighborship between R2-R3 for default vrf".format(action))
 852         temp = {}
 853         for addr_type in ADDR_TYPES:
 854             temp.update(
 855                 {
 856                     addr_type: {
 857                         "unicast": {
 858                             "neighbor": {
 859                                 "r3": {"dest_link": {"r2-link4": {"shutdown": value}}}
 860                             }
 861                         }
 862                     }
 863                 }
 864             )
 865
 866         bgp_disable = {"r2": {"bgp": [{"local_as": 2, "address_family": temp}]}}
 867         result = create_router_bgp(tgen, topo, bgp_disable)
 868         assert result is True, "Testcase {} :Failed \n Error: {}".format(
 869             tc_name, result
 870         )
 871
 872         step(
 873             "Verify that R1 and R2 {} all the routes from default vrfs RIB and"
 874             " FIB".format(status)
 875         )
 876         for dut in ["r1", "r3"]:
 877             step("Verify RIB/FIB for default vrf on {}".format(dut))
 878             for addr_type in ADDR_TYPES:
 879                 static_routes = {
 880                     dut: {
 881                         "static_routes": [
 882                             {
 883                                 "network": [
 884                                     NETWORK1_1[addr_type],
 885                                     NETWORK2_1[addr_type],
 886                                     NETWORK3_1[addr_type],
 887                                 ],
 888                                 "next_hop": "blackhole",
 889                             }
 890                         ]
 891                     }
 892                 }
 893
 894                 if value:
 895                     result = verify_bgp_rib(
 896                         tgen, addr_type, dut, static_routes, expected=False
 897                     )
 898                     assert (
 899                         result is not True
 900                     ), "Testcase {} : Failed \nError {}\n" "Routes {} still in BGP table".format(
 901                         tc_name,
 902                         result,
 903                         static_routes[dut]["static_routes"][0]["network"],
 904                     )
 905
 906                     result = verify_rib(
 907                         tgen, addr_type, dut, static_routes, expected=False
 908                     )
 909                     assert (
 910                         result is not True
 911                     ), "Testcase {} : Failed Error {}" "Routes {} still in Route table".format(
 912                         tc_name,
 913                         result,
 914                         static_routes[dut]["static_routes"][0]["network"],
 915                     )
 916                 else:
 917                     result = verify_bgp_rib(tgen, addr_type, dut, static_routes)
 918                     assert result is True, "Testcase {} : Failed \n Error {}".format(
 919                         tc_name, result
 920                     )
 921
 922                     result = verify_rib(tgen, addr_type, dut, static_routes)
 923                     assert result is True, "Testcase {} : Failed \n Error {}".format(
 924                         tc_name, result
 925                     )
 926
 927     write_test_footer(tc_name)
 928
 929
 930 if __name__ == "__main__":
 931     args = ["-s"] + sys.argv[1:]
 932     sys.exit(pytest.main(args))