tests/topotests/ospf_gr_helper/test_ospf_gr_helper1.py

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