tests/topotests/ospfv3_basic_functionality/test_ospfv3_ecmp_lan.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 import ipaddress
  34
  35 # Save the Current Working Directory to find configuration files.
  36 CWD = os.path.dirname(os.path.realpath(__file__))
  37 sys.path.append(os.path.join(CWD, "../"))
  38 sys.path.append(os.path.join(CWD, "../lib/"))
  39
  40 # pylint: disable=C0413
  41 # Import topogen and topotest helpers
  42 from lib.topogen import Topogen, get_topogen
  43
  44 # Import topoJson from lib, to create topology and initial configuration
  45 from lib.common_config import (
  46     start_topology,
  47     write_test_header,
  48     write_test_footer,
  49     reset_config_on_routers,
  50     verify_rib,
  51     create_static_routes,
  52     step,
  53     create_route_maps,
  54     shutdown_bringup_interface,
  55     create_interfaces_cfg,
  56     get_frr_ipv6_linklocal,
  57 )
  58 from lib.topolog import logger
  59 from lib.topojson import build_config_from_json
  60
  61 from lib.ospf import (
  62     verify_ospf6_neighbor,
  63     config_ospf_interface,
  64     clear_ospf,
  65     verify_ospf6_rib,
  66     create_router_ospf,
  67     verify_ospf6_interface,
  68     verify_ospf6_database,
  69     config_ospf6_interface,
  70 )
  71
  72 from ipaddress import IPv6Address
  73
  74 pytestmark = [pytest.mark.ospfd, pytest.mark.staticd]
  75
  76
  77 # Global variables
  78 topo = None
  79
  80 NETWORK = {
  81     "ipv4": [
  82         "11.0.20.1/32",
  83         "11.0.20.2/32",
  84         "11.0.20.3/32",
  85         "11.0.20.4/32",
  86         "11.0.20.5/32",
  87     ],
  88     "ipv6": ["2::1/128", "2::2/128", "2::3/128", "2::4/128", "2::5/128"],
  89 }
  90 MASK = {"ipv6": "32", "ipv6": "128"}
  91
  92 """
  93 TOPOOLOGY =
  94       Please view in a fixed-width font such as Courier.
  95       Topo : Broadcast Networks
  96         +---+       +---+          +---+           +---+
  97         |R0 +       +R1 +          +R2 +           +R3 |
  98         +-+-+       +-+-+          +-+-+           +-+-+
  99           |           |              |               |
 100           |           |              |               |
 101         --+-----------+--------------+---------------+-----
 102                          Ethernet Segment
 103
 104 TESTCASES =
 105 1.  Verify OSPF ECMP with max path configured as 8
 106     (Edge having 1 uplink port as broadcast network,
 107     connect to 8 TORs - LAN case)
 108
 109  """
 110
 111
 112 def setup_module(mod):
 113     """
 114     Sets up the pytest environment
 115
 116     * `mod`: module name
 117     """
 118     global topo, switch_name
 119     testsuite_run_time = time.asctime(time.localtime(time.time()))
 120     logger.info("Testsuite start time: {}".format(testsuite_run_time))
 121     logger.info("=" * 40)
 122
 123     logger.info("Running setup_module to create topology")
 124
 125     # This function initiates the topology build with Topogen...
 126     json_file = "{}/ospfv3_ecmp_lan.json".format(CWD)
 127     tgen = Topogen(json_file, mod.__name__)
 128     global topo
 129     topo = tgen.json_topo
 130     # ... and here it calls Mininet initialization functions.
 131
 132     # Starting topology, create tmp files which are loaded to routers
 133     #  to start daemons and then start routers
 134     start_topology(tgen)
 135
 136     # Creating configuration from JSON
 137     build_config_from_json(tgen, topo)
 138
 139     # Don't run this test if we have any failure.
 140     if tgen.routers_have_failure():
 141         pytest.skip(tgen.errors)
 142     # Api call verify whether OSPF is converged
 143     ospf_covergence = verify_ospf6_neighbor(tgen, topo, lan=True)
 144     assert ospf_covergence is True, "setup_module :Failed \n Error:  {}".format(
 145         ospf_covergence
 146     )
 147
 148     switch_name = [k for k in topo["switches"].keys()][0]
 149
 150     logger.info("Running setup_module() done")
 151
 152
 153 def teardown_module(mod):
 154     """
 155     Teardown the pytest environment.
 156
 157     * `mod`: module name
 158     """
 159
 160     logger.info("Running teardown_module to delete topology")
 161
 162     tgen = get_topogen()
 163
 164     # Stop toplogy and Remove tmp files
 165
 166     try:
 167         # Stop toplogy and Remove tmp files
 168         tgen.stop_topology()
 169
 170     except OSError:
 171         # OSError exception is raised when mininet tries to stop switch
 172         # though switch is stopped once but mininet tries to stop same
 173         # switch again, where it ended up with exception
 174         pass
 175     logger.info(
 176         "Testsuite end time: {}".format(time.asctime(time.localtime(time.time())))
 177     )
 178     logger.info("=" * 40)
 179
 180
 181 def red_static(dut, config=True):
 182     """Local def for Redstribute static routes inside ospf."""
 183     global topo
 184     tgen = get_topogen()
 185     if config:
 186         ospf_red = {dut: {"ospf6": {"redistribute": [{"redist_type": "static"}]}}}
 187     else:
 188         ospf_red = {
 189             dut: {
 190                 "ospf6": {"redistribute": [{"redist_type": "static", "delete": True}]}
 191             }
 192         }
 193     result = create_router_ospf(tgen, topo, ospf_red)
 194     assert result is True, "Testcase : Failed \n Error: {}".format(result)
 195
 196
 197 def red_connected(dut, config=True):
 198     """Local def for Redstribute connected routes inside ospf."""
 199     global topo
 200     tgen = get_topogen()
 201     if config:
 202         ospf_red = {dut: {"ospf6": {"redistribute": [{"redist_type": "connected"}]}}}
 203     else:
 204         ospf_red = {
 205             dut: {
 206                 "ospf6": {
 207                     "redistribute": [{"redist_type": "connected", "del_action": True}]
 208                 }
 209             }
 210         }
 211     result = create_router_ospf(tgen, topo, ospf_red)
 212     assert result is True, "Testcase: Failed \n Error: {}".format(result)
 213
 214
 215 def get_llip(onrouter, intf):
 216     """
 217     API to get the link local ipv6 address of a particular interface
 218
 219     Parameters
 220     ----------
 221     * `fromnode`: Source node
 222     * `tonode` : interface for which link local ip needs to be returned.
 223
 224     Usage
 225     -----
 226     result = get_llip('r1', 'r2-link0')
 227
 228     Returns
 229     -------
 230     1) link local ipv6 address from the interface.
 231     2) errormsg - when link local ip not found.
 232     """
 233     tgen = get_topogen()
 234     intf = topo["routers"][onrouter]["links"][intf]["interface"]
 235     llip = get_frr_ipv6_linklocal(tgen, onrouter, intf)
 236     if llip:
 237         logger.info("llip ipv6 address to be set as NH is %s", llip)
 238         return llip
 239     return None
 240
 241
 242 def get_glipv6(onrouter, intf):
 243     """
 244     API to get the global ipv6 address of a particular interface
 245
 246     Parameters
 247     ----------
 248     * `onrouter`: Source node
 249     * `intf` : interface for which link local ip needs to be returned.
 250
 251     Usage
 252     -----
 253     result = get_glipv6('r1', 'r2-link0')
 254
 255     Returns
 256     -------
 257     1) global ipv6 address from the interface.
 258     2) errormsg - when link local ip not found.
 259     """
 260     glipv6 = (topo["routers"][onrouter]["links"][intf]["ipv6"]).split("/")[0]
 261     if glipv6:
 262         logger.info("Global ipv6 address to be set as NH is %s", glipv6)
 263         return glipv6
 264     return None
 265
 266
 267 # ##################################
 268 # Test cases start here.
 269 # ##################################
 270
 271
 272 def test_ospfv3_lan_ecmp_tc18_p0(request):
 273     """
 274     OSPF ECMP.
 275
 276     Verify OSPF ECMP with max path configured as 8
 277     (Edge having 1 uplink port as broadcast network,
 278     connect to 8 TORs - LAN case)
 279
 280     """
 281     tc_name = request.node.name
 282     write_test_header(tc_name)
 283     tgen = get_topogen()
 284
 285     # Don't run this test if we have any failure.
 286     if tgen.routers_have_failure():
 287         pytest.skip(tgen.errors)
 288
 289     global topo
 290     step("Bring up the base config as per the topology")
 291     step(". Configure ospf in all the routers on LAN interface.")
 292
 293     reset_config_on_routers(tgen)
 294
 295     step("Verify that OSPF is up with 8 neighborship sessions.")
 296
 297     ospf_covergence = verify_ospf6_neighbor(tgen, topo, lan=True)
 298     assert ospf_covergence is True, "Testcase Failed \n Error:  {}".format(
 299         ospf_covergence
 300     )
 301
 302     step(
 303         "Configure a static route in all the routes and "
 304         "redistribute static/connected in OSPF."
 305     )
 306
 307     for rtr in topo["routers"]:
 308         input_dict = {
 309             rtr: {
 310                 "static_routes": [
 311                     {"network": NETWORK["ipv6"][0], "no_of_ip": 5, "next_hop": "Null0"}
 312                 ]
 313             }
 314         }
 315         result = create_static_routes(tgen, input_dict)
 316         assert result is True, "Testcase {} : Failed \n Error: {}".format(
 317             tc_name, result
 318         )
 319
 320         dut = rtr
 321         red_static(dut)
 322
 323     step(
 324         "Verify that route in R0 in stalled with 8 hops. "
 325         "Verify ospf route table and ip route table."
 326     )
 327
 328     nh = []
 329     for rtr in topo["routers"]:
 330         llip = get_llip(rtr, switch_name)
 331         assert llip is not None, "Testcase {} : Failed \n Error: {}".format(
 332             tc_name, result
 333         )
 334         nh.append(llip)
 335
 336     llip = get_llip("r1", switch_name)
 337     assert llip is not None, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 338
 339     nh.remove(llip)
 340     dut = "r1"
 341     result = verify_ospf6_rib(tgen, dut, input_dict)
 342     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 343
 344     protocol = "ospf6"
 345     result = verify_rib(tgen, "ipv6", dut, input_dict, protocol=protocol)
 346     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 347
 348     step(" clear ip ospf interface on DUT(r0)")
 349     clear_ospf(tgen, "r0", ospf="ospf6")
 350
 351     step(
 352         "Verify that after clearing the ospf interface all the "
 353         "neighbours are up and routes are installed with 8 next hop "
 354         "in ospf and ip route tables on R0"
 355     )
 356
 357     dut = "r0"
 358     ospf_covergence = verify_ospf6_neighbor(tgen, topo, dut=dut, lan=True)
 359     assert ospf_covergence is True, "Testcase Failed \n Error:  {}".format(
 360         ospf_covergence
 361     )
 362
 363     step(" clear ip ospf interface on R2")
 364     clear_ospf(tgen, "r2")
 365
 366     dut = "r2"
 367     ospf_covergence = verify_ospf6_neighbor(tgen, topo, dut=dut, lan=True)
 368     assert ospf_covergence is True, "Testcase Failed \n Error:  {}".format(
 369         ospf_covergence
 370     )
 371
 372     step("Delete static/connected cmd in ospf in all the routes one by one.")
 373     for rtr in topo["routers"]:
 374         input_dict = {
 375             rtr: {
 376                 "static_routes": [
 377                     {
 378                         "network": NETWORK["ipv6"][0],
 379                         "no_of_ip": 5,
 380                         "next_hop": "Null0",
 381                         "delete": True,
 382                     }
 383                 ]
 384             }
 385         }
 386         result = create_static_routes(tgen, input_dict)
 387         assert result is True, "Testcase {} : Failed \n Error: {}".format(
 388             tc_name, result
 389         )
 390
 391     write_test_footer(tc_name)
 392
 393
 394 if __name__ == "__main__":
 395     args = ["-s"] + sys.argv[1:]
 396     sys.exit(pytest.main(args))