tests/topotests/evpn_type5_test_topo1/test_evpn_type5_chaos_topo1.py

   1 #!/usr/bin/env 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 EVPN-Type5 functionality:
  25 1. In absence of an overlay index all IP-Prefixes(RT-5)
  26     are advertised with default values for below parameters:
  27         --> Ethernet Tag ID = GW IP address = ESI=0
  28 2. EVPN CLI output and JSON format validation.
  29 3. RT verification(auto)
  30 """
  31
  32 import os
  33 import re
  34 import sys
  35 import json
  36 import time
  37 import pytest
  38 import platform
  39 from copy import deepcopy
  40 from time import sleep
  41
  42
  43 # Save the Current Working Directory to find configuration files.
  44 CWD = os.path.dirname(os.path.realpath(__file__))
  45 sys.path.append(os.path.join(CWD, "../"))
  46 sys.path.append(os.path.join(CWD, "../lib/"))
  47
  48 # Required to instantiate the topology builder class.
  49
  50 # pylint: disable=C0413
  51 # Import topogen and topotest helpers
  52 from lib.topotest import version_cmp
  53 from lib.topogen import Topogen, get_topogen
  54 from mininet.topo import Topo
  55
  56 from lib.common_config import (
  57     start_topology,
  58     write_test_header,
  59     check_address_types,
  60     write_test_footer,
  61     reset_config_on_routers,
  62     verify_rib,
  63     step,
  64     start_router_daemons,
  65     create_static_routes,
  66     create_vrf_cfg,
  67     create_route_maps,
  68     create_interface_in_kernel,
  69     check_router_status,
  70     configure_vxlan,
  71     configure_brctl,
  72     apply_raw_config,
  73     verify_vrf_vni,
  74     verify_cli_json
  75 )
  76
  77 from lib.topolog import logger
  78 from lib.bgp import (
  79     verify_bgp_convergence,
  80     create_router_bgp,
  81     clear_bgp,
  82     verify_best_path_as_per_bgp_attribute,
  83     verify_attributes_for_evpn_routes,
  84     verify_evpn_routes
  85 )
  86 from lib.topojson import build_topo_from_json, build_config_from_json
  87
  88 # Reading the data from JSON File for topology creation
  89 jsonFile = "{}/evpn_type5_chaos_topo1.json".format(CWD)
  90 try:
  91     with open(jsonFile, "r") as topoJson:
  92         topo = json.load(topoJson)
  93 except IOError:
  94     assert False, "Could not read file {}".format(jsonFile)
  95
  96 # Reading the data from JSON File for topology creation
  97 # Global variables
  98 TCPDUMP_FILE = "evpn_log.txt"
  99 LOGDIR = "/tmp/topotests/"
 100 NETWORK1_1 = {"ipv4": "10.1.1.1/32", "ipv6": "10::1/128"}
 101 NETWORK1_2 = {"ipv4": "40.1.1.1/32", "ipv6": "40::1/128"}
 102 NETWORK1_3 = {"ipv4": "40.1.1.2/32", "ipv6": "40::2/128"}
 103 NETWORK1_4 = {"ipv4": "40.1.1.3/32", "ipv6": "40::3/128"}
 104 NETWORK2_1 = {"ipv4": "20.1.1.1/32", "ipv6": "20::1/128"}
 105 NETWORK3_1 = {"ipv4": "30.1.1.1/32", "ipv6": "30::1/128"}
 106 NETWORK4_1 = {"ipv4": "100.1.1.1/32 ", "ipv6": "100::100/128"}
 107 NEXT_HOP_IP = {"ipv4": "Null0", "ipv6": "Null0"}
 108 VNI_1 = 75100
 109 VNI_2 = 75200
 110 VNI_3 = 75300
 111 MAC_1 = "00:80:48:ba:d1:00"
 112 MAC_2 = "00:80:48:ba:d1:01"
 113 MAC_3 = "00:80:48:ba:d1:02"
 114 BRCTL_1 = "br100"
 115 BRCTL_2 = "br200"
 116 BRCTL_3 = "br300"
 117 VXLAN_1 = "vxlan75100"
 118 VXLAN_2 = "vxlan75200"
 119 VXLAN_3 = "vxlan75300"
 120 BRIDGE_INTF1 = "120.0.0.1"
 121 BRIDGE_INTF2 = "120.0.0.2"
 122 BRIDGE_INTF3 = "120.0.0.3"
 123 MULTICAST_MAC1 = "01:00:5e:00:52:02"
 124
 125 VXLAN = {
 126     "vxlan_name": [VXLAN_1, VXLAN_2, VXLAN_3],
 127     "vxlan_id": [75100, 75200, 75300],
 128     "dstport": 4789,
 129     "local_addr": {"e1": BRIDGE_INTF1, "d1": BRIDGE_INTF2, "d2": BRIDGE_INTF3},
 130     "learning": "no",
 131 }
 132 BRCTL = {
 133     "brctl_name": [BRCTL_1, BRCTL_2, BRCTL_3],
 134     "addvxlan": [VXLAN_1, VXLAN_2, VXLAN_3],
 135     "vrf": ["RED", "BLUE", "GREEN"],
 136     "stp": [0, 0, 0],
 137 }
 138
 139
 140 class CreateTopo(Topo):
 141     """
 142     Test BasicTopo - topology 1
 143
 144     * `Topo`: Topology object
 145     """
 146
 147     def build(self, *_args, **_opts):
 148         """Build function"""
 149         tgen = get_topogen(self)
 150
 151         # Building topology from json file
 152         build_topo_from_json(tgen, topo)
 153
 154
 155 def setup_module(mod):
 156     """
 157     Sets up the pytest environment
 158
 159     * `mod`: module name
 160     """
 161
 162     global topo
 163     testsuite_run_time = time.asctime(time.localtime(time.time()))
 164     logger.info("Testsuite start time: {}".format(testsuite_run_time))
 165     logger.info("=" * 40)
 166
 167     logger.info("Running setup_module to create topology")
 168
 169     # This function initiates the topology build with Topogen...
 170     tgen = Topogen(CreateTopo, mod.__name__)
 171     # ... and here it calls Mininet initialization functions.
 172
 173     # Starting topology, create tmp files which are loaded to routers
 174     #  to start deamons and then start routers
 175     start_topology(tgen)
 176
 177     # Creating configuration from JSON
 178     build_config_from_json(tgen, topo)
 179
 180     if version_cmp(platform.release(), '4.19') < 0:
 181         error_msg = ('EVPN tests will not run (have kernel "{}", '
 182             'but it requires >= 4.19)'.format(platform.release()))
 183         pytest.skip(error_msg)
 184
 185     global BGP_CONVERGENCE
 186     global ADDR_TYPES
 187     ADDR_TYPES = check_address_types()
 188
 189     BGP_CONVERGENCE = verify_bgp_convergence(tgen, topo)
 190     assert BGP_CONVERGENCE is True, "setup_module :Failed \n Error: {}".format(
 191         BGP_CONVERGENCE
 192     )
 193
 194     logger.info("Pre-requisite config for testsuite")
 195     prerequisite_config_for_test_suite(tgen)
 196
 197     logger.info("Running setup_module() done")
 198
 199
 200 def teardown_module():
 201     """Teardown the pytest environment"""
 202
 203     logger.info("Running teardown_module to delete topology")
 204
 205     tgen = get_topogen()
 206
 207     # Stop toplogy and Remove tmp files
 208     tgen.stop_topology()
 209
 210     logger.info(
 211         "Testsuite end time: {}".format(time.asctime(time.localtime(time.time())))
 212     )
 213     logger.info("=" * 40)
 214
 215
 216 #####################################################
 217 #
 218 #   Testcases
 219 #
 220 #####################################################
 221
 222
 223 def prerequisite_config_for_test_suite(tgen):
 224     """
 225     API to do prerequisite config for testsuite
 226
 227     parameters:
 228     -----------
 229     * `tgen`: topogen object
 230     """
 231
 232     step("Configure vxlan, bridge interface")
 233     for dut in ["e1", "d1", "d2"]:
 234         step("[DUT: ]Configure vxlan")
 235         vxlan_input = {
 236             dut: {
 237                 "vxlan": [
 238                     {
 239                         "vxlan_name": VXLAN["vxlan_name"],
 240                         "vxlan_id": VXLAN["vxlan_id"],
 241                         "dstport": VXLAN["dstport"],
 242                         "local_addr": VXLAN["local_addr"][dut],
 243                         "learning": VXLAN["learning"],
 244                     }
 245                 ]
 246             }
 247         }
 248
 249         result = configure_vxlan(tgen, vxlan_input)
 250         assert result is True, "Testcase {} :Failed \n Error: {}".format(
 251             tc_name, result
 252         )
 253
 254         step("Configure bridge interface")
 255         brctl_input = {
 256             dut: {
 257                 "brctl": [
 258                     {
 259                         "brctl_name": BRCTL["brctl_name"],
 260                         "addvxlan": BRCTL["addvxlan"],
 261                         "vrf": BRCTL["vrf"],
 262                         "stp": BRCTL["stp"],
 263                     }
 264                 ]
 265             }
 266         }
 267         result = configure_brctl(tgen, topo, brctl_input)
 268         assert result is True, "Testcase {} :Failed \n Error: {}".format(
 269             tc_name, result
 270         )
 271
 272     step("Configure default routes")
 273     add_default_routes(tgen)
 274
 275
 276 def add_default_routes(tgen):
 277     """
 278     API to do prerequisite config for testsuite
 279
 280     parameters:
 281     -----------
 282     * `tgen`: topogen object
 283     """
 284
 285     step("Add default routes..")
 286
 287     default_routes = {
 288         "e1": {
 289             "static_routes": [
 290                 {
 291                     "network": "{}/32".format(VXLAN["local_addr"]["d1"]),
 292                     "next_hop": topo["routers"]["d1"]["links"]["e1-link1"][
 293                         "ipv4"
 294                     ].split("/")[0],
 295                 },
 296                 {
 297                     "network": "{}/32".format(VXLAN["local_addr"]["d2"]),
 298                     "next_hop": topo["routers"]["d2"]["links"]["e1-link1"][
 299                         "ipv4"
 300                     ].split("/")[0],
 301                 },
 302             ]
 303         },
 304         "d1": {
 305             "static_routes": [
 306                 {
 307                     "network": "{}/32".format(VXLAN["local_addr"]["e1"]),
 308                     "next_hop": topo["routers"]["e1"]["links"]["d1-link1"][
 309                         "ipv4"
 310                     ].split("/")[0],
 311                 },
 312                 {
 313                     "network": "{}/32".format(VXLAN["local_addr"]["d2"]),
 314                     "next_hop": topo["routers"]["e1"]["links"]["d1-link1"][
 315                         "ipv4"
 316                     ].split("/")[0],
 317                 },
 318             ]
 319         },
 320         "d2": {
 321             "static_routes": [
 322                 {
 323                     "network": "{}/32".format(VXLAN["local_addr"]["d1"]),
 324                     "next_hop": topo["routers"]["e1"]["links"]["d2-link1"][
 325                         "ipv4"
 326                     ].split("/")[0],
 327                 },
 328                 {
 329                     "network": "{}/32".format(VXLAN["local_addr"]["e1"]),
 330                     "next_hop": topo["routers"]["e1"]["links"]["d2-link1"][
 331                         "ipv4"
 332                     ].split("/")[0],
 333                 },
 334             ]
 335         },
 336     }
 337
 338     result = create_static_routes(tgen, default_routes)
 339     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 340
 341
 342 def test_verify_overlay_index_p1(request):
 343     """
 344     In absence of an overlay index all IP-Prefixes(RT-5)
 345     are advertised with default values for below parameters:
 346         --> Ethernet Tag ID = GW IP address = ESI=0
 347     """
 348
 349     tgen = get_topogen()
 350     tc_name = request.node.name
 351     write_test_header(tc_name)
 352     check_router_status(tgen)
 353     reset_config_on_routers(tgen)
 354     add_default_routes(tgen)
 355
 356     if tgen.routers_have_failure():
 357         pytest.skip(tgen.errors)
 358
 359     step("Following steps are taken care in base config:")
 360     step(
 361         "Configure BGP neighborship for both address families"
 362         "(IPv4 & IPv6) between Edge-1 and VFN routers(R1 and R2)"
 363     )
 364     step(
 365         "Advertise prefixes from VNF routers R1 and R2 in associated "
 366         "VRFs for both address-family."
 367     )
 368     step("Advertise VRF routes as in EVPN address family from Edge-1 " "router.")
 369
 370     for addr_type in ADDR_TYPES:
 371         input_dict_1 = {
 372             "r1": {
 373                 "static_routes": [
 374                     {
 375                         "network": NETWORK1_1[addr_type],
 376                         "next_hop": NEXT_HOP_IP[addr_type],
 377                         "vrf": "RED",
 378                     }
 379                 ]
 380             },
 381             "r2": {
 382                 "static_routes": [
 383                     {
 384                         "network": NETWORK2_1[addr_type],
 385                         "next_hop": NEXT_HOP_IP[addr_type],
 386                         "vrf": "BLUE",
 387                     },
 388                     {
 389                         "network": NETWORK3_1[addr_type],
 390                         "next_hop": NEXT_HOP_IP[addr_type],
 391                         "vrf": "GREEN",
 392                     }
 393                 ]
 394             }
 395         }
 396
 397         result = create_static_routes(tgen, input_dict_1)
 398         assert result is True, "Testcase {} : Failed \n Error: {}".format(
 399             tc_name, result
 400         )
 401
 402     step("Verify: Prefixes are received in all VRFs on Edge-1 router.")
 403
 404     for addr_type in ADDR_TYPES:
 405         input_routes = {key: topo["routers"][key] for key in ["r1"]}
 406         result = verify_rib(tgen, addr_type, "e1", input_routes)
 407         assert result is True, "Testcase {} :Failed \n Error: {}".format(
 408             tc_name, result
 409         )
 410
 411     for addr_type in ADDR_TYPES:
 412         input_routes = {key: topo["routers"][key] for key in ["r2"]}
 413         result = verify_rib(tgen, addr_type, "e1", input_routes)
 414         assert result is True, "Testcase {} :Failed \n Error: {}".format(
 415             tc_name, result
 416         )
 417
 418     step(
 419         "Verify that EVPN routes, received on DCG-1 and DCG-2 do not "
 420         "carry any overlay index and these indexes are set to default "
 421         "value=0. "
 422     )
 423
 424     for addr_type in ADDR_TYPES:
 425         input_routes = {key: topo["routers"][key] for key in ["r1"]}
 426
 427         result = verify_attributes_for_evpn_routes(
 428             tgen, topo, "d1", input_routes, ethTag=0
 429         )
 430         assert result is True, "Testcase {} :Failed \n Error: {}".format(
 431             tc_name, result
 432         )
 433
 434         result = verify_attributes_for_evpn_routes(
 435             tgen, topo, "d2", input_routes, ethTag=0
 436         )
 437         assert result is True, "Testcase {} :Failed \n Error: {}".format(
 438             tc_name, result
 439         )
 440
 441     write_test_footer(tc_name)
 442
 443
 444 def test_evpn_cli_json_available_p1(request):
 445     """
 446     EVPN CLI output and JSON format validation.
 447     """
 448
 449     tgen = get_topogen()
 450     tc_name = request.node.name
 451     write_test_header(tc_name)
 452     check_router_status(tgen)
 453     reset_config_on_routers(tgen)
 454     add_default_routes(tgen)
 455
 456     if tgen.routers_have_failure():
 457         pytest.skip(tgen.errors)
 458
 459     step("Need to verify below CLIs and associated JSON format " "outputs:")
 460
 461     input_dict = {
 462         "e1": {
 463             "cli": [
 464                 "show evpn vni detail",
 465                 "show bgp l2vpn evpn all overlay",
 466                 "show bgp l2vpn evpn vni"
 467             ]
 468         }
 469     }
 470
 471     result = verify_cli_json(tgen, input_dict)
 472     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
 473
 474     write_test_footer(tc_name)
 475
 476
 477 def test_RT_verification_auto_p0(request):
 478     """
 479     RT verification(auto)
 480     """
 481
 482     tgen = get_topogen()
 483     tc_name = request.node.name
 484     write_test_header(tc_name)
 485     check_router_status(tgen)
 486     reset_config_on_routers(tgen)
 487     add_default_routes(tgen)
 488
 489     if tgen.routers_have_failure():
 490         pytest.skip(tgen.errors)
 491
 492     step(
 493         "Advertise overlapping prefixes from VNFs R1 and R2 in all VRFs "
 494         "RED, GREEN and BLUE 100.1.1.1/32 and 100::100/128"
 495     )
 496
 497     for addr_type in ADDR_TYPES:
 498         input_dict_1 = {
 499             "r1": {
 500                 "static_routes": [
 501                     {
 502                         "network": NETWORK4_1[addr_type],
 503                         "next_hop": NEXT_HOP_IP[addr_type],
 504                         "vrf": "RED",
 505                     }
 506                 ]
 507             },
 508             "r2": {
 509                 "static_routes": [
 510                     {
 511                         "network": NETWORK4_1[addr_type],
 512                         "next_hop": NEXT_HOP_IP[addr_type],
 513                         "vrf": "BLUE",
 514                     },
 515                     {
 516                         "network": NETWORK4_1[addr_type],
 517                         "next_hop": NEXT_HOP_IP[addr_type],
 518                         "vrf": "GREEN",
 519                     }
 520                 ]
 521             }
 522         }
 523
 524         result = create_static_routes(tgen, input_dict_1)
 525         assert result is True, "Testcase {} : Failed \n Error: {}".format(
 526             tc_name, result
 527         )
 528
 529     step(
 530         "Verify that Edge-1 receives same prefixes in all 3 VRFs via "
 531         "corresponding next-hop in associated VRF sh bgp vrf all"
 532     )
 533
 534     for addr_type in ADDR_TYPES:
 535         input_routes = {
 536             "r1": {
 537                 "static_routes": [
 538                     {
 539                         "network": NETWORK4_1[addr_type],
 540                         "next_hop": NEXT_HOP_IP[addr_type],
 541                         "vrf": "RED",
 542                     }
 543                 ]
 544             },
 545             "r2": {
 546                 "static_routes": [
 547                     {
 548                         "network": NETWORK4_1[addr_type],
 549                         "next_hop": NEXT_HOP_IP[addr_type],
 550                         "vrf": "BLUE",
 551                     },
 552                     {
 553                         "network": NETWORK4_1[addr_type],
 554                         "next_hop": NEXT_HOP_IP[addr_type],
 555                         "vrf": "GREEN",
 556                     },
 557                 ]
 558             },
 559         }
 560
 561         result = verify_rib(tgen, addr_type, "e1", input_routes)
 562         assert result is True, "Testcase {} : Failed \n Error: {}".format(
 563             tc_name, result
 564         )
 565
 566     step(
 567         "Configure 4-byte local AS number on Edge-1 and establish EVPN "
 568         "neighborship with DCG-1 & DCG-2."
 569     )
 570
 571     topo_local = deepcopy(topo)
 572
 573     step("Delete BGP config for vrf RED.")
 574
 575     input_dict_vni = {
 576         "e1": {
 577             "vrfs": [
 578                 {"name": "RED", "no_vni": VNI_1},
 579                 {"name": "BLUE", "no_vni": VNI_2},
 580                 {"name": "GREEN", "no_vni": VNI_3},
 581             ]
 582         }
 583     }
 584     result = create_vrf_cfg(tgen, topo, input_dict=input_dict_vni)
 585     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 586
 587     input_dict_2 = {}
 588     for dut in ["e1"]:
 589         temp = {dut: {"bgp": []}}
 590         input_dict_2.update(temp)
 591
 592         INDEX = [0, 1, 2, 3]
 593         VRFS = ["RED", "BLUE", "GREEN", None]
 594         AS_NUM = [100, 100, 100, 100]
 595
 596         for index, vrf, as_num in zip(INDEX, VRFS, AS_NUM):
 597             topo_local["routers"][dut]["bgp"][index]["local_as"] = 4294967293
 598             if vrf:
 599                 temp[dut]["bgp"].append(
 600                     {"local_as": as_num, "vrf": vrf, "delete": True}
 601                 )
 602             else:
 603                 temp[dut]["bgp"].append({"local_as": as_num, "delete": True})
 604
 605     result = create_router_bgp(tgen, topo, input_dict_2)
 606     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 607
 608     result = create_router_bgp(tgen, topo_local["routers"])
 609     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 610
 611     input_dict_vni = {
 612         "e1": {
 613             "vrfs": [
 614                 {"name": "RED", "vni": VNI_1},
 615                 {"name": "BLUE", "vni": VNI_2},
 616                 {"name": "GREEN", "vni": VNI_3},
 617             ]
 618         }
 619     }
 620     result = create_vrf_cfg(tgen, topo, input_dict=input_dict_vni)
 621     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 622
 623     step(
 624         "Verify that all overlapping prefixes across different VRFs are "
 625         "advertised in EVPN with unique RD value(auto derived)."
 626     )
 627     step(
 628         "Verify that FRR uses only the lower 2 bytes of ASN+VNI for auto "
 629         "derived RT value."
 630     )
 631
 632     for addr_type in ADDR_TYPES:
 633         input_routes_1 = {
 634             "r1": {"static_routes": [{"network": NETWORK4_1[addr_type], "vrf": "RED"}]}
 635         }
 636         input_routes_2 = {
 637             "r2": {"static_routes": [{"network": NETWORK4_1[addr_type], "vrf": "BLUE"}]}
 638         }
 639         input_routes_3 = {
 640             "r2": {
 641                 "static_routes": [{"network": NETWORK4_1[addr_type], "vrf": "GREEN"}]
 642             }
 643         }
 644
 645         result = verify_attributes_for_evpn_routes(
 646             tgen, topo, "e1", input_routes_1, rd="auto", rd_peer="e1"
 647         )
 648         assert result is True, "Testcase {} :Failed \n Error: {}".format(
 649             tc_name, result
 650         )
 651
 652         result = verify_attributes_for_evpn_routes(
 653             tgen, topo, "e1", input_routes_1, rt="auto", rt_peer="e1"
 654         )
 655         assert result is True, "Testcase {} :Failed \n Error: {}".format(
 656             tc_name, result
 657         )
 658
 659         result = verify_attributes_for_evpn_routes(
 660             tgen, topo, "e1", input_routes_2, rd="auto", rd_peer="e1"
 661         )
 662         assert result is True, "Testcase {} :Failed \n Error: {}".format(
 663             tc_name, result
 664         )
 665
 666         result = verify_attributes_for_evpn_routes(
 667             tgen, topo, "e1", input_routes_2, rt="auto", rt_peer="e1"
 668         )
 669         assert result is True, "Testcase {} :Failed \n Error: {}".format(
 670             tc_name, result
 671         )
 672
 673         result = verify_attributes_for_evpn_routes(
 674             tgen, topo, "e1", input_routes_3, rd="auto", rd_peer="e1"
 675         )
 676         assert result is True, "Testcase {} :Failed \n Error: {}".format(
 677             tc_name, result
 678         )
 679
 680         result = verify_attributes_for_evpn_routes(
 681             tgen, topo, "e1", input_routes_3, rt="auto", rt_peer="e1"
 682         )
 683         assert result is True, "Testcase {} :Failed \n Error: {}".format(
 684             tc_name, result
 685         )
 686
 687     step(
 688         "Verify that DCG-1(iBGP peer) automatically imports the prefixes"
 689         " from EVPN address-family to respective VRFs."
 690     )
 691     step(
 692         "Verify if DCG-2(eBGP peer) automatically imports the prefixes "
 693         "from EVPN address-family to respective VRFs or not."
 694     )
 695
 696     for addr_type in ADDR_TYPES:
 697         input_routes = {
 698             "r1": {
 699                 "static_routes": [
 700                     {
 701                         "network": NETWORK4_1[addr_type],
 702                         "next_hop": NEXT_HOP_IP[addr_type],
 703                         "vrf": "RED",
 704                     }
 705                 ]
 706             },
 707             "r2": {
 708                 "static_routes": [
 709                     {
 710                         "network": NETWORK4_1[addr_type],
 711                         "next_hop": NEXT_HOP_IP[addr_type],
 712                         "vrf": "BLUE",
 713                     },
 714                     {
 715                         "network": NETWORK4_1[addr_type],
 716                         "next_hop": NEXT_HOP_IP[addr_type],
 717                         "vrf": "GREEN",
 718                     },
 719                 ]
 720             },
 721         }
 722
 723         result = verify_rib(tgen, addr_type, "d1", input_routes)
 724         assert result is True, "Testcase {} : Failed \n Error: {}".format(
 725             tc_name, result
 726         )
 727
 728         result = verify_rib(tgen, addr_type, "d2", input_routes)
 729         assert result is True, "Testcase {} : Failed \n Error: {}".format(
 730             tc_name, result
 731         )
 732
 733     step(
 734         "Change the VNI number for all 3 VRFs on Edge-1 as:"
 735         "RED : 75400, GREEN: 75500, BLUE: 75600"
 736     )
 737
 738     input_dict_vni = {
 739         "e1": {
 740             "vrfs": [
 741                 {"name": "RED", "no_vni": VNI_1},
 742                 {"name": "BLUE", "no_vni": VNI_2},
 743                 {"name": "GREEN", "no_vni": VNI_3},
 744             ]
 745         }
 746     }
 747     result = create_vrf_cfg(tgen, topo, input_dict=input_dict_vni)
 748     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 749
 750     input_dict_vni = {
 751         "e1": {
 752             "vrfs": [
 753                 {"name": "RED", "vni": 75400},
 754                 {"name": "BLUE", "vni": 75500},
 755                 {"name": "GREEN", "vni": 75600},
 756             ]
 757         }
 758     }
 759     result = create_vrf_cfg(tgen, topo, input_dict=input_dict_vni)
 760     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 761
 762     step("Delete configured vxlan")
 763     dut = "e1"
 764     vxlan_input = {
 765         dut: {
 766             "vxlan": [
 767                 {
 768                     "vxlan_name": VXLAN["vxlan_name"],
 769                     "vxlan_id": VXLAN["vxlan_id"],
 770                     "dstport": VXLAN["dstport"],
 771                     "local_addr": VXLAN["local_addr"][dut],
 772                     "learning": VXLAN["learning"],
 773                     "delete": True,
 774                 }
 775             ]
 776         }
 777     }
 778
 779     result = configure_vxlan(tgen, vxlan_input)
 780     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 781
 782     step("Configured vxlan")
 783     VXLAN["vxlan_id"] = [75400, 75500, 75600]
 784     vxlan_input = {
 785         dut: {
 786             "vxlan": [
 787                 {
 788                     "vxlan_name": VXLAN["vxlan_name"],
 789                     "vxlan_id": VXLAN["vxlan_id"],
 790                     "dstport": VXLAN["dstport"],
 791                     "local_addr": VXLAN["local_addr"][dut],
 792                     "learning": VXLAN["learning"],
 793                 }
 794             ]
 795         }
 796     }
 797
 798     result = configure_vxlan(tgen, vxlan_input)
 799     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 800
 801     step("Configure bridge interface")
 802     brctl_input = {
 803         dut: {
 804             "brctl": [
 805                 {
 806                     "brctl_name": BRCTL["brctl_name"],
 807                     "addvxlan": BRCTL["addvxlan"],
 808                     "vrf": BRCTL["vrf"],
 809                     "stp": BRCTL["stp"],
 810                 }
 811             ]
 812         }
 813     }
 814     result = configure_brctl(tgen, topo, brctl_input)
 815     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 816
 817     step(
 818         "Verify on Edge-1 that auto derived RT value has changed for "
 819         "each VRF based on VNI number.."
 820     )
 821
 822     input_dict = {
 823         "e1": {
 824             "vrfs": [
 825                 {"RED": {"vni": 75400}},
 826                 {"BLUE": {"vni": 75500}},
 827                 {"GREEN": {"vni": 75600}},
 828             ]
 829         }
 830     }
 831
 832     result = verify_vrf_vni(tgen, input_dict)
 833     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 834
 835     step(
 836         "Verify on Edge-1 that auto derived RT value has changed for "
 837         "each VRF based on VNI number."
 838     )
 839
 840     for addr_type in ADDR_TYPES:
 841         input_routes = {
 842             "r1": {"static_routes": [{"network": NETWORK4_1[addr_type], "vrf": "RED"}]}
 843         }
 844
 845         result = verify_attributes_for_evpn_routes(
 846             tgen, topo, "e1", input_routes, rt="auto", rt_peer="e1"
 847         )
 848         assert result is True, "Testcase {} :Failed \n Error: {}".format(
 849             tc_name, result
 850         )
 851
 852     step(
 853         "Verify on DCG-2 that prefixes are not imported from EVPN "
 854         "address-family to VRFs as RT values are different on sending("
 855         "edge-1) and receiving(DCG-2) end."
 856     )
 857
 858     for addr_type in ADDR_TYPES:
 859         input_routes = {
 860             "r1": {"static_routes": [{"network": NETWORK4_1[addr_type], "vrf": "RED"}]}
 861         }
 862
 863         result = verify_rib(tgen, addr_type, "d2", input_routes, expected=False)
 864         assert result is not True, "Testcase {} :Failed \n "
 865         "Routes are still present: {}".format(tc_name, result)
 866         logger.info("Expected Behavior: {}".format(result))
 867
 868     step(
 869         "Revert back to original VNI number for all 3 VRFs on Edge-1 "
 870         "as: RED : 75100, GREEN: 75200, BLUE: 75300"
 871     )
 872
 873     input_dict_vni = {
 874         "e1": {
 875             "vrfs": [
 876                 {"name": "RED", "no_vni": 75400},
 877                 {"name": "BLUE", "no_vni": 75500},
 878                 {"name": "GREEN", "no_vni": 75600},
 879             ]
 880         }
 881     }
 882     result = create_vrf_cfg(tgen, topo, input_dict=input_dict_vni)
 883     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 884
 885     input_dict_vni = {
 886         "e1": {
 887             "vrfs": [
 888                 {"name": "RED", "vni": VNI_1},
 889                 {"name": "BLUE", "vni": VNI_2},
 890                 {"name": "GREEN", "vni": VNI_3},
 891             ]
 892         }
 893     }
 894     result = create_vrf_cfg(tgen, topo, input_dict=input_dict_vni)
 895     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 896
 897     step("Delete configured vxlan")
 898     dut = "e1"
 899     vxlan_input = {
 900         dut: {
 901             "vxlan": [
 902                 {
 903                     "vxlan_name": VXLAN["vxlan_name"],
 904                     "vxlan_id": VXLAN["vxlan_id"],
 905                     "dstport": VXLAN["dstport"],
 906                     "local_addr": VXLAN["local_addr"][dut],
 907                     "learning": VXLAN["learning"],
 908                     "delete": True,
 909                 }
 910             ]
 911         }
 912     }
 913     result = configure_vxlan(tgen, vxlan_input)
 914     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 915
 916     step("Configured vxlan")
 917     VXLAN["vxlan_id"] = [75100, 75200, 75300]
 918     vxlan_input = {
 919         dut: {
 920             "vxlan": [
 921                 {
 922                     "vxlan_name": VXLAN["vxlan_name"],
 923                     "vxlan_id": VXLAN["vxlan_id"],
 924                     "dstport": VXLAN["dstport"],
 925                     "local_addr": VXLAN["local_addr"][dut],
 926                     "learning": VXLAN["learning"],
 927                 }
 928             ]
 929         }
 930     }
 931     result = configure_vxlan(tgen, vxlan_input)
 932     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 933
 934     step("Configure bridge interface")
 935     brctl_input = {
 936         dut: {
 937             "brctl": [
 938                 {
 939                     "brctl_name": BRCTL["brctl_name"],
 940                     "addvxlan": BRCTL["addvxlan"],
 941                     "vrf": BRCTL["vrf"],
 942                     "stp": BRCTL["stp"],
 943                 }
 944             ]
 945         }
 946     }
 947     result = configure_brctl(tgen, topo, brctl_input)
 948     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 949
 950     step(
 951         "Verify on Edge-1 that auto derived RT value has changed for "
 952         "each VRF based on VNI number."
 953     )
 954     step(
 955         "Verify that DCG-1(iBGP peer) automatically imports the prefixes"
 956         " from EVPN address-family to respective VRFs."
 957     )
 958
 959     for addr_type in ADDR_TYPES:
 960         input_routes = {
 961             "r1": {"static_routes": [{"network": NETWORK4_1[addr_type], "vrf": "RED"}]}
 962         }
 963
 964         result = verify_attributes_for_evpn_routes(
 965             tgen, topo, "e1", input_routes, rt="auto", rt_peer="e1"
 966         )
 967         assert result is True, "Testcase {} :Failed \n Error: {}".format(
 968             tc_name, result
 969         )
 970
 971         result = verify_rib(tgen, addr_type, "d1", input_routes)
 972         assert result is True, "Testcase {} :Failed \n Error: {}".format(
 973             tc_name, result
 974         )
 975
 976     step("Test with smaller VNI numbers (1-75000)")
 977
 978     input_dict_vni = {"e1": {"vrfs": [{"name": "RED", "no_vni": VNI_1}]}}
 979     result = create_vrf_cfg(tgen, topo, input_dict=input_dict_vni)
 980     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 981
 982     input_dict_vni = {"e1": {"vrfs": [{"name": "RED", "vni": 111}]}}
 983     result = create_vrf_cfg(tgen, topo, input_dict=input_dict_vni)
 984     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 985
 986     step(
 987         "Verify that DCG-2 receives EVPN prefixes along with auto "
 988         "derived RT values(based on smaller VNI numbers)"
 989     )
 990
 991     for addr_type in ADDR_TYPES:
 992         input_routes_1 = {
 993             "r1": {"static_routes": [{"network": NETWORK4_1[addr_type], "vrf": "RED"}]}
 994         }
 995
 996         result = verify_attributes_for_evpn_routes(
 997             tgen, topo, "d2", input_routes_1, rt="auto", rt_peer="e1", expected=False
 998         )
 999         assert result is not True, "Testcase {} :Failed \n "
1000         "Malfaromed Auto-RT value accepted: {}".format(tc_name, result)
1001         logger.info("Expected Behavior: {}".format(result))
1002
1003     step("Configure VNI number more than boundary limit (16777215)")
1004
1005     input_dict_vni = {"e1": {"vrfs": [{"name": "RED", "no_vni": 111}]}}
1006     result = create_vrf_cfg(tgen, topo, input_dict=input_dict_vni)
1007     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
1008
1009     input_dict_vni = {"e1": {"vrfs": [{"name": "RED", "vni": 16777215}]}}
1010     result = create_vrf_cfg(tgen, topo, input_dict=input_dict_vni)
1011     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
1012
1013     step("CLI error for malformed VNI.")
1014     input_dict = {
1015         "e1": {
1016             "vrfs": [{"RED": {"vni": 16777215, "routerMac": "None", "state": "Down"}}]
1017         }
1018     }
1019
1020     result = verify_vrf_vni(tgen, input_dict)
1021     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
1022
1023     for addr_type in ADDR_TYPES:
1024         input_routes_1 = {
1025             "r1": {"static_routes": [{"network": NETWORK4_1[addr_type], "vrf": "RED"}]}
1026         }
1027
1028         result = verify_attributes_for_evpn_routes(
1029             tgen, topo, "d2", input_routes_1, rt="auto", rt_peer="e1", expected=False
1030         )
1031         assert result is not True, "Testcase {} :Failed \n "
1032         "Malfaromed Auto-RT value accepted: {}".format(tc_name, result)
1033         logger.info("Expected Behavior: {}".format(result))
1034
1035     step("Un-configure VNI number more than boundary limit (16777215)")
1036
1037     input_dict_vni = {"e1": {"vrfs": [{"name": "RED", "no_vni": 16777215}]}}
1038     result = create_vrf_cfg(tgen, topo, input_dict=input_dict_vni)
1039     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
1040
1041     write_test_footer(tc_name)
1042
1043
1044 if __name__ == "__main__":
1045     args = ["-s"] + sys.argv[1:]
1046     sys.exit(pytest.main(args))