tests/topotests/ospf_basic_functionality/test_ospf_flood_reduction.py

   1 #!/usr/bin/python
   2
   3 #
   4 # Copyright (c) 2022 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 time import sleep
  30
  31 # Save the Current Working Directory to find configuration files.
  32 CWD = os.path.dirname(os.path.realpath(__file__))
  33 sys.path.append(os.path.join(CWD, "../"))
  34 sys.path.append(os.path.join(CWD, "../lib/"))
  35
  36 pytestmark = [pytest.mark.ospfd, pytest.mark.staticd]
  37 # Global variables
  38 topo = None
  39
  40 # pylint: disable=C0413
  41 # Import topogen and topotest helpers
  42 from lib.topolog import logger
  43 from lib.topojson import build_config_from_json
  44
  45 # Import topoJson from lib, to create topology and initial configuration
  46 from lib.common_config import (
  47     start_topology,
  48     write_test_header,
  49     write_test_footer,
  50     reset_config_on_routers,
  51     create_static_routes,
  52     step,
  53     topo_daemons,
  54     shutdown_bringup_interface,
  55     check_router_status,
  56     start_topology,
  57     write_test_header,
  58     write_test_footer,
  59     reset_config_on_routers,
  60     create_static_routes,
  61     step,
  62     shutdown_bringup_interface,
  63     check_router_status,
  64     start_topology,
  65     write_test_header,
  66     write_test_footer,
  67     reset_config_on_routers,
  68     stop_router,
  69     start_router,
  70     step,
  71     create_static_routes,
  72     kill_router_daemons,
  73     check_router_status,
  74     start_router_daemons,
  75 )
  76 from lib.topolog import logger
  77 from lib.topogen import Topogen, get_topogen
  78
  79 from lib.topojson import build_config_from_json
  80 from lib.ospf import (
  81     verify_ospf_neighbor,
  82     clear_ospf,
  83     create_router_ospf,
  84     verify_ospf_database,
  85     get_ospf_database,
  86 )
  87
  88 # Global variables
  89 topo = None
  90 NETWORK = {
  91     "ipv4": [
  92         "11.0.20.1/32",
  93         "11.0.20.2/32",
  94         "11.0.20.3/32",
  95         "11.0.20.4/32",
  96         "11.0.20.5/32",
  97     ]
  98 }
  99 """
 100 TOPOOLOGY =
 101       Please view in a fixed-width font such as Courier.
 102       +---+  A1       +---+
 103       +R1 +------------+R2 |
 104       +-+-+-           +--++
 105         |  --        --  |
 106         |    -- A0 --    |
 107       A0|      ----      |
 108         |      ----      | A2
 109         |    --    --    |
 110         |  --        --  |
 111       +-+-+-            +-+-+
 112       +R0 +-------------+R3 |
 113       +---+     A3     +---+
 114
 115 TESTCASES =
 116 1. Verify OSPF Flood reduction functionality with ospf enabled on process level.
 117 2. Verify OSPF Flood reduction functionality with ospf enabled on area level.
 118 3. Verify OSPF Flood reduction functionality between different area's.
 119 4. Verify OSPF Flood reduction functionality with ospf enabled on process level with default lsa refresh timer.
 120 5. Chaos - Verify OSPF TTL GTSM and flood  reduction functionality in chaos scenarios.
 121 """
 122
 123
 124 def setup_module(mod):
 125     """
 126     Sets up the pytest environment
 127
 128     * `mod`: module name
 129     """
 130     global topo
 131     testsuite_run_time = time.asctime(time.localtime(time.time()))
 132     logger.info("Testsuite start time: {}".format(testsuite_run_time))
 133     logger.info("=" * 40)
 134
 135     logger.info("Running setup_module to create topology")
 136
 137     # This function initiates the topology build with Topogen...
 138     json_file = "{}/ospf_flood_reduction.json".format(CWD)
 139     tgen = Topogen(json_file, mod.__name__)
 140     global topo
 141     topo = tgen.json_topo
 142     # ... and here it calls Mininet initialization functions.
 143
 144     # get list of daemons needs to be started for this suite.
 145     daemons = topo_daemons(tgen)
 146
 147     # Starting topology, create tmp files which are loaded to routers
 148     #  to start daemons and then start routers
 149     start_topology(tgen)
 150
 151     # Creating configuration from JSON
 152     build_config_from_json(tgen, topo)
 153
 154     # Don't run this test if we have any failure.
 155     if tgen.routers_have_failure():
 156         pytest.skip(tgen.errors)
 157     # Api call verify whether OSPF is converged
 158     ospf_covergence = verify_ospf_neighbor(tgen, topo)
 159     assert ospf_covergence is True, "setup_module :Failed \n Error:" " {}".format(
 160         ospf_covergence
 161     )
 162
 163     logger.info("Running setup_module() done")
 164
 165
 166 def teardown_module(mod):
 167     """
 168     Teardown the pytest environment.
 169
 170     * `mod`: module name
 171     """
 172
 173     logger.info("Running teardown_module to delete topology")
 174
 175     tgen = get_topogen()
 176
 177     # Stop toplogy and Remove tmp files
 178     tgen.stop_topology()
 179
 180     logger.info(
 181         "Testsuite end time: {}".format(time.asctime(time.localtime(time.time())))
 182     )
 183     logger.info("=" * 40)
 184
 185
 186
 187 def red_static(dut, config=True):
 188     """Local def for Redstribute static routes inside ospf."""
 189     global topo
 190     tgen = get_topogen()
 191     if config:
 192         ospf_red = {dut: {"ospf": {"redistribute": [{"redist_type": "static"}]}}}
 193     else:
 194         ospf_red = {
 195             dut: {
 196                 "ospf": {
 197                     "redistribute": [{"redist_type": "static", "del_action": True}]
 198                 }
 199             }
 200         }
 201     result = create_router_ospf(tgen, topo, ospf_red)
 202     assert result is True, "Testcase : Failed \n Error: {}".format(result)
 203
 204
 205 # ##################################
 206 # Test cases start here.
 207 # ##################################
 208 def test_ospf_flood_red_tc1_p0(request):
 209     """Verify OSPF Flood reduction functionality with ospf enabled on process level."""
 210     tc_name = request.node.name
 211     write_test_header(tc_name)
 212     tgen = get_topogen()
 213
 214     # Don't run this test if we have any failure.
 215     if tgen.routers_have_failure():
 216         check_router_status(tgen)
 217
 218     global topo
 219
 220     step("Bring up the base config as per the topology")
 221     reset_config_on_routers(tgen)
 222     red_static("r0")
 223     step("Base config should be up, verify using OSPF convergence on all the routers")
 224
 225     ospf_covergence = verify_ospf_neighbor(tgen, topo)
 226     assert ospf_covergence is True, "Testcase :Failed \n Error:" " {}".format(
 227         ospf_covergence
 228     )
 229
 230     input_dict = {
 231         "r0": {
 232             "static_routes": [
 233                 {
 234                     "network": NETWORK["ipv4"][0],
 235                     "no_of_ip": 5,
 236                     "next_hop": "Null0",
 237                 }
 238             ]
 239         }
 240     }
 241     result = create_static_routes(tgen, input_dict)
 242     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 243
 244     step("Enable flood reduction in process level on R0")
 245     ospf_flood = {"r0": {"ospf": {"flood-reduction": True}}}
 246     result = create_router_ospf(tgen, topo, ospf_flood)
 247     assert result is True, "Testcase : Failed \n Error: {}".format(result)
 248
 249     step("Verify that ospf lsa's are set with dc bit 1.")
 250     dut = "r0"
 251     input_dict_db = {
 252         "routerId": "100.1.1.0",
 253         "areas": {
 254             "0.0.0.0": {
 255                 "routerLinkStates": [
 256                     {
 257                         "lsaId": "100.1.1.0",
 258                         "options": "*|-|DC|-|-|-|E|-",
 259                     },
 260                 ]
 261             }
 262         },
 263     }
 264     result = verify_ospf_database(
 265         tgen, topo, dut, input_dict_db, lsatype="router", rid="100.1.1.0"
 266     )
 267     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 268
 269     step("Configure the custom refresh timer")
 270     ospf_flood = {"r0": {"ospf": {"lsa-refresh": 120}}}
 271     result = create_router_ospf(tgen, topo, ospf_flood)
 272     assert result is True, "Testcase : Failed \n Error: {}".format(result)
 273
 274     step("Enable flood. reduction in all the routers of the topology.")
 275     for rtr in topo["routers"].keys():
 276         ospf_flood = {rtr: {"ospf": {"lsa-refresh": 120, "flood-reduction": True}}}
 277         result = create_router_ospf(tgen, topo, ospf_flood)
 278         assert result is True, "Testcase : Failed \n Error: {}".format(result)
 279
 280     for rtr in topo["routers"]:
 281         clear_ospf(tgen, rtr)
 282
 283     ospf_covergence = verify_ospf_neighbor(tgen, topo)
 284     assert ospf_covergence is True, "Testcase :Failed \n Error:" " {}".format(
 285         ospf_covergence
 286     )
 287
 288     step("Verify that ospf lsa's are set with dc bit 1.")
 289     for rtr in topo["routers"]:
 290         dut = rtr
 291         lsid = "{}".format(topo["routers"][rtr]["ospf"]["router_id"])
 292         input_dict_db = {
 293             "routerId": lsid,
 294             "areas": {
 295                 "0.0.0.0": {
 296                     "routerLinkStates": [
 297                         {
 298                             "lsaId": lsid,
 299                             "options": "*|-|DC|-|-|-|E|-",
 300                         },
 301                     ]
 302                 }
 303             },
 304         }
 305         result = verify_ospf_database(
 306             tgen, topo, dut, input_dict_db, lsatype="router", rid=lsid
 307         )
 308         assert result is True, "Testcase {} : Failed \n Error: {}".format(
 309             tc_name, result
 310         )
 311
 312     step("Wait for 120 secs and verify that LSA's are not refreshed. ")
 313     # get LSA age
 314     dut = "r1"
 315     input_dict_db = {
 316         "routerId": "100.1.1.0",
 317         "areas": {
 318             "0.0.0.0": {
 319                 "routerLinkStates": [
 320                     {"lsaId": "100.1.1.0", "lsaAge": "get"},
 321                 ]
 322             }
 323         },
 324     }
 325     sleep(10)
 326
 327     result1 = get_ospf_database(
 328         tgen, topo, dut, input_dict_db, lsatype="router", rid="100.1.1.0"
 329     )
 330     # this wait is put so that we wait for 5secs to check if lsa is refreshed.
 331     sleep(5)
 332     result2 = get_ospf_database(
 333         tgen, topo, dut, input_dict_db, lsatype="router", rid="100.1.1.0"
 334     )
 335
 336     assert (result1 == result2) is True, "Testcase {} : Failed \n Error: {}".format(
 337         tc_name, result
 338     )
 339
 340     step("Disable flood reduction in R0.")
 341     ospf_flood = {
 342         "r0": {"ospf": {"flood-reduction": True, "del_flood_reduction": True}}
 343     }
 344     result = create_router_ospf(tgen, topo, ospf_flood)
 345     assert result is True, "Testcase : Failed \n Error: {}".format(result)
 346
 347     clear_ospf(tgen, "r0")
 348
 349     ospf_covergence = verify_ospf_neighbor(tgen, topo)
 350     assert ospf_covergence is True, "Testcase :Failed \n Error:" " {}".format(
 351         ospf_covergence
 352     )
 353
 354     step("Verify that ospf lea's are not set with dc bit 1.")
 355     dut = "r0"
 356     input_dict_db = {
 357         "routerId": "100.1.1.0",
 358         "areas": {
 359             "0.0.0.0": {
 360                 "routerLinkStates": [
 361                     {
 362                         "lsaId": "100.1.1.0",
 363                         "options": "*|-|DC|-|-|-|E|-",
 364                     },
 365                 ]
 366             }
 367         },
 368     }
 369     result = verify_ospf_database(
 370         tgen, topo, dut, input_dict_db, lsatype="router", rid="100.1.1.0", expected=False
 371     )
 372     assert result is not True, (
 373         "Testcase {} : Failed \n "
 374         "Expected: OSPF LSA should not be set with DC bit in {} \n "
 375         "Found: {}".format(tc_name, dut, result)
 376     )
 377     step("Wait for 120 secs and verify that LSA's are not refreshed. ")
 378     # get LSA age
 379     dut = "r1"
 380     input_dict_db = {
 381         "routerId": "100.1.1.0",
 382         "areas": {
 383             "0.0.0.0": {
 384                 "routerLinkStates": [
 385                     {"lsaId": "100.1.1.0", "lsaAge": "get"},
 386                 ]
 387             }
 388         },
 389     }
 390     sleep(10)
 391
 392     result1 = get_ospf_database(
 393         tgen, topo, dut, input_dict_db, lsatype="router", rid="100.1.1.0"
 394     )
 395     sleep(5)
 396     result2 = get_ospf_database(
 397         tgen, topo, dut, input_dict_db, lsatype="router", rid="100.1.1.0"
 398     )
 399
 400     if result2 is not result1:
 401         result = True
 402     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 403
 404     step(
 405         "Enable flood reduction on each router with 10 secs delay of between each router."
 406     )
 407     for rtr in topo["routers"].keys():
 408         ospf_flood = {rtr: {"ospf": {"lsa-refresh": 120, "flood-reduction": True}}}
 409         sleep(10)
 410         result = create_router_ospf(tgen, topo, ospf_flood)
 411         assert result is True, "Testcase : Failed \n Error: {}".format(result)
 412
 413     for rtr in topo["routers"]:
 414         clear_ospf(tgen, rtr)
 415
 416     step("Verify that LSA's are not refreshed. Do not age bit should be set to 1.")
 417     dut = "r0"
 418     input_dict_db = {
 419         "routerId": "100.1.1.0",
 420         "areas": {
 421             "0.0.0.0": {
 422                 "routerLinkStates": [
 423                     {
 424                         "lsaId": "100.1.1.0",
 425                         "options": "*|-|DC|-|-|-|E|-",
 426                     },
 427                 ]
 428             }
 429         },
 430     }
 431     result = verify_ospf_database(
 432         tgen, topo, dut, input_dict_db, lsatype="router", rid="100.1.1.0"
 433     )
 434     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 435
 436     step(
 437         "Verify OSPF neighborship when OSPF flood reduction  is configured and ospf process is restarted"
 438     )
 439
 440     step("Kill OSPFd daemon on R0.")
 441     kill_router_daemons(tgen, "r0", ["ospfd"])
 442     start_router_daemons(tgen, "r0", ["ospfd"])
 443
 444     ospf_covergence = verify_ospf_neighbor(tgen, topo)
 445     assert ospf_covergence is True, "Testcase :Failed \n Error:" " {}".format(
 446         ospf_covergence
 447     )
 448
 449     step("Verify that LSA's are not refreshed. Do not age bit should be set to 1.")
 450     dut = "r0"
 451     input_dict_db = {
 452         "routerId": "100.1.1.0",
 453         "areas": {
 454             "0.0.0.0": {
 455                 "routerLinkStates": [
 456                     {
 457                         "lsaId": "100.1.1.0",
 458                         "options": "*|-|DC|-|-|-|E|-",
 459                     },
 460                 ]
 461             }
 462         },
 463     }
 464     result = verify_ospf_database(
 465         tgen, topo, dut, input_dict_db, lsatype="router", rid="100.1.1.0"
 466     )
 467     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 468
 469     write_test_footer(tc_name)
 470
 471
 472 def test_ospf_flood_red_tc2_p0(request):
 473     """Verify OSPF Flood reduction functionality with ospf enabled on area level."""
 474     tc_name = request.node.name
 475     write_test_header(tc_name)
 476     tgen = get_topogen()
 477
 478     # Don't run this test if we have any failure.
 479     if tgen.routers_have_failure():
 480         check_router_status(tgen)
 481
 482     global topo
 483
 484     step("Bring up the base config as per the topology")
 485     reset_config_on_routers(tgen)
 486     red_static("r0")
 487     step("Base config should be up, verify using OSPF convergence on all the routers")
 488
 489     ospf_covergence = verify_ospf_neighbor(tgen, topo)
 490     assert ospf_covergence is True, "Testcase :Failed \n Error:" " {}".format(
 491         ospf_covergence
 492     )
 493
 494     input_dict = {
 495         "r0": {
 496             "static_routes": [
 497                 {
 498                     "network": NETWORK["ipv4"][0],
 499                     "no_of_ip": 5,
 500                     "next_hop": "Null0",
 501                 }
 502             ]
 503         }
 504     }
 505     result = create_static_routes(tgen, input_dict)
 506     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 507
 508     step("Enable flood reduction in area level on R0 in area 0")
 509     ospf_flood = {
 510         "r0": {"ospf": {"area": [{"id": "0.0.0.0", "flood-reduction": True}]}}
 511     }
 512     result = create_router_ospf(tgen, topo, ospf_flood)
 513     assert result is True, "Testcase : Failed \n Error: {}".format(result)
 514
 515     step("Verify that ospf lsa's are set with dc bit 1.")
 516     dut = "r0"
 517     input_dict_db = {
 518         "routerId": "100.1.1.0",
 519         "areas": {
 520             "0.0.0.0": {
 521                 "routerLinkStates": [
 522                     {
 523                         "lsaId": "100.1.1.0",
 524                         "options": "*|-|DC|-|-|-|E|-",
 525                     },
 526                 ]
 527             }
 528         },
 529     }
 530     result = verify_ospf_database(
 531         tgen, topo, dut, input_dict_db, lsatype="router", rid="100.1.1.0"
 532     )
 533     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 534
 535     step("Configure the custom refresh timer")
 536     ospf_flood = {"r0": {"ospf": {"lsa-refresh": 120}}}
 537     result = create_router_ospf(tgen, topo, ospf_flood)
 538     assert result is True, "Testcase : Failed \n Error: {}".format(result)
 539
 540     step("Enable flood. reduction in all the routers of the topology.")
 541     for rtr in topo["routers"].keys():
 542         ospf_flood = {
 543             rtr: {"ospf": {"area": [{"id": "0.0.0.0", "flood-reduction": True}]}}
 544         }
 545         result = create_router_ospf(tgen, topo, ospf_flood)
 546         assert result is True, "Testcase : Failed \n Error: {}".format(result)
 547
 548     for rtr in topo["routers"]:
 549         clear_ospf(tgen, rtr)
 550
 551     ospf_covergence = verify_ospf_neighbor(tgen, topo)
 552     assert ospf_covergence is True, "Testcase :Failed \n Error:" " {}".format(
 553         ospf_covergence
 554     )
 555
 556     step("Verify that ospf lsa's are set with dc bit 1.")
 557     for rtr in topo["routers"]:
 558         dut = rtr
 559         lsid = "{}".format(topo["routers"][rtr]["ospf"]["router_id"])
 560         input_dict_db = {
 561             "routerId": lsid,
 562             "areas": {
 563                 "0.0.0.0": {
 564                     "routerLinkStates": [
 565                         {
 566                             "lsaId": lsid,
 567                             "options": "*|-|DC|-|-|-|E|-",
 568                         },
 569                     ]
 570                 }
 571             },
 572         }
 573         result = verify_ospf_database(
 574             tgen, topo, dut, input_dict_db, lsatype="router", rid=lsid
 575         )
 576         assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 577
 578     step("Wait for 120 secs and verify that LSA's are not refreshed. ")
 579     # get LSA age
 580     dut = "r1"
 581     input_dict_db = {
 582         "routerId": "100.1.1.0",
 583         "areas": {
 584             "0.0.0.0": {
 585                 "routerLinkStates": [
 586                     {"lsaId": "100.1.1.0", "lsaAge": "get"},
 587                 ]
 588             }
 589         },
 590     }
 591     sleep(10)
 592
 593     result1 = get_ospf_database(
 594         tgen, topo, dut, input_dict_db, lsatype="router", rid="100.1.1.0"
 595     )
 596     sleep(5)
 597     result2 = get_ospf_database(
 598         tgen, topo, dut, input_dict_db, lsatype="router", rid="100.1.1.0"
 599     )
 600
 601     assert (result1 == result2) is True, "Testcase {} : Failed \n Error: {}".format(
 602         tc_name, result
 603     )
 604
 605     step("Disable flood reduction in R0.")
 606     ospf_flood = {
 607         "r0": {
 608             "ospf": {
 609                 "area": [{"id": "0.0.0.0", "flood-reduction": True, "delete": True}]
 610             }
 611         }
 612     }
 613     result = create_router_ospf(tgen, topo, ospf_flood)
 614     assert result is True, "Testcase : Failed \n Error: {}".format(result)
 615
 616     clear_ospf(tgen, "r0")
 617
 618     ospf_covergence = verify_ospf_neighbor(tgen, topo)
 619     assert ospf_covergence is True, "Testcase :Failed \n Error:" " {}".format(
 620         ospf_covergence
 621     )
 622
 623     step("Verify that ospf lea's are not set with dc bit 1.")
 624     dut = "r0"
 625     input_dict_db = {
 626         "routerId": "100.1.1.0",
 627         "areas": {
 628             "0.0.0.0": {
 629                 "routerLinkStates": [
 630                     {
 631                         "lsaId": "100.1.1.0",
 632                         "options": "*|-|DC|-|-|-|E|-",
 633                     },
 634                 ]
 635             }
 636         },
 637     }
 638     result = verify_ospf_database(
 639         tgen, topo, dut, input_dict_db, lsatype="router", rid="100.1.1.0", expected=False
 640     )
 641     assert result is not True, (
 642         "Testcase {} : Failed \n "
 643         "Expected: OSPF LSA should not be set with DC bit in {} \n "
 644         "Found: {}".format(tc_name, dut, result)
 645     )
 646     step("Wait for 120 secs and verify that LSA's are not refreshed. ")
 647     # get LSA age
 648     dut = "r1"
 649     input_dict_db = {
 650         "routerId": "100.1.1.0",
 651         "areas": {
 652             "0.0.0.0": {
 653                 "routerLinkStates": [
 654                     {"lsaId": "100.1.1.0", "lsaAge": "get"},
 655                 ]
 656             }
 657         },
 658     }
 659     sleep(10)
 660
 661     result1 = get_ospf_database(
 662         tgen, topo, dut, input_dict_db, lsatype="router", rid="100.1.1.0"
 663     )
 664     sleep(5)
 665     result2 = get_ospf_database(
 666         tgen, topo, dut, input_dict_db, lsatype="router", rid="100.1.1.0"
 667     )
 668
 669     if result2 is not result1:
 670         result = True
 671     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 672
 673     step(
 674         "Enable flood reduction on each router with 10 secs delay of between each router."
 675     )
 676     for rtr in topo["routers"].keys():
 677         ospf_flood = {
 678             rtr: {"ospf": {"area": [{"id": "0.0.0.0", "flood-reduction": True}]}}
 679         }
 680         sleep(10)
 681         result = create_router_ospf(tgen, topo, ospf_flood)
 682         assert result is True, "Testcase : Failed \n Error: {}".format(result)
 683
 684     for rtr in topo["routers"]:
 685         clear_ospf(tgen, rtr)
 686
 687     step("Verify that LSA's are not refreshed. Do not age bit should be set to 1.")
 688     dut = "r0"
 689     input_dict_db = {
 690         "routerId": "100.1.1.0",
 691         "areas": {
 692             "0.0.0.0": {
 693                 "routerLinkStates": [
 694                     {
 695                         "lsaId": "100.1.1.0",
 696                         "options": "*|-|DC|-|-|-|E|-",
 697                     },
 698                 ]
 699             }
 700         },
 701     }
 702     result = verify_ospf_database(
 703         tgen, topo, dut, input_dict_db, lsatype="router", rid="100.1.1.0"
 704     )
 705     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 706
 707     write_test_footer(tc_name)
 708
 709
 710 def test_ospf_flood_red_tc3_p0(request):
 711     """Verify OSPF Flood reduction functionality between different area's"""
 712     tc_name = request.node.name
 713     write_test_header(tc_name)
 714     tgen = get_topogen()
 715
 716     # Don't run this test if we have any failure.
 717     if tgen.routers_have_failure():
 718         check_router_status(tgen)
 719
 720     global topo
 721
 722     step("Bring up the base config as per the topology")
 723     reset_config_on_routers(tgen)
 724     red_static("r0")
 725     step("Base config should be up, verify using OSPF convergence on all the routers")
 726
 727     ospf_covergence = verify_ospf_neighbor(tgen, topo)
 728     assert ospf_covergence is True, "Testcase :Failed \n Error:" " {}".format(
 729         ospf_covergence
 730     )
 731
 732     input_dict = {
 733         "r0": {
 734             "static_routes": [
 735                 {
 736                     "network": NETWORK["ipv4"][0],
 737                     "no_of_ip": 5,
 738                     "next_hop": "Null0",
 739                 }
 740             ]
 741         }
 742     }
 743     result = create_static_routes(tgen, input_dict)
 744     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 745
 746     step("Enable flood reduction in area level on R0 in area 0")
 747     ospf_flood = {
 748         "r0": {"ospf": {"area": [{"id": "0.0.0.0", "flood-reduction": True}]}}
 749     }
 750     result = create_router_ospf(tgen, topo, ospf_flood)
 751     assert result is True, "Testcase : Failed \n Error: {}".format(result)
 752
 753     step("Verify that ospf lsa's are set with dc bit 1.")
 754     dut = "r0"
 755     input_dict_db = {
 756         "routerId": "100.1.1.0",
 757         "areas": {
 758             "0.0.0.0": {
 759                 "routerLinkStates": [
 760                     {
 761                         "lsaId": "100.1.1.0",
 762                         "options": "*|-|DC|-|-|-|E|-",
 763                     },
 764                 ]
 765             }
 766         },
 767     }
 768     result = verify_ospf_database(
 769         tgen, topo, dut, input_dict_db, lsatype="router", rid="100.1.1.0"
 770     )
 771     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 772
 773     step("Configure the custom refresh timer")
 774     ospf_flood = {"r0": {"ospf": {"lsa-refresh": 120}}}
 775     result = create_router_ospf(tgen, topo, ospf_flood)
 776     assert result is True, "Testcase : Failed \n Error: {}".format(result)
 777
 778     step(
 779         "Enable flood. reduction in all the routers of the topology in area 0. Redistribute static route in area 0 R1 and area1 in R2."
 780     )
 781     for rtr in topo["routers"].keys():
 782         ospf_flood = {
 783             rtr: {"ospf": {"area": [{"id": "0.0.0.0", "flood-reduction": True}]}}
 784         }
 785         result = create_router_ospf(tgen, topo, ospf_flood)
 786         assert result is True, "Testcase : Failed \n Error: {}".format(result)
 787
 788     sleep(10)
 789
 790     for rtr in topo["routers"]:
 791         clear_ospf(tgen, rtr)
 792
 793     ospf_covergence = verify_ospf_neighbor(tgen, topo)
 794     assert ospf_covergence is True, "Testcase :Failed \n Error:" " {}".format(
 795         ospf_covergence
 796     )
 797
 798     step("Verify that ospf lsa's are set with dc bit 1.")
 799     for rtr in topo["routers"]:
 800         dut = rtr
 801         lsid = "{}".format(topo["routers"][rtr]["ospf"]["router_id"])
 802         input_dict_db = {
 803             "routerId": lsid,
 804             "areas": {
 805                 "0.0.0.0": {
 806                     "routerLinkStates": [
 807                         {
 808                             "lsaId": lsid,
 809                             "options": "*|-|DC|-|-|-|E|-",
 810                         },
 811                     ]
 812                 }
 813             },
 814         }
 815         result = verify_ospf_database(
 816             tgen, topo, dut, input_dict_db, lsatype="router", rid=lsid
 817         )
 818         assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 819
 820     step("Wait for 120 secs and verify that LSA's are not refreshed. ")
 821     # get LSA age
 822     dut = "r1"
 823     input_dict_db = {
 824         "routerId": "100.1.1.0",
 825         "areas": {
 826             "0.0.0.0": {
 827                 "routerLinkStates": [
 828                     {"lsaId": "100.1.1.0", "lsaAge": "get"},
 829                 ]
 830             }
 831         },
 832     }
 833     sleep(10)
 834
 835     result1 = get_ospf_database(
 836         tgen, topo, dut, input_dict_db, lsatype="router", rid="100.1.1.0"
 837     )
 838     sleep(5)
 839     result2 = get_ospf_database(
 840         tgen, topo, dut, input_dict_db, lsatype="router", rid="100.1.1.0"
 841     )
 842
 843     assert (result1 == result2) is True, "Testcase {} : Failed \n Error: {}".format(
 844         tc_name, result
 845     )
 846
 847     step("Enable flood reduction in area 1.")
 848
 849     ospf_flood = {
 850         "r0": {"ospf": {"area": [{"id": "0.0.0.0", "flood-reduction": True}]}}
 851     }
 852     result = create_router_ospf(tgen, topo, ospf_flood)
 853     assert result is True, "Testcase : Failed \n Error: {}".format(result)
 854
 855     ospf_flood = {
 856         "r1": {
 857             "ospf": {
 858                 "area": [
 859                     {"id": "0.0.0.0", "flood-reduction": True},
 860                     {"id": "0.0.0.1", "flood-reduction": True},
 861                 ]
 862             }
 863         }
 864     }
 865     result = create_router_ospf(tgen, topo, ospf_flood)
 866     assert result is True, "Testcase : Failed \n Error: {}".format(result)
 867
 868     ospf_flood = {
 869         "r2": {
 870             "ospf": {
 871                 "area": [
 872                     {"id": "0.0.0.0", "flood-reduction": True},
 873                     {"id": "0.0.0.1", "flood-reduction": True},
 874                     {"id": "0.0.0.2", "flood-reduction": True},
 875                 ]
 876             }
 877         }
 878     }
 879     result = create_router_ospf(tgen, topo, ospf_flood)
 880     assert result is True, "Testcase : Failed \n Error: {}".format(result)
 881
 882     ospf_flood = {
 883         "r3": {
 884             "ospf": {
 885                 "area": [
 886                     {"id": "0.0.0.0", "flood-reduction": True},
 887                     {"id": "0.0.0.2", "flood-reduction": True},
 888                 ]
 889             }
 890         }
 891     }
 892     result = create_router_ospf(tgen, topo, ospf_flood)
 893     assert result is True, "Testcase : Failed \n Error: {}".format(result)
 894
 895     for rtr in topo["routers"]:
 896         clear_ospf(tgen, rtr)
 897
 898     ospf_covergence = verify_ospf_neighbor(tgen, topo)
 899     assert ospf_covergence is True, "Testcase :Failed \n Error:" " {}".format(
 900         ospf_covergence
 901     )
 902
 903     step("Verify that ospf lea's are  set with dc bit 1.")
 904     dut = "r1"
 905     input_dict_db = {
 906         "routerId": "100.1.1.1",
 907         "areas": {
 908             "0.0.0.0": {
 909                 "routerLinkStates": [
 910                     {
 911                         "lsaId": "100.1.1.1",
 912                         "options": "*|-|DC|-|-|-|E|-",
 913                     },
 914                 ]
 915             }
 916         },
 917     }
 918     result = verify_ospf_database(
 919         tgen, topo, dut, input_dict_db, lsatype="router", rid="100.1.1.1"
 920     )
 921     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 922
 923     step("Wait for 120 secs and verify that LSA's are not refreshed. ")
 924     # get LSA age
 925     dut = "r1"
 926     input_dict_db = {
 927         "routerId": "100.1.1.1",
 928         "areas": {
 929             "0.0.0.0": {
 930                 "routerLinkStates": [
 931                     {"lsaId": "100.1.1.1", "lsaAge": "get"},
 932                 ]
 933             }
 934         },
 935     }
 936     sleep(10)
 937
 938     result1 = get_ospf_database(
 939         tgen, topo, dut, input_dict_db, lsatype="router", rid="100.1.1.1"
 940     )
 941     sleep(5)
 942     result2 = get_ospf_database(
 943         tgen, topo, dut, input_dict_db, lsatype="router", rid="100.1.1.1"
 944     )
 945
 946     if result2 is result1:
 947         result = True
 948     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 949
 950     step("Disable flood reduction in R0.")
 951
 952     ospf_flood = {
 953         "r0": {
 954             "ospf": {
 955                 "area": [{"id": "0.0.0.0", "flood-reduction": True, "delete": True}]
 956             }
 957         }
 958     }
 959     result = create_router_ospf(tgen, topo, ospf_flood)
 960     assert result is True, "Testcase : Failed \n Error: {}".format(result)
 961
 962     ospf_flood = {
 963         "r1": {
 964             "ospf": {
 965                 "area": [
 966                     {"id": "0.0.0.0", "flood-reduction": True, "delete": True},
 967                     {"id": "0.0.0.1", "flood-reduction": True, "delete": True},
 968                 ]
 969             }
 970         }
 971     }
 972     result = create_router_ospf(tgen, topo, ospf_flood)
 973     assert result is True, "Testcase : Failed \n Error: {}".format(result)
 974
 975     ospf_flood = {
 976         "r2": {
 977             "ospf": {
 978                 "area": [
 979                     {"id": "0.0.0.0", "flood-reduction": True, "delete": True},
 980                     {"id": "0.0.0.1", "flood-reduction": True, "delete": True},
 981                     {"id": "0.0.0.2", "flood-reduction": True, "delete": True},
 982                 ]
 983             }
 984         }
 985     }
 986     result = create_router_ospf(tgen, topo, ospf_flood)
 987     assert result is True, "Testcase : Failed \n Error: {}".format(result)
 988
 989     ospf_flood = {
 990         "r3": {
 991             "ospf": {
 992                 "area": [
 993                     {"id": "0.0.0.0", "flood-reduction": True, "delete": True},
 994                     {"id": "0.0.0.2", "flood-reduction": True, "delete": True},
 995                 ]
 996             }
 997         }
 998     }
 999     result = create_router_ospf(tgen, topo, ospf_flood)
