tests/topotests/ospf_gr_helper/test_ospf_gr_helper2.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
  30 # Save the Current Working Directory to find configuration files.
  31 CWD = os.path.dirname(os.path.realpath(__file__))
  32 sys.path.append(os.path.join(CWD, "../"))
  33
  34 # pylint: disable=C0413
  35 # Import topogen and topotest helpers
  36 from lib.topogen import Topogen, get_topogen
  37
  38 # Import topoJson from lib, to create topology and initial configuration
  39 from lib.common_config import (
  40     start_topology,
  41     write_test_header,
  42     write_test_footer,
  43     reset_config_on_routers,
  44     step,
  45     create_interfaces_cfg,
  46     topo_daemons,
  47     scapy_send_raw_packet,
  48 )
  49
  50 from lib.topolog import logger
  51 from lib.topojson import build_config_from_json
  52
  53 from lib.ospf import (
  54     verify_ospf_neighbor,
  55     clear_ospf,
  56     verify_ospf_gr_helper,
  57     create_router_ospf,
  58 )
  59
  60 # Global variables
  61 topo = None
  62 Iters = 5
  63 sw_name = None
  64 intf = None
  65 intf1 = None
  66 pkt = None
  67
  68 """
  69 Topology:
  70
  71       Please view in a fixed-width font such as Courier.
  72       Topo : Broadcast Networks
  73       DUT - HR      RR
  74         +---+       +---+          +---+           +---+
  75         |R0 +       +R1 +          +R2 +           +R3 |
  76         +-+-+       +-+-+          +-+-+           +-+-+
  77           |           |              |               |
  78           |           |              |               |
  79         --+-----------+--------------+---------------+-----
  80                          Ethernet Segment
  81
  82 Testcases:
  83
  84 TC1.    Verify by default helper support is disabled for FRR ospf
  85 TC2.    OSPF GR on Broadcast : Verify DUT enters Helper mode when neighbor
  86         sends grace lsa, helps RR to restart gracefully (RR = DR)
  87 TC3.    OSPF GR on Broadcast : Verify DUT enters Helper mode when neighbor
  88         sends grace lsa, helps RR to restart gracefully (RR = BDR)
  89 TC4.    OSPF GR on Broadcast : Verify DUT enters Helper mode when neighbor
  90         sends grace lsa, helps RR to restart gracefully (RR = DRother)
  91 TC5.    OSPF GR on P2P : Verify DUT enters Helper mode when neighbor sends
  92         grace lsa, helps RR to restart gracefully.
  93 TC6.    Verify all the show commands newly introducted as part of ospf
  94         helper support - Json Key verification wrt to show commands.
  95 TC7.    Verify helper when grace lsa is received with different configured
  96         value in process level (higher, lower, grace lsa timer above 1800)
  97 TC8.    Verify helper functionality when dut is helping RR and new grace lsa
  98         is received from RR.
  99 """
 100
 101
 102 def setup_module(mod):
 103     """
 104     Sets up the pytest environment
 105
 106     * `mod`: module name
 107     """
 108     global topo, intf, intf1, sw_name, pkt
 109     testsuite_run_time = time.asctime(time.localtime(time.time()))
 110     logger.info("Testsuite start time: {}".format(testsuite_run_time))
 111     logger.info("=" * 40)
 112
 113     logger.info("Running setup_module to create topology")
 114
 115     # This function initiates the topology build with Topogen...
 116     json_file = "{}/ospf_gr_helper.json".format(CWD)
 117     tgen = Topogen(json_file, mod.__name__)
 118     global topo
 119     topo = tgen.json_topo
 120     # ... and here it calls Mininet initialization functions.
 121
 122     # get list of daemons needs to be started for this suite.
 123     daemons = topo_daemons(tgen, topo)
 124
 125     # Starting topology, create tmp files which are loaded to routers
 126     #  to start daemons and then start routers
 127     start_topology(tgen, daemons)
 128
 129     # Creating configuration from JSON
 130     build_config_from_json(tgen, topo)
 131
 132     # Don't run this test if we have any failure.
 133     if tgen.routers_have_failure():
 134         pytest.skip(tgen.errors)
 135
 136     ospf_covergence = verify_ospf_neighbor(tgen, topo, lan=True)
 137     assert ospf_covergence is True, "setup_module :Failed \n Error:" " {}".format(
 138         ospf_covergence
 139     )
 140
 141     sw_name = "s1"
 142     intf = topo["routers"]["r0"]["links"][sw_name]["interface"]
 143     intf1 = topo["routers"]["r1"]["links"][sw_name]["interface"]
 144     pkt = topo["routers"]["r1"]["opq_lsa_hex"]
 145
 146     logger.info("Running setup_module() done")
 147
 148
 149 def teardown_module():
 150     """Teardown the pytest environment"""
 151
 152     logger.info("Running teardown_module to delete topology")
 153
 154     tgen = get_topogen()
 155
 156     try:
 157         # Stop toplogy and Remove tmp files
 158         tgen.stop_topology
 159
 160     except OSError:
 161         # OSError exception is raised when mininet tries to stop switch
 162         # though switch is stopped once but mininet tries to stop same
 163         # switch again, where it ended up with exception
 164         pass
 165
 166
 167 def delete_ospf():
 168     """delete ospf process after each test"""
 169     tgen = get_topogen()
 170     step("Delete ospf process")
 171     for rtr in topo["routers"]:
 172         ospf_del = {rtr: {"ospf": {"delete": True}}}
 173         result = create_router_ospf(tgen, topo, ospf_del)
 174         assert result is True, "Testcase: Failed \n Error: {}".format(result)
 175
 176
 177 # ##################################
 178 # Test cases start here.
 179 # ##################################
 180
 181
 182 def test_ospf_gr_helper_tc3_p1(request):
 183     """
 184     OSPF GR on Broadcast : Verify DUT enters Helper mode when neighbor
 185     sends grace lsa, helps RR to restart gracefully (RR = BDR)
 186     """
 187     tc_name = request.node.name
 188     write_test_header(tc_name)
 189     tgen = get_topogen()
 190
 191     # Don't run this test if we have any failure.
 192     if tgen.routers_have_failure():
 193         pytest.skip(tgen.errors)
 194
 195     global topo, intf, intf1, pkt
 196
 197     step("Bring up the base config as per the topology")
 198     step(
 199         "Configure DR priority as 99 in RR , DUT dr priority = 98 "
 200         "& reset ospf process in all the routers"
 201     )
 202     reset_config_on_routers(tgen)
 203     ospf_covergence = verify_ospf_neighbor(tgen, topo, lan=True)
 204     assert (
 205         ospf_covergence is True
 206     ), "OSPF is not after reset config \n Error:" " {}".format(ospf_covergence)
 207     step(
 208         "Configure DR pririty 100 on R0 and clear ospf neighbors " "on all the routers."
 209     )
 210
 211     input_dict = {
 212         "r0": {
 213             "links": {
 214                 sw_name: {
 215                     "interface": topo["routers"]["r0"]["links"][sw_name]["interface"],
 216                     "ospf": {"priority": 100},
 217                 }
 218             }
 219         }
 220     }
 221
 222     result = create_interfaces_cfg(tgen, input_dict)
 223     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 224
 225     step("Clear ospf neighbours in all routers")
 226     for rtr in topo["routers"]:
 227         clear_ospf(tgen, rtr)
 228
 229     step("Verify that DR election is triggered and R0 is elected as DR")
 230     input_dict = {
 231         "r0": {
 232             "ospf": {
 233                 "neighbors": {
 234                     "r1": {"state": "Full", "role": "Backup"},
 235                     "r2": {"state": "Full", "role": "DROther"},
 236                     "r3": {"state": "Full", "role": "DROther"},
 237                 }
 238             }
 239         }
 240     }
 241     dut = "r0"
 242     result = verify_ospf_neighbor(tgen, topo, dut, input_dict, lan=True)
 243     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 244
 245     ospf_gr_r0 = {
 246         "r0": {"ospf": {"graceful-restart": {"helper enable": [], "opaque": True}}}
 247     }
 248     result = create_router_ospf(tgen, topo, ospf_gr_r0)
 249     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 250
 251     ospf_gr_r1 = {
 252         "r1": {"ospf": {"graceful-restart": {"helper enable": [], "opaque": True}}}
 253     }
 254     result = create_router_ospf(tgen, topo, ospf_gr_r1)
 255     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 256
 257     step("Verify that DUT enters into helper mode.")
 258
 259     input_dict = {"activeRestarterCnt": 1}
 260     gracelsa_sent = False
 261     repeat = 0
 262     dut = "r0"
 263     while not gracelsa_sent and repeat < Iters:
 264         gracelsa_sent = scapy_send_raw_packet(tgen, topo, "r1", intf1, pkt)
 265         result = verify_ospf_gr_helper(tgen, topo, dut, input_dict)
 266         if isinstance(result, str):
 267             repeat += 1
 268             gracelsa_sent = False
 269
 270     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 271
 272     delete_ospf()
 273     write_test_footer(tc_name)
 274
 275
 276 def test_ospf_gr_helper_tc4_p1(request):
 277     """
 278     OSPF GR on Broadcast : Verify DUT enters Helper mode when neighbor
 279     sends grace lsa, helps RR to restart gracefully (RR = DRother)
 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, intf, intf1, pkt
 290
 291     step("Bring up the base config as per the topology")
 292     step(
 293         "Configure DR priority as 99 in RR , DUT dr priority = 98 "
 294         "& reset ospf process in all the routers"
 295     )
 296     reset_config_on_routers(tgen)
 297     ospf_covergence = verify_ospf_neighbor(tgen, topo, lan=True)
 298     assert (
 299         ospf_covergence is True
 300     ), "OSPF is not after reset config \n Error:" " {}".format(ospf_covergence)
 301     step(
 302         "Configure DR pririty 100 on R0 and clear ospf neighbors " "on all the routers."
 303     )
 304
 305     input_dict = {
 306         "r0": {
 307             "links": {
 308                 sw_name: {
 309                     "interface": topo["routers"]["r0"]["links"][sw_name]["interface"],
 310                     "ospf": {"priority": 0},
 311                 }
 312             }
 313         }
 314     }
 315
 316     result = create_interfaces_cfg(tgen, input_dict)
 317     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 318
 319     step("Clear ospf neighbours in all routers")
 320     for rtr in topo["routers"]:
 321         clear_ospf(tgen, rtr)
 322
 323     step("Verify that DR election is triggered and R0 is elected as 2-Way")
 324     input_dict = {
 325         "r0": {
 326             "ospf": {
 327                 "neighbors": {
 328                     "r1": {"state": "Full", "role": "DR"},
 329                     "r2": {"state": "2-Way", "role": "DROther"},
 330                     "r3": {"state": "2-Way", "role": "DROther"},
 331                 }
 332             }
 333         }
 334     }
 335     dut = "r0"
 336     result = verify_ospf_neighbor(tgen, topo, dut, input_dict, lan=True)
 337     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 338
 339     ospf_gr_r0 = {
 340         "r0": {"ospf": {"graceful-restart": {"helper enable": [], "opaque": True}}}
 341     }
 342     result = create_router_ospf(tgen, topo, ospf_gr_r0)
 343     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 344
 345     ospf_gr_r1 = {
 346         "r1": {"ospf": {"graceful-restart": {"helper enable": [], "opaque": True}}}
 347     }
 348     result = create_router_ospf(tgen, topo, ospf_gr_r1)
 349     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 350
 351     step("Verify that DUT enters into helper mode.")
 352
 353     input_dict = {"activeRestarterCnt": 1}
 354     gracelsa_sent = False
 355     repeat = 0
 356     dut = "r0"
 357     while not gracelsa_sent and repeat < Iters:
 358         gracelsa_sent = scapy_send_raw_packet(tgen, topo, "r1", intf1, pkt)
 359         result = verify_ospf_gr_helper(tgen, topo, dut, input_dict)
 360         if isinstance(result, str):
 361             repeat += 1
 362             gracelsa_sent = False
 363
 364     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 365
 366     delete_ospf()
 367
 368     write_test_footer(tc_name)
 369
 370
 371 if __name__ == "__main__":
 372     args = ["-s"] + sys.argv[1:]
 373     sys.exit(pytest.main(args))