tests/topotests/ospf_gr_helper/test_ospf_gr_helper1.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_tc1_p0(request):
 181     """Verify by default helper support is disabled for FRR ospf"""
 182
 183     tc_name = request.node.name
 184     write_test_header(tc_name)
 185     tgen = get_topogen()
 186
 187     # Don't run this test if we have any failure.
 188     if tgen.routers_have_failure():
 189         pytest.skip(tgen.errors)
 190
 191     global topo, intf, intf1, pkt
 192
 193     step("Bring up the base config as per the topology")
 194     reset_config_on_routers(tgen)
 195     ospf_covergence = verify_ospf_neighbor(tgen, topo, lan=True)
 196     assert (
 197         ospf_covergence is True
 198     ), "OSPF is not after reset config \n Error:" " {}".format(ospf_covergence)
 199
 200     step("Verify that GR helper route is disabled by default to the in" "the DUT.")
 201     input_dict = {
 202         "helperSupport": "Disabled",
 203         "strictLsaCheck": "Enabled",
 204         "restartSupoort": "Planned and Unplanned Restarts",
 205         "supportedGracePeriod": 1800,
 206     }
 207     dut = "r0"
 208     result = verify_ospf_gr_helper(tgen, topo, dut, input_dict)
 209     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 210
 211     step("Verify that DUT does not enter helper mode upon receiving the " "grace lsa.")
 212
 213     # send grace lsa
 214     scapy_send_raw_packet(tgen, topo, "r1", intf1, pkt)
 215
 216     input_dict = {"activeRestarterCnt": 1}
 217     dut = "r0"
 218     result = verify_ospf_gr_helper(tgen, topo, dut, input_dict, expected=False)
 219     assert (
 220         result is not True
 221     ), "Testcase {} : Failed. DUT entered helper role " " \n Error: {}".format(
 222         tc_name, result
 223     )
 224
 225     step("Configure graceful restart in the DUT")
 226     ospf_gr_r0 = {
 227         "r0": {"ospf": {"graceful-restart": {"helper enable": [], "opaque": True}}}
 228     }
 229     result = create_router_ospf(tgen, topo, ospf_gr_r0)
 230     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 231
 232     step("Verify that GR helper route is enabled in the DUT.")
 233     input_dict = {
 234         "helperSupport": "Enabled",
 235         "strictLsaCheck": "Enabled",
 236         "restartSupoort": "Planned and Unplanned Restarts",
 237         "supportedGracePeriod": 1800,
 238     }
 239     dut = "r0"
 240     result = verify_ospf_gr_helper(tgen, topo, dut, input_dict)
 241     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 242
 243     ospf_gr_r1 = {
 244         "r1": {"ospf": {"graceful-restart": {"helper enable": [], "opaque": True}}}
 245     }
 246     result = create_router_ospf(tgen, topo, ospf_gr_r1)
 247     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 248
 249     step("Perform GR in RR.")
 250     step("Verify that DUT does enter helper mode upon receiving" " the grace lsa.")
 251     input_dict = {"activeRestarterCnt": 1}
 252     gracelsa_sent = False
 253     repeat = 0
 254     dut = "r0"
 255     while not gracelsa_sent and repeat < Iters:
 256         gracelsa_sent = scapy_send_raw_packet(tgen, topo, "r1", intf1, pkt)
 257         result = verify_ospf_gr_helper(tgen, topo, dut, input_dict)
 258         if isinstance(result, str):
 259             repeat += 1
 260             gracelsa_sent = False
 261
 262     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 263
 264     step("Unconfigure the GR helper command.")
 265     ospf_gr_r0 = {
 266         "r0": {
 267             "ospf": {
 268                 "graceful-restart": {
 269                     "helper enable": [],
 270                     "opaque": True,
 271                     "delete": True,
 272                 }
 273             }
 274         }
 275     }
 276     result = create_router_ospf(tgen, topo, ospf_gr_r0)
 277     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 278
 279     input_dict = {"helperSupport": "Disabled"}
 280     dut = "r0"
 281     result = verify_ospf_gr_helper(tgen, topo, dut, input_dict)
 282     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 283
 284     step("Configure gr helper using the router id")
 285     ospf_gr_r0 = {
 286         "r0": {
 287             "ospf": {"graceful-restart": {"helper enable": ["1.1.1.1"], "opaque": True}}
 288         }
 289     }
 290     result = create_router_ospf(tgen, topo, ospf_gr_r0)
 291     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 292
 293     step("Verify that DUT does enter helper mode upon receiving" " the grace lsa.")
 294     input_dict = {"activeRestarterCnt": 1}
 295     gracelsa_sent = False
 296     repeat = 0
 297     dut = "r0"
 298     while not gracelsa_sent and repeat < Iters:
 299         gracelsa_sent = scapy_send_raw_packet(tgen, topo, "r1", intf1, pkt)
 300         result = verify_ospf_gr_helper(tgen, topo, dut, input_dict)
 301         if isinstance(result, str):
 302             repeat += 1
 303             gracelsa_sent = False
 304
 305     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 306
 307     step("Un Configure gr helper using the router id")
 308     ospf_gr_r0 = {
 309         "r0": {
 310             "ospf": {
 311                 "graceful-restart": {
 312                     "helper enable": ["1.1.1.1"],
 313                     "opaque": True,
 314                     "delete": True,
 315                 }
 316             }
 317         }
 318     }
 319     result = create_router_ospf(tgen, topo, ospf_gr_r0)
 320     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 321
 322     step("Verify that GR helper router is disabled in the DUT for" " router id x.x.x.x")
 323     input_dict = {"enabledRouterIds": [{"routerId": "1.1.1.1"}]}
 324     dut = "r0"
 325     result = verify_ospf_gr_helper(tgen, topo, dut, input_dict, expected=False)
 326     assert (
 327         result is not True
 328     ), "Testcase {} : Failed, Helper role enabled for RR\n Error: {}".format(
 329         tc_name, result
 330     )
 331     delete_ospf()
 332     write_test_footer(tc_name)
 333
 334
 335 def test_ospf_gr_helper_tc2_p0(request):
 336     """
 337     OSPF GR on Broadcast : Verify DUT enters Helper mode when neighbor
 338     sends grace lsa, helps RR to restart gracefully (RR = DR)
 339     """
 340     tc_name = request.node.name
 341     write_test_header(tc_name)
 342     tgen = get_topogen()
 343
 344     # Don't run this test if we have any failure.
 345     if tgen.routers_have_failure():
 346         pytest.skip(tgen.errors)
 347
 348     global topo, intf, intf1, pkt
 349
 350     step("Bring up the base config as per the topology")
 351     step(
 352         "Configure DR priority as 99 in RR , DUT dr priority = 98 "
 353         "& reset ospf process in all the routers"
 354     )
 355     reset_config_on_routers(tgen)
 356     ospf_covergence = verify_ospf_neighbor(tgen, topo, lan=True)
 357     assert (
 358         ospf_covergence is True
 359     ), "OSPF is not after reset config \n Error:" " {}".format(ospf_covergence)
 360     ospf_gr_r0 = {
 361         "r0": {"ospf": {"graceful-restart": {"helper enable": [], "opaque": True}}}
 362     }
 363     result = create_router_ospf(tgen, topo, ospf_gr_r0)
 364     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 365
 366     ospf_gr_r1 = {
 367         "r1": {"ospf": {"graceful-restart": {"helper enable": [], "opaque": True}}}
 368     }
 369     result = create_router_ospf(tgen, topo, ospf_gr_r1)
 370     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 371
 372     step("Verify that DUT enters into helper mode.")
 373
 374     input_dict = {"activeRestarterCnt": 1}
 375     gracelsa_sent = False
 376     repeat = 0
 377     dut = "r0"
 378     while not gracelsa_sent and repeat < Iters:
 379         gracelsa_sent = scapy_send_raw_packet(tgen, topo, "r1", intf1, pkt)
 380         result = verify_ospf_gr_helper(tgen, topo, dut, input_dict)
 381         if isinstance(result, str):
 382             repeat += 1
 383             gracelsa_sent = False
 384
 385     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 386
 387     delete_ospf()
 388     write_test_footer(tc_name)
 389
 390
 391 if __name__ == "__main__":
 392     args = ["-s"] + sys.argv[1:]
 393     sys.exit(pytest.main(args))