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