tests/topotests/ospf_basic_functionality/test_ospf_chaos.py

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