tests/topotests/ospf_basic_functionality/test_ospf_chaos.py

   1 #!/usr/bin/python
   2 # SPDX-License-Identifier: ISC
   3
   4 #
   5 # Copyright (c) 2020 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 sys.path.append(os.path.join(CWD, "../lib/"))
  21
  22 # pylint: disable=C0413
  23 # Import topogen and topotest helpers
  24 from lib.topogen import Topogen, get_topogen
  25
  26 # Import topoJson from lib, to create topology and initial configuration
  27 from lib.common_config import (
  28     start_topology,
  29     write_test_header,
  30     write_test_footer,
  31     reset_config_on_routers,
  32     step,
  33     verify_rib,
  34     stop_router,
  35     start_router,
  36     create_static_routes,
  37     start_router_daemons,
  38     kill_router_daemons,
  39 )
  40
  41 from lib.ospf import verify_ospf_neighbor, verify_ospf_rib, create_router_ospf
  42
  43 from lib.topolog import logger
  44 from lib.topojson import build_config_from_json
  45
  46 pytestmark = [pytest.mark.ospfd, pytest.mark.staticd]
  47
  48 # Global variables
  49 topo = None
  50
  51 NETWORK = {
  52     "ipv4": [
  53         "11.0.20.1/32",
  54         "11.0.20.2/32",
  55         "11.0.20.3/32",
  56         "11.0.20.4/32",
  57         "11.0.20.5/32",
  58     ]
  59 }
  60 """
  61 Topology:
  62       Please view in a fixed-width font such as Courier.
  63       +---+  A1       +---+
  64       +R1 +------------+R2 |
  65       +-+-+-           +--++
  66         |  --        --  |
  67         |    -- A0 --    |
  68       A0|      ----      |
  69         |      ----      | A2
  70         |    --    --    |
  71         |  --        --  |
  72       +-+-+-            +-+-+
  73       +R0 +-------------+R3 |
  74       +---+     A3     +---+
  75
  76 TESTCASES =
  77 1. Verify ospf functionality after restart ospfd.
  78 2. Verify ospf functionality after restart FRR service.
  79 3. Verify ospf functionality when staticd is restarted.
  80  """
  81
  82
  83 def setup_module(mod):
  84     """
  85     Sets up the pytest environment
  86
  87     * `mod`: module name
  88     """
  89     testsuite_run_time = time.asctime(time.localtime(time.time()))
  90     logger.info("Testsuite start time: {}".format(testsuite_run_time))
  91     logger.info("=" * 40)
  92
  93     logger.info("Running setup_module to create topology")
  94
  95     # This function initiates the topology build with Topogen...
  96     json_file = "{}/ospf_chaos.json".format(CWD)
  97     tgen = Topogen(json_file, mod.__name__)
  98     global topo
  99     topo = tgen.json_topo
 100     # ... and here it calls Mininet initialization functions.
 101
 102     # Starting topology, create tmp files which are loaded to routers
 103     #  to start daemons and then start routers
 104     start_topology(tgen)
 105
 106     # Creating configuration from JSON
 107     build_config_from_json(tgen, topo)
 108
 109     # Don't run this test if we have any failure.
 110     if tgen.routers_have_failure():
 111         pytest.skip(tgen.errors)
 112
 113     ospf_covergence = verify_ospf_neighbor(tgen, topo)
 114     assert ospf_covergence is True, "setup_module :Failed \n Error:" " {}".format(
 115         ospf_covergence
 116     )
 117
 118     logger.info("Running setup_module() done")
 119
 120
 121 def teardown_module(mod):
 122     """
 123     Teardown the pytest environment.
 124
 125     * `mod`: module name
 126     """
 127
 128     logger.info("Running teardown_module to delete topology")
 129
 130     tgen = get_topogen()
 131
 132     # Stop toplogy and Remove tmp files
 133     tgen.stop_topology()
 134
 135     logger.info(
 136         "Testsuite end time: {}".format(time.asctime(time.localtime(time.time())))
 137     )
 138     logger.info("=" * 40)
 139
 140
 141 # ##################################
 142 # Test cases start here.
 143 # ##################################
 144 def test_ospf_chaos_tc31_p1(request):
 145     """Verify ospf functionality after restart ospfd."""
 146     tc_name = request.node.name
 147     write_test_header(tc_name)
 148     tgen = get_topogen()
 149     global topo
 150     step("Bring up the base config as per the topology")
 151     reset_config_on_routers(tgen)
 152
 153     step(
 154         "Create static routes(10.0.20.1/32) in R1 and redistribute "
 155         "to OSPF using route map."
 156     )
 157
 158     # Create Static routes
 159     input_dict = {
 160         "r0": {
 161             "static_routes": [
 162                 {
 163                     "network": NETWORK["ipv4"][0],
 164                     "no_of_ip": 5,
 165                     "next_hop": "Null0",
 166                 }
 167             ]
 168         }
 169     }
 170     result = create_static_routes(tgen, input_dict)
 171     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 172
 173     ospf_red_r0 = {"r0": {"ospf": {"redistribute": [{"redist_type": "static"}]}}}
 174     result = create_router_ospf(tgen, topo, ospf_red_r0)
 175     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 176
 177     step("Verify OSPF neighbors after base config is done.")
 178     # Api call verify whether OSPF is converged
 179     ospf_covergence = verify_ospf_neighbor(tgen, topo)
 180     assert ospf_covergence is True, "setup_module :Failed \n Error:" " {}".format(
 181         ospf_covergence
 182     )
 183
 184     step("Verify that route is advertised to R1.")
 185     dut = "r1"
 186     protocol = "ospf"
 187     nh = topo["routers"]["r0"]["links"]["r1"]["ipv4"].split("/")[0]
 188     result = verify_ospf_rib(tgen, dut, input_dict, next_hop=nh)
 189     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 190
 191     result = verify_rib(tgen, "ipv4", dut, input_dict, protocol=protocol, next_hop=nh)
 192     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 193
 194     step("Kill OSPFd daemon on R0.")
 195     kill_router_daemons(tgen, "r0", ["ospfd"])
 196
 197     step("Verify OSPF neighbors are down after killing ospfd in R0")
 198     dut = "r0"
 199     # Api call verify whether OSPF is converged
 200     ospf_covergence = verify_ospf_neighbor(tgen, topo, dut=dut, expected=False)
 201     assert ospf_covergence is not True, "setup_module :Failed \n Error:" " {}".format(
 202         ospf_covergence
 203     )
 204
 205     step("Verify that route advertised to R1 are deleted from RIB and FIB.")
 206     dut = "r1"
 207     protocol = "ospf"
 208     result = verify_ospf_rib(tgen, dut, input_dict, expected=False)
 209     assert (
 210         result is not True
 211     ), "Testcase {} : Failed \n " "r1: OSPF routes are present \n Error: {}".format(
 212         tc_name, result
 213     )
 214
 215     result = verify_rib(
 216         tgen, "ipv4", dut, input_dict, protocol=protocol, expected=False
 217     )
 218     assert (
 219         result is not True
 220     ), "Testcase {} : Failed \n " "r1: routes are still present \n Error: {}".format(
 221         tc_name, result
 222     )
 223
 224     step("Bring up OSPFd daemon on R0.")
 225     start_router_daemons(tgen, "r0", ["ospfd"])
 226
 227     step("Verify OSPF neighbors are up after bringing back ospfd in R0")
 228     # Api call verify whether OSPF is converged
 229     ospf_covergence = verify_ospf_neighbor(tgen, topo)
 230     assert ospf_covergence is True, "setup_module :Failed \n Error:" " {}".format(
 231         ospf_covergence
 232     )
 233
 234     step(
 235         "All the neighbours are up and routes are installed before the"
 236         " restart. Verify OSPF route table and ip route table."
 237     )
 238     dut = "r1"
 239     protocol = "ospf"
 240     result = verify_ospf_rib(tgen, dut, input_dict, next_hop=nh)
 241     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 242
 243     result = verify_rib(tgen, "ipv4", dut, input_dict, protocol=protocol, next_hop=nh)
 244     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 245
 246     step("Kill OSPFd daemon on R1.")
 247     kill_router_daemons(tgen, "r1", ["ospfd"])
 248
 249     step("Verify OSPF neighbors are down after killing ospfd in R1")
 250     dut = "r1"
 251     # Api call verify whether OSPF is converged
 252     ospf_covergence = verify_ospf_neighbor(tgen, topo, dut=dut, expected=False)
 253     assert ospf_covergence is not True, "setup_module :Failed \n Error:" " {}".format(
 254         ospf_covergence
 255     )
 256
 257     step("Bring up OSPFd daemon on R1.")
 258     start_router_daemons(tgen, "r1", ["ospfd"])
 259
 260     step("Verify OSPF neighbors are up after bringing back ospfd in R1")
 261     # Api call verify whether OSPF is converged
 262     ospf_covergence = verify_ospf_neighbor(tgen, topo)
 263     assert ospf_covergence is True, "setup_module :Failed \n Error:" " {}".format(
 264         ospf_covergence
 265     )
 266
 267     step(
 268         "All the neighbours are up and routes are installed before the"
 269         " restart. Verify OSPF route table and ip route table."
 270     )
 271
 272     dut = "r1"
 273     protocol = "ospf"
 274     result = verify_ospf_rib(tgen, dut, input_dict, next_hop=nh)
 275     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 276
 277     result = verify_rib(tgen, "ipv4", dut, input_dict, protocol=protocol, next_hop=nh)
 278     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 279
 280     write_test_footer(tc_name)
 281
 282
 283 def test_ospf_chaos_tc32_p1(request):
 284     """Verify ospf functionality after restart FRR service."""
 285     tc_name = request.node.name
 286     write_test_header(tc_name)
 287     tgen = get_topogen()
 288     global topo
 289     step("Bring up the base config as per the topology")
 290     reset_config_on_routers(tgen)
 291
 292     step(
 293         "Create static routes(10.0.20.1/32) in R1 and redistribute "
 294         "to OSPF using route map."
 295     )
 296
 297     # Create Static routes
 298     input_dict = {
 299         "r0": {
 300             "static_routes": [
 301                 {
 302                     "network": NETWORK["ipv4"][0],
 303                     "no_of_ip": 5,
 304                     "next_hop": "Null0",
 305                 }
 306             ]
 307         }
 308     }
 309     result = create_static_routes(tgen, input_dict)
 310     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 311
 312     ospf_red_r0 = {"r0": {"ospf": {"redistribute": [{"redist_type": "static"}]}}}
 313     result = create_router_ospf(tgen, topo, ospf_red_r0)
 314     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 315
 316     step("Verify OSPF neighbors after base config is done.")
 317     # Api call verify whether OSPF is converged
 318     ospf_covergence = verify_ospf_neighbor(tgen, topo)
 319     assert ospf_covergence is True, "setup_module :Failed \n Error:" " {}".format(
 320         ospf_covergence
 321     )
 322
 323     step("Verify that route is advertised to R1.")
 324     dut = "r1"
 325     protocol = "ospf"
 326
 327     nh = topo["routers"]["r0"]["links"]["r1"]["ipv4"].split("/")[0]
 328     result = verify_ospf_rib(tgen, dut, input_dict, next_hop=nh)
 329     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 330
 331     result = verify_rib(tgen, "ipv4", dut, input_dict, protocol=protocol, next_hop=nh)
 332     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 333
 334     step("Restart frr on R0")
 335     stop_router(tgen, "r0")
 336     start_router(tgen, "r0")
 337
 338     step("Verify OSPF neighbors are up after restarting R0")
 339     # Api call verify whether OSPF is converged
 340     ospf_covergence = verify_ospf_neighbor(tgen, topo)
 341     assert ospf_covergence is True, "setup_module :Failed \n Error:" " {}".format(
 342         ospf_covergence
 343     )
 344
 345     step(
 346         "All the neighbours are up and routes are installed before the"
 347         " restart. Verify OSPF route table and ip route table."
 348     )
 349     dut = "r1"
 350     protocol = "ospf"
 351     result = verify_ospf_rib(tgen, dut, input_dict, next_hop=nh)
 352     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 353
 354     result = verify_rib(tgen, "ipv4", dut, input_dict, protocol=protocol, next_hop=nh)
 355     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 356
 357     step("Restart frr on R1")
 358     stop_router(tgen, "r1")
 359     start_router(tgen, "r1")
 360
 361     step("Verify OSPF neighbors are up after restarting R1")
 362     # Api call verify whether OSPF is converged
 363     ospf_covergence = verify_ospf_neighbor(tgen, topo)
 364     assert ospf_covergence is True, "setup_module :Failed \n Error:" " {}".format(
 365         ospf_covergence
 366     )
 367
 368     step(
 369         "All the neighbours are up and routes are installed before the"
 370         " restart. Verify OSPF route table and ip route table."
 371     )
 372     dut = "r1"
 373     protocol = "ospf"
 374     result = verify_ospf_rib(tgen, dut, input_dict, next_hop=nh)
 375     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 376
 377     result = verify_rib(tgen, "ipv4", dut, input_dict, protocol=protocol, next_hop=nh)
 378     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 379
 380     write_test_footer(tc_name)
 381
 382
 383 def test_ospf_chaos_tc34_p1(request):
 384     """
 385     verify ospf functionality when staticd is restarted.
 386
 387     Verify ospf functionalitywhen staticroutes are
 388     redistributed & Staticd is restarted.
 389     """
 390     tc_name = request.node.name
 391     write_test_header(tc_name)
 392     tgen = get_topogen()
 393     global topo
 394     step("Bring up the base config as per the topology")
 395     reset_config_on_routers(tgen)
 396
 397     step(
 398         "Create static routes(10.0.20.1/32) in R1 and redistribute "
 399         "to OSPF using route map."
 400     )
 401
 402     # Create Static routes
 403     input_dict = {
 404         "r0": {
 405             "static_routes": [
 406                 {
 407                     "network": NETWORK["ipv4"][0],
 408                     "no_of_ip": 5,
 409                     "next_hop": "Null0",
 410                 }
 411             ]
 412         }
 413     }
 414     result = create_static_routes(tgen, input_dict)
 415     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 416
 417     ospf_red_r0 = {"r0": {"ospf": {"redistribute": [{"redist_type": "static"}]}}}
 418     result = create_router_ospf(tgen, topo, ospf_red_r0)
 419     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 420
 421     step("Verify OSPF neighbors after base config is done.")
 422     # Api call verify whether OSPF is converged
 423     ospf_covergence = verify_ospf_neighbor(tgen, topo)
 424     assert ospf_covergence is True, "setup_module :Failed \n Error:" " {}".format(
 425         ospf_covergence
 426     )
 427
 428     step("Verify that route is advertised to R1.")
 429     dut = "r1"
 430     protocol = "ospf"
 431     nh = topo["routers"]["r0"]["links"]["r1"]["ipv4"].split("/")[0]
 432     result = verify_ospf_rib(tgen, dut, input_dict, next_hop=nh)
 433     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 434
 435     result = verify_rib(tgen, "ipv4", dut, input_dict, protocol=protocol, next_hop=nh)
 436     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 437
 438     step("Kill staticd daemon on R0.")
 439     kill_router_daemons(tgen, "r0", ["staticd"])
 440
 441     step("Verify that route advertised to R1 are deleted from RIB and FIB.")
 442     dut = "r1"
 443     protocol = "ospf"
 444     result = verify_ospf_rib(tgen, dut, input_dict, expected=False)
 445     assert (
 446         result is not True
 447     ), "Testcase {} : Failed \n " "r1: OSPF routes are present \n Error: {}".format(
 448         tc_name, result
 449     )
 450
 451     result = verify_rib(
 452         tgen, "ipv4", dut, input_dict, protocol=protocol, expected=False
 453     )
 454     assert (
 455         result is not True
 456     ), "Testcase {} : Failed \n " "r1: routes are still present \n Error: {}".format(
 457         tc_name, result
 458     )
 459
 460     step("Bring up staticd daemon on R0.")
 461     start_router_daemons(tgen, "r0", ["staticd"])
 462
 463     step("Verify OSPF neighbors are up after bringing back ospfd in R0")
 464     # Api call verify whether OSPF is converged
 465     ospf_covergence = verify_ospf_neighbor(tgen, topo)
 466     assert ospf_covergence is True, "setup_module :Failed \n Error:" " {}".format(
 467         ospf_covergence
 468     )
 469
 470     step(
 471         "All the neighbours are up and routes are installed before the"
 472         " restart. Verify OSPF route table and ip route table."
 473     )
 474     dut = "r1"
 475     protocol = "ospf"
 476     result = verify_ospf_rib(tgen, dut, input_dict, next_hop=nh)
 477     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 478
 479     result = verify_rib(tgen, "ipv4", dut, input_dict, protocol=protocol, next_hop=nh)
 480     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 481
 482     step("Kill staticd daemon on R1.")
 483     kill_router_daemons(tgen, "r1", ["staticd"])
 484
 485     step("Bring up staticd daemon on R1.")
 486     start_router_daemons(tgen, "r1", ["staticd"])
 487
 488     step("Verify OSPF neighbors are up after bringing back ospfd in R1")
 489     # Api call verify whether OSPF is converged
 490     ospf_covergence = verify_ospf_neighbor(tgen, topo)
 491     assert ospf_covergence is True, "setup_module :Failed \n Error:" " {}".format(
 492         ospf_covergence
 493     )
 494
 495     step(
 496         "All the neighbours are up and routes are installed before the"
 497         " restart. Verify OSPF route table and ip route table."
 498     )
 499
 500     dut = "r1"
 501     protocol = "ospf"
 502     result = verify_ospf_rib(tgen, dut, input_dict, next_hop=nh)
 503     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 504
 505     result = verify_rib(tgen, "ipv4", dut, input_dict, protocol=protocol, next_hop=nh)
 506     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 507
 508     write_test_footer(tc_name)
 509
 510
 511 if __name__ == "__main__":
 512     args = ["-s"] + sys.argv[1:]
 513     sys.exit(pytest.main(args))