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