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