tests/topotests/ospf_basic_functionality/test_ospf_ecmp.py

   1 #!/usr/bin/python
   2
   3 #
   4 # Copyright (c) 2020 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 """OSPF Basic Functionality Automation."""
  25 import os
  26 import sys
  27 import time
  28 import pytest
  29 from time import sleep
  30
  31 # Save the Current Working Directory to find configuration files.
  32 CWD = os.path.dirname(os.path.realpath(__file__))
  33 sys.path.append(os.path.join(CWD, "../"))
  34 sys.path.append(os.path.join(CWD, "../lib/"))
  35
  36 # pylint: disable=C0413
  37 # Import topogen and topotest helpers
  38 from lib.topogen import Topogen, get_topogen
  39
  40 # Import topoJson from lib, to create topology and initial configuration
  41 from lib.common_config import (
  42     start_topology,
  43     write_test_header,
  44     write_test_footer,
  45     reset_config_on_routers,
  46     verify_rib,
  47     create_static_routes,
  48     step,
  49     shutdown_bringup_interface,
  50 )
  51 from lib.topolog import logger
  52
  53 from lib.topojson import build_config_from_json
  54 from lib.ospf import (
  55     verify_ospf_neighbor,
  56     config_ospf_interface,
  57     verify_ospf_rib,
  58     redistribute_ospf,
  59 )
  60
  61 pytestmark = [pytest.mark.ospfd, pytest.mark.staticd]
  62
  63
  64 topo = None
  65
  66
  67 # Global variables
  68 NETWORK = {
  69     "ipv4": [
  70         "11.0.20.1/32",
  71         "11.0.20.2/32",
  72         "11.0.20.3/32",
  73         "11.0.20.4/32",
  74         "11.0.20.5/32",
  75     ]
  76 }
  77 """
  78 TOPOLOGY :
  79       Please view in a fixed-width font such as Courier.
  80       +---+  A1       +---+
  81       +R1 +------------+R2 |
  82       +-+-+-           +--++
  83         |  --        --  |
  84         |    -- A0 --    |
  85       A0|      ----      |
  86         |      ----      | A2
  87         |    --    --    |
  88         |  --        --  |
  89       +-+-+-            +-+-+
  90       +R0 +-------------+R3 |
  91       +---+     A3     +---+
  92
  93 TESTCASES :
  94 1. Verify OSPF ECMP with max path configured as 8 (ECMPconfigured at FRR level)
  95 2. Verify OSPF ECMP with max path configured as 2 (Edge having 2 uplink ports)
  96  """
  97
  98
  99 def setup_module(mod):
 100     """
 101     Sets up the pytest environment
 102
 103     * `mod`: module name
 104     """
 105     testsuite_run_time = time.asctime(time.localtime(time.time()))
 106     logger.info("Testsuite start time: {}".format(testsuite_run_time))
 107     logger.info("=" * 40)
 108
 109     logger.info("Running setup_module to create topology")
 110
 111     # This function initiates the topology build with Topogen...
 112     json_file = "{}/ospf_ecmp.json".format(CWD)
 113     tgen = Topogen(json_file, mod.__name__)
 114     global topo
 115     topo = tgen.json_topo
 116     # ... and here it calls Mininet initialization functions.
 117
 118     # Starting topology, create tmp files which are loaded to routers
 119     #  to start daemons and then start routers
 120     start_topology(tgen)
 121
 122     # Creating configuration from JSON
 123     build_config_from_json(tgen, topo)
 124
 125     # Don't run this test if we have any failure.
 126     if tgen.routers_have_failure():
 127         pytest.skip(tgen.errors)
 128     # Api call verify whether OSPF is converged
 129     ospf_covergence = verify_ospf_neighbor(tgen, topo)
 130     assert ospf_covergence is True, "setup_module :Failed \n Error  {}".format(
 131         ospf_covergence
 132     )
 133
 134     logger.info("Running setup_module() done")
 135
 136
 137 def teardown_module(mod):
 138     """
 139     Teardown the pytest environment.
 140
 141     * `mod`: module name
 142     """
 143
 144     logger.info("Running teardown_module to delete topology")
 145
 146     tgen = get_topogen()
 147
 148     # Stop toplogy and Remove tmp files
 149     tgen.stop_topology()
 150
 151     logger.info(
 152         "Testsuite end time: {}".format(time.asctime(time.localtime(time.time())))
 153     )
 154     logger.info("=" * 40)
 155
 156
 157 # ##################################
 158 # Test cases start here.
 159 # ##################################
 160
 161
 162 def test_ospf_ecmp_tc16_p0(request):
 163     """
 164     Verify OSPF ECMP.
 165
 166     Verify OSPF ECMP with max path configured as 8 (ECMP
 167     configured at FRR level)
 168     """
 169     tc_name = request.node.name
 170     write_test_header(tc_name)
 171     tgen = get_topogen()
 172
 173     # Don't run this test if we have any failure.
 174     if tgen.routers_have_failure():
 175         pytest.skip(tgen.errors)
 176
 177     global topo
 178     step("Bring up the base config as per the topology")
 179     step("Configure 8 interfaces between R1 and R2 and enable ospf in area 0.")
 180     reset_config_on_routers(tgen)
 181     step("Verify that OSPF is up with 8 neighborship sessions.")
 182     dut = "r1"
 183     ospf_covergence = verify_ospf_neighbor(tgen, topo, dut=dut)
 184     assert ospf_covergence is True, "Testcase Failed \n Error  {}".format(
 185         ospf_covergence
 186     )
 187
 188     step("Configure a static route in R0 and redistribute in OSPF.")
 189
 190     input_dict = {
 191         "r0": {
 192             "static_routes": [
 193                 {
 194                     "network": NETWORK["ipv4"][0],
 195                     "no_of_ip": 5,
 196                     "next_hop": "Null0",
 197                 }
 198             ]
 199         }
 200     }
 201     result = create_static_routes(tgen, input_dict)
 202     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 203
 204     dut = "r0"
 205     redistribute_ospf(tgen, topo, dut, "static")
 206
 207     step("Verify that route in R2 in stalled with 8 next hops.")
 208     nh = []
 209     for item in range(1, 7):
 210         nh.append(topo["routers"]["r0"]["links"]["r1-link1"]["ipv4"].split("/")[0])
 211
 212     nh2 = topo["routers"]["r0"]["links"]["r1"]["ipv4"].split("/")[0]
 213
 214     nh.append(nh2)
 215
 216     dut = "r1"
 217     result = verify_ospf_rib(tgen, dut, input_dict, next_hop=nh)
 218     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 219
 220     protocol = "ospf"
 221     result = verify_rib(tgen, "ipv4", dut, input_dict, protocol=protocol, next_hop=nh)
 222     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 223
 224     step("shut no shut all the interfaces on the remote router - R2")
 225     dut = "r1"
 226     for intfr in range(1, 7):
 227         intf = topo["routers"]["r1"]["links"]["r0-link{}".format(intfr)]["interface"]
 228         shutdown_bringup_interface(tgen, dut, intf, False)
 229
 230     result = verify_ospf_rib(tgen, dut, input_dict, next_hop=nh, expected=False)
 231     assert (
 232         result is not True
 233     ), "Testcase {} : Failed \n  r1: OSPF routes are present \n Error: {}".format(
 234         tc_name, result
 235     )
 236
 237     protocol = "ospf"
 238     result = verify_rib(
 239         tgen, "ipv4", dut, input_dict, protocol=protocol, next_hop=nh, expected=False
 240     )
 241     assert (
 242         result is not True
 243     ), "Testcase {} : Failed \n  r1: routes are still present \n Error: {}".format(
 244         tc_name, result
 245     )
 246
 247     for intfr in range(1, 7):
 248         intf = topo["routers"]["r1"]["links"]["r0-link{}".format(intfr)]["interface"]
 249         shutdown_bringup_interface(tgen, dut, intf, True)
 250
 251     result = verify_ospf_rib(tgen, dut, input_dict, next_hop=nh)
 252     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 253
 254     protocol = "ospf"
 255     result = verify_rib(tgen, "ipv4", dut, input_dict, protocol=protocol, next_hop=nh)
 256     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 257
 258     step("shut no shut on all the interfaces on DUT (r1)")
 259     for intfr in range(1, 7):
 260         intf = topo["routers"]["r1"]["links"]["r0-link{}".format(intfr)]["interface"]
 261         shutdown_bringup_interface(tgen, dut, intf, False)
 262
 263     for intfr in range(1, 7):
 264         intf = topo["routers"]["r1"]["links"]["r0-link{}".format(intfr)]["interface"]
 265         shutdown_bringup_interface(tgen, dut, intf, True)
 266
 267     step(
 268         "Verify that all the neighbours are up and routes are installed"
 269         " with 8 next hop in ospf and ip route tables on R1."
 270     )
 271
 272     step("Verify that OSPF is up with 8 neighborship sessions.")
 273     dut = "r1"
 274     ospf_covergence = verify_ospf_neighbor(tgen, topo, dut=dut)
 275     assert ospf_covergence is True, "Testcase Failed \n Error  {}".format(
 276         ospf_covergence
 277     )
 278
 279     dut = "r1"
 280     result = verify_ospf_rib(tgen, dut, input_dict, next_hop=nh)
 281     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 282
 283     protocol = "ospf"
 284     result = verify_rib(tgen, "ipv4", dut, input_dict, protocol=protocol, next_hop=nh)
 285     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 286
 287     step(" Un configure static route on R0")
 288
 289     dut = "r0"
 290     redistribute_ospf(tgen, topo, dut, "static", delete=True)
 291
 292     # Wait for R0 to flush external LSAs.
 293     sleep(10)
 294
 295     step("Verify that route is withdrawn from R2.")
 296     dut = "r1"
 297     result = verify_ospf_rib(
 298         tgen, dut, input_dict, next_hop=nh, retry_timeout=10, expected=False
 299     )
 300     assert (
 301         result is not True
 302     ), "Testcase {} : Failed \n  r1: OSPF routes are present \n Error: {}".format(
 303         tc_name, result
 304     )
 305
 306     protocol = "ospf"
 307     result = verify_rib(
 308         tgen,
 309         "ipv4",
 310         dut,
 311         input_dict,
 312         protocol=protocol,
 313         next_hop=nh,
 314         retry_timeout=10,
 315         expected=False,
 316     )
 317     assert (
 318         result is not True
 319     ), "Testcase {} : Failed \n  r1: routes are still present \n Error: {}".format(
 320         tc_name, result
 321     )
 322
 323     step("Re configure the static route in R0.")
 324     dut = "r0"
 325     redistribute_ospf(tgen, topo, dut, "static")
 326
 327     dut = "r1"
 328     result = verify_ospf_rib(tgen, dut, input_dict, next_hop=nh)
 329     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 330
 331     protocol = "ospf"
 332     result = verify_rib(tgen, "ipv4", dut, input_dict, protocol=protocol, next_hop=nh)
 333     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 334
 335     write_test_footer(tc_name)
 336
 337
 338 def test_ospf_ecmp_tc17_p0(request):
 339     """
 340     Verify OSPF ECMP.
 341
 342     Verify OSPF ECMP with max path configured as 2 (Edge having 2 uplink ports)
 343     """
 344     tc_name = request.node.name
 345     write_test_header(tc_name)
 346     tgen = get_topogen()
 347
 348     # Don't run this test if we have any failure.
 349     if tgen.routers_have_failure():
 350         pytest.skip(tgen.errors)
 351
 352     global topo
 353     step("Bring up the base config as per the topology")
 354     step("Configure 2 interfaces between R1 and R2 & enable ospf in area 0.")
 355     reset_config_on_routers(tgen)
 356     step("Verify that OSPF is up with 2 neighborship sessions.")
 357     dut = "r1"
 358     ospf_covergence = verify_ospf_neighbor(tgen, topo, dut=dut)
 359     assert ospf_covergence is True, "Testcase Failed \n Error  {}".format(
 360         ospf_covergence
 361     )
 362
 363     step("Configure a static route in R0 and redistribute in OSPF.")
 364
 365     input_dict = {
 366         "r0": {
 367             "static_routes": [
 368                 {
 369                     "network": NETWORK["ipv4"][0],
 370                     "no_of_ip": 5,
 371                     "next_hop": "Null0",
 372                 }
 373             ]
 374         }
 375     }
 376     result = create_static_routes(tgen, input_dict)
 377     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 378
 379     dut = "r0"
 380     redistribute_ospf(tgen, topo, dut, "static")
 381
 382     step("Verify that route in R2 in stalled with 2 next hops.")
 383
 384     nh1 = topo["routers"]["r0"]["links"]["r1-link1"]["ipv4"].split("/")[0]
 385     nh2 = topo["routers"]["r0"]["links"]["r1"]["ipv4"].split("/")[0]
 386     nh = [nh1, nh2]
 387
 388     dut = "r1"
 389     result = verify_ospf_rib(tgen, dut, input_dict, next_hop=nh)
 390     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 391
 392     protocol = "ospf"
 393     result = verify_rib(tgen, "ipv4", dut, input_dict, protocol=protocol, next_hop=nh)
 394     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 395
 396     step(" Un configure static route on R0")
 397
 398     dut = "r0"
 399     redistribute_ospf(tgen, topo, dut, "static", delete=True)
 400     # sleep till the route gets withdrawn
 401     sleep(10)
 402
 403     step("Verify that route is withdrawn from R2.")
 404     dut = "r1"
 405     result = verify_ospf_rib(
 406         tgen, dut, input_dict, next_hop=nh, retry_timeout=10, expected=False
 407     )
 408     assert (
 409         result is not True
 410     ), "Testcase {} : Failed \n  r1: OSPF routes are present \n Error: {}".format(
 411         tc_name, result
 412     )
 413
 414     protocol = "ospf"
 415     result = verify_rib(
 416         tgen,
 417         "ipv4",
 418         dut,
 419         input_dict,
 420         protocol=protocol,
 421         next_hop=nh,
 422         retry_timeout=10,
 423         expected=False,
 424     )
 425     assert (
 426         result is not True
 427     ), "Testcase {} : Failed \n  r1: routes are still present \n Error: {}".format(
 428         tc_name, result
 429     )
 430
 431     step("Reconfigure the static route in R0.Change ECMP value to 2.")
 432     dut = "r0"
 433     redistribute_ospf(tgen, topo, dut, "static")
 434
 435     step("Configure cost on R0 as 100")
 436     r0_ospf_cost = {"r0": {"links": {"r1": {"ospf": {"cost": 100}}}}}
 437     result = config_ospf_interface(tgen, topo, r0_ospf_cost)
 438     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 439
 440     dut = "r1"
 441     result = verify_ospf_rib(tgen, dut, input_dict, next_hop=nh)
 442     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 443
 444     protocol = "ospf"
 445     result = verify_rib(tgen, "ipv4", dut, input_dict, protocol=protocol, next_hop=nh)
 446     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 447
 448     write_test_footer(tc_name)
 449
 450
 451 if __name__ == "__main__":
 452     args = ["-s"] + sys.argv[1:]
 453     sys.exit(pytest.main(args))