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")
  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 deamons 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 = {"r1": {"route_maps": {"GSHUT-OUT": [{"set": {"locPrf": 0}}],}}}
 320
 321         static_routes = [NETWORK[addr_type]]
 322         result = verify_bgp_attributes(
 323             tgen, addr_type, dut, static_routes, "GSHUT-OUT", rmap_dict
 324         )
 325         assert result is True, "Test case {} : Failed \n Error: {}".format(
 326             tc_name, result
 327         )
 328
 329     step("Ensure that best path is selected from R4 to R3.")
 330
 331     for addr_type in ADDR_TYPES:
 332         dut = "r3"
 333         next_hop1 = next_hop_per_address_family(
 334             tgen, "r3", "r1", addr_type, NEXT_HOP_IP_1
 335         )
 336         next_hop2 = next_hop_per_address_family(
 337             tgen, "r3", "r4", addr_type, NEXT_HOP_IP_2
 338         )
 339
 340         input_topo = {key: topo["routers"][key] for key in ["r1"]}
 341         result = verify_bgp_rib(
 342             tgen, addr_type, dut, input_topo, next_hop=[next_hop1, next_hop2]
 343         )
 344         assert result is True, "Test case {} : Failed \n Error: {}".format(
 345             tc_name, result
 346         )
 347
 348         result = verify_rib(tgen, addr_type, dut, input_topo, next_hop=next_hop2)
 349         assert result is True, "Test case {} : Failed \n Error: {}".format(
 350             tc_name, result
 351         )
 352
 353     write_test_footer(tc_name)
 354
 355
 356 def test_verify_restarting_zebra_bgpd_staticd_frr_with_eBGP_peers_p0(request):
 357     """
 358     Verify graceful-shutdown functionality when daemons bgpd/zebra/staticd and
 359     frr services are restarted with eBGP peers
 360     """
 361
 362     tc_name = request.node.name
 363     write_test_header(tc_name)
 364     tgen = get_topogen()
 365     reset_config_on_routers(tgen)
 366
 367     step("Done in base config: Configure base config as per the topology")
 368     step("Base config should be up, verify using BGP convergence")
 369     result = verify_bgp_convergence(tgen, topo)
 370     assert result is True, "Test case {} : Failed \n Error: {}".format(tc_name, result)
 371
 372     step("Done in base config: Advertise prefixes from R1")
 373     step("Verify BGP routes are received at R3 with best path from R3 to R1")
 374
 375     for addr_type in ADDR_TYPES:
 376         dut = "r3"
 377         next_hop1 = next_hop_per_address_family(
 378             tgen, "r3", "r1", addr_type, NEXT_HOP_IP_1
 379         )
 380         next_hop2 = next_hop_per_address_family(
 381             tgen, "r3", "r4", addr_type, NEXT_HOP_IP_2
 382         )
 383
 384         input_topo = {key: topo["routers"][key] for key in ["r1"]}
 385         result = verify_bgp_rib(
 386             tgen, addr_type, dut, input_topo, next_hop=[next_hop1, next_hop2]
 387         )
 388         assert result is True, "Test case {} : Failed \n Error: {}".format(
 389             tc_name, result
 390         )
 391
 392         result = verify_rib(tgen, addr_type, dut, input_topo, next_hop=next_hop1)
 393         assert result is True, "Test case {} : Failed \n Error: {}".format(
 394             tc_name, result
 395         )
 396
 397     step("On R1 configure:")
 398     step("Create standard bgp community-list to permit graceful-shutdown:")
 399     input_dict_1 = {
 400         "r1": {
 401             "bgp_community_lists": [
 402                 {
 403                     "community_type": "standard",
 404                     "action": "permit",
 405                     "name": "GSHUT",
 406                     "value": "graceful-shutdown",
 407                 }
 408             ]
 409         }
 410     }
 411
 412     result = create_bgp_community_lists(tgen, input_dict_1)
 413     assert result is True, "Test case {} : Failed \n Error: {}".format(tc_name, result)
 414
 415     step("Create route-map to set community GSHUT in OUT direction")
 416
 417     input_dict_2 = {
 418         "r1": {
 419             "route_maps": {
 420                 "GSHUT-OUT": [
 421                     {
 422                         "action": "permit",
 423                         "seq_id": "10",
 424                         "set": {"community": {"num": "graceful-shutdown"}},
 425                     }
 426                 ]
 427             }
 428         }
 429     }
 430
 431     result = create_route_maps(tgen, input_dict_2)
 432     assert result is True, "Test case {} : Failed \n Error: {}".format(tc_name, result)
 433
 434     input_dict_3 = {
 435         "r1": {
 436             "bgp": {
 437                 "address_family": {
 438                     "ipv4": {
 439                         "unicast": {
 440                             "neighbor": {
 441                                 "r3": {
 442                                     "dest_link": {
 443                                         "r1": {
 444                                             "route_maps": [
 445                                                 {
 446                                                     "name": "GSHUT-OUT",
 447                                                     "direction": "out",
 448                                                 }
 449                                             ]
 450                                         }
 451                                     }
 452                                 }
 453                             }
 454                         }
 455                     },
 456                     "ipv6": {
 457                         "unicast": {
 458                             "neighbor": {
 459                                 "r3": {
 460                                     "dest_link": {
 461                                         "r1": {
 462                                             "route_maps": [
 463                                                 {
 464                                                     "name": "GSHUT-OUT",
 465                                                     "direction": "out",
 466                                                 }
 467                                             ]
 468                                         }
 469                                     }
 470                                 }
 471                             }
 472                         }
 473                     },
 474                 }
 475             }
 476         }
 477     }
 478
 479     result = create_router_bgp(tgen, topo, input_dict_3)
 480     assert result is True, "Test case {} : Failed \n Error: {}".format(tc_name, result)
 481
 482     step(
 483         "FRR is setting local-pref to 0 by-default on receiver GSHUT community, "
 484         "below step is not needed, but keeping for reference"
 485     )
 486     step(
 487         "On R3, apply route-map IN direction to match GSHUT community "
 488         "and set local-preference to 0."
 489     )
 490
 491     step(
 492         "Verify BGP convergence on R3 and ensure all the neighbours state "
 493         "is established"
 494     )
 495
 496     result = verify_bgp_convergence(tgen, topo)
 497     assert result is True, "Test case {} : Failed \n Error: {}".format(tc_name, result)
 498
 499     step("Verify BGP routes on R3:")
 500     step("local pref for routes coming from R1 is set to 0.")
 501
 502     for addr_type in ADDR_TYPES:
 503         rmap_dict = {"r1": {"route_maps": {"GSHUT-OUT": [{"set": {"locPrf": 0}}]}}}
 504
 505         static_routes = [NETWORK[addr_type]]
 506         result = verify_bgp_attributes(
 507             tgen, addr_type, dut, static_routes, "GSHUT-OUT", rmap_dict
 508         )
 509         assert result is True, "Test case {} : Failed \n Error: {}".format(
 510             tc_name, result
 511         )
 512
 513     step("Ensure that best path is selected from R4 to R3.")
 514
 515     for addr_type in ADDR_TYPES:
 516         dut = "r3"
 517         next_hop1 = next_hop_per_address_family(
 518             tgen, "r3", "r1", addr_type, NEXT_HOP_IP_1
 519         )
 520         next_hop2 = next_hop_per_address_family(
 521             tgen, "r3", "r4", addr_type, NEXT_HOP_IP_2
 522         )
 523
 524         input_topo = {key: topo["routers"][key] for key in ["r1"]}
 525         result = verify_bgp_rib(
 526             tgen, addr_type, dut, input_topo, next_hop=[next_hop1, next_hop2]
 527         )
 528         assert result is True, "Test case {} : Failed \n Error: {}".format(
 529             tc_name, result
 530         )
 531
 532         result = verify_rib(tgen, addr_type, dut, input_topo, next_hop=next_hop2)
 533         assert result is True, "Test case {} : Failed \n Error: {}".format(
 534             tc_name, result
 535         )
 536
 537     step("Restart daemons and frr services")
 538
 539     for daemon in ["bgpd", "zebra", "staticd", "frr"]:
 540         if daemon != "frr":
 541             kill_router_daemons(tgen, "r3", ["staticd"])
 542             start_router_daemons(tgen, "r3", ["staticd"])
 543         else:
 544             stop_router(tgen, "r3")
 545             start_router(tgen, "r3")
 546
 547         step(
 548             "Verify BGP convergence on R3 and ensure all the neighbours state "
 549             "is established"
 550         )
 551
 552         result = verify_bgp_convergence(tgen, topo)
 553         assert result is True, "Test case {} : Failed \n Error: {}".format(
 554             tc_name, result
 555         )
 556
 557         step("Verify BGP routes on R3:")
 558         step("local pref for routes coming from R1 is set to 0.")
 559
 560         for addr_type in ADDR_TYPES:
 561             rmap_dict = {"r1": {"route_maps": {"GSHUT-OUT": [{"set": {"locPrf": 0}}]}}}
 562
 563             static_routes = [NETWORK[addr_type]]
 564             result = verify_bgp_attributes(
 565                 tgen, addr_type, dut, static_routes, "GSHUT-OUT", rmap_dict
 566             )
 567             assert result is True, "Test case {} : Failed \n Error: {}".format(
 568                 tc_name, result
 569             )
 570
 571         step("Ensure that best path is selected from R4 to R3.")
 572
 573         for addr_type in ADDR_TYPES:
 574             dut = "r3"
 575             next_hop1 = next_hop_per_address_family(
 576                 tgen, "r3", "r1", addr_type, NEXT_HOP_IP_1
 577             )
 578             next_hop2 = next_hop_per_address_family(
 579                 tgen, "r3", "r4", addr_type, NEXT_HOP_IP_2
 580             )
 581
 582             input_topo = {key: topo["routers"][key] for key in ["r1"]}
 583             result = verify_bgp_rib(
 584                 tgen, addr_type, dut, input_topo, next_hop=[next_hop1, next_hop2]
 585             )
 586             assert result is True, "Test case {} : Failed \n Error: {}".format(
 587                 tc_name, result
 588             )
 589
 590             result = verify_rib(tgen, addr_type, dut, input_topo, next_hop=next_hop2)
 591             assert result is True, "Test case {} : Failed \n Error: {}".format(
 592                 tc_name, result
 593             )
 594
 595     write_test_footer(tc_name)
 596
 597
 598 if __name__ == "__main__":
 599     args = ["-s"] + sys.argv[1:]
 600     sys.exit(pytest.main(args))