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