tests/topotests/ospfv3_basic_functionality/test_ospfv3_ecmp_lan.py

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