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