tests/topotests/bgp_communities_topo1/test_bgp_communities_topo2.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,
   6 # Inc. ("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 Following tests are covered to test bgp community functionality:
  25 1. Verify that BGP well known communities work fine for
  26    eBGP and iBGP peers.
  27    Well known communities tested: no-export, local-AS, internet
  28
  29 """
  30
  31 import os
  32 import sys
  33 import time
  34 import pytest
  35
  36 # Save the Current Working Directory to find configuration files.
  37 CWD = os.path.dirname(os.path.realpath(__file__))
  38 sys.path.append(os.path.join(CWD, "../"))
  39
  40 # pylint: disable=C0413
  41 # Import topogen and topotest helpers
  42 from lib.topogen import Topogen, get_topogen
  43
  44 # Import topoJson from lib, to create topology and initial configuration
  45 from lib.common_config import (
  46     start_topology,
  47     write_test_header,
  48     write_test_footer,
  49     reset_config_on_routers,
  50     verify_rib,
  51     create_static_routes,
  52     check_address_types,
  53     step,
  54     create_route_maps,
  55     create_route_maps,
  56     required_linux_kernel_version,
  57 )
  58
  59 from lib.topolog import logger
  60 from lib.bgp import (
  61     verify_bgp_convergence,
  62     create_router_bgp,
  63     verify_bgp_rib,
  64     verify_bgp_community,
  65 )
  66 from lib.topojson import build_config_from_json
  67
  68 pytestmark = [pytest.mark.bgpd, pytest.mark.staticd]
  69
  70
  71 # Global variables
  72 BGP_CONVERGENCE = False
  73 ADDR_TYPES = check_address_types()
  74 NETWORK = {
  75     "ipv4": ["192.0.2.1/32", "192.0.2.2/32"],
  76     "ipv6": ["2001:DB8::1:1/128", "2001:DB8::1:2/128"],
  77 }
  78
  79
  80 def setup_module(mod):
  81     """
  82     Sets up the pytest environment
  83
  84     * `mod`: module name
  85     """
  86
  87     # Required linux kernel version for this suite to run.
  88     result = required_linux_kernel_version("4.14")
  89     if result is not True:
  90         pytest.skip("Kernel requirements are not met, kernel version should be >= 4.14")
  91
  92     testsuite_run_time = time.asctime(time.localtime(time.time()))
  93     logger.info("Testsuite start time: {}".format(testsuite_run_time))
  94     logger.info("=" * 40)
  95
  96     logger.info("Running setup_module to create topology")
  97
  98     # This function initiates the topology build with Topogen...
  99     json_file = "{}/bgp_communities_topo2.json".format(CWD)
 100     tgen = Topogen(json_file, mod.__name__)
 101     global topo
 102     topo = tgen.json_topo
 103     # ... and here it calls Mininet initialization functions.
 104
 105     # Starting topology, create tmp files which are loaded to routers
 106     #  to start daemons and then start routers
 107     start_topology(tgen)
 108
 109     # Creating configuration from JSON
 110     build_config_from_json(tgen, topo)
 111
 112     # Checking BGP convergence
 113     global BGP_CONVERGENCE
 114     global ADDR_TYPES
 115
 116     # Don't run this test if we have any failure.
 117     if tgen.routers_have_failure():
 118         pytest.skip(tgen.errors)
 119
 120     # Api call verify whether BGP is converged
 121     BGP_CONVERGENCE = verify_bgp_convergence(tgen, topo)
 122     assert BGP_CONVERGENCE is True, "setup_module :Failed \n Error:" " {}".format(
 123         BGP_CONVERGENCE
 124     )
 125
 126     logger.info("Running setup_module() done")
 127
 128
 129 def teardown_module(mod):
 130     """
 131     Teardown the pytest environment
 132
 133     * `mod`: module name
 134     """
 135
 136     logger.info("Running teardown_module to delete topology")
 137
 138     tgen = get_topogen()
 139
 140     # Stop toplogy and Remove tmp files
 141     tgen.stop_topology()
 142
 143     logger.info(
 144         "Testsuite end time: {}".format(time.asctime(time.localtime(time.time())))
 145     )
 146     logger.info("=" * 40)
 147
 148
 149 #####################################################
 150 #
 151 #   Tests starting
 152 #
 153 #####################################################
 154
 155
 156 def test_bgp_no_export_local_as_and_internet_communities_p0(request):
 157     """
 158     Verify that BGP well known communities work fine for
 159     eBGP and iBGP peers.
 160     Well known communities tested: no-export, local-AS, internet
 161     """
 162
 163     tc_name = request.node.name
 164     write_test_header(tc_name)
 165     tgen = get_topogen()
 166
 167     # Don"t run this test if we have any failure.
 168     if tgen.routers_have_failure():
 169         pytest.skip(tgen.errors)
 170
 171     step("Initial config: Configure BGP neighborship between R1 and R3.")
 172     reset_config_on_routers(tgen)
 173
 174     step("Configure static routes on R1 with next-hop as null0")
 175     for addr_type in ADDR_TYPES:
 176         input_dict_4 = {
 177             "r1": {
 178                 "static_routes": [{"network": NETWORK[addr_type], "next_hop": "null0"}]
 179             }
 180         }
 181         result = create_static_routes(tgen, input_dict_4)
 182         assert result is True, "Testcase {} : Failed \n Error: {}".format(
 183             tc_name, result
 184         )
 185
 186     for comm_type in ["no-export", "local-AS", "internet"]:
 187
 188         step("Create a route-map on R1 to set community as {}".format(comm_type))
 189
 190         seq_id = 10
 191         input_rmap = {
 192             "r1": {
 193                 "route_maps": {
 194                     "rmap_wkc": [
 195                         {
 196                             "action": "permit",
 197                             "seq_id": seq_id,
 198                             "set": {"community": {"num": "{}".format(comm_type)}},
 199                         }
 200                     ]
 201                 }
 202             }
 203         }
 204         result = create_route_maps(tgen, input_rmap)
 205         assert result is True, "Testcase {} : Failed \n Error: {}".format(
 206             tc_name, result
 207         )
 208
 209         step("Apply route-map while redistributing static routes into BGP")
 210         input_dict_2 = {
 211             "r1": {
 212                 "bgp": {
 213                     "address_family": {
 214                         "ipv4": {
 215                             "unicast": {
 216                                 "redistribute": [
 217                                     {
 218                                         "redist_type": "static",
 219                                         "attribute": {"route-map": "rmap_wkc"},
 220                                     }
 221                                 ]
 222                             }
 223                         },
 224                         "ipv6": {
 225                             "unicast": {
 226                                 "redistribute": [
 227                                     {
 228                                         "redist_type": "static",
 229                                         "attribute": {"route-map": "rmap_wkc"},
 230                                     }
 231                                 ]
 232                             }
 233                         },
 234                     }
 235                 }
 236             }
 237         }
 238         result = create_router_bgp(tgen, topo, input_dict_2)
 239
 240         step("Verify that BGP prefixes on R1 have community: {}".format(comm_type))
 241         input_dict_4 = {"community": "{}".format(comm_type)}
 242         for addr_type in ADDR_TYPES:
 243             result = verify_bgp_community(
 244                 tgen, addr_type, "r1", NETWORK[addr_type], input_dict_4
 245             )
 246             assert result is True, "Test case {} : Should fail \n Error: {}".format(
 247                 tc_name, result
 248             )
 249
 250         for addr_type in ADDR_TYPES:
 251             input_dict_4 = {
 252                 "r1": {
 253                     "static_routes": [
 254                         {
 255                             "network": NETWORK[addr_type],
 256                             "next_hop": topo["routers"]["r2"]["links"]["r1"][
 257                                 addr_type
 258                             ].split("/")[0],
 259                         }
 260                     ]
 261                 }
 262             }
 263             result = verify_bgp_rib(
 264                 tgen,
 265                 addr_type,
 266                 "r2",
 267                 input_dict_4,
 268                 next_hop=topo["routers"]["r1"]["links"]["r2"][addr_type].split("/")[0],
 269             )
 270             assert result is True, "Testcase {} : Failed \n Error: {}".format(
 271                 tc_name, result
 272             )
 273
 274             if comm_type == "internet":
 275                 step(
 276                     "Verify that these prefixes, originated on R1, are"
 277                     "received on both R2 and R3"
 278                 )
 279
 280                 result = verify_rib(
 281                     tgen,
 282                     addr_type,
 283                     "r3",
 284                     input_dict_4,
 285                     next_hop=topo["routers"]["r1"]["links"]["r3"][addr_type].split("/")[
 286                         0
 287                     ],
 288                 )
 289                 assert result is True, "Testcase {} : Failed \n Error: {}".format(
 290                     tc_name, result
 291                 )
 292             else:
 293                 step(
 294                     "Verify that these prefixes, originated on R1, are not"
 295                     "received on R3 but received on R2"
 296                 )
 297
 298                 result = verify_rib(
 299                     tgen,
 300                     addr_type,
 301                     "r3",
 302                     input_dict_4,
 303                     next_hop=topo["routers"]["r1"]["links"]["r3"][addr_type].split("/")[
 304                         0
 305                     ],
 306                     expected=False,
 307                 )
 308                 assert result is not True, (
 309                     "Testcase {} : Failed \n "
 310                     "Expected: Routes are still present in rib of r3 \n "
 311                     "Found: {}".format(tc_name, result)
 312                 )
 313
 314         step("Remove route-map from redistribute static on R1")
 315         input_dict_2 = {
 316             "r1": {
 317                 "bgp": {
 318                     "address_family": {
 319                         "ipv4": {
 320                             "unicast": {
 321                                 "redistribute": [
 322                                     {"redist_type": "static", "delete": True}
 323                                 ]
 324                             }
 325                         },
 326                         "ipv6": {
 327                             "unicast": {
 328                                 "redistribute": [
 329                                     {"redist_type": "static", "delete": True}
 330                                 ]
 331                             }
 332                         },
 333                     }
 334                 }
 335             }
 336         }
 337         result = create_router_bgp(tgen, topo, input_dict_2)
 338         assert result is True, "Testcase {} : Failed \n Error: {}".format(
 339             tc_name, result
 340         )
 341
 342         step("Configure redistribute static")
 343         input_dict_2 = {
 344             "r1": {
 345                 "bgp": {
 346                     "address_family": {
 347                         "ipv4": {
 348                             "unicast": {"redistribute": [{"redist_type": "static"}]}
 349                         },
 350                         "ipv6": {
 351                             "unicast": {"redistribute": [{"redist_type": "static"}]}
 352                         },
 353                     }
 354                 }
 355             }
 356         }
 357         result = create_router_bgp(tgen, topo, input_dict_2)
 358         assert result is True, "Testcase {} : Failed \n Error: {}".format(
 359             tc_name, result
 360         )
 361
 362         step(
 363             "Verify that these prefixes, originated on R1, are now"
 364             "received on both routers R2 and R3"
 365         )
 366         for addr_type in ADDR_TYPES:
 367             input_dict_4 = {
 368                 "r1": {
 369                     "static_routes": [
 370                         {
 371                             "network": NETWORK[addr_type],
 372                             "next_hop": topo["routers"]["r2"]["links"]["r1"][
 373                                 addr_type
 374                             ].split("/")[0],
 375                         }
 376                     ]
 377                 }
 378             }
 379             result = verify_bgp_rib(
 380                 tgen,
 381                 addr_type,
 382                 "r2",
 383                 input_dict_4,
 384                 next_hop=topo["routers"]["r1"]["links"]["r2"][addr_type].split("/")[0],
 385             )
 386             assert result is True, "Testcase {} : Failed \n Error: {}".format(
 387                 tc_name, result
 388             )
 389
 390             result = verify_bgp_rib(
 391                 tgen,
 392                 addr_type,
 393                 "r3",
 394                 input_dict_4,
 395                 next_hop=topo["routers"]["r1"]["links"]["r3"][addr_type].split("/")[0],
 396             )
 397             assert result is True, "Testcase {} : Failed \n Error: {}".format(
 398                 tc_name, result
 399             )
 400
 401     write_test_footer(tc_name)
 402
 403
 404 if __name__ == "__main__":
 405     args = ["-s"] + sys.argv[1:]
 406     sys.exit(pytest.main(args))