tests/topotests/ospf_gr_helper/test_ospf_gr_helper2.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_tc3_p1(request):
 168     """
 169     OSPF GR on Broadcast : Verify DUT enters Helper mode when neighbor
 170     sends grace lsa, helps RR to restart gracefully (RR = BDR)
 171     """
 172     tc_name = request.node.name
 173     write_test_header(tc_name)
 174     tgen = get_topogen()
 175
 176     # Don't run this test if we have any failure.
 177     if tgen.routers_have_failure():
 178         pytest.skip(tgen.errors)
 179
 180     global topo, intf, intf1, pkt
 181
 182     step("Bring up the base config as per the topology")
 183     step(
 184         "Configure DR priority as 99 in RR , DUT dr priority = 98 "
 185         "& reset ospf process in all the routers"
 186     )
 187     reset_config_on_routers(tgen)
 188     ospf_covergence = verify_ospf_neighbor(tgen, topo, lan=True)
 189     assert (
 190         ospf_covergence is True
 191     ), "OSPF is not after reset config \n Error:" " {}".format(ospf_covergence)
 192     step(
 193         "Configure DR pririty 100 on R0 and clear ospf neighbors " "on all the routers."
 194     )
 195
 196     input_dict = {
 197         "r0": {
 198             "links": {
 199                 sw_name: {
 200                     "interface": topo["routers"]["r0"]["links"][sw_name]["interface"],
 201                     "ospf": {"priority": 100},
 202                 }
 203             }
 204         }
 205     }
 206
 207     result = create_interfaces_cfg(tgen, input_dict)
 208     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 209
 210     step("Clear ospf neighbours in all routers")
 211     for rtr in topo["routers"]:
 212         clear_ospf(tgen, rtr)
 213
 214     step("Verify that DR election is triggered and R0 is elected as DR")
 215     input_dict = {
 216         "r0": {
 217             "ospf": {
 218                 "neighbors": {
 219                     "r1": {"state": "Full", "role": "Backup"},
 220                     "r2": {"state": "Full", "role": "DROther"},
 221                     "r3": {"state": "Full", "role": "DROther"},
 222                 }
 223             }
 224         }
 225     }
 226     dut = "r0"
 227     result = verify_ospf_neighbor(tgen, topo, dut, input_dict, lan=True)
 228     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 229
 230     ospf_gr_r0 = {
 231         "r0": {"ospf": {"graceful-restart": {"helper enable": [], "opaque": True}}}
 232     }
 233     result = create_router_ospf(tgen, topo, ospf_gr_r0)
 234     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 235
 236     ospf_gr_r1 = {
 237         "r1": {"ospf": {"graceful-restart": {"helper enable": [], "opaque": True}}}
 238     }
 239     result = create_router_ospf(tgen, topo, ospf_gr_r1)
 240     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 241
 242     step("Verify that DUT enters into helper mode.")
 243
 244     input_dict = {"activeRestarterCnt": 1}
 245     gracelsa_sent = False
 246     repeat = 0
 247     dut = "r0"
 248     while not gracelsa_sent and repeat < Iters:
 249         gracelsa_sent = scapy_send_raw_packet(tgen, topo, "r1", intf1, pkt)
 250         result = verify_ospf_gr_helper(tgen, topo, dut, input_dict)
 251         if isinstance(result, str):
 252             repeat += 1
 253             gracelsa_sent = False
 254
 255     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 256
 257     delete_ospf()
 258     write_test_footer(tc_name)
 259
 260
 261 def test_ospf_gr_helper_tc4_p1(request):
 262     """
 263     OSPF GR on Broadcast : Verify DUT enters Helper mode when neighbor
 264     sends grace lsa, helps RR to restart gracefully (RR = DRother)
 265     """
 266     tc_name = request.node.name
 267     write_test_header(tc_name)
 268     tgen = get_topogen()
 269
 270     # Don't run this test if we have any failure.
 271     if tgen.routers_have_failure():
 272         pytest.skip(tgen.errors)
 273
 274     global topo, intf, intf1, pkt
 275
 276     step("Bring up the base config as per the topology")
 277     step(
 278         "Configure DR priority as 99 in RR , DUT dr priority = 98 "
 279         "& reset ospf process in all the routers"
 280     )
 281     reset_config_on_routers(tgen)
 282     ospf_covergence = verify_ospf_neighbor(tgen, topo, lan=True)
 283     assert (
 284         ospf_covergence is True
 285     ), "OSPF is not after reset config \n Error:" " {}".format(ospf_covergence)
 286     step(
 287         "Configure DR pririty 100 on R0 and clear ospf neighbors " "on all the routers."
 288     )
 289
 290     input_dict = {
 291         "r0": {
 292             "links": {
 293                 sw_name: {
 294                     "interface": topo["routers"]["r0"]["links"][sw_name]["interface"],
 295                     "ospf": {"priority": 0},
 296                 }
 297             }
 298         }
 299     }
 300
 301     result = create_interfaces_cfg(tgen, input_dict)
 302     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 303
 304     step("Clear ospf neighbours in all routers")
 305     for rtr in topo["routers"]:
 306         clear_ospf(tgen, rtr)
 307
 308     step("Verify that DR election is triggered and R0 is elected as 2-Way")
 309     input_dict = {
 310         "r0": {
 311             "ospf": {
 312                 "neighbors": {
 313                     "r1": {"state": "Full", "role": "DR"},
 314                     "r2": {"state": "2-Way", "role": "DROther"},
 315                     "r3": {"state": "2-Way", "role": "DROther"},
 316                 }
 317             }
 318         }
 319     }
 320     dut = "r0"
 321     result = verify_ospf_neighbor(tgen, topo, dut, input_dict, lan=True)
 322     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 323
 324     ospf_gr_r0 = {
 325         "r0": {"ospf": {"graceful-restart": {"helper enable": [], "opaque": True}}}
 326     }
 327     result = create_router_ospf(tgen, topo, ospf_gr_r0)
 328     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 329
 330     ospf_gr_r1 = {
 331         "r1": {"ospf": {"graceful-restart": {"helper enable": [], "opaque": True}}}
 332     }
 333     result = create_router_ospf(tgen, topo, ospf_gr_r1)
 334     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 335
 336     step("Verify that DUT enters into helper mode.")
 337
 338     input_dict = {"activeRestarterCnt": 1}
 339     gracelsa_sent = False
 340     repeat = 0
 341     dut = "r0"
 342     while not gracelsa_sent and repeat < Iters:
 343         gracelsa_sent = scapy_send_raw_packet(tgen, topo, "r1", intf1, pkt)
 344         result = verify_ospf_gr_helper(tgen, topo, dut, input_dict)
 345         if isinstance(result, str):
 346             repeat += 1
 347             gracelsa_sent = False
 348
 349     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 350
 351     delete_ospf()
 352
 353     write_test_footer(tc_name)
 354
 355
 356 if __name__ == "__main__":
 357     args = ["-s"] + sys.argv[1:]
 358     sys.exit(pytest.main(args))