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