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