tests/topotests/ospf_gr_helper/test_ospf_gr_helper3.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_tc7_p1(request):
 185     """
 186     Test ospf gr helper
 187     Verify helper when grace lsa is received with different configured
 188     value in process level (higher, lower, grace lsa timer above 1800)
 189     """
 190     tc_name = request.node.name
 191     write_test_header(tc_name)
 192     tgen = get_topogen()
 193
 194     # Don't run this test if we have any failure.
 195     if tgen.routers_have_failure():
 196         pytest.skip(tgen.errors)
 197
 198     global topo, intf, intf1, pkt
 199
 200     step("Bring up the base config as per the topology")
 201     step(
 202         "Configure DR priority as 99 in RR , DUT dr priority = 98 "
 203         "& reset ospf process in all the routers"
 204     )
 205     step(
 206         "Enable GR on RR and DUT with grace period on RR = 333"
 207         "and grace period on DUT = 300"
 208     )
 209     reset_config_on_routers(tgen)
 210     ospf_covergence = verify_ospf_neighbor(tgen, topo, lan=True)
 211     assert (
 212         ospf_covergence is True
 213     ), "OSPF is not after reset config \n Error:" " {}".format(ospf_covergence)
 214     ospf_gr_r0 = {
 215         "r0": {"ospf": {"graceful-restart": {"helper enable": [], "opaque": True}}}
 216     }
 217     result = create_router_ospf(tgen, topo, ospf_gr_r0)
 218     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 219
 220     ospf_gr_r1 = {
 221         "r1": {"ospf": {"graceful-restart": {"helper enable": [], "opaque": True}}}
 222     }
 223     result = create_router_ospf(tgen, topo, ospf_gr_r1)
 224     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 225
 226     input_dict = {"supportedGracePeriod": 1800}
 227     dut = "r0"
 228     result = verify_ospf_gr_helper(tgen, topo, dut, input_dict)
 229     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 230
 231     step("Configure grace period = 1801 on RR and restart ospf .")
 232     grace_period_1801 = "01005e00000570708bd051ef080045c0005cbeb10000015907d111010101e00000050204004801010101000000009714000000000000000000000000000100010209030000000101010180000001c8e9002c000100040000016800020001010000000003000411010101"
 233     gracelsa_sent = scapy_send_raw_packet(tgen, topo, "r1", intf1, grace_period_1801)
 234
 235     step("Verify R0 does not enter helper mode.")
 236     input_dict = {"activeRestarterCnt": 1}
 237     dut = "r0"
 238     result = verify_ospf_gr_helper(tgen, topo, dut, input_dict, expected=False)
 239     assert (
 240         result is not True
 241     ), "Testcase {} : Failed. DUT entered helper role " " \n Error: {}".format(
 242         tc_name, result
 243     )
 244
 245     delete_ospf()
 246
 247     write_test_footer(tc_name)
 248
 249
 250 def test_ospf_gr_helper_tc8_p1(request):
 251     """
 252     Test ospf gr helper
 253
 254     Verify helper functionality when dut is helping RR and new grace lsa
 255     is received from RR.
 256     """
 257     tc_name = request.node.name
 258     write_test_header(tc_name)
 259     tgen = get_topogen()
 260
 261     # Don't run this test if we have any failure.
 262     if tgen.routers_have_failure():
 263         pytest.skip(tgen.errors)
 264
 265     global topo, intf, intf1, pkt
 266
 267     step("Bring up the base config as per the topology")
 268     step("Enable GR")
 269     reset_config_on_routers(tgen)
 270     ospf_covergence = verify_ospf_neighbor(tgen, topo, lan=True)
 271     assert (
 272         ospf_covergence is True
 273     ), "OSPF is not after reset config \n Error:" " {}".format(ospf_covergence)
 274     ospf_gr_r0 = {
 275         "r0": {"ospf": {"graceful-restart": {"helper enable": [], "opaque": True}}}
 276     }
 277     result = create_router_ospf(tgen, topo, ospf_gr_r0)
 278     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 279
 280     ospf_gr_r1 = {
 281         "r1": {"ospf": {"graceful-restart": {"helper enable": [], "opaque": True}}}
 282     }
 283     result = create_router_ospf(tgen, topo, ospf_gr_r1)
 284     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 285
 286     input_dict = {"supportedGracePeriod": 1800}
 287     dut = "r0"
 288     result = verify_ospf_gr_helper(tgen, topo, dut, input_dict)
 289     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 290
 291     step("Verify that DUT enters into helper mode.")
 292
 293     input_dict = {"activeRestarterCnt": 1}
 294     gracelsa_sent = False
 295     repeat = 0
 296     dut = "r0"
 297     while not gracelsa_sent and repeat < Iters:
 298         gracelsa_sent = scapy_send_raw_packet(tgen, topo, "r1", intf1, pkt)
 299         result = verify_ospf_gr_helper(tgen, topo, dut, input_dict)
 300         if isinstance(result, str):
 301             repeat += 1
 302             gracelsa_sent = False
 303
 304     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 305
 306     step("Send the Grace LSA again to DUT when RR is in GR.")
 307     input_dict = {"activeRestarterCnt": 1}
 308     gracelsa_sent = False
 309     repeat = 0
 310     dut = "r0"
 311     while not gracelsa_sent and repeat < Iters:
 312         gracelsa_sent = scapy_send_raw_packet(tgen, topo, "r1", intf1, pkt)
 313         result = verify_ospf_gr_helper(tgen, topo, dut, input_dict)
 314         if isinstance(result, str):
 315             repeat += 1
 316             gracelsa_sent = False
 317
 318     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 319
 320     delete_ospf()
 321
 322     write_test_footer(tc_name)
 323
 324
 325 if __name__ == "__main__":
 326     args = ["-s"] + sys.argv[1:]
 327     sys.exit(pytest.main(args))