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