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