1000     assert result is True, "Testcase : Failed \n Error: {}".format(result)
1001
1002     clear_ospf(tgen, "r0")
1003
1004     ospf_covergence = verify_ospf_neighbor(tgen, topo)
1005     assert ospf_covergence is True, "Testcase :Failed \n Error:" " {}".format(
1006         ospf_covergence
1007     )
1008
1009     step("Verify that ospf lea's are not set with dc bit 1.")
1010     dut = "r0"
1011     input_dict_db = {
1012         "routerId": "100.1.1.0",
1013         "areas": {
1014             "0.0.0.0": {
1015                 "routerLinkStates": [
1016                     {
1017                         "lsaId": "100.1.1.0",
1018                         "options": "*|-|DC|-|-|-|E|-",
1019                     },
1020                 ]
1021             }
1022         },
1023     }
1024     result = verify_ospf_database(
1025         tgen, topo, dut, input_dict_db, lsatype="router", rid="100.1.1.0", expected=False
1026     )
1027
1028     assert result is not True, (
1029         "Testcase {} : Failed \n "
1030         "Expected: OSPF LSA should not be set with DC bit in {} \n "
1031         "Found: {}".format(tc_name, dut, result)
1032     )
1033
1034     step("Wait for 120 secs and verify that LSA's are not refreshed. ")
1035     # get LSA age
1036     dut = "r1"
1037     input_dict_db = {
1038         "routerId": "100.1.1.0",
1039         "areas": {
1040             "0.0.0.0": {
1041                 "routerLinkStates": [
1042                     {"lsaId": "100.1.1.0", "lsaAge": "get"},
1043                 ]
1044             }
1045         },
1046     }
1047     sleep(10)
1048
1049     result1 = get_ospf_database(
1050         tgen, topo, dut, input_dict_db, lsatype="router", rid="100.1.1.0"
1051     )
1052     sleep(5)
1053     result2 = get_ospf_database(
1054         tgen, topo, dut, input_dict_db, lsatype="router", rid="100.1.1.0"
1055     )
1056
1057     if result2 is not result1:
1058         result = True
1059     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
1060
1061     write_test_footer(tc_name)
1062
1063
1064 if __name__ == "__main__":
1065     args = ["-s"] + sys.argv[1:]
1066     sys.exit(pytest.main(args))