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