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 re
  38 import sys
  39 import json
  40 import time
  41 import pytest
  42 import platform
  43 from copy import deepcopy
  44 from time import sleep
  45
  46
  47 # Save the Current Working Directory to find configuration files.
  48 CWD = os.path.dirname(os.path.realpath(__file__))
  49 sys.path.append(os.path.join(CWD, "../"))
  50 sys.path.append(os.path.join(CWD, "../lib/"))
  51
  52 # Required to instantiate the topology builder class.
  53
  54 # pylint: disable=C0413
  55 # Import topogen and topotest helpers
  56 from lib.topotest import version_cmp
  57 from lib.topogen import Topogen, get_topogen
  58 from mininet.topo import Topo
  59
  60 from lib.common_config import (
  61     start_topology,
  62     write_test_header,
  63     check_address_types,
  64     write_test_footer,
  65     reset_config_on_routers,
  66     verify_rib,
  67     step,
  68     create_route_maps,
  69     verify_cli_json,
  70     start_router_daemons,
  71     create_static_routes,
  72     stop_router,
  73     start_router,
  74     create_vrf_cfg,
  75     check_router_status,
  76     apply_raw_config,
  77     configure_vxlan,
  78     configure_brctl,
  79     verify_vrf_vni,
  80     create_interface_in_kernel
  81 )
  82
  83 from lib.topolog import logger
  84 from lib.bgp import (
  85     verify_bgp_convergence,
  86     create_router_bgp,
  87     clear_bgp,
  88     verify_best_path_as_per_bgp_attribute,
  89     verify_attributes_for_evpn_routes,
  90     verify_evpn_routes
  91 )
  92 from lib.topojson import build_topo_from_json, build_config_from_json
  93
  94 # Reading the data from JSON File for topology creation
  95 jsonFile = "{}/evpn_type5_topo1.json".format(CWD)
  96 try:
  97     with open(jsonFile, "r") as topoJson:
  98         topo = json.load(topoJson)
  99 except IOError:
 100     assert False, "Could not read file {}".format(jsonFile)
 101
 102 # Global variables
 103 NETWORK1_1 = {"ipv4": "10.1.1.1/32", "ipv6": "10::1/128"}
 104 NETWORK1_2 = {"ipv4": "40.1.1.1/32", "ipv6": "40::1/128"}
 105 NETWORK1_3 = {"ipv4": "40.1.1.2/32", "ipv6": "40::2/128"}
 106 NETWORK1_4 = {"ipv4": "40.1.1.3/32", "ipv6": "40::3/128"}
 107 NETWORK2_1 = {"ipv4": "20.1.1.1/32", "ipv6": "20::1/128"}
 108 NETWORK3_1 = {"ipv4": "30.1.1.1/32", "ipv6": "30::1/128"}
 109 NETWORK4_1 = {"ipv4": "100.1.1.1/32 ", "ipv6": "100::100/128"}
 110 NEXT_HOP_IP = {"ipv4": "Null0", "ipv6": "Null0"}
 111 VNI_1 = 75100
 112 VNI_2 = 75200
 113 VNI_3 = 75300
 114 MAC_1 = "00:80:48:ba:d1:00"
 115 MAC_2 = "00:80:48:ba:d1:01"
 116 MAC_3 = "00:80:48:ba:d1:02"
 117 BRCTL_1 = "br100"
 118 BRCTL_2 = "br200"
 119 BRCTL_3 = "br300"
 120 VXLAN_1 = "vxlan75100"
 121 VXLAN_2 = "vxlan75200"
 122 VXLAN_3 = "vxlan75300"
 123 BRIDGE_INTF1 = "120.0.0.1"
 124 BRIDGE_INTF2 = "120.0.0.2"
 125 BRIDGE_INTF3 = "120.0.0.3"
 126
 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 {} on {} :Failed \n Error: {}".format(
 253             tc_name, dut, 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 {} on {} :Failed \n Error: {}".format(
 271             tc_name, dut, 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_RD_verification_manual_and_auto_p0(request):
 345     """
 346     RD verification (manual/auto).
 347     """
 348
 349     tgen = get_topogen()
 350     tc_name = request.node.name
 351     write_test_header(tc_name)
 352     check_router_status(tgen)
 353     reset_config_on_routers(tgen)
 354     add_default_routes(tgen)
 355
 356     if tgen.routers_have_failure():
 357         pytest.skip(tgen.errors)
 358
 359     step(
 360         "Advertise prefixes from VNF routers R1 and R2 in associated "
 361         "VRFs for both address-family."
 362     )
 363     step(
 364         "Advertise vrf RED's routes in EVPN address family from Edge-1 router"
 365         ", without manual configuration of RD."
 366     )
 367
 368     for addr_type in ADDR_TYPES:
 369         input_dict_1 = {
 370             "r1": {
 371                 "static_routes": [
 372                     {
 373                         "network": NETWORK1_1[addr_type],
 374                         "next_hop": NEXT_HOP_IP[addr_type],
 375                         "vrf": "RED",
 376                     }
 377                 ]
 378             },
 379             "r2": {
 380                 "static_routes": [
 381                     {
 382                         "network": NETWORK2_1[addr_type],
 383                         "next_hop": NEXT_HOP_IP[addr_type],
 384                         "vrf": "BLUE",
 385                     },
 386                     {
 387                         "network": NETWORK3_1[addr_type],
 388                         "next_hop": NEXT_HOP_IP[addr_type],
 389                         "vrf": "GREEN",
 390                     }
 391                 ]
 392             }
 393         }
 394
 395         result = create_static_routes(tgen, input_dict_1)
 396         assert result is True, "Testcase {} : Failed \n Error: {}".format(
 397             tc_name, result
 398         )
 399
 400     step("Verify on DCG-1 and DCG-2:")
 401     step("EVPN route for 10.1.1.1/32 has auto-assigned RD value.")
 402
 403     for dut in ["d1", "d2"]:
 404         input_routes = {key: topo["routers"][key] for key in ["r1"]}
 405         result = verify_attributes_for_evpn_routes(
 406             tgen, topo, dut, input_routes, rd="auto", rd_peer="e1"
 407         )
 408         assert result is True, "Testcase {} on {} :Failed \n Error: {}".format(
 409             tc_name, dut, result
 410         )
 411
 412     step(
 413         "Configure RD for vrf RED manually as 50.50.50.50:50 and "
 414         "advertise vrf RED's routes in EVPN address family from "
 415         "Edge-1 router."
 416     )
 417
 418     input_dict_rd = {
 419         "e1": {
 420             "bgp": [
 421                 {
 422                     "local_as": "100",
 423                     "vrf": "RED",
 424                     "address_family": {"l2vpn": {"evpn": {"rd": "50.50.50.50:50"}}},
 425                 }
 426             ]
 427         }
 428     }
 429
 430     result = create_router_bgp(tgen, topo, input_dict_rd)
 431     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 432
 433     step("EVPN route for vrf RED has RD value as 50.50.50.50:50")
 434     for dut in ["d1", "d2"]:
 435         input_routes = {key: topo["routers"][key] for key in ["r1"]}
 436         result = verify_attributes_for_evpn_routes(
 437             tgen, topo, dut, input_routes, rd="50.50.50.50:50"
 438         )
 439         assert result is True, "Testcase {} on {} :Failed \n Error: {}".format(
 440             tc_name, dut, result
 441         )
 442
 443     step(
 444         "Configure RD for vrf RED manually as 100.100.100.100:100 and "
 445         "advertise vrf RED's routes in EVPN address family from Edge-1 "
 446         "router."
 447     )
 448     input_dict_rd = {
 449         "e1": {
 450             "bgp": [
 451                 {
 452                     "local_as": "100",
 453                     "vrf": "RED",
 454                     "address_family": {
 455                         "l2vpn": {"evpn": {"rd": "100.100.100.100:100"}}
 456                     }
 457                 }
 458             ]
 459         }
 460     }
 461
 462     result = create_router_bgp(tgen, topo, input_dict_rd)
 463     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 464
 465     step(
 466         "EVPN route for vrf RED is overridden with RD value as " "100.100.100.100:100."
 467     )
 468
 469     for dut in ["d1", "d2"]:
 470         input_routes = {key: topo["routers"][key] for key in ["r1"]}
 471         result = verify_attributes_for_evpn_routes(
 472             tgen, topo, dut, input_routes, rd="100.100.100.100:100"
 473         )
 474         assert result is True, "Testcase {} on {} :Failed \n Error: {}".format(
 475             tc_name, dut, result
 476         )
 477
 478     step(
 479         "Configure RD for vrf BLUE manually same as vrf RED "
 480         "(100.100.100.100:100) and advertise vrf RED and BLUE's routes "
 481         "in EVPN address family from Edge-1 router."
 482     )
 483
 484     input_dict_rd = {
 485         "e1": {
 486             "bgp": [
 487                 {
 488                     "local_as": "100",
 489                     "vrf": "BLUE",
 490                     "address_family": {
 491                         "l2vpn": {"evpn": {"rd": "100.100.100.100:100"}}
 492                     },
 493                 }
 494             ]
 495         }
 496     }
 497
 498     result = create_router_bgp(tgen, topo, input_dict_rd)
 499     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 500
 501     step(
 502         "Delete manually configured RD and advertise vrf RED's routes "
 503         "in EVPN address family from Edge-1 router."
 504     )
 505
 506     input_dict_rd = {
 507         "e1": {
 508             "bgp": [
 509                 {
 510                     "local_as": "100",
 511                     "vrf": "RED",
 512                     "address_family": {
 513                         "l2vpn": {"evpn": {"no rd": "100.100.100.100:100"}}
 514                     },
 515                 }
 516             ]
 517         }
 518     }
 519
 520     result = create_router_bgp(tgen, topo, input_dict_rd)
 521     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 522
 523     step(
 524         "Configure same RD value for vrf GREEN, as auto generated RD "
 525         "value for vrf RED on Edge-1 router."
 526     )
 527
 528     input_dict_rd = {
 529         "e1": {
 530             "bgp": [
 531                 {
 532                     "local_as": "100",
 533                     "vrf": "GREEN",
 534                     "address_family": {"l2vpn": {"evpn": {"rd": "10.0.0.33:1"}}},
 535                 }
 536             ]
 537         }
 538     }
 539
 540     result = create_router_bgp(tgen, topo, input_dict_rd)
 541     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 542
 543     step("Delete auto configured RD value from vrf RED in EVPN " "address family.")
 544
 545     input_dict_rd = {
 546         "e1": {
 547             "bgp": [
 548                 {
 549                     "local_as": "100",
 550                     "vrf": "GREEN",
 551                     "address_family": {"l2vpn": {"evpn": {"no rd": "10.0.0.33:1"}}},
 552                 }
 553             ]
 554         }
 555     }
 556
 557     result = create_router_bgp(tgen, topo, input_dict_rd)
 558     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 559
 560     step("Configure RD value as 100.100.100:100")
 561
 562     input_dict_rd = {
 563         "e1": {
 564             "bgp": [
 565                 {
 566                     "local_as": "100",
 567                     "vrf": "GREEN",
 568                     "address_family": {"l2vpn": {"evpn": {"rd": "100.100.100:100"}}},
 569                 }
 570             ]
 571         }
 572     }
 573
 574     result = create_router_bgp(tgen, topo, input_dict_rd)
 575     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 576
 577     write_test_footer(tc_name)
 578
 579
 580 def test_RT_verification_manual_p0(request):
 581     """
 582     RT verification(manual)
 583     """
 584
 585     tgen = get_topogen()
 586     tc_name = request.node.name
 587     write_test_header(tc_name)
 588     check_router_status(tgen)
 589     reset_config_on_routers(tgen)
 590     add_default_routes(tgen)
 591
 592     if tgen.routers_have_failure():
 593         pytest.skip(tgen.errors)
 594
 595     step(
 596         "Advertise prefixes from VNF routers R1 and R2 in associated "
 597         "VRFs for both address-family."
 598     )
 599     step("Advertise VRF routes as in EVPN address family from Edge-1 " "router.")
 600
 601     for addr_type in ADDR_TYPES:
 602         input_dict_1 = {
 603             "r1": {
 604                 "static_routes": [
 605                     {
 606                         "network": NETWORK1_1[addr_type],
 607                         "next_hop": NEXT_HOP_IP[addr_type],
 608                         "vrf": "RED",
 609                     }
 610                 ]
 611             },
 612             "r2": {
 613                 "static_routes": [
 614                     {
 615                         "network": NETWORK2_1[addr_type],
 616                         "next_hop": NEXT_HOP_IP[addr_type],
 617                         "vrf": "BLUE",
 618                     },
 619                     {
 620                         "network": NETWORK3_1[addr_type],
 621                         "next_hop": NEXT_HOP_IP[addr_type],
 622                         "vrf": "GREEN",
 623                     }
 624                 ]
 625             }
 626         }
 627
 628         result = create_static_routes(tgen, input_dict_1)
 629         assert result is True, "Testcase {} : Failed \n Error: {}".format(
 630             tc_name, result
 631         )
 632
 633     step(
 634         "Configure RT for vrf RED manually as export 100:100 "
 635         "and advertise vrf RED's routes in EVPN address family"
 636         " from Edge-1 router."
 637     )
 638
 639     input_dict_rt = {
 640         "e1": {
 641             "bgp": [
 642                 {
 643                     "local_as": "100",
 644                     "vrf": "RED",
 645                     "address_family": {
 646                         "ipv4": {
 647                             "unicast": {"neighbor": {"r1": {"dest_link": {"e1": {}}}}}
 648                         },
 649                         "ipv6": {
 650                             "unicast": {"neighbor": {"r1": {"dest_link": {"e1": {}}}}}
 651                         },
 652                         "l2vpn": {
 653                             "evpn": {"route-target": {"export": [{"value": "100:100"}]}}
 654                         },
 655                     }
 656                 }
 657             ]
 658         }
 659     }
 660
 661     result = create_router_bgp(tgen, topo, input_dict_rt)
 662     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 663
 664     step(
 665         "Verify on dcg-1 and dcg-2, EVPN route for 10.1.1.1/32"
 666         " and 10::1/128 have RT value as 100:100."
 667     )
 668
 669     for dut in ["d1", "d2"]:
 670         input_routes = {key: topo["routers"][key] for key in ["r1"]}
 671         result = verify_attributes_for_evpn_routes(
 672             tgen, topo, dut, input_routes, rt="100:100"
 673         )
 674         assert result is True, "Testcase {} on {} :Failed \n Error: {}".format(
 675             tc_name, dut, result
 676         )
 677
 678     step(
 679         "Configure RT for vrf RED manually as export 500:500 and"
 680         " advertise vrf RED's routes in EVPN address family from"
 681         " e1 router."
 682     )
 683
 684     input_dict_rt = {
 685         "e1": {
 686             "bgp": [
 687                 {
 688                     "local_as": "100",
 689                     "vrf": "RED",
 690                     "address_family": {
 691                         "l2vpn": {
 692                             "evpn": {"route-target": {"export": [{"value": "500:500"}]}}
 693                         }
 694                     },
 695                 }
 696             ]
 697         }
 698     }
 699
 700     result = create_router_bgp(tgen, topo, input_dict_rt)
 701     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 702
 703     step(
 704         "Verify on dcg-1 and dcg-2, EVPN route for 10.1.1.1/32"
 705         " and 10::1/128 have RT value as 500:500."
 706     )
 707
 708     for dut in ["d1", "d2"]:
 709         input_routes = {key: topo["routers"][key] for key in ["r1"]}
 710         result = verify_attributes_for_evpn_routes(
 711             tgen, topo, dut, input_routes, rt=["100:100", "500:500"]
 712         )
 713         assert result is True, "Testcase {} on {} :Failed \n Error: {}".format(
 714             tc_name, dut, result
 715         )
 716
 717     step(
 718         "Import RT value 100:100 and 500:500 in vrf BLUE manually on"
 719         " peer router DCG-1 and DCG-2."
 720     )
 721
 722     input_dict_rt = {
 723         "d1": {
 724             "bgp": [
 725                 {
 726                     "local_as": "100",
 727                     "vrf": "BLUE",
 728                     "address_family": {
 729                         "l2vpn": {
 730                             "evpn": {
 731                                 "route-target": {
 732                                     "import": [
 733                                         {"value": "100:100"},
 734                                         {"value": "500:500"},
 735                                     ]
 736                                 }
 737                             }
 738                         }
 739                     },
 740                 }
 741             ]
 742         },
 743         "d2": {
 744             "bgp": [
 745                 {
 746                     "local_as": "200",
 747                     "vrf": "BLUE",
 748                     "address_family": {
 749                         "l2vpn": {
 750                             "evpn": {
 751                                 "route-target": {
 752                                     "import": [
 753                                         {"value": "100:100"},
 754                                         {"value": "500:500"},
 755                                     ]
 756                                 }
 757                             }
 758                         }
 759                     },
 760                 }
 761             ]
 762         },
 763     }
 764
 765     result = create_router_bgp(tgen, topo, input_dict_rt)
 766     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 767
 768     step(
 769         "EVPN route for 10.1.1.1/32 and 10::1 should be installed "
 770         "in vrf BLUE on DCG-1 and DCG-2 and further advertised to "
 771         "VNF router."
 772     )
 773
 774     for addr_type in ADDR_TYPES:
 775         input_routes = {
 776             "r1": {
 777                 "static_routes": [{"network": [NETWORK1_1[addr_type]], "vrf": "BLUE"}]
 778             }
 779         }
 780         result = verify_rib(tgen, addr_type, "d1", input_routes)
 781         assert result is True, "Testcase {} :Failed \n Error {}".format(tc_name, result)
 782
 783         result = verify_rib(tgen, addr_type, "d2", input_routes)
 784         assert result is True, "Testcase {} :Failed \n Error {}".format(tc_name, result)
 785
 786     step(
 787         "Delete import RT value 500:500 in vrf BLUE manually on "
 788         "peer router DCG-1 and DCG-2."
 789     )
 790
 791     input_dict_rt = {
 792         "d1": {
 793             "bgp": [
 794                 {
 795                     "local_as": "100",
 796                     "vrf": "BLUE",
 797                     "address_family": {
 798                         "l2vpn": {
 799                             "evpn": {
 800                                 "route-target": {
 801                                     "import": [{"value": "500:500", "delete": True}]
 802                                 }
 803                             }
 804                         }
 805                     },
 806                 }
 807             ]
 808         },
 809         "d2": {
 810             "bgp": [
 811                 {
 812                     "local_as": "200",
 813                     "vrf": "BLUE",
 814                     "address_family": {
 815                         "l2vpn": {
 816                             "evpn": {
 817                                 "route-target": {
 818                                     "import": [{"value": "500:500", "delete": True}]
 819                                 }
 820                             }
 821                         }
 822                     },
 823                 }
 824             ]
 825         },
 826     }
 827
 828     result = create_router_bgp(tgen, topo, input_dict_rt)
 829     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 830
 831     for dut in ["d1", "d2"]:
 832         input_routes = {key: topo["routers"][key] for key in ["r1"]}
 833         result = verify_attributes_for_evpn_routes(
 834             tgen, topo, dut, input_routes, rt=["100:100", "500:500"]
 835         )
 836         assert result is True, "Testcase {} on {} :Failed \n Error: {}".format(
 837             tc_name, dut, result
 838         )
 839
 840     step("Delete RT export value 100:100 for vrf RED on Edge-1")
 841
 842     input_dict_rt = {
 843         "e1": {
 844             "bgp": [
 845                 {
 846                     "local_as": "100",
 847                     "vrf": "RED",
 848                     "address_family": {
 849                         "l2vpn": {
 850                             "evpn": {
 851                                 "route-target": {
 852                                     "export": [{"value": "100:100", "delete": True}]
 853                                 }
 854                             }
 855                         }
 856                     },
 857                 }
 858             ]
 859         }
 860     }
 861
 862     result = create_router_bgp(tgen, topo, input_dict_rt)
 863     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 864
 865     step(
 866         "EVPN route for 10.1.1.1/32 and 10::1 should be withdrawn "
 867         "from vrf BLUE on DCG-1,DCG-2 and VNF router."
 868     )
 869
 870     for addr_type in ADDR_TYPES:
 871         input_routes = {
 872             "r1": {
 873                 "static_routes": [{"network": [NETWORK1_1[addr_type]], "vrf": "BLUE"}]
 874             }
 875         }
 876         result = verify_rib(tgen, addr_type, "d1", input_routes, expected=False)
 877         assert result is not True, (
 878             "Testcase {} :Failed \n Expected Behavior: Routes are still "
 879             "present \n Error: {}".format(tc_name, result)
 880         )
 881         logger.info("Expected Behavior: {}".format(result))
 882
 883         result = verify_rib(tgen, addr_type, "d2", input_routes, expected=False)
 884         assert result is not True, (
 885             "Testcase {} :Failed \n Expected Behavior: Routes are still "
 886             "present \n Error: {}".format(tc_name, result)
 887         )
 888         logger.info("Expected Behavior: {}".format(result))
 889
 890     step(
 891         "Configure RT value as 100:100000010000010000101010 to check "
 892         "the boundary value."
 893     )
 894
 895     input_dict_rt = {
 896         "e1": {
 897             "bgp": [
 898                 {
 899                     "local_as": "100",
 900                     "vrf": "RED",
 901                     "address_family": {
 902                         "l2vpn": {
 903                             "evpn": {
 904                                 "route-target": {
 905                                     "export": [
 906                                         {"value": "100:100000010000010000101010"}
 907                                     ]
 908                                 }
 909                             }
 910                         }
 911                     },
 912                 }
 913             ]
 914         }
 915     }
 916
 917     result = create_router_bgp(tgen, topo, input_dict_rt)
 918     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 919
 920     step(
 921         "CLI error: RT value: 100:100000010000010000101010 should not " "be configured"
 922     )
 923
 924     dut = "e1"
 925     input_routes = {key: topo["routers"][key] for key in ["r1"]}
 926     result = verify_attributes_for_evpn_routes(
 927         tgen, topo, dut, input_routes, rt="100:100000010000010000101010", expected=False
 928     )
 929     assert result is not True, (
 930         "Testcase {} :Failed \n Expected Behavior: RT value of out"
 931         " of boundary \n Error: {}".format(tc_name, result)
 932     )
 933     logger.info("Expected Behavior: {}".format(result))
 934
 935     write_test_footer(tc_name)
 936
 937
 938 def test_active_standby_evpn_implementation_p1(request):
 939     """
 940     In an active/standby EVPN implementation, if active DCG goes down,
 941     secondary takes over.
 942     """
 943
 944     tgen = get_topogen()
 945     tc_name = request.node.name
 946     write_test_header(tc_name)
 947     check_router_status(tgen)
 948     reset_config_on_routers(tgen)
 949     add_default_routes(tgen)
 950
 951     if tgen.routers_have_failure():
 952         pytest.skip(tgen.errors)
 953
 954     step(
 955         "Taken care in base config: Configure BGP neighborship for both "
 956         "address families(IPv4 & IPv6) between DCG-1/DCG-2 and VFN routers"
 957         "(R3 and R4)."
 958     )
 959
 960     step(
 961         "BGP neighborships come up within defined VRFs. Please use below "
 962         "command: sh bgp vrf all summary"
 963     )
 964
 965     result = verify_bgp_convergence(tgen, topo, "d1")
 966     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 967
 968     result = verify_bgp_convergence(tgen, topo, "d2")
 969     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
 970
 971     step(
 972         "Advertise prefixes from VNF routers R3 and R4 in associated "
 973         "VRFs for both address-families."
 974     )
 975
 976     for addr_type in ADDR_TYPES:
 977         input_dict_1 = {
 978             "r3": {
 979                 "static_routes": [
 980                     {
 981                         "network": NETWORK1_2[addr_type],
 982                         "next_hop": NEXT_HOP_IP[addr_type],
 983                         "vrf": "RED",
 984                     }
 985                 ]
 986             },
 987             "r4": {
 988                 "static_routes": [
 989                     {
 990                         "network": NETWORK1_3[addr_type],
 991                         "next_hop": NEXT_HOP_IP[addr_type],
 992                         "vrf": "BLUE",
 993                     },
 994                     {
 995                         "network": NETWORK1_4[addr_type],
 996                         "next_hop": NEXT_HOP_IP[addr_type],
 997                         "vrf": "GREEN",
 998                     }
 999                 ]
1000             }
1001         }
1002
1003         result = create_static_routes(tgen, input_dict_1)
1004         assert result is True, "Testcase {} : Failed \n Error: {}".format(
1005             tc_name, result
1006         )
1007
1008     step(
1009         "Redistribute static in (IPv4 and IPv6) address-family "
1010         "on Edge-1 for all VRFs."
1011     )
1012
1013     input_dict_2 = {}
1014     for dut in ["r3", "r4"]:
1015         temp = {dut: {"bgp": []}}
1016         input_dict_2.update(temp)
1017
1018         if dut == "r3":
1019             VRFS = ["RED"]
1020             AS_NUM = [3]
1021         if dut == "r4":
1022             VRFS = ["BLUE", "GREEN"]
1023             AS_NUM = [4, 4]
1024
1025         for vrf, as_num in zip(VRFS, AS_NUM):
1026             temp[dut]["bgp"].append(
1027                 {
1028                     "local_as": as_num,
1029                     "vrf": vrf,
1030                     "address_family": {
1031                         "ipv4": {
1032                             "unicast": {"redistribute": [{"redist_type": "static"}]}
1033                         },
1034                         "ipv6": {
1035                             "unicast": {"redistribute": [{"redist_type": "static"}]}
1036                         },
1037                     },
1038                 }
1039             )
1040
1041     result = create_router_bgp(tgen, topo, input_dict_2)
1042     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
1043
1044     step("Prefixes are received in respective VRFs on DCG-1/DCG-2.")
1045
1046     for addr_type in ADDR_TYPES:
1047         input_routes = {
1048             "r3": {
1049                 "static_routes": [
1050                     {
1051                         "network": NETWORK1_2[addr_type],
1052                         "next_hop": NEXT_HOP_IP[addr_type],
1053                         "vrf": "RED",
1054                     }
1055                 ]
1056             },
1057             "r4": {
1058                 "static_routes": [
1059                     {
1060                         "network": NETWORK1_3[addr_type],
1061                         "next_hop": NEXT_HOP_IP[addr_type],
1062                         "vrf": "BLUE",
1063                     },
1064                     {
1065                         "network": NETWORK1_4[addr_type],
1066                         "next_hop": NEXT_HOP_IP[addr_type],
1067                         "vrf": "GREEN",
1068                     },
1069                 ]
1070             },
1071         }
1072
1073         result = verify_rib(tgen, addr_type, "d1", input_routes)
1074         assert result is True, "Testcase {} :Failed \n Error: {}".format(
1075             tc_name, result
1076         )
1077
1078         result = verify_rib(tgen, addr_type, "d2", input_routes)
1079         assert result is True, "Testcase {} :Failed \n Error: {}".format(
1080             tc_name, result
1081         )
1082
1083     step(
1084         "Taken care in base config: Advertise VRF routes in EVPN "
1085         "address-family from DCG-1 and DCG-2 router."
1086     )
1087
1088     step("Verify on Edge-1 that EVPN routes are installed via next-hop " "as DCG-2.")
1089
1090     for addr_type in ADDR_TYPES:
1091         input_routes = {
1092             "r3": {
1093                 "static_routes": [
1094                     {
1095                         "network": NETWORK1_2[addr_type],
1096                         "next_hop": NEXT_HOP_IP[addr_type],
1097                         "vrf": "RED",
1098                     }
1099                 ]
1100             },
1101             "r4": {
1102                 "static_routes": [
1103                     {
1104                         "network": NETWORK1_3[addr_type],
1105                         "next_hop": NEXT_HOP_IP[addr_type],
1106                         "vrf": "BLUE",
1107                     },
1108                     {
1109                         "network": NETWORK1_4[addr_type],
1110                         "next_hop": NEXT_HOP_IP[addr_type],
1111                         "vrf": "GREEN",
1112                     },
1113                 ]
1114             },
1115         }
1116
1117         if addr_type == "ipv4":
1118             result = verify_rib(
1119                 tgen, addr_type, "e1", input_routes, next_hop=BRIDGE_INTF2
1120             )
1121             assert result is True, "Testcase {} :Failed \n Error: {}".format(
1122                 tc_name, result
1123             )
1124         else:
1125             result = verify_rib(tgen, addr_type, "e1", input_routes)
1126             assert result is True, "Testcase {} :Failed \n Error: {}".format(
1127                 tc_name, result
1128             )
1129
1130     step(
1131         "Configure 'next-hop self' on DCG-1 for peer Edge-1 in EVPN " "address-family."
1132     )
1133
1134     input_dict_3 = {
1135         "d1": {
1136             "bgp": [
1137                 {
1138                     "local_as": "100",
1139                     "address_family": {
1140                         "l2vpn": {
1141                             "evpn": {
1142                                 "neighbor": {
1143                                     "e1": {
1144                                         "ipv4": {"d1-link1": {"next_hop_self": True}}
1145                                     }
1146                                 }
1147                             }
1148                         }
1149                     },
1150                 }
1151             ]
1152         }
1153     }
1154
1155     result = create_router_bgp(tgen, topo, input_dict_3)
1156     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
1157
1158     logger.info(
1159         "Creating route-map so ipv6 glpbal ip wpuld be preferred " "as next-hop"
1160     )
1161
1162     step(
1163         "Verify on Edge-1 that EVPN routes are now preferred via "
1164         "next-hop as DCG-1(iBGP) due to shortest AS-Path."
1165     )
1166
1167     for addr_type in ADDR_TYPES:
1168
1169         logger.info("Verifying only ipv4 routes")
1170         if addr_type != "ipv4":
1171             continue
1172
1173         input_routes = {
1174             "r3": {
1175                 "static_routes": [
1176                     {
1177                         "network": NETWORK1_2[addr_type],
1178                         "next_hop": NEXT_HOP_IP[addr_type],
1179                         "vrf": "RED",
1180                     }
1181                 ]
1182             },
1183             "r4": {
1184                 "static_routes": [
1185                     {
1186                         "network": NETWORK1_3[addr_type],
1187                         "next_hop": NEXT_HOP_IP[addr_type],
1188                         "vrf": "BLUE",
1189                     },
1190                     {
1191                         "network": NETWORK1_4[addr_type],
1192                         "next_hop": NEXT_HOP_IP[addr_type],
1193                         "vrf": "GREEN",
1194                     },
1195                 ]
1196             },
1197         }
1198
1199         next_hop = topo["routers"]["d1"]["links"]["e1-link1"]["ipv4"].split("/")[0]
1200
1201         result = verify_rib(tgen, addr_type, "e1", input_routes, next_hop=next_hop)
1202         assert result is True, "Testcase {} :Failed \n Error: {}".format(
1203             tc_name, result
1204         )
1205
1206     write_test_footer(tc_name)
1207
1208
1209 def test_evpn_routes_from_VNFs_p1(request):
1210     """
1211     EVPN routes are advertised/withdrawn, based on VNFs
1212     advertising/withdrawing IP prefixes.
1213     """
1214
1215     tgen = get_topogen()
1216     tc_name = request.node.name
1217     write_test_header(tc_name)
1218     check_router_status(tgen)
1219     reset_config_on_routers(tgen)
1220     add_default_routes(tgen)
1221
1222     if tgen.routers_have_failure():
1223         pytest.skip(tgen.errors)
1224
1225     step(
1226         "Advertise prefixes from VNF routers R1 and R2 in associated "
1227         "VRFs for both address-family."
1228     )
1229
1230     for addr_type in ADDR_TYPES:
1231         input_dict_1 = {
1232             "r1": {
1233                 "static_routes": [
1234                     {
1235                         "network": NETWORK1_1[addr_type],
1236                         "next_hop": NEXT_HOP_IP[addr_type],
1237                         "vrf": "RED",
1238                     }
1239                 ]
1240             },
1241             "r2": {
1242                 "static_routes": [
1243                     {
1244                         "network": NETWORK2_1[addr_type],
1245                         "next_hop": NEXT_HOP_IP[addr_type],
1246                         "vrf": "BLUE",
1247                     },
1248                     {
1249                         "network": NETWORK3_1[addr_type],
1250                         "next_hop": NEXT_HOP_IP[addr_type],
1251                         "vrf": "GREEN",
1252                     }
1253                 ]
1254             }
1255         }
1256
1257         result = create_static_routes(tgen, input_dict_1)
1258         assert result is True, "Testcase {} : Failed \n Error: {}".format(
1259             tc_name, result
1260         )
1261
1262     step(
1263         "Taken care in base config: Advertise VNFs'(R1 and R2) "
1264         "originated routes in EVPN address-family from Edge-1 to "
1265         "DCG-1 and DCG-2 routers."
1266     )
1267     step(
1268         "Taken care in base config: Advertise IPv4 and IPv6 routes "
1269         "from default vrf in EVPN address-family from Edge-1."
1270     )
1271
1272     step(
1273         "Verify on DCG-2 that VNF routes are received in respective "
1274         "VRFs along with auto derived RD/RT values 'show bgp l2vpn evpn'"
1275     )
1276     for dut in ["d1", "d2"]:
1277         input_routes = {key: topo["routers"][key] for key in ["r1"]}
1278         result = verify_evpn_routes(tgen, topo, dut, input_routes)
1279         assert result is True, "Testcase {} on {} :Failed \n Error: {}".format(
1280             tc_name, dut, result
1281         )
1282
1283         input_routes = {key: topo["routers"][key] for key in ["r2"]}
1284         result = verify_evpn_routes(tgen, topo, dut, input_routes)
1285         assert result is True, "Testcase {} on {} :Failed \n Error: {}".format(
1286             tc_name, dut, result
1287         )
1288
1289     step(
1290         "Verify on R3 and R4 that DCG-2 further advertises all EVPN "
1291         "routes to corresponding VRFs."
1292     )
1293     for addr_type in ADDR_TYPES:
1294         input_routes = {key: topo["routers"][key] for key in ["r1"]}
1295         result = verify_rib(tgen, addr_type, "r3", input_routes)
1296         assert result is True, "Testcase {} :Failed \n Error: {}".format(
1297             tc_name, result
1298         )
1299
1300     for addr_type in ADDR_TYPES:
1301         input_routes = {key: topo["routers"][key] for key in ["r2"]}
1302         result = verify_rib(tgen, addr_type, "r4", input_routes)
1303         assert result is True, "Testcase {} :Failed \n Error: {}".format(
1304             tc_name, result
1305         )
1306
1307     step(
1308         "Verify that DCG-2 receives EVPN routes associated to default "
1309         "VRF and install in default IP routing table as well."
1310     )
1311     for addr_type in ADDR_TYPES:
1312         input_routes = {key: topo["routers"][key] for key in ["r1"]}
1313         result = verify_rib(tgen, addr_type, "d2", input_routes, expected=False)
1314         assert result is True, "Testcase {} :Failed \n Error: {}".format(
1315             tc_name, result
1316         )
1317
1318     for addr_type in ADDR_TYPES:
1319         input_routes = {key: topo["routers"][key] for key in ["r2"]}
1320         result = verify_rib(tgen, addr_type, "d2", input_routes, expected=False)
1321         assert result is True, "Testcase {} :Failed \n Error: {}".format(
1322             tc_name, result
1323         )
1324
1325     step("Withdraw the IP prefixes from VFN(R1).")
1326     dut = "r1"
1327     input_dict_2 = {}
1328     static_routes = topo["routers"][dut]["static_routes"]
1329     for static_route in static_routes:
1330         static_route["delete"] = True
1331         temp = {dut: {"static_routes": [static_route]}}
1332         input_dict_2.update(temp)
1333
1334         result = create_static_routes(tgen, input_dict_2)
1335         assert result is True, "Testcase {} :Failed \n Error: {}".format(
1336             tc_name, result
1337         )
1338
1339     step(
1340         "Verify that DCG-2 removes EVPN routes corresponding to vrf RED and "
1341         "send an withdraw to VNF(R3) as well."
1342     )
1343     for addr_type in ADDR_TYPES:
1344         input_routes = {key: topo["routers"][key] for key in ["r1"]}
1345         result = verify_rib(tgen, addr_type, "d2", input_routes, expected=False)
1346         assert result is not True, "Testcase {} :Failed \n "
1347         "Routes are still present: {}".format(tc_name, result)
1348         logger.info("Expected Behavior: {}".format(result))
1349
1350     for addr_type in ADDR_TYPES:
1351         input_routes = {key: topo["routers"][key] for key in ["r1"]}
1352         result = verify_rib(tgen, addr_type, "r3", input_routes, expected=False)
1353         assert result is not True, "Testcase {} :Failed \n "
1354         "Routes are still present: {}".format(tc_name, result)
1355         logger.info("Expected Behavior: {}".format(result))
1356
1357     step("Re-advertise IP prefixes from VFN(R1).")
1358     step(
1359         "Advertise prefixes from VNF routers R1 and R2 in associated "
1360         "VRFs for both address-family."
1361     )
1362
1363     for addr_type in ADDR_TYPES:
1364         input_dict_1 = {
1365             "r1": {
1366                 "static_routes": [
1367                     {
1368                         "network": NETWORK1_1[addr_type],
1369                         "next_hop": NEXT_HOP_IP[addr_type],
1370                         "vrf": "RED",
1371                     }
1372                 ]
1373             },
1374             "r2": {
1375                 "static_routes": [
1376                     {
1377                         "network": NETWORK2_1[addr_type],
1378                         "next_hop": NEXT_HOP_IP[addr_type],
1379                         "vrf": "BLUE",
1380                     },
1381                     {
1382                         "network": NETWORK3_1[addr_type],
1383                         "next_hop": NEXT_HOP_IP[addr_type],
1384                         "vrf": "GREEN",
1385                     }
1386                 ]
1387             }
1388         }
1389
1390         result = create_static_routes(tgen, input_dict_1)
1391         assert result is True, "Testcase {} : Failed \n Error: {}".format(
1392             tc_name, result
1393         )
1394
1395     step(
1396         "Verify that DCG-2 receives EVPN routes corresponding to vrf RED "
1397         "again and send an update to VNF(R3) as well."
1398     )
1399     for addr_type in ADDR_TYPES:
1400         input_routes = {key: topo["routers"][key] for key in ["r1"]}
1401         result = verify_rib(tgen, addr_type, "d2", input_routes)
1402         assert result is True, "Testcase {} :Failed \n Error: {}".format(
1403             tc_name, result
1404         )
1405
1406     for addr_type in ADDR_TYPES:
1407         input_routes = {key: topo["routers"][key] for key in ["r1"]}
1408         result = verify_rib(tgen, addr_type, "r3", input_routes)
1409         assert result is True, "Testcase {} :Failed \n Error: {}".format(
1410             tc_name, result
1411         )
1412
1413     step("Delete vrf BLUE from router Edge-1")
1414     input_dict_3 = {"e1": {"vrfs": [{"name": "BLUE", "id": "2", "delete": True}]}}
1415
1416     result = create_vrf_cfg(tgen, input_dict_3)
1417     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
1418
1419     step(
1420         "Verify that DCG-2 removes EVPN routes corresponding to "
1421         "vrf BLUE and send an withdraw to VNF(R4) as well."
1422     )
1423     for addr_type in ADDR_TYPES:
1424         input_routes = {
1425             "r2": {"static_routes": [{"network": NETWORK2_1[addr_type], "vrf": "BLUE"}]}
1426         }
1427
1428         result = verify_rib(tgen, addr_type, "d2", input_routes, expected=False)
1429         assert result is not True, "Testcase {} :Failed \n "
1430         "Routes are still present: {}".format(tc_name, result)
1431         logger.info("Expected Behavior: {}".format(result))
1432
1433         result = verify_rib(tgen, addr_type, "r4", input_routes, expected=False)
1434         assert result is not True, "Testcase {} :Failed \n "
1435         "Routes are still present: {}".format(tc_name, result)
1436         logger.info("Expected Behavior: {}".format(result))
1437
1438     step("Add vrf BLUE on router Edge-1 again.")
1439     interface = topo["routers"]["e1"]["links"]["r2-link1"]["interface"]
1440     input_dict_3 = {
1441         "e1": {
1442             "links": {
1443                 "r2-link1": {
1444                     "interface": interface,
1445                     "ipv4": "auto",
1446                     "ipv6": "auto",
1447                     "vrf": "BLUE",
1448                 }
1449             },
1450             "vrfs": [{"name": "BLUE", "id": "2"}],
1451         }
1452     }
1453     result = create_vrf_cfg(tgen, input_dict_3)
1454     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
1455
1456     logger.info(
1457         "After deleting VRFs ipv6 addresses wil be deleted "
1458         "from kernel Adding back ipv6 addresses"
1459     )
1460     dut = "e1"
1461     vrfs = ["BLUE"]
1462
1463     for vrf in vrfs:
1464         for c_link, c_data in topo["routers"][dut]["links"].items():
1465             if "vrf" in c_data:
1466                 if c_data["vrf"] != vrf:
1467                     continue
1468
1469                 intf_name = c_data["interface"]
1470                 intf_ipv6 = c_data["ipv6"]
1471
1472                 create_interface_in_kernel(
1473                     tgen, dut, intf_name, intf_ipv6, vrf, create=False
1474                 )
1475
1476     logger.info("Wait for 60 sec.")
1477     sleep(60)
1478
1479     step(
1480         "Verify that DCG-2 receives EVPN routes corresponding to "
1481         "vrf BLUE again and send an update to VNF(R4) as well."
1482     )
1483     for addr_type in ADDR_TYPES:
1484         input_routes = {
1485             "r2": {"static_routes": [{"network": NETWORK2_1[addr_type], "vrf": "BLUE"}]}
1486         }
1487
1488         result = verify_rib(tgen, addr_type, "d2", input_routes)
1489         assert result is True, "Testcase {} :Failed \n Error: {}".format(
1490             tc_name, result
1491         )
1492
1493         result = verify_rib(tgen, addr_type, "r4", input_routes)
1494         assert result is True, "Testcase {} :Failed \n Error: {}".format(
1495             tc_name, result
1496         )
1497
1498     step("Withdraw IPv6 address-family in EVPN advertisements for " "VRF GREEN")
1499     addr_type = "ipv6"
1500     input_dict_4 = {
1501         "e1": {
1502             "bgp": [
1503                 {
1504                     "local_as": "100",
1505                     "vrf": "GREEN",
1506                     "address_family": {
1507                         "l2vpn": {
1508                             "evpn": {
1509                                 "advertise": {addr_type: {"unicast": {"delete": True}}}
1510                             }
1511                         }
1512                     },
1513                 }
1514             ]
1515         }
1516     }
1517
1518     result = create_router_bgp(tgen, topo, input_dict_4)
1519     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
1520
1521     step(
1522         "Verify that EVPN routes (IPv6)associated with vrf GREEN are "
1523         "withdrawn from DCG-2 and VNF R4."
1524     )
1525     input_routes = {
1526         "r2": {"static_routes": [{"network": NETWORK3_1[addr_type], "vrf": "GREEN"}]}
1527     }
1528
1529     result = verify_rib(tgen, addr_type, "d2", input_routes, expected=False)
1530     assert result is not True, "Testcase {} :Failed \n "
1531     "Routes are still present: {}".format(tc_name, result)
1532     logger.info("Expected Behavior: {}".format(result))
1533
1534     result = verify_rib(tgen, addr_type, "r4", input_routes, expected=False)
1535     assert result is not True, "Testcase {} :Failed \n "
1536     "Routes are still present: {}".format(tc_name, result)
1537     logger.info("Expected Behavior: {}".format(result))
1538
1539     step("Advertise IPv6 address-family in EVPN advertisements " "for VRF GREEN.")
1540     addr_type = "ipv6"
1541     input_dict_4 = {
1542         "e1": {
1543             "bgp": [
1544                 {
1545                     "local_as": "100",
1546                     "vrf": "GREEN",
1547                     "address_family": {
1548                         "l2vpn": {"evpn": {"advertise": {addr_type: {"unicast": {}}}}}
1549                     },
1550                 }
1551             ]
1552         }
1553     }
1554
1555     result = create_router_bgp(tgen, topo, input_dict_4)
1556     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
1557
1558     for addr_type in ADDR_TYPES:
1559         input_routes = {
1560             "r2": {
1561                 "static_routes": [{"network": NETWORK3_1[addr_type], "vrf": "GREEN"}]
1562             }
1563         }
1564
1565         result = verify_rib(tgen, addr_type, "d2", input_routes)
1566         assert result is True, "Testcase {} :Failed \n Error: {}".format(
1567             tc_name, result
1568         )
1569
1570         result = verify_rib(tgen, addr_type, "r4", input_routes)
1571         assert result is True, "Testcase {} :Failed \n Error: {}".format(
1572             tc_name, result
1573         )
1574
1575     write_test_footer(tc_name)
1576
1577
1578 @pytest.mark.parametrize(
1579     "attribute", [{"route-type": "prefix"}, {"vni": VNI_1}, {"rt": "300:300"}]
1580 )
1581 def test_route_map_operations_for_evpn_address_family_p1(request, attribute):
1582     """
1583     Route-map operations for EVPN address family.
1584     """
1585
1586     tgen = get_topogen()
1587     tc_name = request.node.name
1588     write_test_header(tc_name)
1589     check_router_status(tgen)
1590     reset_config_on_routers(tgen)
1591     add_default_routes(tgen)
1592
1593     step(
1594         "Advertise prefixes from VNF routers R1 and R2 in associated "
1595         "VRFs for both address-family."
1596     )
1597
1598     for addr_type in ADDR_TYPES:
1599         input_dict_1 = {
1600             "r1": {
1601                 "static_routes": [
1602                     {
1603                         "network": NETWORK1_1[addr_type],
1604                         "next_hop": NEXT_HOP_IP[addr_type],
1605                         "vrf": "RED",
1606                     }
1607                 ]
1608             },
1609             "r2": {
1610                 "static_routes": [
1611                     {
1612                         "network": NETWORK2_1[addr_type],
1613                         "next_hop": NEXT_HOP_IP[addr_type],
1614                         "vrf": "BLUE",
1615                     },
1616                     {
1617                         "network": NETWORK3_1[addr_type],
1618                         "next_hop": NEXT_HOP_IP[addr_type],
1619                         "vrf": "GREEN",
1620                     }
1621                 ]
1622             }
1623         }
1624
1625         result = create_static_routes(tgen, input_dict_1)
1626         assert result is True, "Testcase {} : Failed \n Error: {}".format(
1627             tc_name, result
1628         )
1629
1630     if tgen.routers_have_failure():
1631         pytest.skip(tgen.errors)
1632
1633     step(
1634         "Advertise VRF routes in EVPN address family from Edge-1 router."
1635         " Configure a route-map on e1 to filter EVPN routes based on"
1636         " below keywords: route-type: prefix"
1637     )
1638
1639     for key, value in attribute.items():
1640         if key == "rt":
1641             logger.info("Creating extcommunity using raw_config")
1642             raw_config = {
1643                 "d2": {
1644                     "raw_config": [
1645                         "bgp extcommunity-list standard ECOMM300 permit {} {}".format(
1646                             key, value
1647                         )
1648                     ]
1649                 }
1650             }
1651             result = apply_raw_config(tgen, raw_config)
1652             assert result is True, "Testcase {} : Failed Error: {}".format(
1653                 tc_name, result
1654             )
1655
1656             input_dict_1 = {
1657                 "e1": {
1658                     "route_maps": {
1659                         "rmap_route_type": [
1660                             {"action": "permit", "set": {"extcommunity": {key: value}}}
1661                         ]
1662                     }
1663                 },
1664                 "d2": {
1665                     "route_maps": {
1666                         "rmap_route_type": [
1667                             {"action": "permit", "match": {"extcommunity": "ECOMM300"}}
1668                         ]
1669                     }
1670                 },
1671             }
1672
1673         else:
1674             input_dict_1 = {
1675                 "e1": {
1676                     "route_maps": {
1677                         "rmap_route_type": [
1678                             {"action": "permit", "match": {"evpn": {key: value}}}
1679                         ]
1680                     }
1681                 },
1682                 "d2": {
1683                     "route_maps": {
1684                         "rmap_route_type": [
1685                             {"action": "permit", "match": {"evpn": {key: value}}}
1686                         ]
1687                     }
1688                 },
1689             }
1690         result = create_route_maps(tgen, input_dict_1)
1691         assert result is True, "Testcase {} : Failed \n Error: {}".format(
1692             tc_name, result
1693         )
1694
1695     input_dict_2 = {
1696         "e1": {
1697             "bgp": [
1698                 {
1699                     "local_as": "100",
1700                     "address_family": {
1701                         "l2vpn": {
1702                             "evpn": {
1703                                 "neighbor": {
1704                                     "d2": {
1705                                         "ipv4": {
1706                                             "e1-link1": {
1707                                                 "route_maps": [
1708                                                     {
1709                                                         "name": "rmap_route_type",
1710                                                         "direction": "out",
1711                                                     }
1712                                                 ]
1713                                             }
1714                                         }
1715                                     }
1716                                 }
1717                             }
1718                         }
1719                     },
1720                 }
1721             ]
1722         },
1723         "d2": {
1724             "bgp": [
1725                 {
1726                     "local_as": "200",
1727                     "address_family": {
1728                         "l2vpn": {
1729                             "evpn": {
1730                                 "neighbor": {
1731                                     "e1": {
1732                                         "ipv4": {
1733                                             "d2-link1": {
1734                                                 "route_maps": [
1735                                                     {
1736                                                         "name": "rmap_route_type",
1737                                                         "direction": "in",
1738                                                     }
1739                                                 ]
1740                                             }
1741                                         }
1742                                     }
1743                                 }
1744                             }
1745                         }
1746                     },
1747                 }
1748             ]
1749         },
1750     }
1751
1752     result = create_router_bgp(tgen, topo, input_dict_2)
1753     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
1754
1755     step(
1756         "Verify on router DCG-2 that EVPN routes corresponding to all "
1757         "VRFs are received. As all EVPN routes are type-5 only."
1758     )
1759
1760     input_routes = {key: topo["routers"][key] for key in ["r1"]}
1761     result = verify_evpn_routes(tgen, topo, "d2", input_routes)
1762     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
1763
1764     input_routes = {key: topo["routers"][key] for key in ["r2"]}
1765     result = verify_evpn_routes(tgen, topo, "d2", input_routes)
1766     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
1767
1768     write_test_footer(tc_name)
1769
1770
1771 @pytest.mark.parametrize("attribute", ["locPrf", "weight", "path"])
1772 def test_bgp_attributes_for_evpn_address_family_p1(request, attribute):
1773     """
1774     BGP attributes for EVPN address-family.
1775     """
1776
1777     tgen = get_topogen()
1778     tc_name = request.node.name
1779     write_test_header(tc_name)
1780     check_router_status(tgen)
1781     reset_config_on_routers(tgen)
1782     add_default_routes(tgen)
1783
1784     if tgen.routers_have_failure():
1785         pytest.skip(tgen.errors)
1786
1787     step(
1788         "Advertise prefixes from VNF routers R1 and R2 in associated "
1789         "VRFs for both address-family."
1790     )
1791
1792     for addr_type in ADDR_TYPES:
1793         input_dict_1 = {
1794             "r1": {
1795                 "static_routes": [
1796                     {
1797                         "network": NETWORK1_1[addr_type],
1798                         "next_hop": NEXT_HOP_IP[addr_type],
1799                         "vrf": "RED",
1800                     }
1801                 ]
1802             },
1803             "r2": {
1804                 "static_routes": [
1805                     {
1806                         "network": NETWORK2_1[addr_type],
1807                         "next_hop": NEXT_HOP_IP[addr_type],
1808                         "vrf": "BLUE",
1809                     },
1810                     {
1811                         "network": NETWORK3_1[addr_type],
1812                         "next_hop": NEXT_HOP_IP[addr_type],
1813                         "vrf": "GREEN",
1814                     }
1815                 ]
1816             }
1817         }
1818
1819         result = create_static_routes(tgen, input_dict_1)
1820         assert result is True, "Testcase {} : Failed \n Error: {}".format(
1821             tc_name, result
1822         )
1823
1824     topo_local = deepcopy(topo)
1825
1826     logger.info("Modifying topology b/w e1 and d1 from iBGP to eBGP")
1827     step("Delete BGP config for vrf RED.")
1828
1829     if attribute == "locPrf":
1830         input_dict_vni = {
1831             "d1": {
1832                 "vrfs": [
1833                     {"name": "RED", "no_vni": VNI_1},
1834                     {"name": "BLUE", "no_vni": VNI_2},
1835                     {"name": "GREEN", "no_vni": VNI_3},
1836                 ]
1837             }
1838         }
1839         result = create_vrf_cfg(tgen, topo, input_dict=input_dict_vni)
1840         assert result is True, "Testcase {} :Failed \n Error: {}".format(
1841             tc_name, result
1842         )
1843
1844         input_dict_2 = {}
1845         for dut in ["d1"]:
1846             temp = {dut: {"bgp": []}}
1847             input_dict_2.update(temp)
1848
1849             INDEX = [0, 1, 2, 3]
1850             VRFS = ["RED", "BLUE", "GREEN", None]
1851             AS_NUM = [100, 100, 100, 100]
1852
1853             for index, vrf, as_num in zip(INDEX, VRFS, AS_NUM):
1854                 topo_local["routers"][dut]["bgp"][index]["local_as"] = 200
1855                 if vrf:
1856                     temp[dut]["bgp"].append(
1857                         {"local_as": as_num, "vrf": vrf, "delete": True}
1858                     )
1859                 else:
1860                     temp[dut]["bgp"].append({"local_as": as_num, "delete": True})
1861
1862         result = create_router_bgp(tgen, topo, input_dict_2)
1863         assert result is True, "Testcase {} on d1 :Failed \n Error: {}".format(
1864             tc_name, result
1865         )
1866
1867         result = create_router_bgp(tgen, topo_local["routers"])
1868         assert result is True, "Testcase {} :Failed \n Error: {}".format(
1869             tc_name, result
1870         )
1871
1872     step("Advertise VRF routes in EVPN address-family from DCG-1 " "and DCG-2 routers.")
1873
1874     for addr_type in ADDR_TYPES:
1875         input_dict_1 = {
1876             "r3": {
1877                 "static_routes": [
1878                     {
1879                         "network": NETWORK1_2[addr_type],
1880                         "next_hop": NEXT_HOP_IP[addr_type],
1881                         "vrf": "RED",
1882                     }
1883                 ]
1884             },
1885             "r4": {
1886                 "static_routes": [
1887                     {
1888                         "network": NETWORK1_3[addr_type],
1889                         "next_hop": NEXT_HOP_IP[addr_type],
1890                         "vrf": "BLUE",
1891                     },
1892                     {
1893                         "network": NETWORK1_4[addr_type],
1894                         "next_hop": NEXT_HOP_IP[addr_type],
1895                         "vrf": "GREEN",
1896                     },
1897                 ]
1898             },
1899         }
1900
1901         result = create_static_routes(tgen, input_dict_1)
1902         assert result is True, "Testcase {} : Failed \n Error: {}".format(
1903             tc_name, result
1904         )
1905
1906     step(
1907         "Redistribute static in (IPv4 and IPv6) address-family "
1908         "on Edge-1 for all VRFs."
1909     )
1910
1911     input_dict_2 = {}
1912     for dut in ["r3", "r4"]:
1913         temp = {dut: {"bgp": []}}
1914         input_dict_2.update(temp)
1915
1916         if dut == "r3":
1917             VRFS = ["RED"]
1918             AS_NUM = [3]
1919         if dut == "r4":
1920             VRFS = ["BLUE", "GREEN"]
1921             AS_NUM = [4, 4]
1922
1923         for vrf, as_num in zip(VRFS, AS_NUM):
1924             temp[dut]["bgp"].append(
1925                 {
1926                     "local_as": as_num,
1927                     "vrf": vrf,
1928                     "address_family": {
1929                         "ipv4": {
1930                             "unicast": {"redistribute": [{"redist_type": "static"}]}
1931                         },
1932                         "ipv6": {
1933                             "unicast": {"redistribute": [{"redist_type": "static"}]}
1934                         },
1935                     },
1936                 }
1937             )
1938
1939     result = create_router_bgp(tgen, topo, input_dict_2)
1940     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
1941
1942     step(
1943         "Verify on router Edge-1 that EVPN routes corresponding to "
1944         "all VRFs are received from both routers DCG-1 and DCG-2"
1945     )
1946
1947     for addr_type in ADDR_TYPES:
1948         input_routes = {
1949             "r3": {
1950                 "static_routes": [
1951                     {
1952                         "network": NETWORK1_2[addr_type],
1953                         "next_hop": NEXT_HOP_IP[addr_type],
1954                         "vrf": "RED",
1955                     }
1956                 ]
1957             },
1958             "r4": {
1959                 "static_routes": [
1960                     {
1961                         "network": NETWORK1_3[addr_type],
1962                         "next_hop": NEXT_HOP_IP[addr_type],
1963                         "vrf": "BLUE",
1964                     },
1965                     {
1966                         "network": NETWORK1_4[addr_type],
1967                         "next_hop": NEXT_HOP_IP[addr_type],
1968                         "vrf": "GREEN",
1969                     },
1970                 ]
1971             },
1972         }
1973
1974         result = verify_rib(tgen, addr_type, "e1", input_routes)
1975         assert result is True, "Testcase {} :Failed \n Error: {}".format(
1976             tc_name, result
1977         )
1978
1979     step(
1980         "Configure a route-map on Edge-1 to modify below BGP attributes "
1981         "for EVPN address-family:"
1982     )
1983
1984     if attribute == "path":
1985         input_dict_1 = {
1986             "e1": {
1987                 "route_maps": {
1988                     "rmap_d1".format(addr_type): [
1989                         {
1990                             "action": "permit",
1991                             "set": {
1992                                 attribute: {
1993                                     "as_num": "123 231 321",
1994                                     "as_action": "prepend",
1995                                 }
1996                             },
1997                         }
1998                     ],
1999                     "rmap_d2".format(addr_type): [
2000                         {
2001                             "action": "permit",
2002                             "set": {
2003                                 attribute: {"as_num": "121", "as_action": "prepend"}
2004                             },
2005                         }
2006                     ],
2007                 }
2008             }
2009         }
2010     else:
2011         input_dict_1 = {
2012             "e1": {
2013                 "route_maps": {
2014                     "rmap_d1".format(addr_type): [
2015                         {"action": "permit", "set": {attribute: 120}}
2016                     ],
2017                     "rmap_d2".format(addr_type): [
2018                         {"action": "permit", "set": {attribute: 150}}
2019                     ],
2020                 }
2021             }
2022         }
2023     result = create_route_maps(tgen, input_dict_1)
2024     assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
2025
2026     input_dict_2 = {
2027         "e1": {
2028             "bgp": [
2029                 {
2030                     "local_as": "100",
2031                     "address_family": {
2032                         "l2vpn": {
2033                             "evpn": {
2034                                 "neighbor": {
2035                                     "d1": {
2036                                         "ipv4": {
2037                                             "e1-link1": {
2038                                                 "route_maps": [
2039                                                     {
2040                                                         "name": "rmap_d1",
2041                                                         "direction": "in",
2042                                                     }
2043                                                 ]
2044                                             }
2045                                         }
2046                                     },
2047                                     "d2": {
2048                                         "ipv4": {
2049                                             "e1-link1": {
2050                                                 "route_maps": [
2051                                                     {
2052                                                         "name": "rmap_d2",
2053                                                         "direction": "in",
2054                                                     }
2055                                                 ]
2056                                             }
2057                                         }
2058                                     },
2059                                 }
2060                             }
2061                         }
2062                     },
2063                 }
2064             ]
2065         }
2066     }
2067
2068     result = create_router_bgp(tgen, topo, input_dict_2)
2069     assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
2070
2071     step(
2072         "Verify on router Edge-1 that EVPN routes are preferred via"
2073         " DCG-1 or DCG-2 based on best path selection criteria "
2074         "(according to the configured BGP attribute values in route-map)."
2075     )
2076
2077     for addr_type in ADDR_TYPES:
2078         input_routes = {
2079             "r3": {
2080                 "static_routes": [
2081                     {
2082                         "network": NETWORK1_2[addr_type],
2083                         "next_hop": NEXT_HOP_IP[addr_type],
2084                         "vrf": "RED",
2085                     }
2086                 ]
2087             },
2088             "r4": {
2089                 "static_routes": [
2090                     {
2091                         "network": NETWORK1_3[addr_type],
2092                         "next_hop": NEXT_HOP_IP[addr_type],
2093                         "vrf": "BLUE",
2094                     },
2095                     {
2096                         "network": NETWORK1_4[addr_type],
2097                         "next_hop": NEXT_HOP_IP[addr_type],
2098                         "vrf": "GREEN",
2099                     },
2100                 ]
2101             },
2102         }
2103
2104         result = verify_best_path_as_per_bgp_attribute(
2105             tgen, addr_type, "e1", input_routes, attribute
2106         )
2107         assert result is True, "Testcase {} : Failed \n Error: {}".format(
2108             tc_name, result
2109         )
2110
2111     write_test_footer(tc_name)
2112
2113
2114 if __name__ == "__main__":
2115     args = ["-s"] + sys.argv[1:]
2116     sys.exit(pytest.main(args))