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