tests/topotests/bgp_vrf_dynamic_route_leak_topo4/test_bgp_vrf_dynamic_route_leak_topo4-2.py

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