tests/topotests/bgp_gshut_topo1/test_ebgp_gshut_topo1.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, Inc.
   6 # ("NetDEF") in this file.
   7 #
   8 # Permission to use, copy, modify, and/or distribute this software
   9 # for any purpose with or without fee is hereby granted, provided
  10 # that the above copyright notice and this permission notice appear
  11 # in all copies.
  12 #
  13 # THE SOFTWARE IS PROVIDED "AS IS" AND VMWARE DISCLAIMS ALL WARRANTIES
  14 # WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
  15 # MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL VMWARE BE LIABLE FOR
  16 # ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY
  17 # DAMAGES WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS,
  18 # WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS
  19 # ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR PERFORMANCE
  20 # OF THIS SOFTWARE.
  21 #
  22
  23
  24 """
  25 Following tests are covered to test ecmp functionality on BGP GSHUT.
  26 1. Verify graceful-shutdown functionality with eBGP peers
  27 2. Verify graceful-shutdown functionality when daemons
  28    bgpd/zebra/staticd and frr services are restarted with eBGP peers
  29 """
  30
  31 import os
  32 import sys
  33 import time
  34 import pytest
  35
  36 # Save the Current Working Directory to find configuration files.
  37 CWD = os.path.dirname(os.path.realpath(__file__))
  38 sys.path.append(os.path.join(CWD, "../"))
  39 sys.path.append(os.path.join(CWD, "../../"))
  40
  41 # pylint: disable=C0413
  42 # Import topogen and topotest helpers
  43 from lib.topogen import Topogen, get_topogen
  44
  45 from lib.common_config import (
  46     start_topology,
  47     write_test_header,
  48     write_test_footer,
  49     verify_rib,
  50     check_address_types,
  51     reset_config_on_routers,
  52     step,
  53     get_frr_ipv6_linklocal,
  54     kill_router_daemons,
  55     start_router_daemons,
  56     stop_router,
  57     start_router,
  58     create_route_maps,
  59     create_bgp_community_lists,
  60     required_linux_kernel_version,
  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_attributes,
  68 )
  69 from lib.topojson import build_config_from_json
  70
  71
  72 pytestmark = [pytest.mark.bgpd, pytest.mark.staticd]
  73
  74
  75 # Global variables
  76 NETWORK = {"ipv4": "100.0.10.1/32", "ipv6": "1::1/128"}
  77 NEXT_HOP_IP_1 = {"ipv4": "10.0.2.1", "ipv6": "fd00:0:0:1::1"}
  78 NEXT_HOP_IP_2 = {"ipv4": "10.0.4.2", "ipv6": "fd00:0:0:3::2"}
  79 PREFERRED_NEXT_HOP = "link_local"
  80 BGP_CONVERGENCE = False
  81
  82
  83 def setup_module(mod):
  84     """
  85     Sets up the pytest environment
  86
  87     * `mod`: module name
  88     """
  89
  90     global ADDR_TYPES
  91
  92     # Required linux kernel version for this suite to run.
  93     result = required_linux_kernel_version("4.16")
  94     if result is not True:
  95         pytest.skip("Kernel requirements are not met, kernel version should be >=4.16")
  96
  97     testsuite_run_time = time.asctime(time.localtime(time.time()))
  98     logger.info("Testsuite start time: {}".format(testsuite_run_time))
  99     logger.info("=" * 40)
 100
 101     logger.info("Running setup_module to create topology")
 102
 103     # This function initiates the topology build with Topogen...
 104     json_file = "{}/ebgp_gshut_topo1.json".format(CWD)
 105     tgen = Topogen(json_file, mod.__name__)
 106     global topo
 107     topo = tgen.json_topo
 108     # ... and here it calls Mininet initialization functions.
 109
 110     # Starting topology, create tmp files which are loaded to routers
 111     #  to start daemons and then start routers
 112     start_topology(tgen)
 113
 114     # Creating configuration from JSON
 115     build_config_from_json(tgen, topo)
 116
 117     # Don't run this test if we have any failure.
 118     if tgen.routers_have_failure():
 119         pytest.skip(tgen.errors)
 120
 121     # Api call verify whether BGP is converged
 122     ADDR_TYPES = check_address_types()
 123
 124     BGP_CONVERGENCE = verify_bgp_convergence(tgen, topo)
 125     assert BGP_CONVERGENCE is True, "setup_module :Failed \n Error:" " {}".format(
 126         BGP_CONVERGENCE
 127     )
 128
 129     logger.info("Running setup_module() done")
 130
 131
 132 def teardown_module():
 133     """
 134     Teardown the pytest environment.
 135
 136     * `mod`: module name
 137     """
 138
 139     logger.info("Running teardown_module to delete topology")
 140
 141     tgen = get_topogen()
 142
 143     # Stop toplogy and Remove tmp files
 144     tgen.stop_topology()
 145
 146
 147 ###########################
 148 # Local APIs
 149 ###########################
 150
 151
 152 def next_hop_per_address_family(
 153     tgen, dut, peer, addr_type, next_hop_dict, preferred_next_hop=PREFERRED_NEXT_HOP
 154 ):
 155     """
 156     This function returns link_local or global next_hop per address-family
 157     """
 158
 159     intferface = topo["routers"][peer]["links"]["{}".format(dut)]["interface"]
 160     if addr_type == "ipv6" and "link_local" in preferred_next_hop:
 161         next_hop = get_frr_ipv6_linklocal(tgen, peer, intf=intferface)
 162     else:
 163         next_hop = next_hop_dict[addr_type]
 164
 165     return next_hop
 166
 167
 168 ###########################
 169 # TESTCASES
 170 ###########################
 171
 172
 173 def test_verify_graceful_shutdown_functionality_with_eBGP_peers_p0(request):
 174     """
 175     Verify graceful-shutdown functionality with eBGP peers
 176     """
 177
 178     tc_name = request.node.name
 179     write_test_header(tc_name)
 180     tgen = get_topogen()
 181     reset_config_on_routers(tgen)
 182
 183     step("Done in base config: Configure base config as per the topology")
 184     step("Base config should be up, verify using BGP convergence")
 185     result = verify_bgp_convergence(tgen, topo)
 186     assert result is True, "Test case {} : Failed \n Error: {}".format(tc_name, result)
 187
 188     step("Done in base config: Advertise prefixes from R1")
 189     step("Verify BGP routes are received at R3 with best path from R3 to R1")
 190
 191     for addr_type in ADDR_TYPES:
 192         dut = "r3"
 193         next_hop1 = next_hop_per_address_family(
 194             tgen, "r3", "r1", addr_type, NEXT_HOP_IP_1
 195         )
 196         next_hop2 = next_hop_per_address_family(
 197             tgen, "r3", "r4", addr_type, NEXT_HOP_IP_2
 198         )
 199
 200         input_topo = {key: topo["routers"][key] for key in ["r1"]}
 201         result = verify_bgp_rib(
 202             tgen, addr_type, dut, input_topo, next_hop=[next_hop1, next_hop2]
 203         )
 204         assert result is True, "Test case {} : Failed \n Error: {}".format(
 205             tc_name, result
 206         )
 207
 208         result = verify_rib(tgen, addr_type, dut, input_topo, next_hop=next_hop1)
 209         assert result is True, "Test case {} : Failed \n Error: {}".format(
 210             tc_name, result
 211         )
 212
 213     step("On R1 configure:")
 214     step("Create standard bgp community-list to permit graceful-shutdown:")
 215     input_dict_1 = {
 216         "r1": {
 217             "bgp_community_lists": [
 218                 {
 219                     "community_type": "standard",
 220                     "action": "permit",
 221                     "name": "GSHUT",
 222                     "value": "graceful-shutdown",
 223                 }
 224             ]
 225         }
 226     }
 227
 228     result = create_bgp_community_lists(tgen, input_dict_1)
 229     assert result is True, "Test case {} : Failed \n Error: {}".format(tc_name, result)
 230
 231     step("Create route-map to set community GSHUT in OUT direction")
 232
 233     input_dict_2 = {
 234         "r1": {
 235             "route_maps": {
 236                 "GSHUT-OUT": [
 237                     {
 238                         "action": "permit",
 239                         "seq_id": "10",
 240                         "set": {"community": {"num": "graceful-shutdown"}},
 241                     }
 242                 ]
 243             }
 244         }
 245     }
 246
 247     result = create_route_maps(tgen, input_dict_2)
 248     assert result is True, "Test case {} : Failed \n Error: {}".format(tc_name, result)
 249
 250     input_dict_3 = {
 251         "r1": {
 252             "bgp": {
 253                 "address_family": {
 254                     "ipv4": {
 255                         "unicast": {
 256                             "neighbor": {
 257                                 "r3": {
 258                                     "dest_link": {
 259                                         "r1": {
 260                                             "route_maps": [
 261                                                 {
 262                                                     "name": "GSHUT-OUT",
 263                                                     "direction": "out",
 264                                                 }
 265                                             ]
 266                                         }
 267                                     }
 268                                 }
 269                             }
 270                         }
 271                     },
 272                     "ipv6": {
 273                         "unicast": {
 274                             "neighbor": {
 275                                 "r3": {
 276                                     "dest_link": {
 277                                         "r1": {
 278                                             "route_maps": [
 279                                                 {
 280                                                     "name": "GSHUT-OUT",
 281                                                     "direction": "out",
 282                                                 }
 283                                             ]
 284                                         }
 285                                     }
 286                                 }
 287                             }
 288                         }
 289                     },
 290                 }
 291             }
 292         }
 293     }
 294
 295     result = create_router_bgp(tgen, topo, input_dict_3)
 296     assert result is True, "Test case {} : Failed \n Error: {}".format(tc_name, result)
 297
 298     step(
 299         "FRR is setting local-pref to 0 by-default on receiver GSHUT community, "
 300         "below step is not needed, but keeping for reference"
 301     )
 302     step(
 303         "On R3, apply route-map IN direction to match GSHUT community "
 304         "and set local-preference to 0."
 305     )
 306
 307     step(
 308         "Verify BGP convergence on R3 and ensure all the neighbours state "
 309         "is established"
 310     )
 311
 312     result = verify_bgp_convergence(tgen, topo)
 313     assert result is True, "Test case {} : Failed \n Error: {}".format(tc_name, result)
 314
 315     step("Verify BGP routes on R3:")
 316     step("local pref for routes coming from R1 is set to 0.")
 317
 318     for addr_type in ADDR_TYPES:
 319         rmap_dict = {
 320             "r1": {
 321                 "route_maps": {
 322                     "GSHUT-OUT": [{"set": {"locPrf": 0}}],
 323                 }
 324             }
 325         }
 326
 327         static_routes = [NETWORK[addr_type]]
 328         result = verify_bgp_attributes(
 329             tgen, addr_type, dut, static_routes, "GSHUT-OUT", rmap_dict
 330         )
 331         assert result is True, "Test case {} : Failed \n Error: {}".format(
 332             tc_name, result
 333         )
 334
 335     step("Ensure that best path is selected from R4 to R3.")
 336
 337     for addr_type in ADDR_TYPES:
 338         dut = "r3"
 339         next_hop1 = next_hop_per_address_family(
 340             tgen, "r3", "r1", addr_type, NEXT_HOP_IP_1
 341         )
 342         next_hop2 = next_hop_per_address_family(
 343             tgen, "r3", "r4", addr_type, NEXT_HOP_IP_2
 344         )
 345
 346         input_topo = {key: topo["routers"][key] for key in ["r1"]}
 347         result = verify_bgp_rib(
 348             tgen, addr_type, dut, input_topo, next_hop=[next_hop1, next_hop2]
 349         )
 350         assert result is True, "Test case {} : Failed \n Error: {}".format(
 351             tc_name, result
 352         )
 353
 354         result = verify_rib(tgen, addr_type, dut, input_topo, next_hop=next_hop2)
 355         assert result is True, "Test case {} : Failed \n Error: {}".format(
 356             tc_name, result
 357         )
 358
 359     write_test_footer(tc_name)
 360
 361
 362 def test_verify_restarting_zebra_bgpd_staticd_frr_with_eBGP_peers_p0(request):
 363     """
 364     Verify graceful-shutdown functionality when daemons bgpd/zebra/staticd and
 365     frr services are restarted with eBGP peers
 366     """
 367
 368     tc_name = request.node.name
 369     write_test_header(tc_name)
 370     tgen = get_topogen()
 371     reset_config_on_routers(tgen)
 372
 373     step("Done in base config: Configure base config as per the topology")
 374     step("Base config should be up, verify using BGP convergence")
 375     result = verify_bgp_convergence(tgen, topo)
 376     assert result is True, "Test case {} : Failed \n Error: {}".format(tc_name, result)
 377
 378     step("Done in base config: Advertise prefixes from R1")
 379     step("Verify BGP routes are received at R3 with best path from R3 to R1")
 380
 381     for addr_type in ADDR_TYPES:
 382         dut = "r3"
 383         next_hop1 = next_hop_per_address_family(
 384             tgen, "r3", "r1", addr_type, NEXT_HOP_IP_1
 385         )
 386         next_hop2 = next_hop_per_address_family(
 387             tgen, "r3", "r4", addr_type, NEXT_HOP_IP_2
 388         )
 389
 390         input_topo = {key: topo["routers"][key] for key in ["r1"]}
 391         result = verify_bgp_rib(
 392             tgen, addr_type, dut, input_topo, next_hop=[next_hop1, next_hop2]
 393         )
 394         assert result is True, "Test case {} : Failed \n Error: {}".format(
 395             tc_name, result
 396         )
 397
 398         result = verify_rib(tgen, addr_type, dut, input_topo, next_hop=next_hop1)
 399         assert result is True, "Test case {} : Failed \n Error: {}".format(
 400             tc_name, result
 401         )
 402
 403     step("On R1 configure:")
 404     step("Create standard bgp community-list to permit graceful-shutdown:")
 405     input_dict_1 = {
 406         "r1": {
 407             "bgp_community_lists": [
 408                 {
 409                     "community_type": "standard",
 410                     "action": "permit",
 411                     "name": "GSHUT",
 412                     "value": "graceful-shutdown",
 413                 }
 414             ]
 415         }
 416     }
 417
 418     result = create_bgp_community_lists(tgen, input_dict_1)
 419     assert result is True, "Test case {} : Failed \n Error: {}".format(tc_name, result)
 420
 421     step("Create route-map to set community GSHUT in OUT direction")
 422
 423     input_dict_2 = {
 424         "r1": {
 425             "route_maps": {
 426                 "GSHUT-OUT": [
 427                     {
 428                         "action": "permit",
 429                         "seq_id": "10",
 430                         "set": {"community": {"num": "graceful-shutdown"}},
 431                     }
 432                 ]
 433             }
 434         }
 435     }
 436
 437     result = create_route_maps(tgen, input_dict_2)
 438     assert result is True, "Test case {} : Failed \n Error: {}".format(tc_name, result)
 439
 440     input_dict_3 = {
 441         "r1": {
 442             "bgp": {
 443                 "address_family": {
 444                     "ipv4": {
 445                         "unicast": {
 446                             "neighbor": {
 447                                 "r3": {
 448                                     "dest_link": {
 449                                         "r1": {
 450                                             "route_maps": [
 451                                                 {
 452                                                     "name": "GSHUT-OUT",
 453                                                     "direction": "out",
 454                                                 }
 455                                             ]
 456                                         }
 457                                     }
 458                                 }
 459                             }
 460                         }
 461                     },
 462                     "ipv6": {
 463                         "unicast": {
 464                             "neighbor": {
 465                                 "r3": {
 466                                     "dest_link": {
 467                                         "r1": {
 468                                             "route_maps": [
 469                                                 {
 470                                                     "name": "GSHUT-OUT",
 471                                                     "direction": "out",
 472                                                 }
 473                                             ]
 474                                         }
 475                                     }
 476                                 }
 477                             }
 478                         }
 479                     },
 480                 }
 481             }
 482         }
 483     }
 484
 485     result = create_router_bgp(tgen, topo, input_dict_3)
 486     assert result is True, "Test case {} : Failed \n Error: {}".format(tc_name, result)
 487
 488     step(
 489         "FRR is setting local-pref to 0 by-default on receiver GSHUT community, "
 490         "below step is not needed, but keeping for reference"
 491     )
 492     step(
 493         "On R3, apply route-map IN direction to match GSHUT community "
 494         "and set local-preference to 0."
 495     )
 496
 497     step(
 498         "Verify BGP convergence on R3 and ensure all the neighbours state "
 499         "is established"
 500     )
 501
 502     result = verify_bgp_convergence(tgen, topo)
 503     assert result is True, "Test case {} : Failed \n Error: {}".format(tc_name, result)
 504
 505     step("Verify BGP routes on R3:")
 506     step("local pref for routes coming from R1 is set to 0.")
 507
 508     for addr_type in ADDR_TYPES:
 509         rmap_dict = {"r1": {"route_maps": {"GSHUT-OUT": [{"set": {"locPrf": 0}}]}}}
 510
 511         static_routes = [NETWORK[addr_type]]
 512         result = verify_bgp_attributes(
 513             tgen, addr_type, dut, static_routes, "GSHUT-OUT", rmap_dict
 514         )
 515         assert result is True, "Test case {} : Failed \n Error: {}".format(
 516             tc_name, result
 517         )
 518
 519     step("Ensure that best path is selected from R4 to R3.")
 520
 521     for addr_type in ADDR_TYPES:
 522         dut = "r3"
 523         next_hop1 = next_hop_per_address_family(
 524             tgen, "r3", "r1", addr_type, NEXT_HOP_IP_1
 525         )
 526         next_hop2 = next_hop_per_address_family(
 527             tgen, "r3", "r4", addr_type, NEXT_HOP_IP_2
 528         )
 529
 530         input_topo = {key: topo["routers"][key] for key in ["r1"]}
 531         result = verify_bgp_rib(
 532             tgen, addr_type, dut, input_topo, next_hop=[next_hop1, next_hop2]
 533         )
 534         assert result is True, "Test case {} : Failed \n Error: {}".format(
 535             tc_name, result
 536         )
 537
 538         result = verify_rib(tgen, addr_type, dut, input_topo, next_hop=next_hop2)
 539         assert result is True, "Test case {} : Failed \n Error: {}".format(
 540             tc_name, result
 541         )
 542
 543     step("Restart daemons and frr services")
 544
 545     for daemon in ["bgpd", "zebra", "staticd", "frr"]:
 546         if daemon != "frr":
 547             kill_router_daemons(tgen, "r3", ["staticd"])
 548             start_router_daemons(tgen, "r3", ["staticd"])
 549         else:
 550             stop_router(tgen, "r3")
 551             start_router(tgen, "r3")
 552
 553         step(
 554             "Verify BGP convergence on R3 and ensure all the neighbours state "
 555             "is established"
 556         )
 557
 558         result = verify_bgp_convergence(tgen, topo)
 559         assert result is True, "Test case {} : Failed \n Error: {}".format(
 560             tc_name, result
 561         )
 562
 563         step("Verify BGP routes on R3:")
 564         step("local pref for routes coming from R1 is set to 0.")
 565
 566         for addr_type in ADDR_TYPES:
 567             rmap_dict = {"r1": {"route_maps": {"GSHUT-OUT": [{"set": {"locPrf": 0}}]}}}
 568
 569             static_routes = [NETWORK[addr_type]]
 570             result = verify_bgp_attributes(
 571                 tgen, addr_type, dut, static_routes, "GSHUT-OUT", rmap_dict
 572             )
 573             assert result is True, "Test case {} : Failed \n Error: {}".format(
 574                 tc_name, result
 575             )
 576
 577         step("Ensure that best path is selected from R4 to R3.")
 578
 579         for addr_type in ADDR_TYPES:
 580             dut = "r3"
 581             next_hop1 = next_hop_per_address_family(
 582                 tgen, "r3", "r1", addr_type, NEXT_HOP_IP_1
 583             )
 584             next_hop2 = next_hop_per_address_family(
 585                 tgen, "r3", "r4", addr_type, NEXT_HOP_IP_2
 586             )
 587
 588             input_topo = {key: topo["routers"][key] for key in ["r1"]}
 589             result = verify_bgp_rib(
 590                 tgen, addr_type, dut, input_topo, next_hop=[next_hop1, next_hop2]
 591             )
 592             assert result is True, "Test case {} : Failed \n Error: {}".format(
 593                 tc_name, result
 594             )
 595
 596             result = verify_rib(tgen, addr_type, dut, input_topo, next_hop=next_hop2)
 597             assert result is True, "Test case {} : Failed \n Error: {}".format(
 598                 tc_name, result
 599             )
 600
 601     write_test_footer(tc_name)
 602
 603
 604 if __name__ == "__main__":
 605     args = ["-s"] + sys.argv[1:]
 606     sys.exit(pytest.main(args))