tests/topotests/bgp_vrf_lite_best_path_test/test_bgp_vrf_lite_best_path_topo2.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 VRF Lite:
  25 1. Verify that locally imported routes are selected as best path over eBGP imported routes
  26     peers.
  27 2. Verify ECMP for imported routes from different VRFs.
  28 """
  29
  30 import os
  31 import sys
  32 import time
  33 import pytest
  34 import platform
  35 from time import sleep
  36
  37 # Save the Current Working Directory to find configuration files.
  38 CWD = os.path.dirname(os.path.realpath(__file__))
  39 sys.path.append(os.path.join(CWD, "../"))
  40 sys.path.append(os.path.join(CWD, "../lib/"))
  41
  42 # Required to instantiate the topology builder class.
  43
  44 # pylint: disable=C0413
  45 # Import topogen and topotest helpers
  46 from lib.topogen import Topogen, get_topogen
  47 from lib.topotest import version_cmp
  48
  49 from lib.common_config import (
  50     start_topology,
  51     write_test_header,
  52     check_address_types,
  53     write_test_footer,
  54     reset_config_on_routers,
  55     verify_rib,
  56     step,
  57     create_static_routes,
  58     check_router_status,
  59     apply_raw_config
  60 )
  61
  62 from lib.topolog import logger
  63 from lib.bgp import (
  64     verify_bgp_convergence,
  65     create_router_bgp,
  66     verify_bgp_rib,
  67     verify_bgp_bestpath
  68 )
  69 from lib.topojson import build_config_from_json
  70
  71 pytestmark = [pytest.mark.bgpd, pytest.mark.staticd]
  72
  73 # Global variables
  74 NETWORK1_1 = {"ipv4": "11.11.11.1/32", "ipv6": "11:11::1/128"}
  75 NETWORK1_2 = {"ipv4": "11.11.11.11/32", "ipv6": "11:11::11/128"}
  76 NETWORK1_3 = {"ipv4": "10.10.10.1/32", "ipv6": "10:10::1/128"}
  77 NETWORK1_4 = {"ipv4": "10.10.10.100/32", "ipv6": "10:10::100/128"}
  78 NETWORK1_5 = {"ipv4": "110.110.110.1/32", "ipv6": "110:110::1/128"}
  79 NETWORK1_6 = {"ipv4": "110.110.110.100/32", "ipv6": "110:110::100/128"}
  80
  81 NETWORK2_1 = {"ipv4": "22.22.22.2/32", "ipv6": "22:22::2/128"}
  82 NETWORK2_2 = {"ipv4": "22.22.22.22/32", "ipv6": "22:22::22/128"}
  83 NETWORK2_3 = {"ipv4": "20.20.20.20/32", "ipv6": "20:20::20/128"}
  84 NETWORK2_4 = {"ipv4": "20.20.20.200/32", "ipv6": "20:20::200/128"}
  85 NETWORK2_5 = {"ipv4": "220.220.220.20/32", "ipv6": "220:220::20/128"}
  86 NETWORK2_6 = {"ipv4": "220.220.220.200/32", "ipv6": "220:220::200/128"}
  87
  88 NETWORK3_1 = {"ipv4": "30.30.30.3/32", "ipv6": "30:30::3/128"}
  89 NETWORK3_2 = {"ipv4": "30.30.30.30/32", "ipv6": "30:30::30/128"}
  90
  91 PREFIX_LIST = {
  92     "ipv4": ["11.11.11.1", "22.22.22.2", "22.22.22.22"],
  93     "ipv6": ["11:11::1", "22:22::2", "22:22::22"],
  94 }
  95 PREFERRED_NEXT_HOP = "global"
  96 VRF_LIST = ["RED", "BLUE", "GREEN"]
  97 COMM_VAL_1 = "100:100"
  98 COMM_VAL_2 = "500:500"
  99 COMM_VAL_3 = "600:600"
 100 BESTPATH = {
 101     "ipv4": "0.0.0.0",
 102     "ipv6": "::"
 103 }
 104
 105 def setup_module(mod):
 106     """
 107     Sets up the pytest environment
 108
 109     * `mod`: module name
 110     """
 111
 112     testsuite_run_time = time.asctime(time.localtime(time.time()))
 113     logger.info("Testsuite start time: {}".format(testsuite_run_time))
 114     logger.info("=" * 40)
 115
 116     logger.info("Running setup_module to create topology")
 117
 118     # This function initiates the topology build with Topogen...
 119     json_file = "{}/bgp_vrf_lite_best_path_topo2.json".format(CWD)
 120     tgen = Topogen(json_file, mod.__name__)
 121     global topo
 122     topo = tgen.json_topo
 123     # ... and here it calls Mininet initialization functions.
 124
 125     # Starting topology, create tmp files which are loaded to routers
 126     #  to start daemons and then start routers
 127     start_topology(tgen)
 128
 129     # Run these tests for kernel version 4.19 or above
 130     if version_cmp(platform.release(), "4.19") < 0:
 131         error_msg = (
 132             "BGP vrf dynamic route leak tests will not run "
 133             '(have kernel "{}", but it requires >= 4.19)'.format(platform.release())
 134         )
 135         pytest.skip(error_msg)
 136
 137     # Creating configuration from JSON
 138     build_config_from_json(tgen, topo)
 139
 140     global BGP_CONVERGENCE
 141     global ADDR_TYPES
 142     ADDR_TYPES = check_address_types()
 143
 144     BGP_CONVERGENCE = verify_bgp_convergence(tgen, topo)
 145     assert BGP_CONVERGENCE is True, "setup_module : Failed \n Error: {}".format(
 146         BGP_CONVERGENCE
 147     )
 148
 149     logger.info("Running setup_module() done")
 150
 151
 152 def teardown_module():
 153     """Teardown the pytest environment"""
 154
 155     logger.info("Running teardown_module to delete topology")
 156
 157     tgen = get_topogen()
 158
 159     # Stop toplogy and Remove tmp files
 160     tgen.stop_topology()
 161
 162     logger.info(
 163         "Testsuite end time: {}".format(time.asctime(time.localtime(time.time())))
 164     )
 165     logger.info("=" * 40)
 166
 167
 168 #####################################################
 169 #
 170 #   Testcases
 171 #
 172 #####################################################
 173
 174 def test_dynamic_import_ecmp_imported_routed_diffrent_vrfs_p0(request):
 175     """
 176     Verify ECMP for imported routes from different VRFs.
 177     """
 178
 179     tgen = get_topogen()
 180     tc_name = request.node.name
 181     write_test_header(tc_name)
 182     if tgen.routers_have_failure():
 183         check_router_status(tgen)
 184     reset_config_on_routers(tgen)
 185
 186     step("Configure same static routes in tenant vrfs RED and GREEN on router "
 187          "R3 and redistribute in respective BGP process")
 188
 189     for vrf_name in ["RED", "GREEN"]:
 190         for addr_type in ADDR_TYPES:
 191             if vrf_name == "GREEN":
 192                 next_hop_vrf = topo["routers"]["r1"]["links"][
 193                     "r3-link3"][addr_type].split("/")[0]
 194             else:
 195                 next_hop_vrf = topo["routers"]["r2"]["links"][
 196                     "r3-link1"][addr_type].split("/")[0]
 197             static_routes = {
 198                 "r3": {
 199                     "static_routes": [
 200                         {
 201                             "network": [NETWORK1_1[addr_type]],
 202                             "next_hop": next_hop_vrf,
 203                             "vrf": vrf_name
 204                         }
 205                     ]
 206                 }
 207             }
 208
 209             result = create_static_routes(tgen, static_routes)
 210             assert result is True, "Testcase {} :Failed \n Error: {}". \
 211                 format(tc_name, result)
 212
 213         step("Redistribute static route on BGP VRF : {}".format(vrf_name))
 214         temp = {}
 215         for addr_type in ADDR_TYPES:
 216             temp.update({
 217                 addr_type: {
 218                     "unicast": {
 219                         "redistribute": [{
 220                             "redist_type": "static"
 221                         }]
 222                     }
 223                 }
 224             })
 225
 226         redist_dict = {"r3": {"bgp": [{
 227             "vrf": vrf_name, "local_as": 3, "address_family": temp
 228         }]}}
 229
 230         result = create_router_bgp(tgen, topo, redist_dict)
 231         assert result is True, "Testcase {} :Failed \n Error: {}". \
 232             format(tc_name, result)
 233
 234     step("Verify that configured static routes are installed in respective "
 235          "BGP table for vrf RED & GREEN")
 236     for vrf_name in ["RED", "GREEN"]:
 237         for addr_type in ADDR_TYPES:
 238             if vrf_name == "GREEN":
 239                 next_hop_vrf = topo["routers"]["r1"]["links"][
 240                     "r3-link3"][addr_type].split("/")[0]
 241             else:
 242                 next_hop_vrf = topo["routers"]["r2"]["links"][
 243                     "r3-link1"][addr_type].split("/")[0]
 244             static_routes = {
 245                 "r3": {
 246                     "static_routes": [
 247                         {
 248                             "network": [NETWORK1_1[addr_type]],
 249                             "vrf": vrf_name
 250                         }
 251                     ]
 252                 }
 253             }
 254
 255             result = verify_bgp_rib(tgen, addr_type, "r3", static_routes,
 256                                     next_hop=next_hop_vrf)
 257             assert result is True, "Testcase {} : Failed \n Error {}". \
 258                 format(tc_name, result)
 259
 260             result = verify_rib(tgen, addr_type, "r3", static_routes,
 261                                 next_hop=next_hop_vrf)
 262             assert result is True, "Testcase {} : Failed \n Error {}". \
 263                 format(tc_name, result)
 264
 265     step("Import vrf RED and GREEN into default vrf and Configure ECMP")
 266     bgp_val = []
 267     for vrf_name in ["RED", "GREEN"]:
 268         temp = {}
 269         for addr_type in ADDR_TYPES:
 270             temp.update({
 271                 addr_type: {
 272                     "unicast": {
 273                         "import": {
 274                             "vrf": vrf_name
 275                         },
 276                         "maximum_paths": {
 277                             "ebgp": 2
 278                         }
 279                     }
 280                 }
 281             })
 282
 283         bgp_val.append({
 284             "local_as": 3, "address_family": temp
 285         })
 286
 287     import_dict = {"r3": {"bgp": bgp_val}}
 288
 289     result = create_router_bgp(tgen, topo, import_dict)
 290     assert result is True, "Testcase {} :Failed \n Error: {}". \
 291         format(tc_name, result)
 292
 293     step("Configure bgp bestpath on router r3")
 294     r3_raw_config = {
 295         "r3": {
 296             "raw_config": [
 297                 "router bgp 3",
 298                 "bgp bestpath as-path multipath-relax"
 299             ]
 300         }
 301     }
 302     result = apply_raw_config(tgen, r3_raw_config)
 303     assert result is True, "Testcase {} :Failed \n Error: {}". \
 304         format(tc_name, result)
 305
 306     step("Verify that routes are imported with two different next-hop vrfs "
 307          "and IPs. Additionally R3 must do ECMP for both the routes.")
 308
 309     for addr_type in ADDR_TYPES:
 310         next_hop_vrf = [
 311             topo["routers"]["r2"]["links"]["r3-link1"][addr_type]. \
 312                 split("/")[0],
 313             topo["routers"]["r1"]["links"]["r3-link3"][addr_type]. \
 314                 split("/")[0]
 315             ]
 316         static_routes = {
 317             "r3": {
 318                 "static_routes": [
 319                     {
 320                         "network": [NETWORK1_1[addr_type]],
 321                     }
 322                 ]
 323             }
 324         }
 325
 326         result = verify_bgp_rib(tgen, addr_type, "r3", static_routes,
 327                                 next_hop=next_hop_vrf)
 328         assert result is True, "Testcase {} : Failed \n Error {}". \
 329             format(tc_name, result)
 330
 331         result = verify_rib(tgen, addr_type, "r3", static_routes,
 332                             next_hop=next_hop_vrf)
 333         assert result is True, "Testcase {} : Failed \n Error {}". \
 334             format(tc_name, result)
 335
 336     step("Now change the next-hop of static routes in vrf RED and GREEN to "
 337          "same IP address")
 338     for addr_type in ADDR_TYPES:
 339         next_hop_vrf = topo["routers"]["r1"]["links"][
 340             "r3-link3"][addr_type].split("/")[0]
 341         static_routes = {
 342             "r3": {
 343                 "static_routes": [
 344                     {
 345                         "network": [NETWORK1_1[addr_type]],
 346                         "next_hop": next_hop_vrf,
 347                         "vrf": "RED"
 348                     },
 349                     {
 350                         "network": [NETWORK1_1[addr_type]],
 351                         "next_hop":  topo["routers"]["r2"]["links"][
 352                                     "r3-link1"][addr_type].split("/")[0],
 353                         "vrf": "RED",
 354                         "delete": True
 355                     }
 356                 ]
 357             }
 358         }
 359
 360         result = create_static_routes(tgen, static_routes)
 361         assert result is True, "Testcase {} :Failed \n Error: {}". \
 362             format(tc_name, result)
 363
 364     step("Verify that now routes are imported with two different next-hop "
 365          "vrfs but same IPs. Additionally R3 must do ECMP for both the routes")
 366
 367     for addr_type in ADDR_TYPES:
 368         next_hop_vrf = [
 369             topo["routers"]["r1"]["links"]["r3-link3"][addr_type].\
 370                 split("/")[0],
 371             topo["routers"]["r1"]["links"]["r3-link3"][addr_type]. \
 372                 split("/")[0]
 373             ]
 374         static_routes = {
 375             "r3": {
 376                 "static_routes": [
 377                     {
 378                         "network": [NETWORK1_1[addr_type]],
 379                     }
 380                 ]
 381             }
 382         }
 383
 384         result = verify_bgp_rib(tgen, addr_type, "r3", static_routes,
 385                                 next_hop=next_hop_vrf)
 386         assert result is True, "Testcase {} : Failed \n Error {}". \
 387             format(tc_name, result)
 388
 389         result = verify_rib(tgen, addr_type, "r3", static_routes,
 390                             next_hop=next_hop_vrf)
 391         assert result is True, "Testcase {} : Failed \n Error {}". \
 392             format(tc_name, result)
 393
 394     write_test_footer(tc_name)
 395
 396
 397 def test_locally_imported_routes_selected_as_bestpath_over_ebgp_imported_routes_p0(request):
 398     """
 399     Verify ECMP for imported routes from different VRFs.
 400     """
 401
 402     tgen = get_topogen()
 403     tc_name = request.node.name
 404     write_test_header(tc_name)
 405     if tgen.routers_have_failure():
 406         check_router_status(tgen)
 407     reset_config_on_routers(tgen)
 408
 409     step("Configure same static routes on R2 and R3 vrfs and redistribute in BGP "
 410          "for GREEN and RED vrf instances")
 411     for dut, network in zip(["r2", "r3"], [
 412         [NETWORK1_1, NETWORK1_2], [NETWORK1_1, NETWORK1_2]]):
 413         for vrf_name, network_vrf in zip(["RED", "GREEN"], network):
 414             step("Configure static route for VRF : {} on {}".format(vrf_name,
 415                                                                     dut))
 416             for addr_type in ADDR_TYPES:
 417                 static_routes = {
 418                     dut: {
 419                         "static_routes": [
 420                             {
 421                                 "network": [network_vrf[addr_type]],
 422                                 "next_hop": "blackhole",
 423                                 "vrf": vrf_name
 424                             }
 425                         ]
 426                     }
 427                 }
 428
 429                 result = create_static_routes(tgen, static_routes)
 430                 assert result is True, "Testcase {} :Failed \n Error: {}". \
 431                     format(tc_name, result)
 432
 433     for dut, as_num in zip(["r2", "r3"], ["2", "3"]):
 434         for vrf_name in ["RED", "GREEN"]:
 435             step("Redistribute static route on BGP VRF : {}".format(vrf_name))
 436             temp = {}
 437             for addr_type in ADDR_TYPES:
 438                 temp.update({
 439                     addr_type: {
 440                         "unicast": {
 441                             "redistribute": [{
 442                                 "redist_type": "static"
 443                             }]
 444                         }
 445                     }
 446                 })
 447
 448             redist_dict = {dut: {"bgp": [{
 449                 "vrf": vrf_name, "local_as": as_num, "address_family": temp
 450             }]}}
 451
 452             result = create_router_bgp(tgen, topo, redist_dict)
 453             assert result is True, "Testcase {} :Failed \n Error: {}". \
 454                 format(tc_name, result)
 455
 456     step("Verify that R2 and R3 has installed redistributed routes in default "
 457          "and RED vrfs and GREEN respectively:")
 458     for dut, network in zip(["r2", "r3"],
 459                             [[NETWORK1_1, NETWORK1_2],
 460                              [NETWORK1_1, NETWORK1_2]]):
 461         for vrf_name, network_vrf in zip(["RED", "GREEN"], network):
 462             for addr_type in ADDR_TYPES:
 463                 static_routes = {
 464                     dut: {
 465                         "static_routes": [
 466                             {
 467                                 "network": [network_vrf[addr_type]],
 468                                 "next_hop": "blackhole",
 469                                 "vrf": vrf_name
 470                             }
 471                         ]
 472                     }
 473                 }
 474                 result = verify_bgp_rib(tgen, addr_type, dut, static_routes)
 475                 assert result is True, "Testcase {} : Failed \n Error {}". \
 476                     format(tc_name, result)
 477
 478     step("Import vrf RED's route in vrf GREEN on R3")
 479     temp = {}
 480     for addr_type in ADDR_TYPES:
 481         temp.update({
 482             addr_type: {
 483                 "unicast": {
 484                     "import": {
 485                         "vrf": "RED"
 486                     }
 487                 }
 488             }
 489         })
 490
 491     import_dict = {"r3": {"bgp": [{
 492         "vrf": "GREEN", "local_as": 3, "address_family": temp
 493     }]}}
 494
 495     result = create_router_bgp(tgen, topo, import_dict)
 496     assert result is True, "Testcase {} :Failed \n Error: {}". \
 497         format(tc_name, result)
 498
 499     step("Verify that locally imported routes are installed over eBGP imported"
 500         " routes from VRF RED into VRF GREEN")
 501     for addr_type in ADDR_TYPES:
 502         static_routes = {
 503             "r3": {
 504                 "static_routes": [
 505                     {
 506                         "network": [NETWORK1_2[addr_type]],
 507                         "next_hop": "blackhole",
 508                         "vrf": "GREEN"
 509                     }
 510                 ]
 511             }
 512         }
 513
 514         input_routes = {
 515             "r3": {
 516                 addr_type: [
 517                     {
 518                         "network": NETWORK1_2[addr_type],
 519                         "bestpath": BESTPATH[addr_type],
 520                         "vrf": "GREEN"
 521                     }
 522                 ]
 523             }
 524         }
 525
 526         result = verify_bgp_bestpath(tgen, addr_type, input_routes)
 527         assert result is True, "Testcase {} : Failed \n Error {}". \
 528             format(tc_name, result)
 529
 530         result = verify_rib(tgen, addr_type, "r3", static_routes)
 531         assert result is True, "Testcase {} : Failed \n Error {}". \
 532             format(tc_name, result)
 533
 534     write_test_footer(tc_name)
 535
 536
 537 if __name__ == "__main__":
 538     args = ["-s"] + sys.argv[1:]
 539     sys.exit(pytest.main(args))