tests/topotests/ospf_basic_functionality/test_ospf_ecmp_lan.py

   1 #!/usr/bin/python
   2 # SPDX-License-Identifier: ISC
   3
   4 #
   5 # Copyright (c) 2020 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
  17 # Save the Current Working Directory to find configuration files.
  18 CWD = os.path.dirname(os.path.realpath(__file__))
  19 sys.path.append(os.path.join(CWD, "../"))
  20 sys.path.append(os.path.join(CWD, "../lib/"))
  21
  22 # pylint: disable=C0413
  23 # Import topogen and topotest helpers
  24 from lib.topogen import Topogen, get_topogen
  25
  26 # Import topoJson from lib, to create topology and initial configuration
  27 from lib.common_config import (
  28     start_topology,
  29     write_test_header,
  30     write_test_footer,
  31     reset_config_on_routers,
  32     verify_rib,
  33     create_static_routes,
  34     step,
  35 )
  36 from lib.topolog import logger
  37 from lib.topojson import build_config_from_json
  38
  39 from lib.ospf import (
  40     verify_ospf_neighbor,
  41     clear_ospf,
  42     verify_ospf_rib,
  43     redistribute_ospf,
  44 )
  45
  46 pytestmark = [pytest.mark.ospfd, pytest.mark.staticd]
  47
  48
  49 # Global variables
  50 topo = None
  51 # Reading the data from JSON File for topology creation
  52 NETWORK = {
  53     "ipv4": [
  54         "11.0.20.1/32",
  55         "11.0.20.2/32",
  56         "11.0.20.3/32",
  57         "11.0.20.4/32",
  58         "11.0.20.5/32",
  59     ],
  60     "ipv6": ["1::1/128", "1::2/128", "1::3/128", "1::4/128", "1::5/128"],
  61 }
  62 MASK = {"ipv4": "32", "ipv6": "128"}
  63 NEXT_HOP = {
  64     "ipv4": ["10.0.0.1", "10.0.1.1", "10.0.2.1", "10.0.3.1", "10.0.4.1"],
  65     "ipv6": ["Null0", "Null0", "Null0", "Null0", "Null0"],
  66 }
  67 """
  68 TOPOOLOGY =
  69       Please view in a fixed-width font such as Courier.
  70       Topo : Broadcast Networks
  71         +---+       +---+          +---+           +---+
  72         |R0 +       +R1 +          +R2 +           +R3 |
  73         +-+-+       +-+-+          +-+-+           +-+-+
  74           |           |              |               |
  75           |           |              |               |
  76         --+-----------+--------------+---------------+-----
  77                          Ethernet Segment
  78
  79 TESTCASES =
  80 1.  Verify OSPF ECMP with max path configured as 8
  81     (Edge having 1 uplink port as broadcast network,
  82     connect to 8 TORs - LAN case)
  83
  84  """
  85
  86
  87 def setup_module(mod):
  88     """
  89     Sets up the pytest environment
  90
  91     * `mod`: module name
  92     """
  93     testsuite_run_time = time.asctime(time.localtime(time.time()))
  94     logger.info("Testsuite start time: {}".format(testsuite_run_time))
  95     logger.info("=" * 40)
  96
  97     logger.info("Running setup_module to create topology")
  98
  99     # This function initiates the topology build with Topogen...
 100     json_file = "{}/ospf_ecmp_lan.json".format(CWD)
 101     tgen = Topogen(json_file, mod.__name__)
 102     global topo
 103     topo = tgen.json_topo
 104     # ... and here it calls Mininet initialization functions.
 105
 106     # Starting topology, create tmp files which are loaded to routers
 107     #  to start daemons and then start routers
 108     start_topology(tgen)
 109
 110     # Creating configuration from JSON
 111     build_config_from_json(tgen, topo)
 112
 113     # Don't run this test if we have any failure.
 114     if tgen.routers_have_failure():
 115         pytest.skip(tgen.errors)
 116     # Api call verify whether OSPF is converged
 117     ospf_covergence = verify_ospf_neighbor(tgen, topo, lan=True)
 118     assert ospf_covergence is True, "setup_module :Failed \n Error:" " {}".format(
 119         ospf_covergence
 120     )
 121
 122     logger.info("Running setup_module() done")
 123
 124
 125 def teardown_module():
 126     """Teardown the pytest environment"""
 127
 128     logger.info("Running teardown_module to delete topology")
 129
 130     tgen = get_topogen()
 131
 132     try:
 133         # Stop toplogy and Remove tmp files
 134         tgen.stop_topology()
 135
 136     except OSError:
 137         # OSError exception is raised when mininet tries to stop switch
 138         # though switch is stopped once but mininet tries to stop same
 139         # switch again, where it ended up with exception
 140         pass
 141
 142
 143 # ##################################
 144 # Test cases start here.
 145 # ##################################
 146
 147
 148 def test_ospf_lan_ecmp_tc18_p0(request):
 149     """
 150     OSPF ECMP.
 151
 152     Verify OSPF ECMP with max path configured as 8
 153     (Edge having 1 uplink port as broadcast network,
 154     connect to 8 TORs - LAN case)
 155
 156     """
 157     tc_name = request.node.name
 158     write_test_header(tc_name)
 159     tgen = get_topogen()
 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     global topo
 166     step("Bring up the base config as per the topology")
 167     step(". Configure ospf in all the routers on LAN interface.")
 168     reset_config_on_routers(tgen)
 169     step("Verify that OSPF is up with 8 neighborship sessions.")
 170
 171     ospf_covergence = verify_ospf_neighbor(tgen, topo, lan=True)
 172     assert ospf_covergence is True, "setup_module :Failed \n Error:" " {}".format(
 173         ospf_covergence
 174     )
 175
 176     step(
 177         "Configure a static route in all the routes and "
 178         "redistribute static/connected in OSPF."
 179     )
 180
 181     for rtr in topo["routers"]:
 182         input_dict = {
 183             rtr: {
 184                 "static_routes": [
 185                     {"network": NETWORK["ipv4"][0], "no_of_ip": 5, "next_hop": "Null0"}
 186                 ]
 187             }
 188         }
 189         result = create_static_routes(tgen, input_dict)
 190         assert result is True, "Testcase {} : Failed \n Error: {}".format(
 191             tc_name, result
 192         )
 193
 194         dut = rtr
 195         redistribute_ospf(tgen, topo, dut, "static")
 196
 197     step(
 198         "Verify that route in R0 in stalled with 8 hops. "
 199         "Verify ospf route table and ip route table."
 200     )
 201
 202     nh = []
 203     for rtr in topo["routers"]:
 204         nh.append(topo["routers"][rtr]["links"]["s1"]["ipv4"].split("/")[0])
 205     nh.remove(topo["routers"]["r1"]["links"]["s1"]["ipv4"].split("/")[0])
 206     dut = "r1"
 207     result = verify_ospf_rib(tgen, dut, input_dict, next_hop=nh)
 208     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 209
 210     protocol = "ospf"
 211     result = verify_rib(tgen, "ipv4", dut, input_dict, protocol=protocol, next_hop=nh)
 212     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 213
 214     step(" clear ip ospf interface on DUT(r0)")
 215     clear_ospf(tgen, "r0")
 216
 217     step(
 218         "Verify that after clearing the ospf interface all the "
 219         "neighbours are up and routes are installed with 8 next hop "
 220         "in ospf and ip route tables on R0"
 221     )
 222
 223     dut = "r0"
 224     ospf_covergence = verify_ospf_neighbor(tgen, topo, dut=dut, lan=True)
 225     assert ospf_covergence is True, "setup_module :Failed \n Error:" " {}".format(
 226         ospf_covergence
 227     )
 228
 229     step(" clear ip ospf interface on R2")
 230     clear_ospf(tgen, "r2")
 231
 232     dut = "r2"
 233     ospf_covergence = verify_ospf_neighbor(tgen, topo, dut=dut, lan=True)
 234     assert ospf_covergence is True, "setup_module :Failed \n Error:" " {}".format(
 235         ospf_covergence
 236     )
 237
 238     step("Delete static/connected cmd in ospf in all the routes one by one.")
 239     for rtr in topo["routers"]:
 240         input_dict = {
 241             rtr: {
 242                 "static_routes": [
 243                     {
 244                         "network": NETWORK["ipv4"][0],
 245                         "no_of_ip": 5,
 246                         "next_hop": "Null0",
 247                         "delete": True,
 248                     }
 249                 ]
 250             }
 251         }
 252         result = create_static_routes(tgen, input_dict)
 253         assert result is True, "Testcase {} : Failed \n Error: {}".format(
 254             tc_name, result
 255         )
 256
 257     step("Verify that all the routes are withdrawn from R0")
 258     dut = "r1"
 259     result = verify_ospf_rib(
 260         tgen, dut, input_dict, next_hop=nh, retry_timeout=10, expected=False
 261     )
 262     assert (
 263         result is not True
 264     ), "Testcase {} : Failed \n " "r1: OSPF routes are present \n Error: {}".format(
 265         tc_name, result
 266     )
 267
 268     protocol = "ospf"
 269     result = verify_rib(
 270         tgen,
 271         "ipv4",
 272         dut,
 273         input_dict,
 274         protocol=protocol,
 275         next_hop=nh,
 276         retry_timeout=10,
 277         expected=False,
 278     )
 279     assert (
 280         result is not True
 281     ), "Testcase {} : Failed \n " "r1: routes are still present \n Error: {}".format(
 282         tc_name, result
 283     )
 284
 285     write_test_footer(tc_name)
 286
 287
 288 if __name__ == "__main__":
 289     args = ["-s"] + sys.argv[1:]
 290     sys.exit(pytest.main(args))