tests/topotests/bgp_basic_functionality_topo1/test_bgp_basic_functionality.py

   1 #!/usr/bin/env python
   2
   3 #
   4 # Copyright (c) 2019 by VMware, Inc. ("VMware")
   5 # Used Copyright (c) 2018 by Network Device Education Foundation,
   6 # Inc. ("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 Following tests are covered to test BGP basic functionality:
  25
  26 Test steps
  27 - Create topology (setup module)
  28   Creating 4 routers topology, r1, r2, r3 are in IBGP and
  29   r3, r4 are in EBGP
  30 - Bring up topology
  31 - Verify for bgp to converge
  32 - Modify/Delete and verify router-id
  33 - Modify and verify bgp timers
  34 - Create and verify static routes
  35 - Modify and verify admin distance for existing static routes
  36 - Test advertise network using network command
  37 - Verify clear bgp
  38 - Test bgp convergence with loopback interface
  39 - Test advertise network using network command
  40 - Verify routes not installed in zebra when /32 routes received
  41    with loopback BGP session subnet
  42 """
  43 # XXX clean up in later commit to avoid conflict on rebase
  44 # pylint: disable=C0413
  45
  46 import os
  47 import sys
  48 import time
  49 import pytest
  50 from copy import deepcopy
  51
  52 # Save the Current Working Directory to find configuration files.
  53 CWD = os.path.dirname(os.path.realpath(__file__))
  54 sys.path.append(os.path.join(CWD, "../"))
  55 sys.path.append(os.path.join(CWD, "../lib/"))
  56
  57 # Required to instantiate the topology builder class.
  58
  59 from lib.bgp import (
  60     clear_bgp_and_verify,
  61     create_router_bgp,
  62     modify_as_number,
  63     verify_as_numbers,
  64     verify_bgp_convergence,
  65     verify_bgp_rib,
  66     verify_bgp_timers_and_functionality,
  67     verify_router_id,
  68 )
  69 from lib.common_config import (
  70     addKernelRoute,
  71     apply_raw_config,
  72     check_address_types,
  73     create_prefix_lists,
  74     create_route_maps,
  75     create_static_routes,
  76     required_linux_kernel_version,
  77     reset_config_on_routers,
  78     start_topology,
  79     step,
  80     verify_admin_distance_for_static_routes,
  81     verify_bgp_community,
  82     verify_fib_routes,
  83     verify_rib,
  84     write_test_footer,
  85     write_test_header,
  86 )
  87
  88 # pylint: disable=C0413
  89 # Import topogen and topotest helpers
  90 from lib.topogen import Topogen, get_topogen
  91 from lib.topojson import build_config_from_json
  92 from lib.topolog import logger
  93
  94 pytestmark = [pytest.mark.bgpd, pytest.mark.staticd]
  95
  96
  97 # Global Variable
  98 KEEPALIVETIMER = 2
  99 HOLDDOWNTIMER = 6
 100 r1_ipv4_loopback = "1.0.1.0/24"
 101 r2_ipv4_loopback = "1.0.2.0/24"
 102 r3_ipv4_loopback = "1.0.3.0/24"
 103 r4_ipv4_loopback = "1.0.4.0/24"
 104 r1_ipv6_loopback = "2001:db8:f::1:0/120"
 105 r2_ipv6_loopback = "2001:db8:f::2:0/120"
 106 r3_ipv6_loopback = "2001:db8:f::3:0/120"
 107 r4_ipv6_loopback = "2001:db8:f::4:0/120"
 108 NETWORK = {
 109     "ipv4": ["100.1.1.1/32", "100.1.1.2/32"],
 110     "ipv6": ["100::1/128", "100::2/128"],
 111 }
 112
 113
 114 def setup_module(mod):
 115     """
 116     Sets up the pytest environment
 117
 118     * `mod`: module name
 119     """
 120
 121     # Required linux kernel version for this suite to run.
 122     result = required_linux_kernel_version("4.15")
 123     if result is not True:
 124         pytest.skip("Kernel requirements are not met, kernel version should be >=4.15")
 125
 126     testsuite_run_time = time.asctime(time.localtime(time.time()))
 127     logger.info("Testsuite start time: {}".format(testsuite_run_time))
 128     logger.info("=" * 40)
 129
 130     logger.info("Running setup_module to create topology")
 131
 132     # This function initiates the topology build with Topogen...
 133     json_file = "{}/bgp_basic_functionality.json".format(CWD)
 134     tgen = Topogen(json_file, mod.__name__)
 135     global topo
 136     topo = tgen.json_topo
 137     # ... and here it calls Mininet initialization functions.
 138
 139     # Starting topology, create tmp files which are loaded to routers
 140     #  to start daemons and then start routers
 141     start_topology(tgen)
 142
 143     # Creating configuration from JSON
 144     build_config_from_json(tgen, topo)
 145
 146     global ADDR_TYPES
 147     global BGP_CONVERGENCE
 148     ADDR_TYPES = check_address_types()
 149     BGP_CONVERGENCE = verify_bgp_convergence(tgen, topo)
 150     assert BGP_CONVERGENCE is True, "setup_module :Failed \n Error: {}".format(
 151         BGP_CONVERGENCE
 152     )
 153
 154     logger.info("Running setup_module() done")
 155
 156
 157 def teardown_module():
 158     """Teardown the pytest environment"""
 159
 160     logger.info("Running teardown_module to delete topology")
 161
 162     tgen = get_topogen()
 163
 164     # Stop toplogy and Remove tmp files
 165     tgen.stop_topology()
 166
 167     logger.info(
 168         "Testsuite end time: {}".format(time.asctime(time.localtime(time.time())))
 169     )
 170     logger.info("=" * 40)
 171
 172
 173 #####################################################
 174 #
 175 #   Testcases
 176 #
 177 #####################################################
 178
 179
 180 def test_modify_and_delete_router_id(request):
 181     """Test to modify, delete and verify router-id."""
 182
 183     tgen = get_topogen()
 184     if BGP_CONVERGENCE is not True:
 185         pytest.skip("skipped because of BGP Convergence failure")
 186
 187     # test case name
 188     tc_name = request.node.name
 189     write_test_header(tc_name)
 190
 191     # Creating configuration from JSON
 192     reset_config_on_routers(tgen)
 193
 194     # Modify router id
 195     input_dict = {
 196         "r1": {"bgp": {"router_id": "12.12.12.12"}},
 197         "r2": {"bgp": {"router_id": "22.22.22.22"}},
 198         "r3": {"bgp": {"router_id": "33.33.33.33"}},
 199     }
 200     result = create_router_bgp(tgen, topo, input_dict)
 201     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 202
 203     # Verifying router id once modified
 204     result = verify_router_id(tgen, topo, input_dict)
 205     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 206
 207     # Delete router id
 208     input_dict = {
 209         "r1": {"bgp": {"del_router_id": True}},
 210         "r2": {"bgp": {"del_router_id": True}},
 211         "r3": {"bgp": {"del_router_id": True}},
 212     }
 213     result = create_router_bgp(tgen, topo, input_dict)
 214     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 215
 216     # Verifying router id once deleted
 217     # Once router-id is deleted, highest interface ip should become
 218     # router-id
 219     result = verify_router_id(tgen, topo, input_dict)
 220     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 221
 222     write_test_footer(tc_name)
 223
 224
 225 def test_bgp_config_with_4byte_as_number(request):
 226     """
 227     Configure BGP with 4 byte ASN and verify it works fine
 228     """
 229
 230     tgen = get_topogen()
 231     if BGP_CONVERGENCE is not True:
 232         pytest.skip("skipped because of BGP Convergence failure")
 233
 234     # test case name
 235     tc_name = request.node.name
 236     write_test_header(tc_name)
 237
 238     # Creating configuration from JSON
 239     reset_config_on_routers(tgen)
 240
 241     input_dict = {
 242         "r1": {"bgp": {"local_as": 131079}},
 243         "r2": {"bgp": {"local_as": 131079}},
 244         "r3": {"bgp": {"local_as": 131079}},
 245         "r4": {"bgp": {"local_as": 131080}},
 246     }
 247     result = modify_as_number(tgen, topo, input_dict)
 248     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 249
 250     result = verify_as_numbers(tgen, topo, input_dict)
 251     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 252
 253     write_test_footer(tc_name)
 254
 255
 256 def test_BGP_config_with_invalid_ASN_p2(request):
 257     """
 258     Configure BGP with invalid ASN(ex - 0, reserved ASN) and verify test case
 259     ended up with error
 260     """
 261
 262     tgen = get_topogen()
 263     global BGP_CONVERGENCE
 264
 265     if BGP_CONVERGENCE != True:
 266         pytest.skip("skipped because of BGP Convergence failure")
 267
 268     # test case name
 269     tc_name = request.node.name
 270     write_test_header(tc_name)
 271
 272     # Creating configuration from JSON
 273     reset_config_on_routers(tgen)
 274
 275     # Api call to modify AS number
 276     input_dict = {
 277         "r1": {
 278             "bgp": {
 279                 "local_as": 0,
 280             }
 281         },
 282         "r2": {
 283             "bgp": {
 284                 "local_as": 0,
 285             }
 286         },
 287         "r3": {
 288             "bgp": {
 289                 "local_as": 0,
 290             }
 291         },
 292         "r4": {
 293             "bgp": {
 294                 "local_as": 64000,
 295             }
 296         },
 297     }
 298     result = modify_as_number(tgen, topo, input_dict)
 299     assert (
 300         result is not True
 301     ), "Expected BGP config is not created because of invalid ASNs: {}".format(result)
 302
 303     # Creating configuration from JSON
 304     reset_config_on_routers(tgen)
 305
 306     result = verify_bgp_convergence(tgen, topo)
 307     if result != True:
 308         assert False, "Testcase " + tc_name + " :Failed \n Error: {}".format(result)
 309
 310     write_test_footer(tc_name)
 311
 312
 313 def test_BGP_config_with_2byteAS_and_4byteAS_number_p1(request):
 314     """
 315     Configure BGP with 4 byte and 2 byte ASN and verify BGP is converged
 316     """
 317
 318     tgen = get_topogen()
 319     global BGP_CONVERGENCE
 320
 321     if BGP_CONVERGENCE != True:
 322         pytest.skip("skipped because of BGP Convergence failure")
 323
 324     # test case name
 325     tc_name = request.node.name
 326     write_test_header(tc_name)
 327
 328     # Creating configuration from JSON
 329     reset_config_on_routers(tgen)
 330
 331     result = verify_bgp_convergence(tgen, topo)
 332     if result != True:
 333         assert False, "Testcase " + tc_name + " :Failed \n Error: {}".format(result)
 334
 335     # Api call to modify AS number
 336     input_dict = {
 337         "r1": {"bgp": {"local_as": 131079}},
 338         "r2": {"bgp": {"local_as": 131079}},
 339         "r3": {"bgp": {"local_as": 131079}},
 340         "r4": {"bgp": {"local_as": 111}},
 341     }
 342     result = modify_as_number(tgen, topo, input_dict)
 343     if result != True:
 344         assert False, "Testcase " + tc_name + " :Failed \n Error: {}".format(result)
 345
 346     result = verify_as_numbers(tgen, topo, input_dict)
 347     if result != True:
 348         assert False, "Testcase " + tc_name + " :Failed \n Error: {}".format(result)
 349
 350     # Api call verify whether BGP is converged
 351     result = verify_bgp_convergence(tgen, topo)
 352     if result != True:
 353         assert False, "Testcase " + tc_name + " :Failed \n Error: {}".format(result)
 354
 355     write_test_footer(tc_name)
 356
 357
 358 def test_bgp_timers_functionality(request):
 359     """
 360     Test to modify bgp timers and verify timers functionality.
 361     """
 362
 363     tgen = get_topogen()
 364     if BGP_CONVERGENCE is not True:
 365         pytest.skip("skipped because of BGP Convergence failure")
 366
 367     # test case name
 368     tc_name = request.node.name
 369     write_test_header(tc_name)
 370
 371     # Creating configuration from JSON
 372     reset_config_on_routers(tgen)
 373
 374     # Api call to modify BGP timerse
 375     input_dict = {
 376         "r1": {
 377             "bgp": {
 378                 "address_family": {
 379                     "ipv4": {
 380                         "unicast": {
 381                             "neighbor": {
 382                                 "r2": {
 383                                     "dest_link": {
 384                                         "r1": {
 385                                             "keepalivetimer": KEEPALIVETIMER,
 386                                             "holddowntimer": HOLDDOWNTIMER,
 387                                         }
 388                                     }
 389                                 }
 390                             }
 391                         }
 392                     }
 393                 }
 394             }
 395         }
 396     }
 397     result = create_router_bgp(tgen, topo, deepcopy(input_dict))
 398     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 399
 400     # Api call to clear bgp, so timer modification would take place
 401     clear_bgp_and_verify(tgen, topo, "r1")
 402
 403     # Verifying bgp timers functionality
 404     result = verify_bgp_timers_and_functionality(tgen, topo, input_dict)
 405     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 406
 407     write_test_footer(tc_name)
 408
 409
 410 def test_static_routes(request):
 411     """Test to create and verify static routes."""
 412
 413     tgen = get_topogen()
 414     if BGP_CONVERGENCE is not True:
 415         pytest.skip("skipped because of BGP Convergence failure")
 416
 417     # test case name
 418     tc_name = request.node.name
 419     write_test_header(tc_name)
 420
 421     # Creating configuration from JSON
 422     reset_config_on_routers(tgen)
 423
 424     # Api call to create static routes
 425     input_dict = {
 426         "r1": {
 427             "static_routes": [
 428                 {
 429                     "network": "10.0.20.1/32",
 430                     "no_of_ip": 9,
 431                     "admin_distance": 100,
 432                     "next_hop": "10.0.0.2",
 433                 }
 434             ]
 435         }
 436     }
 437     result = create_static_routes(tgen, input_dict)
 438     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 439
 440     # Api call to redistribute static routes
 441     input_dict_1 = {
 442         "r1": {
 443             "bgp": {
 444                 "address_family": {
 445                     "ipv4": {
 446                         "unicast": {
 447                             "redistribute": [
 448                                 {"redist_type": "static"},
 449                                 {"redist_type": "connected"},
 450                             ]
 451                         }
 452                     }
 453                 }
 454             }
 455         }
 456     }
 457
 458     result = create_router_bgp(tgen, topo, input_dict_1)
 459     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 460
 461     # Verifying RIB routes
 462     dut = "r3"
 463     protocol = "bgp"
 464     next_hop = ["10.0.0.2", "10.0.0.5"]
 465     result = verify_rib(
 466         tgen, "ipv4", dut, input_dict, next_hop=next_hop, protocol=protocol
 467     )
 468     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 469
 470     write_test_footer(tc_name)
 471
 472
 473 def test_admin_distance_for_existing_static_routes(request):
 474     """Test to modify and verify admin distance for existing static routes."""
 475
 476     tgen = get_topogen()
 477     if BGP_CONVERGENCE is not True:
 478         pytest.skip("skipped because of BGP Convergence failure")
 479
 480     # test case name
 481     tc_name = request.node.name
 482     write_test_header(tc_name)
 483
 484     # Creating configuration from JSON
 485     reset_config_on_routers(tgen)
 486
 487     input_dict = {
 488         "r1": {
 489             "static_routes": [
 490                 {
 491                     "network": "10.0.20.1/32",
 492                     "admin_distance": 10,
 493                     "next_hop": "10.0.0.2",
 494                 }
 495             ]
 496         }
 497     }
 498     result = create_static_routes(tgen, input_dict)
 499     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 500
 501     # Verifying admin distance  once modified
 502     result = verify_admin_distance_for_static_routes(tgen, input_dict)
 503     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 504
 505     write_test_footer(tc_name)
 506
 507
 508 def test_advertise_network_using_network_command(request):
 509     """Test advertise networks using network command."""
 510
 511     tgen = get_topogen()
 512     if BGP_CONVERGENCE is not True:
 513         pytest.skip("skipped because of BGP Convergence failure")
 514
 515     # test case name
 516     tc_name = request.node.name
 517     write_test_header(tc_name)
 518
 519     # Creating configuration from JSON
 520     reset_config_on_routers(tgen)
 521
 522     # Api call to advertise networks
 523     input_dict = {
 524         "r1": {
 525             "bgp": {
 526                 "address_family": {
 527                     "ipv4": {
 528                         "unicast": {
 529                             "advertise_networks": [
 530                                 {"network": "20.0.0.0/32", "no_of_network": 10},
 531                                 {"network": "30.0.0.0/32", "no_of_network": 10},
 532                             ]
 533                         }
 534                     }
 535                 }
 536             }
 537         }
 538     }
 539
 540     result = create_router_bgp(tgen, topo, input_dict)
 541     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 542
 543     # Verifying RIB routes
 544     dut = "r2"
 545     protocol = "bgp"
 546     result = verify_rib(tgen, "ipv4", dut, input_dict, protocol=protocol)
 547     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 548
 549     write_test_footer(tc_name)
 550
 551
 552 def test_clear_bgp_and_verify(request):
 553     """
 554     Created few static routes and verified all routes are learned via BGP
 555     cleared BGP and verified all routes are intact
 556     """
 557
 558     tgen = get_topogen()
 559     if BGP_CONVERGENCE is not True:
 560         pytest.skip("skipped because of BGP Convergence failure")
 561
 562     # test case name
 563     tc_name = request.node.name
 564     write_test_header(tc_name)
 565
 566     # Creating configuration from JSON
 567     reset_config_on_routers(tgen)
 568
 569     # clear ip bgp
 570     result = clear_bgp_and_verify(tgen, topo, "r1")
 571     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 572
 573     write_test_footer(tc_name)
 574
 575
 576 def test_BGP_attributes_with_vrf_default_keyword_p0(request):
 577     """
 578     TC_9:
 579     Verify BGP functionality for default vrf with
 580     "vrf default" keyword.
 581     """
 582
 583     tc_name = request.node.name
 584     write_test_header(tc_name)
 585     tgen = get_topogen()
 586
 587     if tgen.routers_have_failure():
 588         pytest.skip(tgen.errors)
 589
 590     # Creating configuration from JSON
 591     reset_config_on_routers(tgen)
 592
 593     step("Configure static routes and redistribute in BGP on R3")
 594     for addr_type in ADDR_TYPES:
 595         input_dict = {
 596             "r3": {
 597                 "static_routes": [
 598                     {
 599                         "network": NETWORK[addr_type][0],
 600                         "no_of_ip": 4,
 601                         "next_hop": "Null0",
 602                     }
 603                 ]
 604             }
 605         }
 606
 607         result = create_static_routes(tgen, input_dict)
 608         assert result is True, "Testcase {} : Failed \n Error: {}".format(
 609             tc_name, result
 610         )
 611
 612     input_dict_2 = {
 613         "r3": {
 614             "bgp": {
 615                 "address_family": {
 616                     "ipv4": {"unicast": {"redistribute": [{"redist_type": "static"}]}},
 617                     "ipv6": {"unicast": {"redistribute": [{"redist_type": "static"}]}},
 618                 }
 619             }
 620         }
 621     }
 622     result = create_router_bgp(tgen, topo, input_dict_2)
 623     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 624
 625     step(
 626         "Create a route-map to match a specific prefix and modify"
 627         "BGP attributes for matched prefix"
 628     )
 629     input_dict_2 = {
 630         "r3": {
 631             "prefix_lists": {
 632                 "ipv4": {
 633                     "ABC": [
 634                         {
 635                             "seqid": 10,
 636                             "action": "permit",
 637                             "network": NETWORK["ipv4"][0],
 638                         }
 639                     ]
 640                 },
 641                 "ipv6": {
 642                     "XYZ": [
 643                         {
 644                             "seqid": 100,
 645                             "action": "permit",
 646                             "network": NETWORK["ipv6"][0],
 647                         }
 648                     ]
 649                 },
 650             }
 651         }
 652     }
 653     result = create_prefix_lists(tgen, input_dict_2)
 654     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 655
 656     for addr_type in ADDR_TYPES:
 657         if addr_type == "ipv4":
 658             pf_list = "ABC"
 659         else:
 660             pf_list = "XYZ"
 661
 662         input_dict_6 = {
 663             "r3": {
 664                 "route_maps": {
 665                     "BGP_ATTR_{}".format(addr_type): [
 666                         {
 667                             "action": "permit",
 668                             "seq_id": 10,
 669                             "match": {addr_type: {"prefix_lists": pf_list}},
 670                             "set": {
 671                                 "aspath": {"as_num": 500, "as_action": "prepend"},
 672                                 "localpref": 500,
 673                                 "origin": "egp",
 674                                 "community": {"num": "500:500", "action": "additive"},
 675                                 "large_community": {
 676                                     "num": "500:500:500",
 677                                     "action": "additive",
 678                                 },
 679                             },
 680                         },
 681                         {"action": "permit", "seq_id": 20},
 682                     ]
 683                 },
 684                 "BGP_ATTR_{}".format(addr_type): [
 685                     {
 686                         "action": "permit",
 687                         "seq_id": 100,
 688                         "match": {addr_type: {"prefix_lists": pf_list}},
 689                         "set": {
 690                             "aspath": {"as_num": 500, "as_action": "prepend"},
 691                             "localpref": 500,
 692                             "origin": "egp",
 693                             "community": {"num": "500:500", "action": "additive"},
 694                             "large_community": {
 695                                 "num": "500:500:500",
 696                                 "action": "additive",
 697                             },
 698                         },
 699                     },
 700                     {"action": "permit", "seq_id": 200},
 701                 ],
 702             }
 703         }
 704
 705         result = create_route_maps(tgen, input_dict_6)
 706         assert result is True, "Testcase {} : Failed \n Error: {}".format(
 707             tc_name, result
 708         )
 709
 710     step("Apply the route-map on R3 in outbound direction for peer R4")
 711
 712     input_dict_7 = {
 713         "r3": {
 714             "bgp": {
 715                 "address_family": {
 716                     "ipv4": {
 717                         "unicast": {
 718                             "neighbor": {
 719                                 "r4": {
 720                                     "dest_link": {
 721                                         "r3": {
 722                                             "route_maps": [
 723                                                 {
 724                                                     "name": "BGP_ATTR_ipv4",
 725                                                     "direction": "out",
 726                                                 }
 727                                             ]
 728                                         }
 729                                     }
 730                                 }
 731                             }
 732                         }
 733                     },
 734                     "ipv6": {
 735                         "unicast": {
 736                             "neighbor": {
 737                                 "r4": {
 738                                     "dest_link": {
 739                                         "r3": {
 740                                             "route_maps": [
 741                                                 {
 742                                                     "name": "BGP_ATTR_ipv6",
 743                                                     "direction": "out",
 744                                                 }
 745                                             ]
 746                                         }
 747                                     }
 748                                 }
 749                             }
 750                         }
 751                     },
 752                 }
 753             }
 754         }
 755     }
 756
 757     result = create_router_bgp(tgen, topo, input_dict_7)
 758     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 759
 760     step(
 761         "verify modified attributes for specific prefix with 'vrf default'"
 762         "keyword on R4"
 763     )
 764     for addr_type in ADDR_TYPES:
 765         dut = "r4"
 766         input_dict = {
 767             "r3": {
 768                 "static_routes": [
 769                     {
 770                         "network": NETWORK[addr_type][0],
 771                         "vrf": "default",
 772                         "largeCommunity": "500:500:500",
 773                     }
 774                 ]
 775             }
 776         }
 777
 778         result = verify_bgp_rib(tgen, addr_type, dut, input_dict)
 779         assert result is True, "Testcase {} : Failed \n Error: {}".format(
 780             tc_name, result
 781         )
 782         result = verify_rib(tgen, addr_type, dut, input_dict)
 783         assert result is True, "Testcase {} : Failed \n Error: {}".format(
 784             tc_name, result
 785         )
 786
 787     for addr_type in ADDR_TYPES:
 788         dut = "r4"
 789         input_dict = {
 790             "r3": {
 791                 "static_routes": [
 792                     {
 793                         "network": NETWORK[addr_type][0],
 794                         "vrf": "default",
 795                         "community": "500:500",
 796                     }
 797                 ]
 798             }
 799         }
 800
 801         result = verify_bgp_rib(tgen, addr_type, dut, input_dict)
 802         assert result is True, "Testcase {} : Failed \n Error: {}".format(
 803             tc_name, result
 804         )
 805         result = verify_rib(tgen, addr_type, dut, input_dict)
 806         assert result is True, "Testcase {} : Failed \n Error: {}".format(
 807             tc_name, result
 808         )
 809
 810         input_dict_4 = {"largeCommunity": "500:500:500", "community": "500:500"}
 811
 812         result = verify_bgp_community(
 813             tgen, addr_type, dut, [NETWORK[addr_type][0]], input_dict_4
 814         )
 815         assert result is True, "Test case {} : Should fail \n Error: {}".format(
 816             tc_name, result
 817         )
 818
 819     write_test_footer(tc_name)
 820
 821
 822 def test_bgp_with_loopback_interface(request):
 823     """
 824     Test BGP with loopback interface
 825
 826     Adding keys:value pair  "dest_link": "lo" and "source_link": "lo"
 827     peer dict of input json file for all router's creating config using
 828     loopback interface. Once BGP neighboship is up then verifying BGP
 829     convergence
 830     """
 831
 832     tgen = get_topogen()
 833     if BGP_CONVERGENCE is not True:
 834         pytest.skip("skipped because of BGP Convergence failure")
 835
 836     # test case name
 837     tc_name = request.node.name
 838     write_test_header(tc_name)
 839
 840     # Creating configuration from JSON
 841     reset_config_on_routers(tgen)
 842
 843     for routerN in sorted(topo["routers"].keys()):
 844         for bgp_neighbor in topo["routers"][routerN]["bgp"]["address_family"]["ipv4"][
 845             "unicast"
 846         ]["neighbor"].keys():
 847
 848             # Adding ['source_link'] = 'lo' key:value pair
 849             topo["routers"][routerN]["bgp"]["address_family"]["ipv4"]["unicast"][
 850                 "neighbor"
 851             ][bgp_neighbor]["dest_link"] = {
 852                 "lo": {
 853                     "source_link": "lo",
 854                 }
 855             }
 856
 857     # Creating configuration from JSON
 858     build_config_from_json(tgen, topo)
 859
 860     input_dict = {
 861         "r1": {
 862             "static_routes": [
 863                 {"network": "1.0.2.17/32", "next_hop": "10.0.0.2"},
 864                 {"network": "1.0.3.17/32", "next_hop": "10.0.0.6"},
 865             ]
 866         },
 867         "r2": {
 868             "static_routes": [
 869                 {"network": "1.0.1.17/32", "next_hop": "10.0.0.1"},
 870                 {"network": "1.0.3.17/32", "next_hop": "10.0.0.10"},
 871             ]
 872         },
 873         "r3": {
 874             "static_routes": [
 875                 {"network": "1.0.1.17/32", "next_hop": "10.0.0.5"},
 876                 {"network": "1.0.2.17/32", "next_hop": "10.0.0.9"},
 877                 {"network": "1.0.4.17/32", "next_hop": "10.0.0.14"},
 878             ]
 879         },
 880         "r4": {"static_routes": [{"network": "1.0.3.17/32", "next_hop": "10.0.0.13"}]},
 881     }
 882     result = create_static_routes(tgen, input_dict)
 883     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 884
 885     # Api call verify whether BGP is converged
 886     result = verify_bgp_convergence(tgen, topo)
 887     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 888
 889     write_test_footer(tc_name)
 890
 891
 892 def test_bgp_with_loopback_with_same_subnet_p1(request):
 893     """
 894     Verify routes not installed in zebra when /32 routes received
 895     with loopback BGP session subnet
 896     """
 897
 898     tgen = get_topogen()
 899     if BGP_CONVERGENCE is not True:
 900         pytest.skip("skipped because of BGP Convergence failure")
 901
 902     # test case name
 903     tc_name = request.node.name
 904     write_test_header(tc_name)
 905
 906     # Creating configuration from JSON
 907     reset_config_on_routers(tgen)
 908     step("Delete BGP seesion created initially")
 909     input_dict_r1 = {
 910         "r1": {"bgp": {"delete": True}},
 911         "r2": {"bgp": {"delete": True}},
 912         "r3": {"bgp": {"delete": True}},
 913         "r4": {"bgp": {"delete": True}},
 914     }
 915     result = create_router_bgp(tgen, topo, input_dict_r1)
 916     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 917
 918     step("Create BGP session over loop address")
 919     topo_modify = deepcopy(topo)
 920
 921     for routerN in sorted(topo["routers"].keys()):
 922         for addr_type in ADDR_TYPES:
 923             for bgp_neighbor in topo_modify["routers"][routerN]["bgp"][
 924                 "address_family"
 925             ][addr_type]["unicast"]["neighbor"].keys():
 926
 927                 # Adding ['source_link'] = 'lo' key:value pair
 928                 topo_modify["routers"][routerN]["bgp"]["address_family"][addr_type][
 929                     "unicast"
 930                 ]["neighbor"][bgp_neighbor]["dest_link"] = {
 931                     "lo": {"source_link": "lo", "ebgp_multihop": 2}
 932                 }
 933
 934     result = create_router_bgp(tgen, topo_modify["routers"])
 935     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 936
 937     step("Disable IPv6 BGP nbr from ipv4 address family")
 938     raw_config = {
 939         "r1": {
 940             "raw_config": [
 941                 "router bgp {}".format(topo["routers"]["r1"]["bgp"]["local_as"]),
 942                 "address-family ipv4 unicast",
 943                 "no neighbor {} activate".format(
 944                     topo["routers"]["r2"]["links"]["lo"]["ipv6"].split("/")[0]
 945                 ),
 946                 "no neighbor {} activate".format(
 947                     topo["routers"]["r3"]["links"]["lo"]["ipv6"].split("/")[0]
 948                 ),
 949             ]
 950         },
 951         "r2": {
 952             "raw_config": [
 953                 "router bgp {}".format(topo["routers"]["r2"]["bgp"]["local_as"]),
 954                 "address-family ipv4 unicast",
 955                 "no neighbor {} activate".format(
 956                     topo["routers"]["r1"]["links"]["lo"]["ipv6"].split("/")[0]
 957                 ),
 958                 "no neighbor {} activate".format(
 959                     topo["routers"]["r3"]["links"]["lo"]["ipv6"].split("/")[0]
 960                 ),
 961             ]
 962         },
 963         "r3": {
 964             "raw_config": [
 965                 "router bgp {}".format(topo["routers"]["r3"]["bgp"]["local_as"]),
 966                 "address-family ipv4 unicast",
 967                 "no neighbor {} activate".format(
 968                     topo["routers"]["r1"]["links"]["lo"]["ipv6"].split("/")[0]
 969                 ),
 970                 "no neighbor {} activate".format(
 971                     topo["routers"]["r2"]["links"]["lo"]["ipv6"].split("/")[0]
 972                 ),
 973                 "no neighbor {} activate".format(
 974                     topo["routers"]["r4"]["links"]["lo"]["ipv6"].split("/")[0]
 975                 ),
 976             ]
 977         },
 978         "r4": {
 979             "raw_config": [
 980                 "router bgp {}".format(topo["routers"]["r4"]["bgp"]["local_as"]),
 981                 "address-family ipv4 unicast",
 982                 "no neighbor {} activate".format(
 983                     topo["routers"]["r3"]["links"]["lo"]["ipv6"].split("/")[0]
 984                 ),
 985             ]
 986         },
 987     }
 988
 989     step("Configure kernel routes")
 990     result = apply_raw_config(tgen, raw_config)
 991     assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
 992
 993     r1_ipv4_lo = topo["routers"]["r1"]["links"]["lo"]["ipv4"]
 994     r1_ipv6_lo = topo["routers"]["r1"]["links"]["lo"]["ipv6"]
 995     r2_ipv4_lo = topo["routers"]["r2"]["links"]["lo"]["ipv4"]
 996     r2_ipv6_lo = topo["routers"]["r2"]["links"]["lo"]["ipv6"]
 997     r3_ipv4_lo = topo["routers"]["r3"]["links"]["lo"]["ipv4"]
 998     r3_ipv6_lo = topo["routers"]["r3"]["links"]["lo"]["ipv6"]
 999     r4_ipv4_lo = topo["routers"]["r4"]["links"]["lo"]["ipv4"]
1000     r4_ipv6_lo = topo["routers"]["r4"]["links"]["lo"]["ipv6"]
1001
1002     r1_r2 = topo["routers"]["r1"]["links"]["r2"]["ipv6"].split("/")[0]
1003     r2_r1 = topo["routers"]["r2"]["links"]["r1"]["ipv6"].split("/")[0]
1004     r1_r3 = topo["routers"]["r1"]["links"]["r3"]["ipv6"].split("/")[0]
1005     r3_r1 = topo["routers"]["r3"]["links"]["r1"]["ipv6"].split("/")[0]
1006     r2_r3 = topo["routers"]["r2"]["links"]["r3"]["ipv6"].split("/")[0]
1007     r3_r2 = topo["routers"]["r3"]["links"]["r2"]["ipv6"].split("/")[0]
1008     r3_r4 = topo["routers"]["r3"]["links"]["r4"]["ipv6"].split("/")[0]
1009     r4_r3 = topo["routers"]["r4"]["links"]["r3"]["ipv6"].split("/")[0]
1010
1011     r1_r2_ipv4 = topo["routers"]["r1"]["links"]["r2"]["ipv4"].split("/")[0]
1012     r2_r1_ipv4 = topo["routers"]["r2"]["links"]["r1"]["ipv4"].split("/")[0]
1013     r1_r3_ipv4 = topo["routers"]["r1"]["links"]["r3"]["ipv4"].split("/")[0]
1014     r3_r1_ipv4 = topo["routers"]["r3"]["links"]["r1"]["ipv4"].split("/")[0]
1015     r2_r3_ipv4 = topo["routers"]["r2"]["links"]["r3"]["ipv4"].split("/")[0]
1016     r3_r2_ipv4 = topo["routers"]["r3"]["links"]["r2"]["ipv4"].split("/")[0]
1017     r3_r4_ipv4 = topo["routers"]["r3"]["links"]["r4"]["ipv4"].split("/")[0]
1018     r4_r3_ipv4 = topo["routers"]["r4"]["links"]["r3"]["ipv4"].split("/")[0]
1019
1020     r1_r2_intf = topo["routers"]["r1"]["links"]["r2"]["interface"]
1021     r2_r1_intf = topo["routers"]["r2"]["links"]["r1"]["interface"]
1022     r1_r3_intf = topo["routers"]["r1"]["links"]["r3"]["interface"]
1023     r3_r1_intf = topo["routers"]["r3"]["links"]["r1"]["interface"]
1024     r2_r3_intf = topo["routers"]["r2"]["links"]["r3"]["interface"]
1025     r3_r2_intf = topo["routers"]["r3"]["links"]["r2"]["interface"]
1026     r3_r4_intf = topo["routers"]["r3"]["links"]["r4"]["interface"]
1027     r4_r3_intf = topo["routers"]["r4"]["links"]["r3"]["interface"]
1028
1029     ipv4_list = [
1030         ("r1", r1_r2_intf, r2_ipv4_loopback),
1031         ("r1", r1_r3_intf, r3_ipv4_loopback),
1032         ("r2", r2_r1_intf, r1_ipv4_loopback),
1033         ("r2", r2_r3_intf, r3_ipv4_loopback),
1034         ("r3", r3_r1_intf, r1_ipv4_loopback),
1035         ("r3", r3_r2_intf, r2_ipv4_loopback),
1036         ("r3", r3_r4_intf, r4_ipv4_loopback),
1037         ("r4", r4_r3_intf, r3_ipv4_loopback),
1038     ]
1039
1040     ipv6_list = [
1041         ("r1", r1_r2_intf, r2_ipv6_loopback, r2_r1),
1042         ("r1", r1_r3_intf, r3_ipv6_loopback, r3_r1),
1043         ("r2", r2_r1_intf, r1_ipv6_loopback, r1_r2),
1044         ("r2", r2_r3_intf, r3_ipv6_loopback, r3_r2),
1045         ("r3", r3_r1_intf, r1_ipv6_loopback, r1_r3),
1046         ("r3", r3_r2_intf, r2_ipv6_loopback, r2_r3),
1047         ("r3", r3_r4_intf, r4_ipv6_loopback, r4_r3),
1048         ("r4", r4_r3_intf, r3_ipv6_loopback, r3_r4),
1049     ]
1050
1051     for dut, intf, loop_addr in ipv4_list:
1052         result = addKernelRoute(tgen, dut, intf, loop_addr)
1053         assert result is True, "Testcase {}:Failed \n Error: {}".format(tc_name, result)
1054
1055     for dut, intf, loop_addr, next_hop in ipv6_list:
1056         result = addKernelRoute(tgen, dut, intf, loop_addr, next_hop)
1057         assert result is True, "Testcase {}:Failed \n Error: {}".format(tc_name, result)
1058
1059     step("Configure static routes")
1060
1061     input_dict = {
1062         "r1": {
1063             "static_routes": [
1064                 {"network": r2_ipv4_loopback, "next_hop": r2_r1_ipv4},
1065                 {"network": r3_ipv4_loopback, "next_hop": r3_r1_ipv4},
1066                 {"network": r2_ipv6_loopback, "next_hop": r2_r1},
1067                 {"network": r3_ipv6_loopback, "next_hop": r3_r1},
1068             ]
1069         },
1070         "r2": {
1071             "static_routes": [
1072                 {"network": r1_ipv4_loopback, "next_hop": r1_r2_ipv4},
1073                 {"network": r3_ipv4_loopback, "next_hop": r3_r2_ipv4},
1074                 {"network": r1_ipv6_loopback, "next_hop": r1_r2},
1075                 {"network": r3_ipv6_loopback, "next_hop": r3_r2},
1076             ]
1077         },
1078         "r3": {
1079             "static_routes": [
1080                 {"network": r1_ipv4_loopback, "next_hop": r1_r3_ipv4},
1081                 {"network": r2_ipv4_loopback, "next_hop": r2_r3_ipv4},
1082                 {"network": r4_ipv4_loopback, "next_hop": r4_r3_ipv4},
1083                 {"network": r1_ipv6_loopback, "next_hop": r1_r3},
1084                 {"network": r2_ipv6_loopback, "next_hop": r2_r3},
1085                 {"network": r4_ipv6_loopback, "next_hop": r4_r3},
1086             ]
1087         },
1088         "r4": {
1089             "static_routes": [
1090                 {"network": r3_ipv4_loopback, "next_hop": r3_r4_ipv4},
1091                 {"network": r3_ipv6_loopback, "next_hop": r3_r4},
1092             ]
1093         },
1094     }
1095     result = create_static_routes(tgen, input_dict)
1096     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
1097
1098     step("Verify BGP session convergence")
1099
1100     result = verify_bgp_convergence(tgen, topo_modify)
1101     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
1102
1103     step("Configure redistribute connected on R2 and R4")
1104     input_dict_1 = {
1105         "r2": {
1106             "bgp": {
1107                 "address_family": {
1108                     "ipv4": {
1109                         "unicast": {"redistribute": [{"redist_type": "connected"}]}
1110                     },
1111                     "ipv6": {
1112                         "unicast": {"redistribute": [{"redist_type": "connected"}]}
1113                     },
1114                 }
1115             }
1116         },
1117         "r4": {
1118             "bgp": {
1119                 "address_family": {
1120                     "ipv4": {
1121                         "unicast": {"redistribute": [{"redist_type": "connected"}]}
1122                     },
1123                     "ipv6": {
1124                         "unicast": {"redistribute": [{"redist_type": "connected"}]}
1125                     },
1126                 }
1127             }
1128         },
1129     }
1130
1131     result = create_router_bgp(tgen, topo, input_dict_1)
1132     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
1133
1134     step("Verify Ipv4 and Ipv6 network installed in R1 RIB but not in FIB")
1135     input_dict_r1 = {
1136         "r1": {
1137             "static_routes": [
1138                 {"network": "1.0.2.17/32"},
1139                 {"network": "2001:db8:f::2:17/128"},
1140             ]
1141         }
1142     }
1143
1144     dut = "r1"
1145     protocol = "bgp"
1146     for addr_type in ADDR_TYPES:
1147         result = verify_fib_routes(
1148             tgen, addr_type, dut, input_dict_r1, expected=False
1149         )  # pylint: disable=E1123
1150         assert result is not True, (
1151             "Testcase {} : Failed \n "
1152             "Expected: Routes should not be present in {} FIB \n "
1153             "Found: {}".format(tc_name, dut, result)
1154         )
1155
1156     step("Verify Ipv4 and Ipv6 network installed in r3 RIB but not in FIB")
1157     input_dict_r3 = {
1158         "r3": {
1159             "static_routes": [
1160                 {"network": "1.0.4.17/32"},
1161                 {"network": "2001:db8:f::4:17/128"},
1162             ]
1163         }
1164     }
1165     dut = "r3"
1166     protocol = "bgp"
1167     for addr_type in ADDR_TYPES:
1168         result = verify_fib_routes(
1169             tgen, addr_type, dut, input_dict_r1, expected=False
1170         )  # pylint: disable=E1123
1171         assert result is not True, (
1172             "Testcase {} : Failed \n "
1173             "Expected: Routes should not be present in {} FIB \n "
1174             "Found: {}".format(tc_name, dut, result)
1175         )
1176
1177     write_test_footer(tc_name)
1178
1179
1180 if __name__ == "__main__":
1181     args = ["-s"] + sys.argv[1:]
1182     sys.exit(pytest.main(args))