tests/topotests/ospf_basic_functionality/test_ospf_asbr_summary_type7_lsa.py

   1 #!/usr/bin/python
   2 # SPDX-License-Identifier: ISC
   3
   4 #
   5 # Copyright (c) 2020 by VMware, Inc. ("VMware")
   6 # Used Copyright (c) 2018 by Network Device Education Foundation, Inc.
   7 # ("NetDEF") in this file.
   8 #
   9
  10
  11 """OSPF Summarisation Functionality Automation."""
  12 import os
  13 import sys
  14 import time
  15 import pytest
  16
  17 # Save the Current Working Directory to find configuration files.
  18 CWD = os.path.dirname(os.path.realpath(__file__))
  19 sys.path.append(os.path.join(CWD, "../"))
  20 sys.path.append(os.path.join(CWD, "../lib/"))
  21
  22 # pylint: disable=C0413
  23 # Import topogen and topotest helpers
  24 from lib.topogen import Topogen, get_topogen
  25 from time import sleep
  26
  27 # Import topoJson from lib, to create topology and initial configuration
  28 from lib.common_config import (
  29     start_topology,
  30     write_test_header,
  31     write_test_footer,
  32     reset_config_on_routers,
  33     verify_rib,
  34     create_static_routes,
  35     step,
  36 )
  37 from lib.topolog import logger
  38 from lib.topojson import build_config_from_json
  39 from lib.ospf import (
  40     verify_ospf_neighbor,
  41     verify_ospf_rib,
  42     create_router_ospf,
  43     verify_ospf_summary,
  44 )
  45
  46 pytestmark = [pytest.mark.ospfd, pytest.mark.staticd]
  47
  48
  49 # Global variables
  50 topo = None
  51
  52 NETWORK = {
  53     "ipv4": [
  54         "11.0.20.1/32",
  55         "11.0.20.2/32",
  56         "11.0.20.3/32",
  57         "11.0.20.4/32",
  58         "11.0.20.5/32",
  59     ]
  60 }
  61 NETWORK2 = {
  62     "ipv4": [
  63         "12.0.20.1/32",
  64         "12.0.20.2/32",
  65         "12.0.20.3/32",
  66         "12.0.20.4/32",
  67         "12.0.20.5/32",
  68     ]
  69 }
  70 NETWORK3 = {
  71     "ipv4": [
  72         "13.0.20.1/32",
  73         "13.0.20.2/32",
  74         "13.0.20.3/32",
  75         "13.0.20.4/32",
  76         "13.0.20.5/32",
  77     ]
  78 }
  79 SUMMARY = {"ipv4": ["11.0.20.1/8", "12.0.0.0/8", "13.0.0.0/8", "11.0.0.0/8"]}
  80 """
  81 TOPOOLOGY =
  82       Please view in a fixed-width font such as Courier.
  83       +---+  A1       +---+
  84       +R1 +------------+R2 |
  85       +-+-+-           +--++
  86         |  --        --  |
  87         |    -- A0 --    |
  88       A0|      ----      |
  89         |      ----      | A2
  90         |    --    --    |
  91         |  --        --  |
  92       +-+-+-            +-+-+
  93       +R0 +-------------+R3 |
  94       +---+     A3     +---+
  95
  96
  97
  98 TESTCASES =
  99 1. OSPF summarisation with type7 LSAs.
 100
 101 """
 102
 103
 104 def setup_module(mod):
 105     """
 106     Sets up the pytest environment
 107
 108     * `mod`: module name
 109     """
 110     testsuite_run_time = time.asctime(time.localtime(time.time()))
 111     logger.info("Testsuite start time: {}".format(testsuite_run_time))
 112     logger.info("=" * 40)
 113
 114     logger.info("Running setup_module to create topology")
 115
 116     # This function initiates the topology build with Topogen...
 117     json_file = "{}/ospf_asbr_summary_type7_lsa.json".format(CWD)
 118     tgen = Topogen(json_file, mod.__name__)
 119     global topo
 120     topo = tgen.json_topo
 121     # ... and here it calls Mininet initialization functions.
 122
 123     # Starting topology, create tmp files which are loaded to routers
 124     #  to start daemons and then start routers
 125     start_topology(tgen)
 126
 127     # Creating configuration from JSON
 128     build_config_from_json(tgen, topo)
 129
 130     # Don't run this test if we have any failure.
 131     if tgen.routers_have_failure():
 132         pytest.skip(tgen.errors)
 133     # Api call verify whether OSPF is converged
 134     ospf_covergence = verify_ospf_neighbor(tgen, topo)
 135     assert ospf_covergence is True, "setup_module :Failed \n Error  {}".format(
 136         ospf_covergence
 137     )
 138
 139     logger.info("Running setup_module() done")
 140
 141
 142 def teardown_module(mod):
 143     """
 144     Teardown the pytest environment.
 145
 146     * `mod`: module name
 147     """
 148
 149     logger.info("Running teardown_module to delete topology")
 150
 151     tgen = get_topogen()
 152
 153     # Stop toplogy and Remove tmp files
 154     tgen.stop_topology()
 155
 156     logger.info(
 157         "Testsuite end time: {}".format(time.asctime(time.localtime(time.time())))
 158     )
 159     logger.info("=" * 40)
 160
 161
 162 def red_static(dut, config=True):
 163     """Local def for Redstribute static routes inside ospf."""
 164     global topo
 165     tgen = get_topogen()
 166     if config:
 167         ospf_red = {dut: {"ospf": {"redistribute": [{"redist_type": "static"}]}}}
 168     else:
 169         ospf_red = {
 170             dut: {"ospf": {"redistribute": [{"redist_type": "static", "delete": True}]}}
 171         }
 172     result = create_router_ospf(tgen, topo, ospf_red)
 173     assert result is True, "Testcase : Failed \n Error: {}".format(result)
 174
 175
 176 def red_connected(dut, config=True):
 177     """Local def for Redstribute connected routes inside ospf."""
 178     global topo
 179     tgen = get_topogen()
 180     if config:
 181         ospf_red = {dut: {"ospf": {"redistribute": [{"redist_type": "connected"}]}}}
 182     else:
 183         ospf_red = {
 184             dut: {
 185                 "ospf": {
 186                     "redistribute": [{"redist_type": "connected", "del_action": True}]
 187                 }
 188             }
 189         }
 190     result = create_router_ospf(tgen, topo, ospf_red)
 191     assert result is True, "Testcase: Failed \n Error: {}".format(result)
 192
 193
 194 # ##################################
 195 # Test cases start here.
 196 # ##################################
 197
 198
 199 def test_ospf_type5_summary_tc44_p0(request):
 200     """OSPF summarisation with type7 LSAs"""
 201
 202     tc_name = request.node.name
 203     write_test_header(tc_name)
 204     tgen = get_topogen()
 205
 206     # Don't run this test if we have any failure.
 207     if tgen.routers_have_failure():
 208         pytest.skip(tgen.errors)
 209
 210     step("Bring up the base config as per the topology")
 211     step("Configure area 1 as NSSA Area")
 212
 213     reset_config_on_routers(tgen)
 214
 215     dut = "r0"
 216     protocol = "ospf"
 217
 218     red_static(dut)
 219     input_dict_static_rtes = {
 220         "r0": {
 221             "static_routes": [
 222                 {"network": NETWORK["ipv4"], "next_hop": "blackhole"},
 223                 {"network": NETWORK2["ipv4"], "next_hop": "blackhole"},
 224             ]
 225         }
 226     }
 227     result = create_static_routes(tgen, input_dict_static_rtes)
 228     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 229
 230     step("Verify that routes are learnt on R1.")
 231     dut = "r1"
 232
 233     result = verify_ospf_rib(tgen, dut, input_dict_static_rtes)
 234     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 235     result = verify_rib(tgen, "ipv4", dut, input_dict_static_rtes, protocol=protocol)
 236     assert (
 237         result is True
 238     ), "Testcase {} : Failed. Error: Routes is missing in RIB".format(tc_name)
 239
 240     step("Configure External Route summary in R0 to summarise 5  routes to one route.")
 241
 242     ospf_summ_r0 = {
 243         "r0": {
 244             "ospf": {
 245                 "summary-address": [
 246                     {"prefix": SUMMARY["ipv4"][0].split("/")[0], "mask": "8"}
 247                 ]
 248             }
 249         }
 250     }
 251     result = create_router_ospf(tgen, topo, ospf_summ_r0)
 252     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 253
 254     step(
 255         "Verify that external routes are summarised to configured summary "
 256         "address on R0 after 5 secs of delay timer expiry and only one "
 257         "route is sent to R1."
 258     )
 259
 260     step("Configure summary & redistribute static/connected route with  metric type 2")
 261
 262     input_dict_summary = {"r0": {"static_routes": [{"network": SUMMARY["ipv4"][3]}]}}
 263     dut = "r1"
 264
 265     result = verify_ospf_rib(tgen, dut, input_dict_summary)
 266     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 267
 268     result = verify_rib(tgen, "ipv4", dut, input_dict_summary, protocol=protocol)
 269     assert (
 270         result is True
 271     ), "Testcase {} : Failed. Error: Routes is missing in RIB".format(tc_name)
 272
 273     step("Verify that show ip ospf summary should show the summaries.")
 274     input_dict = {
 275         SUMMARY["ipv4"][3]: {
 276             "summaryAddress": SUMMARY["ipv4"][3],
 277             "metricType": "E2",
 278             "metric": 20,
 279             "tag": 0,
 280             "externalRouteCount": 5,
 281         }
 282     }
 283     dut = "r0"
 284     result = verify_ospf_summary(tgen, topo, dut, input_dict)
 285     assert (
 286         result is True
 287     ), "Testcase {} : Failed. Error: Summary missing in OSPF DB".format(tc_name)
 288
 289     step("Learn type 7 lsa from neighbours")
 290
 291     dut = "r1"
 292     protocol = "ospf"
 293
 294     red_static(dut)
 295     input_dict_static_rtes = {
 296         "r1": {
 297             "static_routes": [{"network": NETWORK3["ipv4"], "next_hop": "blackhole"}]
 298         }
 299     }
 300     result = create_static_routes(tgen, input_dict_static_rtes)
 301     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 302
 303     step("Verify that routes are learnt on R0.")
 304     dut = "r0"
 305
 306     result = verify_ospf_rib(tgen, dut, input_dict_static_rtes)
 307     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 308     result = verify_rib(tgen, "ipv4", dut, input_dict_static_rtes, protocol=protocol)
 309     assert (
 310         result is True
 311     ), "Testcase {} : Failed. Error: Routes is missing in RIB".format(tc_name)
 312
 313     ospf_summ_r0 = {
 314         "r0": {
 315             "ospf": {
 316                 "summary-address": [
 317                     {"prefix": SUMMARY["ipv4"][2].split("/")[0], "mask": "8"}
 318                 ]
 319             }
 320         }
 321     }
 322     result = create_router_ospf(tgen, topo, ospf_summ_r0)
 323     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 324
 325     step("Verify that type7 LSAs received from neighbor are not summarised.")
 326     input_dict = {
 327         "13.0.0.0/8": {
 328             "summaryAddress": "13.0.0.0/8",
 329             "metricType": "E2",
 330             "metric": 20,
 331             "tag": 0,
 332             "externalRouteCount": 0,
 333         }
 334     }
 335     dut = "r0"
 336     result = verify_ospf_summary(tgen, topo, dut, input_dict)
 337     assert (
 338         result is True
 339     ), "Testcase {} : Failed. Error: Summary missing in OSPF DB".format(tc_name)
 340
 341     step("Verify that already originated summary is intact.")
 342     input_dict = {
 343         SUMMARY["ipv4"][3]: {
 344             "summaryAddress": SUMMARY["ipv4"][3],
 345             "metricType": "E2",
 346             "metric": 20,
 347             "tag": 0,
 348             "externalRouteCount": 5,
 349         }
 350     }
 351     dut = "r0"
 352     result = verify_ospf_summary(tgen, topo, dut, input_dict)
 353     assert (
 354         result is True
 355     ), "Testcase {} : Failed. Error: Summary missing in OSPF DB".format(tc_name)
 356
 357     dut = "r1"
 358     aggr_timer = {"r1": {"ospf": {"aggr_timer": 6}}}
 359     result = create_router_ospf(tgen, topo, aggr_timer)
 360     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 361     ospf_summ_r0 = {
 362         "r0": {
 363             "ospf": {
 364                 "summary-address": [
 365                     {"prefix": SUMMARY["ipv4"][2].split("/")[0], "mask": "8"}
 366                 ]
 367             }
 368         }
 369     }
 370     result = create_router_ospf(tgen, topo, ospf_summ_r0)
 371     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 372
 373     step(
 374         "wait for 6+1 seconds as ospf aggregation start after 6 secs as "
 375         "per the above aggr_timer command"
 376     )
 377     sleep(7)
 378     dut = "r1"
 379     aggr_timer = {"r1": {"ospf": {"del_aggr_timer": 6}}}
 380     result = create_router_ospf(tgen, topo, aggr_timer)
 381     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 382
 383     write_test_footer(tc_name)
 384
 385
 386 if __name__ == "__main__":
 387     args = ["-s"] + sys.argv[1:]
 388     sys.exit(pytest.main(args))