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