tests/topotests/bgp_vrf_dynamic_route_leak_topo4/test_bgp_vrf_dynamic_route_leak_topo4-3.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_tenant_to_default_vrf_p0(request):
 167     """
 168     Verify that with multiple tenant VRFs, dynamic import works fine between
 169     Tenant VRFs to default VRF.
 170
 171     When next-hop IPs and prefixes are same across all VRFs.
 172     When next-hop IPs and prefixes are different across all VRFs.
 173     """
 174
 175     tgen = get_topogen()
 176     tc_name = request.node.name
 177     write_test_header(tc_name)
 178     reset_config_on_routers(tgen)
 179     if tgen.routers_have_failure():
 180         pytest.skip(tgen.errors)
 181
 182     step(
 183         "Configure static routes on R3 for each vrf and redistribute in "
 184         "respective BGP instance"
 185     )
 186
 187     for vrf_name, network in zip(VRF_LIST, [NETWORK1_1, NETWORK2_1, NETWORK3_1]):
 188         step("Configure static route for VRF : {}".format(vrf_name))
 189         for addr_type in ADDR_TYPES:
 190             static_routes = {
 191                 "r3": {
 192                     "static_routes": [
 193                         {
 194                             "network": [network[addr_type]],
 195                             "next_hop": "blackhole",
 196                             "vrf": vrf_name,
 197                         }
 198                     ]
 199                 }
 200             }
 201
 202             result = create_static_routes(tgen, static_routes)
 203             assert result is True, "Testcase {} :Failed \n Error: {}".format(
 204                 tc_name, result
 205             )
 206
 207         step("Redistribute static route on BGP VRF : {}".format(vrf_name))
 208         temp = {}
 209         for addr_type in ADDR_TYPES:
 210             temp.update(
 211                 {addr_type: {"unicast": {"redistribute": [{"redist_type": "static"}]}}}
 212             )
 213
 214         redist_dict = {
 215             "r3": {"bgp": [{"vrf": vrf_name, "local_as": 3, "address_family": temp}]}
 216         }
 217
 218         result = create_router_bgp(tgen, topo, redist_dict)
 219         assert result is True, "Testcase {} :Failed \n Error: {}".format(
 220             tc_name, result
 221         )
 222
 223     for vrf_name, network in zip(VRF_LIST, [NETWORK1_1, NETWORK2_1, NETWORK3_1]):
 224         step(
 225             "Verify that R3 has installed redistributed routes in respective "
 226             "vrfs: {}".format(vrf_name)
 227         )
 228         for addr_type in ADDR_TYPES:
 229             static_routes = {
 230                 "r3": {
 231                     "static_routes": [
 232                         {
 233                             "network": [network[addr_type]],
 234                             "next_hop": "blackhole",
 235                             "vrf": vrf_name,
 236                         }
 237                     ]
 238                 }
 239             }
 240
 241             result = verify_rib(tgen, addr_type, "r3", static_routes)
 242             assert result is True, "Testcase {} : Failed \n Error {}".format(
 243                 tc_name, result
 244             )
 245
 246     step("Import all tenant vrfs(GREEN+BLUE+RED) in default vrf on R3")
 247
 248     for vrf_name in ["RED", "BLUE", "GREEN"]:
 249         temp = {}
 250         for addr_type in ADDR_TYPES:
 251             temp.update({addr_type: {"unicast": {"import": {"vrf": vrf_name}}}})
 252
 253         redist_dict = {"r3": {"bgp": [{"local_as": 3, "address_family": temp}]}}
 254
 255         result = create_router_bgp(tgen, topo, redist_dict)
 256         assert result is True, "Testcase {} :Failed \n Error: {}".format(
 257             tc_name, result
 258         )
 259
 260     step(
 261         "Verify on R3 that it installs all the routes(imported from tenant "
 262         "VRFs) in default vrf. Additionally, verify that R1 & R2 also "
 263         "receive these routes from R3 and install in default vrf, next-hop "
 264         "pointing to R3"
 265     )
 266
 267     for vrf_name, network in zip(VRF_LIST, [NETWORK1_1, NETWORK2_1, NETWORK3_1]):
 268         for addr_type in ADDR_TYPES:
 269             static_routes = {
 270                 "r3": {
 271                     "static_routes": [
 272                         {
 273                             "network": [network[addr_type]],
 274                             "next_hop": "blackhole",
 275                         }
 276                     ]
 277                 }
 278             }
 279
 280             for dut in ["r2", "r1"]:
 281                 next_hop_val = topo["routers"]["r3"]["links"]["{}-link4".format(dut)][
 282                     addr_type
 283                 ].split("/")[0]
 284
 285                 result = verify_bgp_rib(
 286                     tgen, addr_type, dut, static_routes, next_hop=next_hop_val
 287                 )
 288                 assert result is True, "Testcase {} : Failed \n Error {}".format(
 289                     tc_name, result
 290                 )
 291
 292                 result = verify_rib(
 293                     tgen, addr_type, dut, static_routes, next_hop=next_hop_val
 294                 )
 295                 assert result is True, "Testcase {} : Failed \n Error {}".format(
 296                     tc_name, result
 297                 )
 298
 299             result = verify_bgp_rib(tgen, addr_type, "r3", static_routes)
 300             assert result is True, "Testcase {} : Failed \n Error {}".format(
 301                 tc_name, result
 302             )
 303
 304             result = verify_rib(tgen, addr_type, "r3", static_routes)
 305             assert result is True, "Testcase {} : Failed \n Error {}".format(
 306                 tc_name, result
 307             )
 308
 309     for action, value, status in zip(
 310         ["Remove", "Add"], [True, False], ["withdraw", "re-install"]
 311     ):
 312         step(
 313             "{} import vrf GREEN/BLUE/RED/all command from default vrf "
 314             "instance on R3".format(action)
 315         )
 316         for vrf_name in ["RED", "BLUE", "GREEN"]:
 317             temp = {}
 318             for addr_type in ADDR_TYPES:
 319                 temp.update(
 320                     {
 321                         addr_type: {
 322                             "unicast": {"import": {"vrf": vrf_name, "delete": value}}
 323                         }
 324                     }
 325                 )
 326
 327             import_dict = {"r3": {"bgp": [{"local_as": 3, "address_family": temp}]}}
 328
 329             result = create_router_bgp(tgen, topo, import_dict)
 330             assert result is True, "Testcase {} :Failed \n Error: {}".format(
 331                 tc_name, result
 332             )
 333
 334         step(
 335             "Verify that R1,R2 & R3 {} imported routes from GREEN/BLUE/RED/all"
 336             " in default vrf's RIB".format(status)
 337         )
 338         for dut in ["r1", "r2", "r3"]:
 339             for addr_type in ADDR_TYPES:
 340                 static_routes = {
 341                     dut: {
 342                         "static_routes": [
 343                             {
 344                                 "network": [
 345                                     NETWORK1_1[addr_type],
 346                                     NETWORK2_1[addr_type],
 347                                     NETWORK3_1[addr_type],
 348                                 ],
 349                                 "next_hop": "blackhole",
 350                             }
 351                         ]
 352                     }
 353                 }
 354
 355                 if value:
 356                     result = verify_bgp_rib(
 357                         tgen, addr_type, dut, static_routes, expected=False
 358                     )
 359                     assert (
 360                         result is not True
 361                     ), "Testcase {} : Failed \nError {}\n" "Routes {} still in BGP table".format(
 362                         tc_name,
 363                         result,
 364                         static_routes[dut]["static_routes"][0]["network"],
 365                     )
 366
 367                     result = verify_rib(
 368                         tgen, addr_type, dut, static_routes, expected=False
 369                     )
 370                     assert (
 371                         result is not True
 372                     ), "Testcase {} : Failed \nError {}\n" "Routes {} still in BGP table".format(
 373                         tc_name,
 374                         result,
 375                         static_routes[dut]["static_routes"][0]["network"],
 376                     )
 377                 else:
 378                     result = verify_bgp_rib(tgen, addr_type, dut, static_routes)
 379                     assert result is True, "Testcase {} : Failed \n Error {}".format(
 380                         tc_name, result
 381                     )
 382
 383                     result = verify_rib(tgen, addr_type, dut, static_routes)
 384                     assert result is True, "Testcase {} : Failed \n Error {}".format(
 385                         tc_name, result
 386                     )
 387
 388     for action, value in zip(["Shut", "No shut"], [True, False]):
 389         step(
 390             "{} the neighborship between R1-R3 and R1-R2 for vrf RED, GREEN "
 391             "and BLUE".format(action)
 392         )
 393         bgp_disable = {"r3": {"bgp": []}}
 394         for vrf_name in ["RED", "GREEN", "BLUE"]:
 395             temp = {}
 396             for addr_type in ADDR_TYPES:
 397                 temp.update(
 398                     {
 399                         addr_type: {
 400                             "unicast": {
 401                                 "neighbor": {
 402                                     "r1": {
 403                                         "dest_link": {"r3-link4": {"shutdown": value}}
 404                                     },
 405                                     "r2": {
 406                                         "dest_link": {"r3-link4": {"shutdown": value}}
 407                                     },
 408                                 }
 409                             }
 410                         }
 411                     }
 412                 )
 413
 414             bgp_disable["r3"]["bgp"].append(
 415                 {"vrf": vrf_name, "local_as": 3, "address_family": temp}
 416             )
 417         result = create_router_bgp(tgen, topo, bgp_disable)
 418         assert result is True, "Testcase {} :Failed \n Error: {}".format(
 419             tc_name, result
 420         )
 421
 422         step(
 423             "Verify that when peering is shutdown for tenant vrfs, it "
 424             "doesn't impact the RIB/FIB of default vrf on router R1 and R2"
 425         )
 426         for dut in ["r1", "r2"]:
 427             step("Verify RIB/FIB for default vrf on {}".format(dut))
 428             for addr_type in ADDR_TYPES:
 429                 static_routes = {
 430                     dut: {
 431                         "static_routes": [
 432                             {
 433                                 "network": [
 434                                     NETWORK1_1[addr_type],
 435                                     NETWORK2_1[addr_type],
 436                                     NETWORK3_1[addr_type],
 437                                 ],
 438                                 "next_hop": "blackhole",
 439                             }
 440                         ]
 441                     }
 442                 }
 443                 result = verify_bgp_rib(tgen, addr_type, dut, static_routes)
 444                 assert result is True, "Testcase {} : Failed \n Error {}".format(
 445                     tc_name, result
 446                 )
 447
 448                 result = verify_rib(tgen, addr_type, dut, static_routes)
 449                 assert result is True, "Testcase {} : Failed \n Error {}".format(
 450                     tc_name, result
 451                 )
 452
 453     for action, value, status in zip(
 454         ["Shut", "No shut"], [True, False], ["Withdraw", "Reinstall"]
 455     ):
 456         step(
 457             "{} the neighborship between R1-R3 and R2-R3 for default vrf".format(action)
 458         )
 459         temp = {}
 460         for addr_type in ADDR_TYPES:
 461             temp.update(
 462                 {
 463                     addr_type: {
 464                         "unicast": {
 465                             "neighbor": {
 466                                 "r1": {"dest_link": {"r3-link4": {"shutdown": value}}},
 467                                 "r2": {"dest_link": {"r3-link4": {"shutdown": value}}},
 468                             }
 469                         }
 470                     }
 471                 }
 472             )
 473
 474         bgp_disable = {"r3": {"bgp": [{"local_as": 3, "address_family": temp}]}}
 475         result = create_router_bgp(tgen, topo, bgp_disable)
 476         assert result is True, "Testcase {} :Failed \n Error: {}".format(
 477             tc_name, result
 478         )
 479
 480         step(
 481             "Verify that R1 and R2 {} all the routes from default vrf's RIB"
 482             " and FIB".format(status)
 483         )
 484         for dut in ["r1", "r2"]:
 485             step("Verify RIB/FIB for default vrf on {}".format(dut))
 486             for addr_type in ADDR_TYPES:
 487                 static_routes = {
 488                     dut: {
 489                         "static_routes": [
 490                             {
 491                                 "network": [
 492                                     NETWORK1_1[addr_type],
 493                                     NETWORK2_1[addr_type],
 494                                     NETWORK3_1[addr_type],
 495                                 ],
 496                                 "next_hop": "blackhole",
 497                             }
 498                         ]
 499                     }
 500                 }
 501
 502                 if value:
 503                     result = verify_bgp_rib(
 504                         tgen, addr_type, dut, static_routes, expected=False
 505                     )
 506                     assert (
 507                         result is not True
 508                     ), "Testcase {} : Failed \nError {}\n" "Routes {} still in BGP table".format(
 509                         tc_name,
 510                         result,
 511                         static_routes[dut]["static_routes"][0]["network"],
 512                     )
 513
 514                     result = verify_rib(
 515                         tgen, addr_type, dut, static_routes, expected=False
 516                     )
 517                     assert (
 518                         result is not True
 519                     ), "Testcase {} : Failed Error {}" "Routes {} still in Route table".format(
 520                         tc_name,
 521                         result,
 522                         static_routes[dut]["static_routes"][0]["network"],
 523                     )
 524                 else:
 525                     result = verify_bgp_rib(tgen, addr_type, dut, static_routes)
 526                     assert result is True, "Testcase {} : Failed \n Error {}".format(
 527                         tc_name, result
 528                     )
 529
 530                     result = verify_rib(tgen, addr_type, dut, static_routes)
 531                     assert result is True, "Testcase {} : Failed \n Error {}".format(
 532                         tc_name, result
 533                     )
 534
 535     step("Remove import command from router R3 and configure the same on R2")
 536     temp = {}
 537     for vrf_name in VRF_LIST:
 538         for addr_type in ADDR_TYPES:
 539             temp.update(
 540                 {addr_type: {"unicast": {"import": {"vrf": vrf_name, "delete": True}}}}
 541             )
 542
 543         import_dict = {"r3": {"bgp": [{"local_as": 3, "address_family": temp}]}}
 544
 545         result = create_router_bgp(tgen, topo, import_dict)
 546         assert result is True, "Testcase {} :Failed \n Error: {}".format(
 547             tc_name, result
 548         )
 549
 550     step(
 551         "Verify that once import commands are removed from R3, all imported "
 552         "routes are withdrawn from RIB/FIB of default vrf on R1/R2/R3"
 553     )
 554
 555     for dut in ["r1", "r2", "r3"]:
 556         step("Verify RIB/FIB for default vrf on {}".format(dut))
 557         for addr_type in ADDR_TYPES:
 558             static_routes = {
 559                 dut: {
 560                     "static_routes": [
 561                         {
 562                             "network": [
 563                                 NETWORK1_1[addr_type],
 564                                 NETWORK2_1[addr_type],
 565                                 NETWORK3_1[addr_type],
 566                             ],
 567                             "next_hop": "blackhole",
 568                         }
 569                     ]
 570                 }
 571             }
 572             result = verify_bgp_rib(tgen, addr_type, dut, static_routes, expected=False)
 573             assert result is not True, (
 574                 "Testcase {} : Failed \nError {}\n"
 575                 "Routes {} still in BGP table".format(
 576                     tc_name, result, static_routes[dut]["static_routes"][0]["network"]
 577                 )
 578             )
 579
 580             result = verify_rib(tgen, addr_type, dut, static_routes, expected=False)
 581             assert result is not True, "Testcase {} : Failed \n Error {}".format(
 582                 tc_name, result
 583             )
 584
 585     for vrf_name, network in zip(VRF_LIST, [NETWORK1_1, NETWORK2_1, NETWORK3_1]):
 586         step("Configure static route for VRF : {} on r2".format(vrf_name))
 587         for addr_type in ADDR_TYPES:
 588             static_routes = {
 589                 "r2": {
 590                     "static_routes": [
 591                         {
 592                             "network": [network[addr_type]],
 593                             "next_hop": "blackhole",
 594                             "vrf": vrf_name,
 595                         }
 596                     ]
 597                 }
 598             }
 599
 600             result = create_static_routes(tgen, static_routes)
 601             assert result is True, "Testcase {} :Failed \n Error: {}".format(
 602                 tc_name, result
 603             )
 604
 605         step("Redistribute static route on BGP VRF : {}".format(vrf_name))
 606         temp = {}
 607         for addr_type in ADDR_TYPES:
 608             temp.update(
 609                 {addr_type: {"unicast": {"redistribute": [{"redist_type": "static"}]}}}
 610             )
 611
 612         redist_dict = {
 613             "r2": {"bgp": [{"vrf": vrf_name, "local_as": 2, "address_family": temp}]}
 614         }
 615
 616         result = create_router_bgp(tgen, topo, redist_dict)
 617         assert result is True, "Testcase {} :Failed \n Error: {}".format(
 618             tc_name, result
 619         )
 620
 621         step("Remove redistribute static route on BGP VRF : {} on r3".format(vrf_name))
 622         temp = {}
 623         for addr_type in ADDR_TYPES:
 624             temp.update(
 625                 {
 626                     addr_type: {
 627                         "unicast": {
 628                             "redistribute": [{"redist_type": "static", "delete": True}]
 629                         }
 630                     }
 631                 }
 632             )
 633
 634         redist_dict = {
 635             "r3": {"bgp": [{"vrf": vrf_name, "local_as": 3, "address_family": temp}]}
 636         }
 637
 638         result = create_router_bgp(tgen, topo, redist_dict)
 639         assert result is True, "Testcase {} :Failed \n Error: {}".format(
 640             tc_name, result
 641         )
 642
 643     for vrf_name in ["RED", "BLUE", "GREEN"]:
 644         temp = {}
 645         for addr_type in ADDR_TYPES:
 646             temp.update({addr_type: {"unicast": {"import": {"vrf": vrf_name}}}})
 647
 648         import_dict = {"r2": {"bgp": [{"local_as": 2, "address_family": temp}]}}
 649
 650         result = create_router_bgp(tgen, topo, import_dict)
 651         assert result is True, "Testcase {} :Failed \n Error: {}".format(
 652             tc_name, result
 653         )
 654
 655     step(
 656         "Verify after import commands are re-configured on R2's vrf RED, all "
 657         "those routes are installed again in default vrf of R1,R2,R3"
 658     )
 659     for dut in ["r1", "r2", "r3"]:
 660         step("Verify RIB/FIB for vrf RED on {}".format(dut))
 661         for addr_type in ADDR_TYPES:
 662             static_routes = {
 663                 dut: {
 664                     "static_routes": [
 665                         {
 666                             "network": [
 667                                 NETWORK1_1[addr_type],
 668                                 NETWORK2_1[addr_type],
 669                                 NETWORK3_1[addr_type],
 670                             ],
 671                             "next_hop": "blackhole",
 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("Remove import vrf RED/GREEN/BLUE/all one by one from default vrf" " on R2")
 687     for vrf_name in ["RED", "BLUE", "GREEN"]:
 688         temp = {}
 689         for addr_type in ADDR_TYPES:
 690             temp.update(
 691                 {addr_type: {"unicast": {"import": {"vrf": vrf_name, "delete": True}}}}
 692             )
 693
 694         import_dict = {"r2": {"bgp": [{"local_as": 2, "address_family": temp}]}}
 695
 696         result = create_router_bgp(tgen, topo, import_dict)
 697         assert result is True, "Testcase {} :Failed \n Error: {}".format(
 698             tc_name, result
 699         )
 700
 701     step(
 702         "Verify that R1, R2 and R3 withdraws imported routes from default "
 703         "vrf's RIB and FIB "
 704     )
 705     for dut in ["r1", "r2", "r3"]:
 706         for addr_type in ADDR_TYPES:
 707             static_routes = {
 708                 dut: {
 709                     "static_routes": [
 710                         {
 711                             "network": [
 712                                 NETWORK1_1[addr_type],
 713                                 NETWORK2_1[addr_type],
 714                                 NETWORK3_1[addr_type],
 715                             ],
 716                             "next_hop": "blackhole",
 717                         }
 718                     ]
 719                 }
 720             }
 721             result = verify_bgp_rib(tgen, addr_type, dut, static_routes, expected=False)
 722             assert result is not True, (
 723                 "Testcase {} : Failed \nError {}\n"
 724                 "Routes {} still in BGP table".format(
 725                     tc_name, result, static_routes[dut]["static_routes"][0]["network"]
 726                 )
 727             )
 728
 729             result = verify_rib(tgen, addr_type, dut, static_routes, expected=False)
 730             assert result is not True, "Testcase {} : Failed \n Error {}".format(
 731                 tc_name, result
 732             )
 733
 734     step("Add import vrf RED/GREEN/BLUE/all one by one from default vrf on R2")
 735     for vrf_name in ["RED", "BLUE", "GREEN"]:
 736         temp = {}
 737         for addr_type in ADDR_TYPES:
 738             temp.update({addr_type: {"unicast": {"import": {"vrf": vrf_name}}}})
 739
 740         import_dict = {"r2": {"bgp": [{"local_as": 2, "address_family": temp}]}}
 741
 742         result = create_router_bgp(tgen, topo, import_dict)
 743         assert result is True, "Testcase {} :Failed \n Error: {}".format(
 744             tc_name, result
 745         )
 746
 747     for dut in ["r1", "r2", "r3"]:
 748         step("Verify that {} reinstall imported routes from vrf RED's RIB".format(dut))
 749         for addr_type in ADDR_TYPES:
 750             static_routes = {
 751                 dut: {
 752                     "static_routes": [
 753                         {
 754                             "network": [
 755                                 NETWORK1_1[addr_type],
 756                                 NETWORK2_1[addr_type],
 757                                 NETWORK3_1[addr_type],
 758                             ],
 759                             "next_hop": "blackhole",
 760                         }
 761                     ]
 762                 }
 763             }
 764             result = verify_bgp_rib(tgen, addr_type, dut, static_routes)
 765             assert result is True, "Testcase {} : Failed \n Error {}".format(
 766                 tc_name, result
 767             )
 768
 769             result = verify_rib(tgen, addr_type, dut, static_routes)
 770             assert result is True, "Testcase {} : Failed \n Error {}".format(
 771                 tc_name, result
 772             )
 773
 774     for action, value in zip(["Shut", "No shut"], [True, False]):
 775         step(
 776             "{} the neighborship between R2-R3 for vrf GREEN, BLUE and RED".format(
 777                 action
 778             )
 779         )
 780         bgp_disable = {"r2": {"bgp": []}}
 781         for vrf_name in ["GREEN", "BLUE", "RED"]:
 782             temp = {}
 783             for addr_type in ADDR_TYPES:
 784                 temp.update(
 785                     {
 786                         addr_type: {
 787                             "unicast": {
 788                                 "neighbor": {
 789                                     "r3": {
 790                                         "dest_link": {"r2-link4": {"shutdown": value}}
 791                                     }
 792                                 }
 793                             }
 794                         }
 795                     }
 796                 )
 797
 798             bgp_disable["r2"]["bgp"].append(
 799                 {"vrf": vrf_name, "local_as": 2, "address_family": temp}
 800             )
 801         result = create_router_bgp(tgen, topo, bgp_disable)
 802         assert result is True, "Testcase {} :Failed \n Error: {}".format(
 803             tc_name, result
 804         )
 805
 806         step("Verify RIB/FIB of vrf RED will be unchanged on all 3 routers")
 807         for dut in ["r1", "r2", "r3"]:
 808             step("Verify RIB/FIB for vrf RED on {}".format(dut))
 809             for addr_type in ADDR_TYPES:
 810                 static_routes = {
 811                     dut: {
 812                         "static_routes": [
 813                             {
 814                                 "network": [
 815                                     NETWORK1_1[addr_type],
 816                                     NETWORK2_1[addr_type],
 817                                     NETWORK3_1[addr_type],
 818                                 ],
 819                                 "next_hop": "blackhole",
 820                             }
 821                         ]
 822                     }
 823                 }
 824
 825                 result = verify_bgp_rib(tgen, addr_type, dut, static_routes)
 826                 assert result is True, "Testcase {} : Failed \n Error {}".format(
 827                     tc_name, result
 828                 )
 829
 830                 result = verify_rib(tgen, addr_type, dut, static_routes)
 831                 assert result is True, "Testcase {} : Failed \n Error {}".format(
 832                     tc_name, result
 833                 )
 834
 835     for action, value, status in zip(
 836         ["Shut", "No shut"], [True, False], ["Withdraw", "Reinstall"]
 837     ):
 838         step("{} the neighborship between R2-R3 for default vrf".format(action))
 839         temp = {}
 840         for addr_type in ADDR_TYPES:
 841             temp.update(
 842                 {
 843                     addr_type: {
 844                         "unicast": {
 845                             "neighbor": {
 846                                 "r3": {"dest_link": {"r2-link4": {"shutdown": value}}}
 847                             }
 848                         }
 849                     }
 850                 }
 851             )
 852
 853         bgp_disable = {"r2": {"bgp": [{"local_as": 2, "address_family": temp}]}}
 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 default vrfs RIB and"
 861             " FIB".format(status)
 862         )
 863         for dut in ["r1", "r3"]:
 864             step("Verify RIB/FIB for default vrf on {}".format(dut))
 865             for addr_type in ADDR_TYPES:
 866                 static_routes = {
 867                     dut: {
 868                         "static_routes": [
 869                             {
 870                                 "network": [
 871                                     NETWORK1_1[addr_type],
 872                                     NETWORK2_1[addr_type],
 873                                     NETWORK3_1[addr_type],
 874                                 ],
 875                                 "next_hop": "blackhole",
 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))