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