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 pytest
  32 import re
  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
  44 from lib.common_config import (
  45     write_test_header,
  46     write_test_footer,
  47     verify_rib,
  48     create_static_routes,
  49     check_address_types,
  50     reset_config_on_routers,
  51     shutdown_bringup_interface,
  52     apply_raw_config,
  53     start_topology,
  54 )
  55 from lib.topolog import logger
  56 from lib.topojson import build_config_from_json
  57 from lib.bgp import create_router_bgp, verify_bgp_convergence
  58
  59 pytestmark = [pytest.mark.bgpd, pytest.mark.staticd]
  60
  61
  62 # Global variables
  63 NEXT_HOPS = {"ipv4": [], "ipv6": []}
  64 NETWORK = {"ipv4": "192.168.1.10/32", "ipv6": "fd00:0:0:1::10/128"}
  65 NEXT_HOP_IP = {"ipv4": "10.0.0.1", "ipv6": "fd00::1"}
  66 BGP_CONVERGENCE = False
  67
  68
  69 def setup_module(mod):
  70     """
  71     Sets up the pytest environment.
  72
  73     * `mod`: module name
  74     """
  75     global ADDR_TYPES
  76
  77     testsuite_run_time = time.asctime(time.localtime(time.time()))
  78     logger.info("Testsuite start time: {}".format(testsuite_run_time))
  79     logger.info("=" * 40)
  80
  81     # This function initiates the topology build with Topogen...
  82     json_file = "{}/ibgp_ecmp_topo3.json".format(CWD)
  83     tgen = Topogen(json_file, mod.__name__)
  84     global topo
  85     topo = tgen.json_topo
  86
  87     # Starting topology, create tmp files which are loaded to routers
  88     #  to start daemons and then start routers
  89     start_topology(tgen)
  90
  91     # Creating configuration from JSON
  92     build_config_from_json(tgen, topo)
  93
  94     # Don't run this test if we have any failure.
  95     if tgen.routers_have_failure():
  96         pytest.skip(tgen.errors)
  97
  98     # Api call verify whether BGP is converged
  99     ADDR_TYPES = check_address_types()
 100
 101     BGP_CONVERGENCE = verify_bgp_convergence(tgen, topo)
 102     assert BGP_CONVERGENCE is True, "setup_module :Failed \n Error:" " {}".format(
 103         BGP_CONVERGENCE
 104     )
 105
 106     # STATIC_ROUTE = True
 107     logger.info("Running setup_module() done")
 108
 109
 110 def teardown_module():
 111     get_topogen().stop_topology()
 112
 113
 114 def static_or_nw(tgen, topo, tc_name, test_type, dut):
 115
 116     if test_type == "redist_static":
 117         input_dict_static = {
 118             dut: {
 119                 "static_routes": [
 120                     {"network": NETWORK["ipv4"], "next_hop": NEXT_HOP_IP["ipv4"]},
 121                     {"network": NETWORK["ipv6"], "next_hop": NEXT_HOP_IP["ipv6"]},
 122                 ]
 123             }
 124         }
 125         logger.info("Configuring static route on router %s", dut)
 126         result = create_static_routes(tgen, input_dict_static)
 127         assert result is True, "Testcase {} : Failed \n Error: {}".format(
 128             tc_name, result
 129         )
 130
 131         input_dict_2 = {
 132             dut: {
 133                 "bgp": {
 134                     "address_family": {
 135                         "ipv4": {
 136                             "unicast": {"redistribute": [{"redist_type": "static"}]}
 137                         },
 138                         "ipv6": {
 139                             "unicast": {"redistribute": [{"redist_type": "static"}]}
 140                         },
 141                     }
 142                 }
 143             }
 144         }
 145
 146         logger.info("Configuring redistribute static route on router %s", dut)
 147         result = create_router_bgp(tgen, topo, input_dict_2)
 148         assert result is True, "Testcase {} : Failed \n Error: {}".format(
 149             tc_name, result
 150         )
 151
 152     elif test_type == "advertise_nw":
 153         input_dict_nw = {
 154             dut: {
 155                 "bgp": {
 156                     "address_family": {
 157                         "ipv4": {
 158                             "unicast": {
 159                                 "advertise_networks": [{"network": NETWORK["ipv4"]}]
 160                             }
 161                         },
 162                         "ipv6": {
 163                             "unicast": {
 164                                 "advertise_networks": [{"network": NETWORK["ipv6"]}]
 165                             }
 166                         },
 167                     }
 168                 }
 169             }
 170         }
 171
 172         logger.info(
 173             "Advertising networks %s %s from router %s",
 174             NETWORK["ipv4"],
 175             NETWORK["ipv6"],
 176             dut,
 177         )
 178         result = create_router_bgp(tgen, topo, input_dict_nw)
 179         assert result is True, "Testcase {} : Failed \n Error: {}".format(
 180             tc_name, result
 181         )
 182
 183
 184 @pytest.mark.parametrize("test_type", ["redist_static"])
 185 def test_ecmp_fast_convergence(request, test_type, tgen, topo):
 186     """This test is to verify bgp fast-convergence cli functionality"""
 187
 188     tc_name = request.node.name
 189     write_test_header(tc_name)
 190
 191     # Verifying RIB routes
 192     dut = "r3"
 193     protocol = "bgp"
 194
 195     reset_config_on_routers(tgen)
 196     static_or_nw(tgen, topo, tc_name, test_type, "r2")
 197
 198     for addr_type in ADDR_TYPES:
 199         input_dict = {"r3": {"static_routes": [{"network": NETWORK[addr_type]}]}}
 200
 201         logger.info("Verifying %s routes on r3", addr_type)
 202         result = verify_rib(
 203             tgen,
 204             addr_type,
 205             dut,
 206             input_dict,
 207             protocol=protocol,
 208         )
 209         assert result is True, "Testcase {} : Failed \n Error: {}".format(
 210             tc_name, result
 211         )
 212
 213     intf1 = topo["routers"]["r2"]["links"]["r3-link1"]["interface"]
 214     intf2 = topo["routers"]["r2"]["links"]["r3-link2"]["interface"]
 215
 216     logger.info("Shutdown one of the link b/w r2 and r3")
 217     shutdown_bringup_interface(tgen, "r2", intf1, False)
 218
 219     logger.info("Verify bgp neighbors are still up")
 220     result = verify_bgp_convergence(tgen, topo)
 221     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 222
 223     logger.info("Shutdown another link b/w r2 and r3")
 224     shutdown_bringup_interface(tgen, "r2", intf2, False)
 225
 226     logger.info("Wait for 10 sec and make sure bgp neighbors are still up")
 227     sleep(10)
 228     result = verify_bgp_convergence(tgen, topo)
 229     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 230
 231     logger.info("No shut links b/w r2 and r3")
 232     shutdown_bringup_interface(tgen, "r2", intf1, True)
 233     shutdown_bringup_interface(tgen, "r2", intf2, True)
 234
 235     logger.info("Ensure that the links are still up")
 236     result = verify_bgp_convergence(tgen, topo)
 237
 238     logger.info("Enable bgp fast-convergence cli")
 239     raw_config = {
 240         "r2": {
 241             "raw_config": [
 242                 "router bgp {}".format(topo["routers"]["r2"]["bgp"]["local_as"]),
 243                 "bgp fast-convergence",
 244             ]
 245         }
 246     }
 247     result = apply_raw_config(tgen, raw_config)
 248     assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
 249
 250     logger.info("Ensure BGP has processed the cli")
 251     r2 = tgen.gears["r2"]
 252     output = r2.vtysh_cmd("show run")
 253     verify = re.search(r"fast-convergence", output)
 254     assert verify is not None, "r2 does not have the fast convergence command yet"
 255
 256     logger.info("Shutdown one link b/w r2 and r3")
 257     shutdown_bringup_interface(tgen, "r2", intf1, False)
 258
 259     logger.info("Verify bgp neighbors goes down immediately")
 260     result = verify_bgp_convergence(tgen, topo, dut="r2", expected=False)
 261     assert result is not True, (
 262         "Testcase {} : Failed \n "
 263         "Expected: BGP should not be converged for {} \n "
 264         "Found: {}".format(tc_name, "r2", result)
 265     )
 266
 267     logger.info("Shutdown second link b/w r2 and r3")
 268     shutdown_bringup_interface(tgen, "r2", intf2, False)
 269
 270     logger.info("Verify bgp neighbors goes down immediately")
 271     result = verify_bgp_convergence(tgen, topo, dut="r2", expected=False)
 272     assert result is not True, (
 273         "Testcase {} : Failed \n "
 274         "Expected: BGP should not be converged for {} \n "
 275         "Found: {}".format(tc_name, "r2", result)
 276     )
 277
 278     write_test_footer(tc_name)
 279
 280
 281 if __name__ == "__main__":
 282     args = ["-s"] + sys.argv[1:]
 283     sys.exit(pytest.main(args))