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(
 241         "Configure External Route summary in R0 to summarise 5" " routes to one route."
 242     )
 243
 244     ospf_summ_r0 = {
 245         "r0": {
 246             "ospf": {
 247                 "summary-address": [
 248                     {"prefix": SUMMARY["ipv4"][0].split("/")[0], "mask": "8"}
 249                 ]
 250             }
 251         }
 252     }
 253     result = create_router_ospf(tgen, topo, ospf_summ_r0)
 254     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 255
 256     step(
 257         "Verify that external routes are summarised to configured summary "
 258         "address on R0 after 5 secs of delay timer expiry and only one "
 259         "route is sent to R1."
 260     )
 261
 262     step(
 263         "Configure summary & redistribute static/connected route with " "metric type 2"
 264     )
 265
 266     input_dict_summary = {"r0": {"static_routes": [{"network": SUMMARY["ipv4"][3]}]}}
 267     dut = "r1"
 268
 269     result = verify_ospf_rib(tgen, dut, input_dict_summary)
 270     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 271
 272     result = verify_rib(tgen, "ipv4", dut, input_dict_summary, protocol=protocol)
 273     assert (
 274         result is True
 275     ), "Testcase {} : Failed" "Error: Routes is missing in RIB".format(tc_name)
 276
 277     step("Verify that show ip ospf summary should show the summaries.")
 278     input_dict = {
 279         SUMMARY["ipv4"][3]: {
 280             "summaryAddress": SUMMARY["ipv4"][3],
 281             "metricType": "E2",
 282             "metric": 20,
 283             "tag": 0,
 284             "externalRouteCount": 5,
 285         }
 286     }
 287     dut = "r0"
 288     result = verify_ospf_summary(tgen, topo, dut, input_dict)
 289     assert (
 290         result is True
 291     ), "Testcase {} : Failed" "Error: Summary missing in OSPF DB".format(tc_name)
 292
 293     step("Learn type 7 lsa from neighbours")
 294
 295     dut = "r1"
 296     protocol = "ospf"
 297
 298     red_static(dut)
 299     input_dict_static_rtes = {
 300         "r1": {
 301             "static_routes": [{"network": NETWORK3["ipv4"], "next_hop": "blackhole"}]
 302         }
 303     }
 304     result = create_static_routes(tgen, input_dict_static_rtes)
 305     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 306
 307     step("Verify that routes are learnt on R0.")
 308     dut = "r0"
 309
 310     result = verify_ospf_rib(tgen, dut, input_dict_static_rtes)
 311     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 312     result = verify_rib(tgen, "ipv4", dut, input_dict_static_rtes, protocol=protocol)
 313     assert (
 314         result is True
 315     ), "Testcase {} : Failed" "Error: Routes is missing in RIB".format(tc_name)
 316
 317     ospf_summ_r0 = {
 318         "r0": {
 319             "ospf": {
 320                 "summary-address": [
 321                     {"prefix": SUMMARY["ipv4"][2].split("/")[0], "mask": "8"}
 322                 ]
 323             }
 324         }
 325     }
 326     result = create_router_ospf(tgen, topo, ospf_summ_r0)
 327     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 328
 329     step("Verify that type7 LSAs received from neighbor are not summarised.")
 330     input_dict = {
 331         "13.0.0.0/8": {
 332             "summaryAddress": "13.0.0.0/8",
 333             "metricType": "E2",
 334             "metric": 20,
 335             "tag": 0,
 336             "externalRouteCount": 0,
 337         }
 338     }
 339     dut = "r0"
 340     result = verify_ospf_summary(tgen, topo, dut, input_dict)
 341     assert (
 342         result is True
 343     ), "Testcase {} : Failed" "Error: Summary missing in OSPF DB".format(tc_name)
 344
 345     step("Verify that already originated summary is intact.")
 346     input_dict = {
 347         SUMMARY["ipv4"][3]: {
 348             "summaryAddress": SUMMARY["ipv4"][3],
 349             "metricType": "E2",
 350             "metric": 20,
 351             "tag": 0,
 352             "externalRouteCount": 5,
 353         }
 354     }
 355     dut = "r0"
 356     result = verify_ospf_summary(tgen, topo, dut, input_dict)
 357     assert (
 358         result is True
 359     ), "Testcase {} : Failed" "Error: Summary missing in OSPF DB".format(tc_name)
 360
 361     dut = "r1"
 362     aggr_timer = {"r1": {"ospf": {"aggr_timer": 6}}}
 363     result = create_router_ospf(tgen, topo, aggr_timer)
 364     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 365     ospf_summ_r0 = {
 366         "r0": {
 367             "ospf": {
 368                 "summary-address": [
 369                     {"prefix": SUMMARY["ipv4"][2].split("/")[0], "mask": "8"}
 370                 ]
 371             }
 372         }
 373     }
 374     result = create_router_ospf(tgen, topo, ospf_summ_r0)
 375     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 376
 377     step(
 378         "wait for 6+1 seconds as ospf aggregation start after 6 secs as "
 379         "per the above aggr_timer command"
 380     )
 381     sleep(7)
 382     dut = "r1"
 383     aggr_timer = {"r1": {"ospf": {"del_aggr_timer": 6}}}
 384     result = create_router_ospf(tgen, topo, aggr_timer)
 385     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 386
 387     write_test_footer(tc_name)
 388
 389
 390 if __name__ == "__main__":
 391     args = ["-s"] + sys.argv[1:]
 392     sys.exit(pytest.main(args))