tests/topotests/bgp_ecmp_topo3/test_ibgp_ecmp_topo3.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, 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 iBGP.
  26 1. Verify bgp fast-convergence functionality
  27 """
  28 import os
  29 import sys
  30 import time
  31 import json
  32 import pytest
  33 from time import sleep
  34
  35 # Save the Current Working Directory to find configuration files.
  36 CWD = os.path.dirname(os.path.realpath(__file__))
  37 sys.path.append(os.path.join(CWD, "../"))
  38 sys.path.append(os.path.join(CWD, "../../"))
  39
  40 # pylint: disable=C0413
  41 # Import topogen and topotest helpers
  42 from lib.topogen import Topogen, get_topogen
  43 from mininet.topo import Topo
  44
  45 from lib.common_config import (
  46     start_topology,
  47     write_test_header,
  48     write_test_footer,
  49     verify_rib,
  50     create_static_routes,
  51     check_address_types,
  52     interface_status,
  53     reset_config_on_routers,
  54     required_linux_kernel_version,
  55     shutdown_bringup_interface,
  56     apply_raw_config,
  57 )
  58 from lib.topolog import logger
  59 from lib.bgp import verify_bgp_convergence, create_router_bgp, clear_bgp
  60 from lib.topojson import build_topo_from_json, build_config_from_json
  61
  62
  63 pytestmark = [pytest.mark.bgpd, pytest.mark.staticd]
  64
  65
  66 # Reading the data from JSON File for topology and configuration creation
  67 jsonFile = "{}/ibgp_ecmp_topo3.json".format(CWD)
  68
  69 try:
  70     with open(jsonFile, "r") as topoJson:
  71         topo = json.load(topoJson)
  72 except IOError:
  73     assert False, "Could not read file {}".format(jsonFile)
  74
  75 # Global variables
  76 NEXT_HOPS = {"ipv4": [], "ipv6": []}
  77 NETWORK = {"ipv4": "192.168.1.10/32", "ipv6": "fd00:0:0:1::10/128"}
  78 NEXT_HOP_IP = {"ipv4": "10.0.0.1", "ipv6": "fd00::1"}
  79 BGP_CONVERGENCE = False
  80
  81
  82 class CreateTopo(Topo):
  83     """
  84     Test topology builder.
  85
  86     * `Topo`: Topology object
  87     """
  88
  89     def build(self, *_args, **_opts):
  90         """Build function."""
  91         tgen = get_topogen(self)
  92
  93         # Building topology from json file
  94         build_topo_from_json(tgen, topo)
  95
  96
  97 def setup_module(mod):
  98     """
  99     Sets up the pytest environment.
 100
 101     * `mod`: module name
 102     """
 103     global NEXT_HOPS, INTF_LIST_R3, INTF_LIST_R2, TEST_STATIC
 104     global ADDR_TYPES
 105
 106     testsuite_run_time = time.asctime(time.localtime(time.time()))
 107     logger.info("Testsuite start time: {}".format(testsuite_run_time))
 108     logger.info("=" * 40)
 109
 110     logger.info("Running setup_module to create topology")
 111
 112     # This function initiates the topology build with Topogen...
 113     tgen = Topogen(CreateTopo, mod.__name__)
 114
 115     # Starting topology, create tmp files which are loaded to routers
 116     #  to start deamons and then start routers
 117     start_topology(tgen)
 118
 119     # Creating configuration from JSON
 120     build_config_from_json(tgen, topo)
 121
 122     # Don't run this test if we have any failure.
 123     if tgen.routers_have_failure():
 124         pytest.skip(tgen.errors)
 125
 126     # Api call verify whether BGP is converged
 127     ADDR_TYPES = check_address_types()
 128
 129     BGP_CONVERGENCE = verify_bgp_convergence(tgen, topo)
 130     assert BGP_CONVERGENCE is True, "setup_module :Failed \n Error:" " {}".format(
 131         BGP_CONVERGENCE
 132     )
 133
 134     # STATIC_ROUTE = True
 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 def static_or_nw(tgen, topo, tc_name, test_type, dut):
 154
 155     if test_type == "redist_static":
 156         input_dict_static = {
 157             dut: {
 158                 "static_routes": [
 159                     {"network": NETWORK["ipv4"], "next_hop": NEXT_HOP_IP["ipv4"]},
 160                     {"network": NETWORK["ipv6"], "next_hop": NEXT_HOP_IP["ipv6"]},
 161                 ]
 162             }
 163         }
 164         logger.info("Configuring static route on router %s", dut)
 165         result = create_static_routes(tgen, input_dict_static)
 166         assert result is True, "Testcase {} : Failed \n Error: {}".format(
 167             tc_name, result
 168         )
 169
 170         input_dict_2 = {
 171             dut: {
 172                 "bgp": {
 173                     "address_family": {
 174                         "ipv4": {
 175                             "unicast": {"redistribute": [{"redist_type": "static"}]}
 176                         },
 177                         "ipv6": {
 178                             "unicast": {"redistribute": [{"redist_type": "static"}]}
 179                         },
 180                     }
 181                 }
 182             }
 183         }
 184
 185         logger.info("Configuring redistribute static route on router %s", dut)
 186         result = create_router_bgp(tgen, topo, input_dict_2)
 187         assert result is True, "Testcase {} : Failed \n Error: {}".format(
 188             tc_name, result
 189         )
 190
 191     elif test_type == "advertise_nw":
 192         input_dict_nw = {
 193             dut: {
 194                 "bgp": {
 195                     "address_family": {
 196                         "ipv4": {
 197                             "unicast": {
 198                                 "advertise_networks": [{"network": NETWORK["ipv4"]}]
 199                             }
 200                         },
 201                         "ipv6": {
 202                             "unicast": {
 203                                 "advertise_networks": [{"network": NETWORK["ipv6"]}]
 204                             }
 205                         },
 206                     }
 207                 }
 208             }
 209         }
 210
 211         logger.info(
 212             "Advertising networks %s %s from router %s",
 213             NETWORK["ipv4"],
 214             NETWORK["ipv6"],
 215             dut,
 216         )
 217         result = create_router_bgp(tgen, topo, input_dict_nw)
 218         assert result is True, "Testcase {} : Failed \n Error: {}".format(
 219             tc_name, result
 220         )
 221
 222
 223 @pytest.mark.parametrize("test_type", ["redist_static"])
 224 def test_ecmp_fast_convergence(request, test_type):
 225     """This test is to verify bgp fast-convergence cli functionality"""
 226
 227     tc_name = request.node.name
 228     write_test_header(tc_name)
 229     tgen = get_topogen()
 230
 231     # Verifying RIB routes
 232     dut = "r3"
 233     protocol = "bgp"
 234
 235     reset_config_on_routers(tgen)
 236     static_or_nw(tgen, topo, tc_name, test_type, "r2")
 237
 238     for addr_type in ADDR_TYPES:
 239         input_dict = {"r3": {"static_routes": [{"network": NETWORK[addr_type]}]}}
 240
 241         logger.info("Verifying %s routes on r3", addr_type)
 242         result = verify_rib(
 243             tgen,
 244             addr_type,
 245             dut,
 246             input_dict,
 247             protocol=protocol,
 248         )
 249         assert result is True, "Testcase {} : Failed \n Error: {}".format(
 250             tc_name, result
 251         )
 252
 253     intf1 = topo["routers"]["r2"]["links"]["r3-link1"]["interface"]
 254     intf2 = topo["routers"]["r2"]["links"]["r3-link2"]["interface"]
 255
 256     logger.info("Shutdown one of the link b/w r2 and r3")
 257     shutdown_bringup_interface(tgen, "r2", intf1, False)
 258
 259     logger.info("Verify bgp neighbors are still up")
 260     result = verify_bgp_convergence(tgen, topo)
 261     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 262
 263     logger.info("Shutdown another link b/w r2 and r3")
 264     shutdown_bringup_interface(tgen, "r2", intf2, False)
 265
 266     logger.info("Wait for 10 sec and make sure bgp neighbors are still up")
 267     sleep(10)
 268     result = verify_bgp_convergence(tgen, topo)
 269     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 270
 271     logger.info("No shut links b/w r2 and r3")
 272     shutdown_bringup_interface(tgen, "r2", intf1, True)
 273     shutdown_bringup_interface(tgen, "r2", intf2, True)
 274
 275     logger.info("Enable bgp fast-convergence cli")
 276     raw_config = {
 277        "r2": {
 278           "raw_config": [
 279             "router bgp {}".format(topo["routers"]["r2"]["bgp"]["local_as"]),
 280             "bgp fast-convergence",
 281           ]
 282        }
 283     }
 284     result = apply_raw_config(tgen, raw_config)
 285     assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
 286
 287     logger.info("Shutdown one link b/w r2 and r3")
 288     shutdown_bringup_interface(tgen, "r2", intf1, False)
 289
 290     logger.info("Verify bgp neighbors goes down immediately")
 291     result = verify_bgp_convergence(tgen, topo, dut="r2", expected=False)
 292     assert result is not True, "Testcase {} : Failed \n Error: {}".format(
 293         tc_name, result
 294     )
 295
 296     logger.info("Shutdown second link b/w r2 and r3")
 297     shutdown_bringup_interface(tgen, "r2", intf2, False)
 298
 299     logger.info("Verify bgp neighbors goes down immediately")
 300     result = verify_bgp_convergence(tgen, topo, dut="r2", expected=False)
 301     assert result is not True, "Testcase {} : Failed \n Error: {}".format(
 302         tc_name, result
 303     )
 304
 305     write_test_footer(tc_name)
 306
 307
 308 if __name__ == "__main__":
 309     args = ["-s"] + sys.argv[1:]
 310     sys.exit(pytest.main(args))