tests/topotests/ospfv3_basic_functionality/test_ospfv3_ecmp.py

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