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