tests/topotests/bgp_local_asn/test_bgp_local_asn_agg.py

   1 #!/usr/bin/env python3
   2 # SPDX-License-Identifier: ISC
   3 #
   4 # Copyright (c) 2022 by VMware, Inc. ("VMware")
   5 # Used Copyright (c) 2018 by Network Device Education Foundation,
   6 # Inc. ("NetDEF") in this file.
   7 #
   8
   9
  10 """
  11 Following tests are covered to test BGP Multi-VRF Dynamic Route Leaking:
  12 1. Verify the BGP Local AS functionality by aggregating routes  in between eBGP Peers.
  13 """
  14
  15 import os
  16 import sys
  17 import time
  18 import pytest
  19
  20 # Save the Current Working Directory to find configuration files.
  21 CWD = os.path.dirname(os.path.realpath(__file__))
  22 sys.path.append(os.path.join(CWD, "../"))
  23 sys.path.append(os.path.join(CWD, "../lib/"))
  24
  25 # pylint: disable=C0413
  26 # Import topogen and topotest helpers
  27 from lib.topogen import Topogen, get_topogen
  28 from lib.topotest import version_cmp
  29
  30 from lib.common_config import (
  31     start_topology,
  32     write_test_header,
  33     write_test_footer,
  34     reset_config_on_routers,
  35     verify_rib,
  36     step,
  37     check_address_types,
  38     check_router_status
  39 )
  40
  41 from lib.topolog import logger
  42 from lib.bgp import (
  43     verify_bgp_convergence,
  44     verify_bgp_rib,
  45     create_router_bgp,
  46 )
  47 from lib.topojson import build_config_from_json
  48
  49 pytestmark = [pytest.mark.bgpd, pytest.mark.staticd]
  50
  51 # Global variables
  52 BGP_CONVERGENCE = False
  53 ADDR_TYPES = check_address_types()
  54 NETWORK_1_1 = {"ipv4": "10.1.1.0/32", "ipv6": "10:1::1:0/128"}
  55 NETWORK_1_2 = {"ipv4": "10.1.2.0/32", "ipv6": "10:1::2:0/128"}
  56 AGGREGATE_NW = {"ipv4": "10.1.0.0/16", "ipv6": "10:1::/96"}
  57 NEXT_HOP_IP = {"ipv4": "Null0", "ipv6": "Null0"}
  58
  59
  60 def setup_module(mod):
  61     """
  62     Sets up the pytest environment
  63
  64     * `mod`: module name
  65     """
  66
  67     testsuite_run_time = time.asctime(time.localtime(time.time()))
  68     logger.info("Testsuite start time: {}".format(testsuite_run_time))
  69     logger.info("=" * 40)
  70
  71     logger.info("Running setup_module to create topology")
  72
  73     # This function initiates the topology build with Topogen...
  74     json_file = "{}/bgp_local_asn_agg.json".format(CWD)
  75     tgen = Topogen(json_file, mod.__name__)
  76     global topo
  77     topo = tgen.json_topo
  78     # ... and here it calls Mininet initialization functions.
  79
  80     # Starting topology, create tmp files which are loaded to routers
  81     #  to start daemons and then start routers
  82     start_topology(tgen)
  83
  84     # Creating configuration from JSON
  85     build_config_from_json(tgen, topo)
  86
  87     global BGP_CONVERGENCE
  88     global ADDR_TYPES
  89     ADDR_TYPES = check_address_types()
  90
  91     BGP_CONVERGENCE = verify_bgp_convergence(tgen, topo)
  92     assert BGP_CONVERGENCE is True, "setup_module : Failed \n Error: {}".format(
  93         BGP_CONVERGENCE
  94     )
  95
  96     logger.info("Running setup_module() done")
  97
  98
  99 def teardown_module():
 100     """Teardown the pytest environment"""
 101
 102     logger.info("Running teardown_module to delete topology")
 103
 104     tgen = get_topogen()
 105
 106     # Stop toplogy and Remove tmp files
 107     tgen.stop_topology()
 108
 109     logger.info(
 110         "Testsuite end time: {}".format(time.asctime(time.localtime(time.time())))
 111     )
 112     logger.info("=" * 40)
 113
 114
 115 ####################################################################################################################
 116 #
 117 #   Testcases
 118 #
 119 ####################################################################################################################
 120
 121
 122 def test_verify_bgp_local_as_agg_in_EBGP_p0(request):
 123     """
 124     Verify the BGP Local AS functionality by aggregating routes  in between eBGP Peers.
 125     """
 126     tgen = get_topogen()
 127     global BGP_CONVERGENCE
 128
 129     if BGP_CONVERGENCE != True:
 130         pytest.skip("skipped because of BGP Convergence failure")
 131
 132     # test case name
 133     tc_name = request.node.name
 134     write_test_header(tc_name)
 135     if tgen.routers_have_failure():
 136         check_router_status(tgen)
 137     reset_config_on_routers(tgen)
 138
 139     step("Base config is done as part of JSON")
 140     step("Configure local-as at R3 towards R4.")
 141     for addr_type in ADDR_TYPES:
 142         for neighbor in ["r2", "r4"]:
 143             input_dict_r3 = {
 144                 "r3": {
 145                     "bgp": {
 146                         "local_as": "300",
 147                         "address_family": {
 148                             addr_type: {
 149                                 "unicast": {
 150                                     "neighbor": {
 151                                         neighbor: {
 152                                             "dest_link": {
 153                                                 "r3": {"local_asn": {"local_as": "110"}}
 154                                             }
 155                                         }
 156                                     }
 157                                 }
 158                             }
 159                         },
 160                     }
 161                 }
 162             }
 163             result = create_router_bgp(tgen, topo, input_dict_r3)
 164             assert result is True, "Testcase {} :Failed \n Error: {}".format(
 165                 tc_name, result
 166             )
 167
 168     for addr_type in ADDR_TYPES:
 169         for dut, asn, neighbor in zip(["r2", "r4"], ["200", "400"], ["r3", "r3"]):
 170             input_dict_r2_r4 = {
 171                 dut: {
 172                     "bgp": {
 173                         "local_as": asn,
 174                         "address_family": {
 175                             addr_type: {
 176                                 "unicast": {
 177                                     "neighbor": {
 178                                         neighbor: {
 179                                             "dest_link": {
 180                                                 dut: {"local_asn": {"remote_as": "110"}}
 181                                             }
 182                                         }
 183                                     }
 184                                 }
 185                             }
 186                         },
 187                     }
 188                 }
 189             }
 190             result = create_router_bgp(tgen, topo, input_dict_r2_r4)
 191             assert result is True, "Testcase {} :Failed \n Error: {}".format(
 192                 tc_name, result
 193             )
 194
 195     step("BGP neighborship is verified by following commands in R3 routers")
 196     BGP_CONVERGENCE = verify_bgp_convergence(tgen, topo)
 197     assert BGP_CONVERGENCE is True, "BGP convergence :Failed \n Error: {}".format(
 198         BGP_CONVERGENCE
 199     )
 200
 201     step("Done in base config: Advertise prefix 10.1.1.0/24 from Router-1(AS-100).")
 202     step(
 203         "Done in base config: Advertise an ipv6 prefix 10:1::1:0/120 from Router-1(AS-100)."
 204     )
 205     step("Verify that Static routes are redistributed in BGP process")
 206     for addr_type in ADDR_TYPES:
 207         input_static_verify_r1 = {
 208             "r1": {"static_routes": [{"network": NETWORK_1_1[addr_type]}]}
 209         }
 210
 211         input_static_verify_r2 = {
 212             "r2": {"static_routes": [{"network": NETWORK_1_2[addr_type]}]}
 213         }
 214         result = verify_rib(tgen, addr_type, "r1", input_static_verify_r1)
 215         assert result is True, "Testcase {}: Failed \n Error: {}".format(
 216             tc_name, result
 217         )
 218
 219         result = verify_rib(tgen, addr_type, "r2", input_static_verify_r2)
 220         assert result is True, "Testcase {}: Failed \n Error: {}".format(
 221             tc_name, result
 222         )
 223
 224     step("Configure aggregate-address to summarise all the advertised routes.")
 225     for addr_type in ADDR_TYPES:
 226         route_aggregate = {
 227             "r3": {
 228                 "bgp": {
 229                     "address_family": {
 230                         addr_type: {
 231                             "unicast": {
 232                                 "aggregate_address": [
 233                                     {
 234                                         "network": AGGREGATE_NW[addr_type],
 235                                         "summary": True,
 236                                         "as_set": True,
 237                                     }
 238                                 ]
 239                             }
 240                         }
 241                     }
 242                 }
 243             }
 244         }
 245
 246         result = create_router_bgp(tgen, topo, route_aggregate)
 247         assert result is True, "Testcase {} :Failed \n Error: {}".format(
 248             tc_name, result
 249         )
 250
 251     step(
 252         "Verify that we see a summarised route on advertising router R3 "
 253         "and receiving router R4 for both AFIs"
 254     )
 255
 256     for addr_type in ADDR_TYPES:
 257         input_static_agg_r1 = {
 258             "r1": {"static_routes": [{"network": AGGREGATE_NW[addr_type]}]}
 259         }
 260         input_static_r1 = {
 261             "r1": {"static_routes": [{"network": [NETWORK_1_1[addr_type]]}]}
 262         }
 263
 264         input_static_r2 = {
 265             "r2": {"static_routes": [{"network": [NETWORK_1_2[addr_type]]}]}
 266         }
 267
 268         for dut in ["r3", "r4"]:
 269             result = verify_rib(tgen, addr_type, dut, input_static_agg_r1)
 270             assert result is True, "Testcase {}: Failed \n Error: {}".format(
 271                 tc_name, result
 272             )
 273
 274         for dut, input_routes in zip(["r1", "r2"], [input_static_r1, input_static_r2]):
 275             result = verify_rib(tgen, addr_type, dut, input_routes)
 276             assert result is True, "Testcase {}: Failed \n Error: {}".format(
 277                 tc_name, result
 278             )
 279
 280     step(
 281         "Verify that AS-110 is got added in the AS list 110 {100,110,200} by following "
 282         "commands at R3 router."
 283     )
 284     dut = "r3"
 285     aspath = "{100,110,200}"
 286     for addr_type in ADDR_TYPES:
 287         input_static_agg_r1 = {
 288             "r1": {"static_routes": [{"network": AGGREGATE_NW[addr_type]}]}
 289         }
 290         result = verify_bgp_rib(
 291             tgen, addr_type, dut, input_static_agg_r1, aspath=aspath
 292         )
 293         assert result is True, "Testcase {} : Failed \n Error: {}".format(
 294             tc_name, result
 295         )
 296
 297     step("Configure local-as with no-prepend at R3 towards R4 & R2.")
 298     for addr_type in ADDR_TYPES:
 299         for neighbor in ["r2", "r4"]:
 300             input_dict_r3 = {
 301                 "r3": {
 302                     "bgp": {
 303                         "local_as": "300",
 304                         "address_family": {
 305                             addr_type: {
 306                                 "unicast": {
 307                                     "neighbor": {
 308                                         neighbor: {
 309                                             "dest_link": {
 310                                                 "r3": {
 311                                                     "local_asn": {
 312                                                         "local_as": "110",
 313                                                         "no_prepend": True,
 314                                                     }
 315                                                 }
 316                                             }
 317                                         }
 318                                     }
 319                                 }
 320                             }
 321                         },
 322                     }
 323                 }
 324             }
 325             result = create_router_bgp(tgen, topo, input_dict_r3)
 326             assert result is True, "Testcase {} :Failed \n Error: {}".format(
 327                 tc_name, result
 328             )
 329
 330     step("BGP neighborship is verified by following commands in R3 routers")
 331     BGP_CONVERGENCE = verify_bgp_convergence(tgen, topo)
 332     assert BGP_CONVERGENCE is True, "BGP convergence :Failed \n Error: {}".format(
 333         BGP_CONVERGENCE
 334     )
 335
 336     dut = "r3"
 337     aspath = "{100,200}"
 338     for addr_type in ADDR_TYPES:
 339         input_static_agg_r1 = {
 340             "r1": {"static_routes": [{"network": AGGREGATE_NW[addr_type]}]}
 341         }
 342         result = verify_bgp_rib(
 343             tgen, addr_type, dut, input_static_agg_r1, aspath=aspath
 344         )
 345         assert result is True, "Testcase {} : Failed \n Error: {}".format(
 346             tc_name, result
 347         )
 348
 349     step("Configure local-as with no-prepend and replace-as at R3 towards R4 & R2.")
 350     for addr_type in ADDR_TYPES:
 351         for neighbor in ["r2", "r4"]:
 352             input_dict_r3 = {
 353                 "r3": {
 354                     "bgp": {
 355                         "local_as": "300",
 356                         "address_family": {
 357                             addr_type: {
 358                                 "unicast": {
 359                                     "neighbor": {
 360                                         neighbor: {
 361                                             "dest_link": {
 362                                                 "r3": {
 363                                                     "local_asn": {
 364                                                         "local_as": "110",
 365                                                         "no_prepend": True,
 366                                                         "replace_as": True,
 367                                                     }
 368                                                 }
 369                                             }
 370                                         }
 371                                     }
 372                                 }
 373                             }
 374                         },
 375                     }
 376                 }
 377             }
 378             result = create_router_bgp(tgen, topo, input_dict_r3)
 379             assert result is True, "Testcase {} :Failed \n Error: {}".format(
 380                 tc_name, result
 381             )
 382
 383     step("BGP neighborship is verified by following commands in R3 routers")
 384     BGP_CONVERGENCE = verify_bgp_convergence(tgen, topo)
 385     assert BGP_CONVERGENCE is True, "BGP convergence :Failed \n Error: {}".format(
 386         BGP_CONVERGENCE
 387     )
 388
 389     dut = "r4"
 390     aspath = "110 {100,200}"
 391     for addr_type in ADDR_TYPES:
 392         input_static_agg_r1 = {
 393             "r1": {"static_routes": [{"network": AGGREGATE_NW[addr_type]}]}
 394         }
 395         result = verify_bgp_rib(
 396             tgen, addr_type, dut, input_static_agg_r1, aspath=aspath
 397         )
 398         assert result is True, "Testcase {} : Failed \n Error: {}".format(
 399             tc_name, result
 400         )
 401
 402     write_test_footer(tc_name)
 403
 404
 405 if __name__ == "__main__":
 406     args = ["-s"] + sys.argv[1:]
 407     sys.exit(pytest.main(args))