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