tests/topotests/ospf_basic_functionality/test_ospf_ecmp.py

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