tests/topotests/evpn_type5_test_topo1/test_evpn_type5_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. RD verification (manual/auto).
  27 2. RT verification(manual)
  28 3. In an active/standby EVPN implementation, if active DCG goes down,
  29         secondary takes over.
  30 4. EVPN routes are advertised/withdrawn, based on VNFs
  31         advertising/withdrawing IP prefixes.
  32 5. Route-map operations for EVPN address family.
  33 6. BGP attributes for EVPN address-family.
  34 """
  35
  36 import os
  37 import sys
  38 import json
  39 import time
  40 import pytest
  41 import platform
  42 from copy import deepcopy
  43
  44
  45 # Save the Current Working Directory to find configuration files.
  46 CWD = os.path.dirname(os.path.realpath(__file__))
  47 sys.path.append(os.path.join(CWD, "../"))
  48 sys.path.append(os.path.join(CWD, "../lib/"))
  49
  50 # Required to instantiate the topology builder class.
  51
  52 # pylint: disable=C0413
  53 # Import topogen and topotest helpers
  54 from lib.topotest import version_cmp
  55 from lib.topogen import Topogen, get_topogen
  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     create_route_maps,
  66     create_static_routes,
  67     create_vrf_cfg,
  68     check_router_status,
  69     apply_raw_config,
  70     configure_vxlan,
  71     configure_brctl,
  72     create_interface_in_kernel,
  73     kill_router_daemons,
  74     start_router_daemons,
  75 )
  76
  77 from lib.topolog import logger
  78 from lib.bgp import (
  79     verify_bgp_convergence,
  80     create_router_bgp,
  81     verify_best_path_as_per_bgp_attribute,
  82     verify_attributes_for_evpn_routes,
  83     verify_evpn_routes,
  84 )
  85 from lib.topojson import build_topo_from_json, build_config_from_json
  86
  87 pytestmark = [pytest.mark.bgpd, pytest.mark.staticd]
  88
  89 # Global variables
  90 NETWORK1_1 = {"ipv4": "10.1.1.1/32", "ipv6": "10::1/128"}
  91 NETWORK1_2 = {"ipv4": "40.1.1.1/32", "ipv6": "40::1/128"}
  92 NETWORK1_3 = {"ipv4": "40.1.1.2/32", "ipv6": "40::2/128"}
  93 NETWORK1_4 = {"ipv4": "40.1.1.3/32", "ipv6": "40::3/128"}
  94 NETWORK2_1 = {"ipv4": "20.1.1.1/32", "ipv6": "20::1/128"}
  95 NETWORK3_1 = {"ipv4": "30.1.1.1/32", "ipv6": "30::1/128"}
  96 NETWORK4_1 = {"ipv4": "100.1.1.1/32 ", "ipv6": "100::100/128"}
  97 NEXT_HOP_IP = {"ipv4": "Null0", "ipv6": "Null0"}
  98 VNI_1 = 75100
  99 VNI_2 = 75200
 100 VNI_3 = 75300
 101 MAC_1 = "00:80:48:ba:d1:00"
 102 MAC_2 = "00:80:48:ba:d1:01"
 103 MAC_3 = "00:80:48:ba:d1:02"
 104 BRCTL_1 = "br100"
 105 BRCTL_2 = "br200"
 106 BRCTL_3 = "br300"
 107 VXLAN_1 = "vxlan75100"
 108 VXLAN_2 = "vxlan75200"
 109 VXLAN_3 = "vxlan75300"
 110 BRIDGE_INTF1 = "120.0.0.1"
 111 BRIDGE_INTF2 = "120.0.0.2"
 112 BRIDGE_INTF3 = "120.0.0.3"
 113
 114 VXLAN = {
 115     "vxlan_name": [VXLAN_1, VXLAN_2, VXLAN_3],
 116     "vxlan_id": [75100, 75200, 75300],
 117     "dstport": 4789,
 118     "local_addr": {"e1": BRIDGE_INTF1, "d1": BRIDGE_INTF2, "d2": BRIDGE_INTF3},
 119     "learning": "no",
 120 }
 121 BRCTL = {
 122     "brctl_name": [BRCTL_1, BRCTL_2, BRCTL_3],
 123     "addvxlan": [VXLAN_1, VXLAN_2, VXLAN_3],
 124     "vrf": ["RED", "BLUE", "GREEN"],
 125     "stp": [0, 0, 0],
 126 }
 127
 128
 129 def setup_module(mod):
 130     """
 131     Sets up the pytest environment
 132
 133     * `mod`: module name
 134     """
 135
 136     global topo
 137     testsuite_run_time = time.asctime(time.localtime(time.time()))
 138     logger.info("Testsuite start time: {}".format(testsuite_run_time))
 139     logger.info("=" * 40)
 140
 141     logger.info("Running setup_module to create topology")
 142
 143     # This function initiates the topology build with Topogen...
 144     json_file = "{}/evpn_type5_topo1.json".format(CWD)
 145     tgen = Topogen(json_file, mod.__name__)
 146     topo = tgen.json_topo
 147
 148     # ... and here it calls Mininet initialization functions.
 149
 150     # Starting topology, create tmp files which are loaded to routers
 151     #  to start daemons and then start routers
 152     start_topology(tgen)
 153
 154     # Creating configuration from JSON
 155     build_config_from_json(tgen, topo)
 156
 157     if version_cmp(platform.release(), "4.19") < 0:
 158         error_msg = (
 159             'EVPN tests will not run (have kernel "{}", '
 160             "but it requires >= 4.19)".format(platform.release())
 161         )
 162         pytest.skip(error_msg)
 163
 164     global BGP_CONVERGENCE
 165     global ADDR_TYPES
 166     ADDR_TYPES = check_address_types()
 167
 168     BGP_CONVERGENCE = verify_bgp_convergence(tgen, topo)
 169     assert BGP_CONVERGENCE is True, "setup_module :Failed \n Error: {}".format(
 170         BGP_CONVERGENCE
 171     )
 172
 173     logger.info("Pre-requisite config for testsuite")
 174     prerequisite_config_for_test_suite(tgen)
 175
 176     logger.info("Running setup_module() done")
 177
 178
 179 def teardown_module():
 180     """Teardown the pytest environment"""
 181
 182     logger.info("Running teardown_module to delete topology")
 183
 184     tgen = get_topogen()
 185
 186     # Stop toplogy and Remove tmp files
 187     tgen.stop_topology()
 188
 189     logger.info(
 190         "Testsuite end time: {}".format(time.asctime(time.localtime(time.time())))
 191     )
 192     logger.info("=" * 40)
 193
 194
 195 #####################################################
 196 #
 197 #   Testcases
 198 #
 199 #####################################################
 200
 201
 202 def prerequisite_config_for_test_suite(tgen):
 203     """
 204     API to do prerequisite config for testsuite
 205
 206     parameters:
 207     -----------
 208     * `tgen`: topogen object
 209     """
 210
 211     step("Configure vxlan, bridge interface")
 212     for dut in ["e1", "d1", "d2"]:
 213         step("[DUT: ]Configure vxlan")
 214         vxlan_input = {
 215             dut: {
 216                 "vxlan": [
 217                     {
 218                         "vxlan_name": VXLAN["vxlan_name"],
 219                         "vxlan_id": VXLAN["vxlan_id"],
 220                         "dstport": VXLAN["dstport"],
 221                         "local_addr": VXLAN["local_addr"][dut],
 222                         "learning": VXLAN["learning"],
 223                     }
 224                 ]
 225             }
 226         }
 227
 228         result = configure_vxlan(tgen, vxlan_input)
 229         assert result is True, "Testcase :Failed \n Error: {}".format(result)
 230
 231         step("Configure bridge interface")
 232         brctl_input = {
 233             dut: {
 234                 "brctl": [
 235                     {
 236                         "brctl_name": BRCTL["brctl_name"],
 237                         "addvxlan": BRCTL["addvxlan"],
 238                         "vrf": BRCTL["vrf"],
 239                         "stp": BRCTL["stp"],
 240                     }
 241                 ]
 242             }
 243         }
 244         result = configure_brctl(tgen, topo, brctl_input)
 245         assert result is True, "Testcase :Failed \n Error: {}".format(result)
 246
 247     step("Configure default routes")
 248     add_default_routes(tgen)
 249
 250
 251 def add_default_routes(tgen):
 252     """
 253     API to do prerequisite config for testsuite
 254
 255     parameters:
 256     -----------
 257     * `tgen`: topogen object
 258     """
 259
 260     step("Add default routes..")
 261
 262     default_routes = {
 263         "e1": {
 264             "static_routes": [
 265                 {
 266                     "network": "{}/32".format(VXLAN["local_addr"]["d1"]),
 267                     "next_hop": topo["routers"]["d1"]["links"]["e1-link1"][
 268                         "ipv4"
 269                     ].split("/")[0],
 270                 },
 271                 {
 272                     "network": "{}/32".format(VXLAN["local_addr"]["d2"]),
 273                     "next_hop": topo["routers"]["d2"]["links"]["e1-link1"][
 274                         "ipv4"
 275                     ].split("/")[0],
 276                 },
 277             ]
 278         },
 279         "d1": {
 280             "static_routes": [
 281                 {
 282                     "network": "{}/32".format(VXLAN["local_addr"]["e1"]),
 283                     "next_hop": topo["routers"]["e1"]["links"]["d1-link1"][
 284                         "ipv4"
 285                     ].split("/")[0],
 286                 },
 287                 {
 288                     "network": "{}/32".format(VXLAN["local_addr"]["d2"]),
 289                     "next_hop": topo["routers"]["e1"]["links"]["d1-link1"][
 290                         "ipv4"
 291                     ].split("/")[0],
 292                 },
 293             ]
 294         },
 295         "d2": {
 296             "static_routes": [
 297                 {
 298                     "network": "{}/32".format(VXLAN["local_addr"]["d1"]),
 299                     "next_hop": topo["routers"]["e1"]["links"]["d2-link1"][
 300                         "ipv4"
 301                     ].split("/")[0],
 302                 },
 303                 {
 304                     "network": "{}/32".format(VXLAN["local_addr"]["e1"]),
 305                     "next_hop": topo["routers"]["e1"]["links"]["d2-link1"][
 306                         "ipv4"
 307                     ].split("/")[0],
 308                 },
 309             ]
 310         },
 311     }
 312
 313     result = create_static_routes(tgen, default_routes)
 314     assert result is True, "Testcase :Failed \n Error: {}".format(result)
 315
 316
 317 def test_RD_verification_manual_and_auto_p0(request):
 318     """
 319     RD verification (manual/auto).
 320     """
 321
 322     tgen = get_topogen()
 323     tc_name = request.node.name
 324     write_test_header(tc_name)
 325     check_router_status(tgen)
 326     reset_config_on_routers(tgen)
 327     add_default_routes(tgen)
 328
 329     if tgen.routers_have_failure():
 330         pytest.skip(tgen.errors)
 331
 332     step(
 333         "Advertise prefixes from VNF routers R1 and R2 in associated "
 334         "VRFs for both address-family."
 335     )
 336     step(
 337         "Advertise vrf RED's routes in EVPN address family from Edge-1 router"
 338         ", without manual configuration of RD."
 339     )
 340
 341     for addr_type in ADDR_TYPES:
 342         input_dict_1 = {
 343             "r1": {
 344                 "static_routes": [
 345                     {
 346                         "network": NETWORK1_1[addr_type],
 347                         "next_hop": NEXT_HOP_IP[addr_type],
 348                         "vrf": "RED",
 349                     }
 350                 ]
 351             },
 352             "r2": {
 353                 "static_routes": [
 354                     {
 355                         "network": NETWORK2_1[addr_type],
 356                         "next_hop": NEXT_HOP_IP[addr_type],
 357                         "vrf": "BLUE",
 358                     },
 359                     {
 360                         "network": NETWORK3_1[addr_type],
 361                         "next_hop": NEXT_HOP_IP[addr_type],
 362                         "vrf": "GREEN",
 363                     },
 364                 ]
 365             },
 366         }
 367
 368         result = create_static_routes(tgen, input_dict_1)
 369         assert result is True, "Testcase {} : Failed \n Error: {}".format(
 370             tc_name, result
 371         )
 372
 373     step("Verify on DCG-1 and DCG-2:")
 374     step("EVPN route for 10.1.1.1/32 has auto-assigned RD value.")
 375
 376     for dut in ["d1", "d2"]:
 377         input_routes = {key: topo["routers"][key] for key in ["r1"]}
 378         result = verify_attributes_for_evpn_routes(
 379             tgen, topo, dut, input_routes, rd="auto", rd_peer="e1"
 380         )
 381         assert result is True, "Testcase {} on {} :Failed \n Error: {}".format(
 382             tc_name, dut, result
 383         )
 384
 385     step(
 386         "Configure RD for vrf RED manually as 50.50.50.50:50 and "
 387         "advertise vrf RED's routes in EVPN address family from "
 388         "Edge-1 router."
 389     )
 390
 391     input_dict_rd = {
 392         "e1": {
 393             "bgp": [
 394                 {
 395                     "local_as": "100",
 396                     "vrf": "RED",
 397                     "address_family": {"l2vpn": {"evpn": {"rd": "50.50.50.50:50"}}},
 398                 }
 399             ]
 400         }
 401     }
 402
 403     result = create_router_bgp(tgen, topo, input_dict_rd)
 404     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 405
 406     step("EVPN route for vrf RED has RD value as 50.50.50.50:50")
 407     for dut in ["d1", "d2"]:
 408         input_routes = {key: topo["routers"][key] for key in ["r1"]}
 409         result = verify_attributes_for_evpn_routes(
 410             tgen, topo, dut, input_routes, rd="50.50.50.50:50"
 411         )
 412         assert result is True, "Testcase {} on {} :Failed \n Error: {}".format(
 413             tc_name, dut, result
 414         )
 415
 416     step(
 417         "Configure RD for vrf RED manually as 100.100.100.100:100 and "
 418         "advertise vrf RED's routes in EVPN address family from Edge-1 "
 419         "router."
 420     )
 421     input_dict_rd = {
 422         "e1": {
 423             "bgp": [
 424                 {
 425                     "local_as": "100",
 426                     "vrf": "RED",
 427                     "address_family": {
 428                         "l2vpn": {"evpn": {"rd": "100.100.100.100:100"}}
 429                     },
 430                 }
 431             ]
 432         }
 433     }
 434
 435     result = create_router_bgp(tgen, topo, input_dict_rd)
 436     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 437
 438     step(
 439         "EVPN route for vrf RED is overridden with RD value as " "100.100.100.100:100."
 440     )
 441
 442     for dut in ["d1", "d2"]:
 443         input_routes = {key: topo["routers"][key] for key in ["r1"]}
 444         result = verify_attributes_for_evpn_routes(
 445             tgen, topo, dut, input_routes, rd="100.100.100.100:100"
 446         )
 447         assert result is True, "Testcase {} on {} :Failed \n Error: {}".format(
 448             tc_name, dut, result
 449         )
 450
 451     step(
 452         "Configure RD for vrf BLUE manually same as vrf RED "
 453         "(100.100.100.100:100) and advertise vrf RED and BLUE's routes "
 454         "in EVPN address family from Edge-1 router."
 455     )
 456
 457     input_dict_rd = {
 458         "e1": {
 459             "bgp": [
 460                 {
 461                     "local_as": "100",
 462                     "vrf": "BLUE",
 463                     "address_family": {
 464                         "l2vpn": {"evpn": {"rd": "100.100.100.100:100"}}
 465                     },
 466                 }
 467             ]
 468         }
 469     }
 470
 471     result = create_router_bgp(tgen, topo, input_dict_rd)
 472     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 473
 474     step(
 475         "Delete manually configured RD and advertise vrf RED's routes "
 476         "in EVPN address family from Edge-1 router."
 477     )
 478
 479     input_dict_rd = {
 480         "e1": {
 481             "bgp": [
 482                 {
 483                     "local_as": "100",
 484                     "vrf": "RED",
 485                     "address_family": {
 486                         "l2vpn": {"evpn": {"no rd": "100.100.100.100:100"}}
 487                     },
 488                 }
 489             ]
 490         }
 491     }
 492
 493     result = create_router_bgp(tgen, topo, input_dict_rd)
 494     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 495
 496     step(
 497         "Configure same RD value for vrf GREEN, as auto generated RD "
 498         "value for vrf RED on Edge-1 router."
 499     )
 500
 501     input_dict_rd = {
 502         "e1": {
 503             "bgp": [
 504                 {
 505                     "local_as": "100",
 506                     "vrf": "GREEN",
 507                     "address_family": {"l2vpn": {"evpn": {"rd": "10.0.0.33:1"}}},
 508                 }
 509             ]
 510         }
 511     }
 512
 513     result = create_router_bgp(tgen, topo, input_dict_rd)
 514     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 515
 516     step("Delete auto configured RD value from vrf RED in EVPN " "address family.")
 517
 518     input_dict_rd = {
 519         "e1": {
 520             "bgp": [
 521                 {
 522                     "local_as": "100",
 523                     "vrf": "GREEN",
 524                     "address_family": {"l2vpn": {"evpn": {"no rd": "10.0.0.33:1"}}},
 525                 }
 526             ]
 527         }
 528     }
 529
 530     result = create_router_bgp(tgen, topo, input_dict_rd)
 531     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 532
 533     step("Configure RD value as 100.100.100:100")
 534
 535     input_dict_rd = {
 536         "e1": {
 537             "bgp": [
 538                 {
 539                     "local_as": "100",
 540                     "vrf": "GREEN",
 541                     "address_family": {"l2vpn": {"evpn": {"rd": "100.100.100:100"}}},
 542                 }
 543             ]
 544         }
 545     }
 546
 547     result = create_router_bgp(tgen, topo, input_dict_rd)
 548     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 549
 550     write_test_footer(tc_name)
 551
 552
 553 def test_RT_verification_manual_p0(request):
 554     """
 555     RT verification(manual)
 556     """
 557
 558     tgen = get_topogen()
 559     tc_name = request.node.name
 560     write_test_header(tc_name)
 561     check_router_status(tgen)
 562     reset_config_on_routers(tgen)
 563     add_default_routes(tgen)
 564
 565     if tgen.routers_have_failure():
 566         pytest.skip(tgen.errors)
 567
 568     step(
 569         "Advertise prefixes from VNF routers R1 and R2 in associated "
 570         "VRFs for both address-family."
 571     )
 572     step("Advertise VRF routes as in EVPN address family from Edge-1 " "router.")
 573
 574     for addr_type in ADDR_TYPES:
 575         input_dict_1 = {
 576             "r1": {
 577                 "static_routes": [
 578                     {
 579                         "network": NETWORK1_1[addr_type],
 580                         "next_hop": NEXT_HOP_IP[addr_type],
 581                         "vrf": "RED",
 582                     }
 583                 ]
 584             },
 585             "r2": {
 586                 "static_routes": [
 587                     {
 588                         "network": NETWORK2_1[addr_type],
 589                         "next_hop": NEXT_HOP_IP[addr_type],
 590                         "vrf": "BLUE",
 591                     },
 592                     {
 593                         "network": NETWORK3_1[addr_type],
 594                         "next_hop": NEXT_HOP_IP[addr_type],
 595                         "vrf": "GREEN",
 596                     },
 597                 ]
 598             },
 599         }
 600
 601         result = create_static_routes(tgen, input_dict_1)
 602         assert result is True, "Testcase {} : Failed \n Error: {}".format(
 603             tc_name, result
 604         )
 605
 606     step(
 607         "Configure RT for vrf RED manually as export 100:100 "
 608         "and advertise vrf RED's routes in EVPN address family"
 609         " from Edge-1 router."
 610     )
 611
 612     input_dict_rt = {
 613         "e1": {
 614             "bgp": [
 615                 {
 616                     "local_as": "100",
 617                     "vrf": "RED",
 618                     "address_family": {
 619                         "ipv4": {
 620                             "unicast": {"neighbor": {"r1": {"dest_link": {"e1": {}}}}}
 621                         },
 622                         "ipv6": {
 623                             "unicast": {"neighbor": {"r1": {"dest_link": {"e1": {}}}}}
 624                         },
 625                         "l2vpn": {
 626                             "evpn": {"route-target": {"export": [{"value": "100:100"}]}}
 627                         },
 628                     },
 629                 }
 630             ]
 631         }
 632     }
 633
 634     result = create_router_bgp(tgen, topo, input_dict_rt)
 635     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 636
 637     step(
 638         "Verify on dcg-1 and dcg-2, EVPN route for 10.1.1.1/32"
 639         " and 10::1/128 have RT value as 100:100."
 640     )
 641
 642     for dut in ["d1", "d2"]:
 643         input_routes = {key: topo["routers"][key] for key in ["r1"]}
 644         result = verify_attributes_for_evpn_routes(
 645             tgen, topo, dut, input_routes, rt="100:100"
 646         )
 647         assert result is True, "Testcase {} on {} :Failed \n Error: {}".format(
 648             tc_name, dut, result
 649         )
 650
 651     step(
 652         "Configure RT for vrf RED manually as export 500:500 and"
 653         " advertise vrf RED's routes in EVPN address family from"
 654         " e1 router."
 655     )
 656
 657     input_dict_rt = {
 658         "e1": {
 659             "bgp": [
 660                 {
 661                     "local_as": "100",
 662                     "vrf": "RED",
 663                     "address_family": {
 664                         "l2vpn": {
 665                             "evpn": {"route-target": {"export": [{"value": "500:500"}]}}
 666                         }
 667                     },
 668                 }
 669             ]
 670         }
 671     }
 672
 673     result = create_router_bgp(tgen, topo, input_dict_rt)
 674     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 675
 676     step(
 677         "Verify on dcg-1 and dcg-2, EVPN route for 10.1.1.1/32"
 678         " and 10::1/128 have RT value as 500:500."
 679     )
 680
 681     for dut in ["d1", "d2"]:
 682         input_routes = {key: topo["routers"][key] for key in ["r1"]}
 683         result = verify_attributes_for_evpn_routes(
 684             tgen, topo, dut, input_routes, rt=["100:100", "500:500"]
 685         )
 686         assert result is True, "Testcase {} on {} :Failed \n Error: {}".format(
 687             tc_name, dut, result
 688         )
 689
 690     step(
 691         "Import RT value 100:100 and 500:500 in vrf BLUE manually on"
 692         " peer router DCG-1 and DCG-2."
 693     )
 694
 695     input_dict_rt = {
 696         "d1": {
 697             "bgp": [
 698                 {
 699                     "local_as": "100",
 700                     "vrf": "BLUE",
 701                     "address_family": {
 702                         "l2vpn": {
 703                             "evpn": {
 704                                 "route-target": {
 705                                     "import": [
 706                                         {"value": "100:100"},
 707                                         {"value": "500:500"},
 708                                     ]
 709                                 }
 710                             }
 711                         }
 712                     },
 713                 }
 714             ]
 715         },
 716         "d2": {
 717             "bgp": [
 718                 {
 719                     "local_as": "200",
 720                     "vrf": "BLUE",
 721                     "address_family": {
 722                         "l2vpn": {
 723                             "evpn": {
 724                                 "route-target": {
 725                                     "import": [
 726                                         {"value": "100:100"},
 727                                         {"value": "500:500"},
 728                                     ]
 729                                 }
 730                             }
 731                         }
 732                     },
 733                 }
 734             ]
 735         },
 736     }
 737
 738     result = create_router_bgp(tgen, topo, input_dict_rt)
 739     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 740
 741     step(
 742         "EVPN route for 10.1.1.1/32 and 10::1 should be installed "
 743         "in vrf BLUE on DCG-1 and DCG-2 and further advertised to "
 744         "VNF router."
 745     )
 746
 747     for addr_type in ADDR_TYPES:
 748         input_routes = {
 749             "r1": {
 750                 "static_routes": [{"network": [NETWORK1_1[addr_type]], "vrf": "BLUE"}]
 751             }
 752         }
 753         result = verify_rib(tgen, addr_type, "d1", input_routes)
 754         assert result is True, "Testcase {} :Failed \n Error {}".format(tc_name, result)
 755
 756         result = verify_rib(tgen, addr_type, "d2", input_routes)
 757         assert result is True, "Testcase {} :Failed \n Error {}".format(tc_name, result)
 758
 759     step(
 760         "Delete import RT value 500:500 in vrf BLUE manually on "
 761         "peer router DCG-1 and DCG-2."
 762     )
 763
 764     input_dict_rt = {
 765         "d1": {
 766             "bgp": [
 767                 {
 768                     "local_as": "100",
 769                     "vrf": "BLUE",
 770                     "address_family": {
 771                         "l2vpn": {
 772                             "evpn": {
 773                                 "route-target": {
 774                                     "import": [{"value": "500:500", "delete": True}]
 775                                 }
 776                             }
 777                         }
 778                     },
 779                 }
 780             ]
 781         },
 782         "d2": {
 783             "bgp": [
 784                 {
 785                     "local_as": "200",
 786                     "vrf": "BLUE",
 787                     "address_family": {
 788                         "l2vpn": {
 789                             "evpn": {
 790                                 "route-target": {
 791                                     "import": [{"value": "500:500", "delete": True}]
 792                                 }
 793                             }
 794                         }
 795                     },
 796                 }
 797             ]
 798         },
 799     }
 800
 801     result = create_router_bgp(tgen, topo, input_dict_rt)
 802     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 803
 804     for dut in ["d1", "d2"]:
 805         input_routes = {key: topo["routers"][key] for key in ["r1"]}
 806         result = verify_attributes_for_evpn_routes(
 807             tgen, topo, dut, input_routes, rt=["100:100", "500:500"]
 808         )
 809         assert result is True, "Testcase {} on {} :Failed \n Error: {}".format(
 810             tc_name, dut, result
 811         )
 812
 813     step("Delete RT export value 100:100 for vrf RED on Edge-1")
 814
 815     input_dict_rt = {
 816         "e1": {
 817             "bgp": [
 818                 {
 819                     "local_as": "100",
 820                     "vrf": "RED",
 821                     "address_family": {
 822                         "l2vpn": {
 823                             "evpn": {
 824                                 "route-target": {
 825                                     "export": [{"value": "100:100", "delete": True}]
 826                                 }
 827                             }
 828                         }
 829                     },
 830                 }
 831             ]
 832         }
 833     }
 834
 835     result = create_router_bgp(tgen, topo, input_dict_rt)
 836     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 837
 838     step(
 839         "EVPN route for 10.1.1.1/32 and 10::1 should be withdrawn "
 840         "from vrf BLUE on DCG-1,DCG-2 and VNF router."
 841     )
 842
 843     for addr_type in ADDR_TYPES:
 844         input_routes = {
 845             "r1": {
 846                 "static_routes": [{"network": [NETWORK1_1[addr_type]], "vrf": "BLUE"}]
 847             }
 848         }
 849         result = verify_rib(tgen, addr_type, "d1", input_routes, expected=False)
 850         assert result is not True, (
 851             "Testcase {} : Failed \n "
 852             "Expected: Routes should not be present in {} RIB \n "
 853             "Found: {}".format(tc_name, "d1", result)
 854         )
 855
 856         result = verify_rib(tgen, addr_type, "d2", input_routes, expected=False)
 857         assert result is not True, (
 858             "Testcase {} : Failed \n "
 859             "Expected: Routes should not be present in {} RIB \n "
 860             "Found: {}".format(tc_name, "d2", result)
 861         )
 862
 863     step(
 864         "Configure RT value as 100:100000010000010000101010 to check "
 865         "the boundary value."
 866     )
 867
 868     input_dict_rt = {
 869         "e1": {
 870             "bgp": [
 871                 {
 872                     "local_as": "100",
 873                     "vrf": "RED",
 874                     "address_family": {
 875                         "l2vpn": {
 876                             "evpn": {
 877                                 "route-target": {
 878                                     "export": [
 879                                         {"value": "100:100000010000010000101010"}
 880                                     ]
 881                                 }
 882                             }
 883                         }
 884                     },
 885                 }
 886             ]
 887         }
 888     }
 889
 890     result = create_router_bgp(tgen, topo, input_dict_rt)
 891     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 892
 893     step(
 894         "CLI error: RT value: 100:100000010000010000101010 should not " "be configured"
 895     )
 896
 897     dut = "e1"
 898     input_routes = {key: topo["routers"][key] for key in ["r1"]}
 899     result = verify_attributes_for_evpn_routes(
 900         tgen, topo, dut, input_routes, rt="100:100000010000010000101010", expected=False
 901     )
 902     assert result is not True, (
 903         "Testcase {} : Failed \n "
 904         "Expected: RT value out of boundary error in {} \n "
 905         "Found: {}".format(tc_name, dut, result)
 906     )
 907
 908     write_test_footer(tc_name)
 909
 910
 911 def test_active_standby_evpn_implementation_p1(request):
 912     """
 913     In an active/standby EVPN implementation, if active DCG goes down,
 914     secondary takes over.
 915     """
 916
 917     tgen = get_topogen()
 918     tc_name = request.node.name
 919     write_test_header(tc_name)
 920     check_router_status(tgen)
 921     reset_config_on_routers(tgen)
 922     add_default_routes(tgen)
 923
 924     if tgen.routers_have_failure():
 925         pytest.skip(tgen.errors)
 926
 927     step(
 928         "Taken care in base config: Configure BGP neighborship for both "
 929         "address families(IPv4 & IPv6) between DCG-1/DCG-2 and VFN routers"
 930         "(R3 and R4)."
 931     )
 932
 933     step(
 934         "BGP neighborships come up within defined VRFs. Please use below "
 935         "command: sh bgp vrf all summary"
 936     )
 937
 938     result = verify_bgp_convergence(tgen, topo, "d1")
 939     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 940
 941     result = verify_bgp_convergence(tgen, topo, "d2")
 942     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 943
 944     step(
 945         "Advertise prefixes from VNF routers R3 and R4 in associated "
 946         "VRFs for both address-families."
 947     )
 948
 949     for addr_type in ADDR_TYPES:
 950         input_dict_1 = {
 951             "r3": {
 952                 "static_routes": [
 953                     {
 954                         "network": NETWORK1_2[addr_type],
 955                         "next_hop": NEXT_HOP_IP[addr_type],
 956                         "vrf": "RED",
 957                     }
 958                 ]
 959             },
 960             "r4": {
 961                 "static_routes": [
 962                     {
 963                         "network": NETWORK1_3[addr_type],
 964                         "next_hop": NEXT_HOP_IP[addr_type],
 965                         "vrf": "BLUE",
 966                     },
 967                     {
 968                         "network": NETWORK1_4[addr_type],
 969                         "next_hop": NEXT_HOP_IP[addr_type],
 970                         "vrf": "GREEN",
 971                     },
 972                 ]
 973             },
 974         }
 975
 976         result = create_static_routes(tgen, input_dict_1)
 977         assert result is True, "Testcase {} : Failed \n Error: {}".format(
 978             tc_name, result
 979         )
 980
 981     step(
 982         "Redistribute static in (IPv4 and IPv6) address-family "
 983         "on Edge-1 for all VRFs."
 984     )
 985
 986     input_dict_2 = {}
 987     for dut in ["r3", "r4"]:
 988         temp = {dut: {"bgp": []}}
 989         input_dict_2.update(temp)
 990
 991         if dut == "r3":
 992             VRFS = ["RED"]
 993             AS_NUM = [3]
 994         if dut == "r4":
 995             VRFS = ["BLUE", "GREEN"]
 996             AS_NUM = [4, 4]
 997
 998         for vrf, as_num in zip(VRFS, AS_NUM):
 999             temp[dut]["bgp"].append(
1000                 {
1001                     "local_as": as_num,
1002                     "vrf": vrf,
1003                     "address_family": {
1004                         "ipv4": {
1005                             "unicast": {"redistribute": [{"redist_type": "static"}]}
1006                         },
1007                         "ipv6": {
1008                             "unicast": {"redistribute": [{"redist_type": "static"}]}
1009                         },
1010                     },
1011                 }
1012             )
1013
1014     result = create_router_bgp(tgen, topo, input_dict_2)
1015     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
1016
1017     step("Prefixes are received in respective VRFs on DCG-1/DCG-2.")
1018
1019     for addr_type in ADDR_TYPES:
1020         input_routes = {
1021             "r3": {
1022                 "static_routes": [
1023                     {
1024                         "network": NETWORK1_2[addr_type],
1025                         "next_hop": NEXT_HOP_IP[addr_type],
1026                         "vrf": "RED",
1027                     }
1028                 ]
1029             },
1030             "r4": {
1031                 "static_routes": [
1032                     {
1033                         "network": NETWORK1_3[addr_type],
1034                         "next_hop": NEXT_HOP_IP[addr_type],
1035                         "vrf": "BLUE",
1036                     },
1037                     {
1038                         "network": NETWORK1_4[addr_type],
1039                         "next_hop": NEXT_HOP_IP[addr_type],
1040                         "vrf": "GREEN",
1041                     },
1042                 ]
1043             },
1044         }
1045
1046         result = verify_rib(tgen, addr_type, "d1", input_routes)
1047         assert result is True, "Testcase {} :Failed \n Error: {}".format(
1048             tc_name, result
1049         )
1050
1051         result = verify_rib(tgen, addr_type, "d2", input_routes)
1052         assert result is True, "Testcase {} :Failed \n Error: {}".format(
1053             tc_name, result
1054         )
1055
1056     step(
1057         "Taken care in base config: Advertise VRF routes in EVPN "
1058         "address-family from DCG-1 and DCG-2 router."
1059     )
1060
1061     step("Verify on Edge-1 that EVPN routes are installed via next-hop " "as DCG-2.")
1062
1063     for addr_type in ADDR_TYPES:
1064         input_routes = {
1065             "r3": {
1066                 "static_routes": [
1067                     {
1068                         "network": NETWORK1_2[addr_type],
1069                         "next_hop": NEXT_HOP_IP[addr_type],
1070                         "vrf": "RED",
1071                     }
1072                 ]
1073             },
1074             "r4": {
1075                 "static_routes": [
1076                     {
1077                         "network": NETWORK1_3[addr_type],
1078                         "next_hop": NEXT_HOP_IP[addr_type],
1079                         "vrf": "BLUE",
1080                     },
1081                     {
1082                         "network": NETWORK1_4[addr_type],
1083                         "next_hop": NEXT_HOP_IP[addr_type],
1084                         "vrf": "GREEN",
1085                     },
1086                 ]
1087             },
1088         }
1089
1090         if addr_type == "ipv4":
1091             result = verify_rib(
1092                 tgen, addr_type, "e1", input_routes, next_hop=BRIDGE_INTF2
1093             )
1094             assert result is True, "Testcase {} :Failed \n Error: {}".format(
1095                 tc_name, result
1096             )
1097         else:
1098             result = verify_rib(tgen, addr_type, "e1", input_routes)
1099             assert result is True, "Testcase {} :Failed \n Error: {}".format(
1100                 tc_name, result
1101             )
1102
1103     step(
1104         "Configure 'next-hop self' on DCG-1 for peer Edge-1 in EVPN " "address-family."
1105     )
1106
1107     input_dict_3 = {
1108         "d1": {
1109             "bgp": [
1110                 {
1111                     "local_as": "100",
1112                     "address_family": {
1113                         "l2vpn": {
1114                             "evpn": {
1115                                 "neighbor": {
1116                                     "e1": {
1117                                         "ipv4": {"d1-link1": {"next_hop_self": True}}
1118                                     }
1119                                 }
1120                             }
1121                         }
1122                     },
1123                 }
1124             ]
1125         }
1126     }
1127
1128     result = create_router_bgp(tgen, topo, input_dict_3)
1129     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
1130
1131     logger.info(
1132         "Creating route-map so ipv6 glpbal ip wpuld be preferred " "as next-hop"
1133     )
1134
1135     step(
1136         "Verify on Edge-1 that EVPN routes are now preferred via "
1137         "next-hop as DCG-1(iBGP) due to shortest AS-Path."
1138     )
1139
1140     for addr_type in ADDR_TYPES:
1141
1142         logger.info("Verifying only ipv4 routes")
1143         if addr_type != "ipv4":
1144             continue
1145
1146         input_routes = {
1147             "r3": {
1148                 "static_routes": [
1149                     {
1150                         "network": NETWORK1_2[addr_type],
1151                         "next_hop": NEXT_HOP_IP[addr_type],
1152                         "vrf": "RED",
1153                     }
1154                 ]
1155             },
1156             "r4": {
1157                 "static_routes": [
1158                     {
1159                         "network": NETWORK1_3[addr_type],
1160                         "next_hop": NEXT_HOP_IP[addr_type],
1161                         "vrf": "BLUE",
1162                     },
1163                     {
1164                         "network": NETWORK1_4[addr_type],
1165                         "next_hop": NEXT_HOP_IP[addr_type],
1166                         "vrf": "GREEN",
1167                     },
1168                 ]
1169             },
1170         }
1171
1172         next_hop = topo["routers"]["d1"]["links"]["e1-link1"]["ipv4"].split("/")[0]
1173
1174         result = verify_rib(tgen, addr_type, "e1", input_routes, next_hop=next_hop)
1175         assert result is True, "Testcase {} :Failed \n Error: {}".format(
1176             tc_name, result
1177         )
1178
1179     write_test_footer(tc_name)
1180
1181
1182 def test_evpn_routes_from_VNFs_p1(request):
1183     """
1184     EVPN routes are advertised/withdrawn, based on VNFs
1185     advertising/withdrawing IP prefixes.
1186     """
1187
1188     tgen = get_topogen()
1189     tc_name = request.node.name
1190     write_test_header(tc_name)
1191     check_router_status(tgen)
1192     reset_config_on_routers(tgen)
1193     add_default_routes(tgen)
1194
1195     if tgen.routers_have_failure():
1196         pytest.skip(tgen.errors)
1197
1198     step(
1199         "Advertise prefixes from VNF routers R1 and R2 in associated "
1200         "VRFs for both address-family."
1201     )
1202
1203     for addr_type in ADDR_TYPES:
1204         input_dict_1 = {
1205             "r1": {
1206                 "static_routes": [
1207                     {
1208                         "network": NETWORK1_1[addr_type],
1209                         "next_hop": NEXT_HOP_IP[addr_type],
1210                         "vrf": "RED",
1211                     }
1212                 ]
1213             },
1214             "r2": {
1215                 "static_routes": [
1216                     {
1217                         "network": NETWORK2_1[addr_type],
1218                         "next_hop": NEXT_HOP_IP[addr_type],
1219                         "vrf": "BLUE",
1220                     },
1221                     {
1222                         "network": NETWORK3_1[addr_type],
1223                         "next_hop": NEXT_HOP_IP[addr_type],
1224                         "vrf": "GREEN",
1225                     },
1226                 ]
1227             },
1228         }
1229
1230         result = create_static_routes(tgen, input_dict_1)
1231         assert result is True, "Testcase {} : Failed \n Error: {}".format(
1232             tc_name, result
1233         )
1234
1235     step(
1236         "Taken care in base config: Advertise VNFs'(R1 and R2) "
1237         "originated routes in EVPN address-family from Edge-1 to "
1238         "DCG-1 and DCG-2 routers."
1239     )
1240     step(
1241         "Taken care in base config: Advertise IPv4 and IPv6 routes "
1242         "from default vrf in EVPN address-family from Edge-1."
1243     )
1244
1245     step(
1246         "Verify on DCG-2 that VNF routes are received in respective "
1247         "VRFs along with auto derived RD/RT values 'show bgp l2vpn evpn'"
1248     )
1249     for dut in ["d1", "d2"]:
1250         input_routes = {key: topo["routers"][key] for key in ["r1"]}
1251         result = verify_evpn_routes(tgen, topo, dut, input_routes)
1252         assert result is True, "Testcase {} on {} :Failed \n Error: {}".format(
1253             tc_name, dut, result
1254         )
1255
1256         input_routes = {key: topo["routers"][key] for key in ["r2"]}
1257         result = verify_evpn_routes(tgen, topo, dut, input_routes)
1258         assert result is True, "Testcase {} on {} :Failed \n Error: {}".format(
1259             tc_name, dut, result
1260         )
1261
1262     step(
1263         "Verify on R3 and R4 that DCG-2 further advertises all EVPN "
1264         "routes to corresponding VRFs."
1265     )
1266     for addr_type in ADDR_TYPES:
1267         input_routes = {key: topo["routers"][key] for key in ["r1"]}
1268         result = verify_rib(tgen, addr_type, "r3", input_routes)
1269         assert result is True, "Testcase {} :Failed \n Error: {}".format(
1270             tc_name, result
1271         )
1272
1273     for addr_type in ADDR_TYPES:
1274         input_routes = {key: topo["routers"][key] for key in ["r2"]}
1275         result = verify_rib(tgen, addr_type, "r4", input_routes)
1276         assert result is True, "Testcase {} :Failed \n Error: {}".format(
1277             tc_name, result
1278         )
1279
1280     step(
1281         "Verify that DCG-2 receives EVPN routes associated to default "
1282         "VRF and install in default IP routing table as well."
1283     )
1284     for addr_type in ADDR_TYPES:
1285         input_routes = {key: topo["routers"][key] for key in ["r1"]}
1286         result = verify_rib(tgen, addr_type, "d2", input_routes)
1287         assert result is True, "Testcase {} :Failed \n Error: {}".format(
1288             tc_name, result
1289         )
1290
1291     for addr_type in ADDR_TYPES:
1292         input_routes = {key: topo["routers"][key] for key in ["r2"]}
1293         result = verify_rib(tgen, addr_type, "d2", input_routes)
1294         assert result is True, "Testcase {} :Failed \n Error: {}".format(
1295             tc_name, result
1296         )
1297
1298     step("Withdraw the IP prefixes from VFN(R1).")
1299     dut = "r1"
1300     input_dict_2 = {}
1301     static_routes = topo["routers"][dut]["static_routes"]
1302     for static_route in static_routes:
1303         static_route["delete"] = True
1304         temp = {dut: {"static_routes": [static_route]}}
1305         input_dict_2.update(temp)
1306
1307         result = create_static_routes(tgen, input_dict_2)
1308         assert result is True, "Testcase {} :Failed \n Error: {}".format(
1309             tc_name, result
1310         )
1311
1312     step(
1313         "Verify that DCG-2 removes EVPN routes corresponding to vrf RED and "
1314         "send an withdraw to VNF(R3) as well."
1315     )
1316     for addr_type in ADDR_TYPES:
1317         input_routes = {key: topo["routers"][key] for key in ["r1"]}
1318         result = verify_rib(tgen, addr_type, "d2", input_routes, expected=False)
1319         assert result is not True, (
1320             "Testcase {} : Failed \n "
1321             "Expected: Routes should not be present in {} RIB \n "
1322             "Found: {}".format(tc_name, "d2", result)
1323         )
1324
1325     for addr_type in ADDR_TYPES:
1326         input_routes = {key: topo["routers"][key] for key in ["r1"]}
1327         result = verify_rib(tgen, addr_type, "r3", input_routes, expected=False)
1328         assert result is not True, (
1329             "Testcase {} : Failed \n "
1330             "Expected: Routes should not be present in {} RIB \n "
1331             "Found: {}".format(tc_name, "r3", result)
1332         )
1333
1334     step("Re-advertise IP prefixes from VFN(R1).")
1335     step(
1336         "Advertise prefixes from VNF routers R1 and R2 in associated "
1337         "VRFs for both address-family."
1338     )
1339
1340     for addr_type in ADDR_TYPES:
1341         input_dict_1 = {
1342             "r1": {
1343                 "static_routes": [
1344                     {
1345                         "network": NETWORK1_1[addr_type],
1346                         "next_hop": NEXT_HOP_IP[addr_type],
1347                         "vrf": "RED",
1348                     }
1349                 ]
1350             },
1351             "r2": {
1352                 "static_routes": [
1353                     {
1354                         "network": NETWORK2_1[addr_type],
1355                         "next_hop": NEXT_HOP_IP[addr_type],
1356                         "vrf": "BLUE",
1357                     },
1358                     {
1359                         "network": NETWORK3_1[addr_type],
1360                         "next_hop": NEXT_HOP_IP[addr_type],
1361                         "vrf": "GREEN",
1362                     },
1363                 ]
1364             },
1365         }
1366
1367         result = create_static_routes(tgen, input_dict_1)
1368         assert result is True, "Testcase {} : Failed \n Error: {}".format(
1369             tc_name, result
1370         )
1371
1372     step(
1373         "Verify that DCG-2 receives EVPN routes corresponding to vrf RED "
1374         "again and send an update to VNF(R3) as well."
1375     )
1376     for addr_type in ADDR_TYPES:
1377         input_routes = {key: topo["routers"][key] for key in ["r1"]}
1378         result = verify_rib(tgen, addr_type, "d2", input_routes)
1379         assert result is True, "Testcase {} :Failed \n Error: {}".format(
1380             tc_name, result
1381         )
1382
1383     for addr_type in ADDR_TYPES:
1384         input_routes = {key: topo["routers"][key] for key in ["r1"]}
1385         result = verify_rib(tgen, addr_type, "r3", input_routes)
1386         assert result is True, "Testcase {} :Failed \n Error: {}".format(
1387             tc_name, result
1388         )
1389
1390     step("Delete vrf BLUE from router Edge-1")
1391     input_dict_3 = {"e1": {"vrfs": [{"name": "BLUE", "id": "2", "delete": True}]}}
1392
1393     result = create_vrf_cfg(tgen, input_dict_3)
1394     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
1395
1396     step(
1397         "Verify that DCG-2 removes EVPN routes corresponding to "
1398         "vrf BLUE and send an withdraw to VNF(R4) as well."
1399     )
1400     for addr_type in ADDR_TYPES:
1401         input_routes = {
1402             "r2": {"static_routes": [{"network": NETWORK2_1[addr_type], "vrf": "BLUE"}]}
1403         }
1404
1405         result = verify_rib(tgen, addr_type, "d2", input_routes, expected=False)
1406         assert result is not True, (
1407             "Testcase {} : Failed \n "
1408             "Expected: Routes should not be present in {} RIB \n "
1409             "Found: {}".format(tc_name, "d2", result)
1410         )
1411
1412         result = verify_rib(tgen, addr_type, "r4", input_routes, expected=False)
1413         assert result is not True, (
1414             "Testcase {} : Failed \n "
1415             "Expected: Routes should not be present in {} RIB \n "
1416             "Found: {}".format(tc_name, "r4", result)
1417         )
1418
1419     step("Add vrf BLUE on router Edge-1 again.")
1420     interface = topo["routers"]["e1"]["links"]["r2-link1"]["interface"]
1421     input_dict_3 = {
1422         "e1": {
1423             "links": {
1424                 "r2-link1": {
1425                     "interface": interface,
1426                     "ipv4": "auto",
1427                     "ipv6": "auto",
1428                     "vrf": "BLUE",
1429                 }
1430             },
1431             "vrfs": [{"name": "BLUE", "id": "2"}],
1432         }
1433     }
1434     result = create_vrf_cfg(tgen, input_dict_3)
1435     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
1436
1437     logger.info(
1438         "After deleting VRFs ipv6 addresses wil be deleted "
1439         "from kernel Adding back ipv6 addresses"
1440     )
1441     dut = "e1"
1442     vrfs = ["BLUE"]
1443
1444     for vrf in vrfs:
1445         for c_link, c_data in topo["routers"][dut]["links"].items():
1446             if "vrf" in c_data:
1447                 if c_data["vrf"] != vrf:
1448                     continue
1449
1450                 intf_name = c_data["interface"]
1451                 intf_ipv6 = c_data["ipv6"]
1452
1453                 create_interface_in_kernel(
1454                     tgen, dut, intf_name, intf_ipv6, vrf, create=False
1455                 )
1456
1457     result = verify_bgp_convergence(tgen, topo, dut)
1458     assert result is True, "Failed to converge on {}".format(dut)
1459
1460     step(
1461         "Verify that DCG-2 receives EVPN routes corresponding to "
1462         "vrf BLUE again and send an update to VNF(R4) as well."
1463     )
1464     for addr_type in ADDR_TYPES:
1465         input_routes = {
1466             "r2": {"static_routes": [{"network": NETWORK2_1[addr_type], "vrf": "BLUE"}]}
1467         }
1468
1469         result = verify_rib(tgen, addr_type, "d2", input_routes)
1470         assert result is True, "Testcase {} :Failed \n Error: {}".format(
1471             tc_name, result
1472         )
1473
1474         result = verify_rib(tgen, addr_type, "r4", input_routes)
1475         assert result is True, "Testcase {} :Failed \n Error: {}".format(
1476             tc_name, result
1477         )
1478
1479     step("Withdraw IPv6 address-family in EVPN advertisements for " "VRF GREEN")
1480     addr_type = "ipv6"
1481     input_dict_4 = {
1482         "e1": {
1483             "bgp": [
1484                 {
1485                     "local_as": "100",
1486                     "vrf": "GREEN",
1487                     "address_family": {
1488                         "l2vpn": {
1489                             "evpn": {
1490                                 "advertise": {addr_type: {"unicast": {"delete": True}}}
1491                             }
1492                         }
1493                     },
1494                 }
1495             ]
1496         }
1497     }
1498
1499     result = create_router_bgp(tgen, topo, input_dict_4)
1500     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
1501
1502     step(
1503         "Verify that EVPN routes (IPv6)associated with vrf GREEN are "
1504         "withdrawn from DCG-2 and VNF R4."
1505     )
1506     input_routes = {
1507         "r2": {"static_routes": [{"network": NETWORK3_1[addr_type], "vrf": "GREEN"}]}
1508     }
1509
1510     result = verify_rib(tgen, addr_type, "d2", input_routes, expected=False)
1511     assert result is not True, (
1512         "Testcase {} : Failed \n "
1513         "Expected: Routes should not be present in {} RIB \n "
1514         "Found: {}".format(tc_name, "d2", result)
1515     )
1516
1517     result = verify_rib(tgen, addr_type, "r4", input_routes, expected=False)
1518     assert result is not True, (
1519         "Testcase {} : Failed \n "
1520         "Expected: Routes should not be present in {} RIB \n "
1521         "Found: {}".format(tc_name, "r4", result)
1522     )
1523
1524     step("Advertise IPv6 address-family in EVPN advertisements " "for VRF GREEN.")
1525     addr_type = "ipv6"
1526     input_dict_4 = {
1527         "e1": {
1528             "bgp": [
1529                 {
1530                     "local_as": "100",
1531                     "vrf": "GREEN",
1532                     "address_family": {
1533                         "l2vpn": {"evpn": {"advertise": {addr_type: {"unicast": {}}}}}
1534                     },
1535                 }
1536             ]
1537         }
1538     }
1539
1540     result = create_router_bgp(tgen, topo, input_dict_4)
1541     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
1542
1543     for addr_type in ADDR_TYPES:
1544         input_routes = {
1545             "r2": {
1546                 "static_routes": [{"network": NETWORK3_1[addr_type], "vrf": "GREEN"}]
1547             }
1548         }
1549
1550         result = verify_rib(tgen, addr_type, "d2", input_routes)
1551         assert result is True, "Testcase {} :Failed \n Error: {}".format(
1552             tc_name, result
1553         )
1554
1555         result = verify_rib(tgen, addr_type, "r4", input_routes)
1556         assert result is True, "Testcase {} :Failed \n Error: {}".format(
1557             tc_name, result
1558         )
1559
1560     write_test_footer(tc_name)
1561
1562
1563 @pytest.mark.parametrize(
1564     "attribute", [{"route-type": "prefix"}, {"vni": VNI_1}, {"rt": "300:300"}]
1565 )
1566 def test_route_map_operations_for_evpn_address_family_p1(request, attribute):
1567     """
1568     Route-map operations for EVPN address family.
1569     """
1570
1571     tgen = get_topogen()
1572     tc_name = request.node.name
1573     write_test_header(tc_name)
1574     check_router_status(tgen)
1575     reset_config_on_routers(tgen)
1576     add_default_routes(tgen)
1577
1578     step(
1579         "Advertise prefixes from VNF routers R1 and R2 in associated "
1580         "VRFs for both address-family."
1581     )
1582
1583     for addr_type in ADDR_TYPES:
1584         input_dict_1 = {
1585             "r1": {
1586                 "static_routes": [
1587                     {
1588                         "network": NETWORK1_1[addr_type],
1589                         "next_hop": NEXT_HOP_IP[addr_type],
1590                         "vrf": "RED",
1591                     }
1592                 ]
1593             },
1594             "r2": {
1595                 "static_routes": [
1596                     {
1597                         "network": NETWORK2_1[addr_type],
1598                         "next_hop": NEXT_HOP_IP[addr_type],
1599                         "vrf": "BLUE",
1600                     },
1601                     {
1602                         "network": NETWORK3_1[addr_type],
1603                         "next_hop": NEXT_HOP_IP[addr_type],
1604                         "vrf": "GREEN",
1605                     },
1606                 ]
1607             },
1608         }
1609
1610         result = create_static_routes(tgen, input_dict_1)
1611         assert result is True, "Testcase {} : Failed \n Error: {}".format(
1612             tc_name, result
1613         )
1614
1615     if tgen.routers_have_failure():
1616         pytest.skip(tgen.errors)
1617
1618     step(
1619         "Advertise VRF routes in EVPN address family from Edge-1 router."
1620         " Configure a route-map on e1 to filter EVPN routes based on"
1621         " below keywords: route-type: prefix"
1622     )
1623
1624     for key, value in attribute.items():
1625         if key == "rt":
1626             logger.info("Creating extcommunity using raw_config")
1627             raw_config = {
1628                 "d2": {
1629                     "raw_config": [
1630                         "bgp extcommunity-list standard ECOMM300 permit {} {}".format(
1631                             key, value
1632                         )
1633                     ]
1634                 }
1635             }
1636             result = apply_raw_config(tgen, raw_config)
1637             assert result is True, "Testcase {} : Failed Error: {}".format(
1638                 tc_name, result
1639             )
1640
1641             input_dict_1 = {
1642                 "e1": {
1643                     "route_maps": {
1644                         "rmap_route_type": [
1645                             {"action": "permit", "set": {"extcommunity": {key: value}}}
1646                         ]
1647                     }
1648                 },
1649                 "d2": {
1650                     "route_maps": {
1651                         "rmap_route_type": [
1652                             {"action": "permit", "match": {"extcommunity": "ECOMM300"}}
1653                         ]
1654                     }
1655                 },
1656             }
1657
1658         else:
1659             input_dict_1 = {
1660                 "e1": {
1661                     "route_maps": {
1662                         "rmap_route_type": [
1663                             {"action": "permit", "match": {"evpn": {key: value}}}
1664                         ]
1665                     }
1666                 },
1667                 "d2": {
1668                     "route_maps": {
1669                         "rmap_route_type": [
1670                             {"action": "permit", "match": {"evpn": {key: value}}}
1671                         ]
1672                     }
1673                 },
1674             }
1675         result = create_route_maps(tgen, input_dict_1)
1676         assert result is True, "Testcase {} : Failed \n Error: {}".format(
1677             tc_name, result
1678         )
1679
1680     input_dict_2 = {
1681         "e1": {
1682             "bgp": [
1683                 {
1684                     "local_as": "100",
1685                     "address_family": {
1686                         "l2vpn": {
1687                             "evpn": {
1688                                 "neighbor": {
1689                                     "d2": {
1690                                         "ipv4": {
1691                                             "e1-link1": {
1692                                                 "route_maps": [
1693                                                     {
1694                                                         "name": "rmap_route_type",
1695                                                         "direction": "out",
1696                                                     }
1697                                                 ]
1698                                             }
1699                                         }
1700                                     }
1701                                 }
1702                             }
1703                         }
1704                     },
1705                 }
1706             ]
1707         },
1708         "d2": {
1709             "bgp": [
1710                 {
1711                     "local_as": "200",
1712                     "address_family": {
1713                         "l2vpn": {
1714                             "evpn": {
1715                                 "neighbor": {
1716                                     "e1": {
1717                                         "ipv4": {
1718                                             "d2-link1": {
1719                                                 "route_maps": [
1720                                                     {
1721                                                         "name": "rmap_route_type",
1722                                                         "direction": "in",
1723                                                     }
1724                                                 ]
1725                                             }
1726                                         }
1727                                     }
1728                                 }
1729                             }
1730                         }
1731                     },
1732                 }
1733             ]
1734         },
1735     }
1736
1737     result = create_router_bgp(tgen, topo, input_dict_2)
1738     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
1739
1740     step(
1741         "Verify on router DCG-2 that EVPN routes corresponding to all "
1742         "VRFs are received. As all EVPN routes are type-5 only."
1743     )
1744
1745     input_routes = {key: topo["routers"][key] for key in ["r1"]}
1746     result = verify_evpn_routes(tgen, topo, "d2", input_routes)
1747     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
1748
1749     input_routes = {key: topo["routers"][key] for key in ["r2"]}
1750     result = verify_evpn_routes(tgen, topo, "d2", input_routes)
1751     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
1752
1753     write_test_footer(tc_name)
1754
1755
1756 def test_evpn_address_family_with_graceful_restart_p0(request):
1757     """
1758     Verify Graceful-restart function for EVPN address-family.
1759     """
1760
1761     tgen = get_topogen()
1762     tc_name = request.node.name
1763     write_test_header(tc_name)
1764     check_router_status(tgen)
1765     reset_config_on_routers(tgen)
1766     add_default_routes(tgen)
1767
1768     if tgen.routers_have_failure():
1769         pytest.skip(tgen.errors)
1770
1771     for addr_type in ADDR_TYPES:
1772         input_dict_1 = {
1773             "r3": {
1774                 "static_routes": [
1775                     {
1776                         "network": NETWORK1_2[addr_type],
1777                         "next_hop": NEXT_HOP_IP[addr_type],
1778                         "vrf": "RED",
1779                     }
1780                 ]
1781             },
1782             "r4": {
1783                 "static_routes": [
1784                     {
1785                         "network": NETWORK1_3[addr_type],
1786                         "next_hop": NEXT_HOP_IP[addr_type],
1787                         "vrf": "BLUE",
1788                     },
1789                     {
1790                         "network": NETWORK1_4[addr_type],
1791                         "next_hop": NEXT_HOP_IP[addr_type],
1792                         "vrf": "GREEN",
1793                     },
1794                 ]
1795             },
1796         }
1797
1798         result = create_static_routes(tgen, input_dict_1)
1799         assert result is True, "Testcase {} : Failed \n Error: {}".format(
1800             tc_name, result
1801         )
1802
1803     step(
1804         "Redistribute static in (IPv4 and IPv6) address-family "
1805         "on Edge-1 for all VRFs."
1806     )
1807
1808     input_dict_2 = {}
1809     for dut in ["r3", "r4"]:
1810         temp = {dut: {"bgp": []}}
1811         input_dict_2.update(temp)
1812
1813         if dut == "r3":
1814             VRFS = ["RED"]
1815             AS_NUM = [3]
1816         if dut == "r4":
1817             VRFS = ["BLUE", "GREEN"]
1818             AS_NUM = [4, 4]
1819
1820         for vrf, as_num in zip(VRFS, AS_NUM):
1821             temp[dut]["bgp"].append(
1822                 {
1823                     "local_as": as_num,
1824                     "vrf": vrf,
1825                     "address_family": {
1826                         "ipv4": {
1827                             "unicast": {"redistribute": [{"redist_type": "static"}]}
1828                         },
1829                         "ipv6": {
1830                             "unicast": {"redistribute": [{"redist_type": "static"}]}
1831                         },
1832                     },
1833                 }
1834             )
1835
1836     result = create_router_bgp(tgen, topo, input_dict_2)
1837     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
1838
1839     step(
1840         "Verify on router Edge-1 that EVPN routes corresponding to "
1841         "all VRFs are received from both routers DCG-1 and DCG-2"
1842     )
1843
1844     for addr_type in ADDR_TYPES:
1845         input_routes = {
1846             "r3": {
1847                 "static_routes": [
1848                     {
1849                         "network": NETWORK1_2[addr_type],
1850                         "next_hop": NEXT_HOP_IP[addr_type],
1851                         "vrf": "RED",
1852                     }
1853                 ]
1854             },
1855             "r4": {
1856                 "static_routes": [
1857                     {
1858                         "network": NETWORK1_3[addr_type],
1859                         "next_hop": NEXT_HOP_IP[addr_type],
1860                         "vrf": "BLUE",
1861                     },
1862                     {
1863                         "network": NETWORK1_4[addr_type],
1864                         "next_hop": NEXT_HOP_IP[addr_type],
1865                         "vrf": "GREEN",
1866                     },
1867                 ]
1868             },
1869         }
1870
1871         result = verify_rib(tgen, addr_type, "e1", input_routes)
1872         assert result is True, "Testcase {} :Failed \n Error: {}".format(
1873             tc_name, result
1874         )
1875
1876     step(
1877         "Configure DCG-2 as GR restarting node for EVPN session between"
1878         " DCG-2 and EDGE-1, following by a session reset using 'clear bgp *'"
1879         " command."
1880     )
1881
1882     input_dict_gr = {
1883         "d2": {
1884             "bgp": [
1885                 {
1886                     "local_as": "200",
1887                     "graceful-restart": {
1888                         "graceful-restart": True,
1889                     },
1890                 }
1891             ]
1892         }
1893     }
1894
1895     result = create_router_bgp(tgen, topo, input_dict_gr)
1896     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
1897
1898     step(
1899         "Verify that DCG-2 changes it's role to GR-restarting router "
1900         "and EDGE-1 becomes the GR-helper."
1901     )
1902
1903     step("Kill BGPd daemon on DCG-2.")
1904     kill_router_daemons(tgen, "d2", ["bgpd"])
1905
1906     step(
1907         "Verify that EDGE-1 keep stale entries for EVPN RT-5 routes "
1908         "received from DCG-2 before the restart."
1909     )
1910
1911     for addr_type in ADDR_TYPES:
1912         input_routes = {
1913             "r4": {
1914                 "static_routes": [
1915                     {
1916                         "network": NETWORK1_3[addr_type],
1917                         "next_hop": NEXT_HOP_IP[addr_type],
1918                         "vrf": "BLUE",
1919                     },
1920                     {
1921                         "network": NETWORK1_4[addr_type],
1922                         "next_hop": NEXT_HOP_IP[addr_type],
1923                         "vrf": "GREEN",
1924                     },
1925                 ]
1926             }
1927         }
1928         result = verify_evpn_routes(tgen, topo, "e1", input_routes)
1929         assert result is True, "Testcase {} :Failed \n Error: {}".format(
1930             tc_name, result
1931         )
1932
1933     step(
1934         "Verify that DCG-2 keeps BGP routes in Zebra until BGPd "
1935         "comes up or end of 'rib-stale-time'"
1936     )
1937
1938     step("Start BGPd daemon on DCG-2.")
1939     start_router_daemons(tgen, "d2", ["bgpd"])
1940
1941     step("Verify that EDGE-1 removed all the stale entries.")
1942     for addr_type in ADDR_TYPES:
1943         input_routes = {
1944             "r4": {
1945                 "static_routes": [
1946                     {
1947                         "network": NETWORK1_3[addr_type],
1948                         "next_hop": NEXT_HOP_IP[addr_type],
1949                         "vrf": "BLUE",
1950                     },
1951                     {
1952                         "network": NETWORK1_4[addr_type],
1953                         "next_hop": NEXT_HOP_IP[addr_type],
1954                         "vrf": "GREEN",
1955                     },
1956                 ]
1957             }
1958         }
1959         result = verify_evpn_routes(tgen, topo, "e1", input_routes)
1960         assert result is True, "Testcase {} :Failed \n Error: {}".format(
1961             tc_name, result
1962         )
1963
1964     step(
1965         "Verify that DCG-2 refresh zebra with EVPN routes. "
1966         "(no significance of 'rib-stale-time'"
1967     )
1968
1969     for addr_type in ADDR_TYPES:
1970         input_routes = {
1971             "r4": {
1972                 "static_routes": [
1973                     {
1974                         "network": NETWORK1_3[addr_type],
1975                         "next_hop": NEXT_HOP_IP[addr_type],
1976                         "vrf": "BLUE",
1977                     },
1978                     {
1979                         "network": NETWORK1_4[addr_type],
1980                         "next_hop": NEXT_HOP_IP[addr_type],
1981                         "vrf": "GREEN",
1982                     },
1983                 ]
1984             }
1985         }
1986         result = verify_rib(tgen, addr_type, "d2", input_routes)
1987         assert result is True, "Testcase {} :Failed \n Error: {}".format(
1988             tc_name, result
1989         )
1990
1991     write_test_footer(tc_name)
1992
1993
1994 @pytest.mark.parametrize("attribute", ["locPrf", "weight", "path"])
1995 def test_bgp_attributes_for_evpn_address_family_p1(request, attribute):
1996     """
1997     BGP attributes for EVPN address-family.
1998     """
1999
2000     tgen = get_topogen()
2001     tc_name = request.node.name
2002     write_test_header(tc_name)
2003     check_router_status(tgen)
2004     reset_config_on_routers(tgen)
2005     add_default_routes(tgen)
2006
2007     if tgen.routers_have_failure():
2008         pytest.skip(tgen.errors)
2009
2010     step(
2011         "Advertise prefixes from VNF routers R1 and R2 in associated "
2012         "VRFs for both address-family."
2013     )
2014
2015     for addr_type in ADDR_TYPES:
2016         input_dict_1 = {
2017             "r1": {
2018                 "static_routes": [
2019                     {
2020                         "network": NETWORK1_1[addr_type],
2021                         "next_hop": NEXT_HOP_IP[addr_type],
2022                         "vrf": "RED",
2023                     }
2024                 ]
2025             },
2026             "r2": {
2027                 "static_routes": [
2028                     {
2029                         "network": NETWORK2_1[addr_type],
2030                         "next_hop": NEXT_HOP_IP[addr_type],
2031                         "vrf": "BLUE",
2032                     },
2033                     {
2034                         "network": NETWORK3_1[addr_type],
2035                         "next_hop": NEXT_HOP_IP[addr_type],
2036                         "vrf": "GREEN",
2037                     },
2038                 ]
2039             },
2040         }
2041
2042         result = create_static_routes(tgen, input_dict_1)
2043         assert result is True, "Testcase {} : Failed \n Error: {}".format(
2044             tc_name, result
2045         )
2046
2047     topo_local = deepcopy(topo)
2048
2049     logger.info("Modifying topology b/w e1 and d1 from iBGP to eBGP")
2050     step("Delete BGP config for vrf RED.")
2051
2052     if attribute == "locPrf":
2053         input_dict_vni = {
2054             "d1": {
2055                 "vrfs": [
2056                     {"name": "RED", "no_vni": VNI_1},
2057                     {"name": "BLUE", "no_vni": VNI_2},
2058                     {"name": "GREEN", "no_vni": VNI_3},
2059                 ]
2060             }
2061         }
2062         result = create_vrf_cfg(tgen, topo, input_dict=input_dict_vni)
2063         assert result is True, "Testcase {} :Failed \n Error: {}".format(
2064             tc_name, result
2065         )
2066
2067         input_dict_2 = {}
2068         for dut in ["d1"]:
2069             temp = {dut: {"bgp": []}}
2070             input_dict_2.update(temp)
2071
2072             INDEX = [0, 1, 2, 3]
2073             VRFS = ["RED", "BLUE", "GREEN", None]
2074             AS_NUM = [100, 100, 100, 100]
2075
2076             for index, vrf, as_num in zip(INDEX, VRFS, AS_NUM):
2077                 topo_local["routers"][dut]["bgp"][index]["local_as"] = 200
2078                 if vrf:
2079                     temp[dut]["bgp"].append(
2080                         {"local_as": as_num, "vrf": vrf, "delete": True}
2081                     )
2082                 else:
2083                     temp[dut]["bgp"].append({"local_as": as_num, "delete": True})
2084
2085         result = create_router_bgp(tgen, topo, input_dict_2)
2086         assert result is True, "Testcase {} on d1 :Failed \n Error: {}".format(
2087             tc_name, result
2088         )
2089
2090         result = create_router_bgp(tgen, topo_local["routers"])
2091         assert result is True, "Testcase {} :Failed \n Error: {}".format(
2092             tc_name, result
2093         )
2094
2095     step("Advertise VRF routes in EVPN address-family from DCG-1 " "and DCG-2 routers.")
2096
2097     for addr_type in ADDR_TYPES:
2098         input_dict_1 = {
2099             "r3": {
2100                 "static_routes": [
2101                     {
2102                         "network": NETWORK1_2[addr_type],
2103                         "next_hop": NEXT_HOP_IP[addr_type],
2104                         "vrf": "RED",
2105                     }
2106                 ]
2107             },
2108             "r4": {
2109                 "static_routes": [
2110                     {
2111                         "network": NETWORK1_3[addr_type],
2112                         "next_hop": NEXT_HOP_IP[addr_type],
2113                         "vrf": "BLUE",
2114                     },
2115                     {
2116                         "network": NETWORK1_4[addr_type],
2117                         "next_hop": NEXT_HOP_IP[addr_type],
2118                         "vrf": "GREEN",
2119                     },
2120                 ]
2121             },
2122         }
2123
2124         result = create_static_routes(tgen, input_dict_1)
2125         assert result is True, "Testcase {} : Failed \n Error: {}".format(
2126             tc_name, result
2127         )
2128
2129     step(
2130         "Redistribute static in (IPv4 and IPv6) address-family "
2131         "on Edge-1 for all VRFs."
2132     )
2133
2134     input_dict_2 = {}
2135     for dut in ["r3", "r4"]:
2136         temp = {dut: {"bgp": []}}
2137         input_dict_2.update(temp)
2138
2139         if dut == "r3":
2140             VRFS = ["RED"]
2141             AS_NUM = [3]
2142         if dut == "r4":
2143             VRFS = ["BLUE", "GREEN"]
2144             AS_NUM = [4, 4]
2145
2146         for vrf, as_num in zip(VRFS, AS_NUM):
2147             temp[dut]["bgp"].append(
2148                 {
2149                     "local_as": as_num,
2150                     "vrf": vrf,
2151                     "address_family": {
2152                         "ipv4": {
2153                             "unicast": {"redistribute": [{"redist_type": "static"}]}
2154                         },
2155                         "ipv6": {
2156                             "unicast": {"redistribute": [{"redist_type": "static"}]}
2157                         },
2158                     },
2159                 }
2160             )
2161
2162     result = create_router_bgp(tgen, topo, input_dict_2)
2163     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
2164
2165     step(
2166         "Verify on router Edge-1 that EVPN routes corresponding to "
2167         "all VRFs are received from both routers DCG-1 and DCG-2"
2168     )
2169
2170     for addr_type in ADDR_TYPES:
2171         input_routes = {
2172             "r3": {
2173                 "static_routes": [
2174                     {
2175                         "network": NETWORK1_2[addr_type],
2176                         "next_hop": NEXT_HOP_IP[addr_type],
2177                         "vrf": "RED",
2178                     }
2179                 ]
2180             },
2181             "r4": {
2182                 "static_routes": [
2183                     {
2184                         "network": NETWORK1_3[addr_type],
2185                         "next_hop": NEXT_HOP_IP[addr_type],
2186                         "vrf": "BLUE",
2187                     },
2188                     {
2189                         "network": NETWORK1_4[addr_type],
2190                         "next_hop": NEXT_HOP_IP[addr_type],
2191                         "vrf": "GREEN",
2192                     },
2193                 ]
2194             },
2195         }
2196
2197         result = verify_rib(tgen, addr_type, "e1", input_routes)
2198         assert result is True, "Testcase {} :Failed \n Error: {}".format(
2199             tc_name, result
2200         )
2201
2202     step(
2203         "Configure a route-map on Edge-1 to modify below BGP attributes "
2204         "for EVPN address-family:"
2205     )
2206
2207     if attribute == "path":
2208         input_dict_1 = {
2209             "e1": {
2210                 "route_maps": {
2211                     "rmap_d1": [
2212                         {
2213                             "action": "permit",
2214                             "set": {
2215                                 attribute: {
2216                                     "as_num": "123 231 321",
2217                                     "as_action": "prepend",
2218                                 }
2219                             },
2220                         }
2221                     ],
2222                     "rmap_d2": [
2223                         {
2224                             "action": "permit",
2225                             "set": {
2226                                 attribute: {"as_num": "121", "as_action": "prepend"}
2227                             },
2228                         }
2229                     ],
2230                 }
2231             }
2232         }
2233     else:
2234         input_dict_1 = {
2235             "e1": {
2236                 "route_maps": {
2237                     "rmap_d1": [{"action": "permit", "set": {attribute: 120}}],
2238                     "rmap_d2": [{"action": "permit", "set": {attribute: 150}}],
2239                 }
2240             }
2241         }
2242     result = create_route_maps(tgen, input_dict_1)
2243     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
2244
2245     input_dict_2 = {
2246         "e1": {
2247             "bgp": [
2248                 {
2249                     "local_as": "100",
2250                     "address_family": {
2251                         "l2vpn": {
2252                             "evpn": {
2253                                 "neighbor": {
2254                                     "d1": {
2255                                         "ipv4": {
2256                                             "e1-link1": {
2257                                                 "route_maps": [
2258                                                     {
2259                                                         "name": "rmap_d1",
2260                                                         "direction": "in",
2261                                                     }
2262                                                 ]
2263                                             }
2264                                         }
2265                                     },
2266                                     "d2": {
2267                                         "ipv4": {
2268                                             "e1-link1": {
2269                                                 "route_maps": [
2270                                                     {
2271                                                         "name": "rmap_d2",
2272                                                         "direction": "in",
2273                                                     }
2274                                                 ]
2275                                             }
2276                                         }
2277                                     },
2278                                 }
2279                             }
2280                         }
2281                     },
2282                 }
2283             ]
2284         }
2285     }
2286
2287     result = create_router_bgp(tgen, topo, input_dict_2)
2288     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
2289
2290     step(
2291         "Verify on router Edge-1 that EVPN routes are preferred via"
2292         " DCG-1 or DCG-2 based on best path selection criteria "
2293         "(according to the configured BGP attribute values in route-map)."
2294     )
2295
2296     for addr_type in ADDR_TYPES:
2297         input_routes = {
2298             "r3": {
2299                 "static_routes": [
2300                     {
2301                         "network": NETWORK1_2[addr_type],
2302                         "next_hop": NEXT_HOP_IP[addr_type],
2303                         "vrf": "RED",
2304                     }
2305                 ]
2306             },
2307             "r4": {
2308                 "static_routes": [
2309                     {
2310                         "network": NETWORK1_3[addr_type],
2311                         "next_hop": NEXT_HOP_IP[addr_type],
2312                         "vrf": "BLUE",
2313                     },
2314                     {
2315                         "network": NETWORK1_4[addr_type],
2316                         "next_hop": NEXT_HOP_IP[addr_type],
2317                         "vrf": "GREEN",
2318                     },
2319                 ]
2320             },
2321         }
2322
2323         result = verify_best_path_as_per_bgp_attribute(
2324             tgen, addr_type, "e1", input_routes, attribute
2325         )
2326         assert result is True, "Testcase {} : Failed \n Error: {}".format(
2327             tc_name, result
2328         )
2329
2330     write_test_footer(tc_name)
2331
2332
2333 if __name__ == "__main__":
2334     args = ["-s"] + sys.argv[1:]
2335     sys.exit(pytest.main(args))