4 # Copyright (c) 2020 by VMware, Inc. ("VMware")
5 # Used Copyright (c) 2018 by Network Device Education Foundation,
6 # Inc. ("NetDEF") in this file.
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
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
24 Following tests are covered to test BGP Multi-VRF Dynamic Route Leaking:
26 1. Verify that Changing route-map configurations(match/set clauses) on
27 the fly it takes immediate effect.
28 2. Verify BGP best path selection algorithm works fine when
29 routes are imported from ISR to default vrf and vice versa.
39 # Save the Current Working Directory to find configuration files.
40 CWD
= os
.path
.dirname(os
.path
.realpath(__file__
))
41 sys
.path
.append(os
.path
.join(CWD
, "../"))
42 sys
.path
.append(os
.path
.join(CWD
, "../lib/"))
44 # Required to instantiate the topology builder class.
46 # pylint: disable=C0413
47 # Import topogen and topotest helpers
48 from lib
.topogen
import Topogen
, get_topogen
49 from lib
.topotest
import version_cmp
51 from lib
.common_config
import (
59 create_bgp_community_lists
,
61 get_frr_ipv6_linklocal
,
62 shutdown_bringup_interface
,
65 from lib
.topolog
import logger
67 verify_bgp_convergence
,
70 verify_bgp_attributes
,
71 verify_best_path_as_per_bgp_attribute
,
74 from lib
.topojson
import build_topo_from_json
, build_config_from_json
76 pytestmark
= [pytest
.mark
.bgpd
, pytest
.mark
.staticd
]
79 NETWORK1_1
= {"ipv4": "11.11.11.1/32", "ipv6": "11:11::1/128"}
80 NETWORK3_3
= {"ipv4": "50.50.50.5/32", "ipv6": "50:50::5/128"}
81 NETWORK3_4
= {"ipv4": "50.50.50.50/32", "ipv6": "50:50::50/128"}
83 PREFERRED_NEXT_HOP
= "global"
86 def setup_module(mod
):
88 Sets up the pytest environment
94 testsuite_run_time
= time
.asctime(time
.localtime(time
.time()))
95 logger
.info("Testsuite start time: {}".format(testsuite_run_time
))
98 logger
.info("Running setup_module to create topology")
100 # This function initiates the topology build with Topogen...
101 json_file
= "{}/bgp_vrf_dynamic_route_leak_topo2.json".format(CWD
)
102 tgen
= Topogen(json_file
, mod
.__name
__)
103 topo
= tgen
.json_topo
104 # ... and here it calls Mininet initialization functions.
106 # Starting topology, create tmp files which are loaded to routers
107 # to start daemons and then start routers
110 # Run these tests for kernel version 4.19 or above
111 if version_cmp(platform
.release(), "4.19") < 0:
113 "BGP vrf dynamic route leak tests will not run "
114 '(have kernel "{}", but it requires >= 4.19)'.format(platform
.release())
116 pytest
.skip(error_msg
)
118 # Creating configuration from JSON
119 build_config_from_json(tgen
, topo
)
121 global BGP_CONVERGENCE
123 ADDR_TYPES
= check_address_types()
125 BGP_CONVERGENCE
= verify_bgp_convergence(tgen
, topo
)
126 assert BGP_CONVERGENCE
is True, "setup_module : Failed \n Error: {}".format(
130 logger
.info("Running setup_module() done")
133 def teardown_module():
134 """Teardown the pytest environment"""
136 logger
.info("Running teardown_module to delete topology")
140 # Stop toplogy and Remove tmp files
144 "Testsuite end time: {}".format(time
.asctime(time
.localtime(time
.time())))
146 logger
.info("=" * 40)
149 #####################################################
153 #####################################################
156 def test_bgp_best_path_with_dynamic_import_p0(request
):
159 1.5.6. Verify BGP best path selection algorithm works fine when
160 routes are imported from ISR to default vrf and vice versa.
164 tc_name
= request
.node
.name
165 write_test_header(tc_name
)
166 build_config_from_json(tgen
, topo
)
168 if tgen
.routers_have_failure():
169 check_router_status(tgen
)
171 for addr_type
in ADDR_TYPES
:
174 "Redistribute configured static routes into BGP process" " on R1/R2 and R3"
178 DUT
= ["r1", "r2", "r3", "r4"]
179 VRFS
= ["ISR", "ISR", "default", "default"]
180 AS_NUM
= [100, 100, 300, 400]
182 for dut
, vrf
, as_num
in zip(DUT
, VRFS
, AS_NUM
):
183 temp
= {dut
: {"bgp": []}}
184 input_dict_1
.update(temp
)
186 temp
[dut
]["bgp"].append(
192 "unicast": {"redistribute": [{"redist_type": "static"}]}
198 result
= create_router_bgp(tgen
, topo
, input_dict_1
)
199 assert result
is True, "Testcase {} :Failed \n Error: {}".format(
203 for addr_type
in ADDR_TYPES
:
205 step("Import from default vrf into vrf ISR on R1 and R2 as below")
209 VRFS
= ["ISR", "ISR"]
212 for dut
, vrf
, as_num
in zip(DUT
, VRFS
, AS_NUM
):
213 temp
= {dut
: {"bgp": []}}
214 input_dict_vrf
.update(temp
)
216 temp
[dut
]["bgp"].append(
221 addr_type
: {"unicast": {"import": {"vrf": "default"}}}
226 result
= create_router_bgp(tgen
, topo
, input_dict_vrf
)
227 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
231 input_dict_default
= {}
233 VRFS
= ["default", "default"]
236 for dut
, vrf
, as_num
in zip(DUT
, VRFS
, AS_NUM
):
237 temp
= {dut
: {"bgp": []}}
238 input_dict_default
.update(temp
)
240 temp
[dut
]["bgp"].append(
245 addr_type
: {"unicast": {"import": {"vrf": "ISR"}}}
250 result
= create_router_bgp(tgen
, topo
, input_dict_default
)
251 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
256 "Verify ECMP/Next-hop/Imported routes Vs Locally originated "
257 "routes/eBGP routes vs iBGP routes --already covered in almost"
261 for addr_type
in ADDR_TYPES
:
263 step("Verify Pre-emption")
266 "r3": {"static_routes": [{"network": [NETWORK3_3
[addr_type
]]}]}
269 intf_r3_r1
= topo
["routers"]["r3"]["links"]["r1-link1"]["interface"]
270 intf_r4_r1
= topo
["routers"]["r4"]["links"]["r1-link1"]["interface"]
272 if addr_type
== "ipv6" and "link_local" in PREFERRED_NEXT_HOP
:
273 nh_r3_r1
= get_frr_ipv6_linklocal(tgen
, "r3", intf
=intf_r3_r1
)
274 nh_r4_r1
= get_frr_ipv6_linklocal(tgen
, "r4", intf
=intf_r4_r1
)
276 nh_r3_r1
= topo
["routers"]["r3"]["links"]["r1-link1"][addr_type
].split("/")[
279 nh_r4_r1
= topo
["routers"]["r4"]["links"]["r1-link1"][addr_type
].split("/")[
283 result
= verify_bgp_rib(
284 tgen
, addr_type
, "r1", input_routes_r3
, next_hop
=[nh_r4_r1
]
286 assert result
is True, "Testcase {} : Failed \n Error {}".format(
290 step("Shutdown interface connected to r1 from r4:")
291 shutdown_bringup_interface(tgen
, "r4", intf_r4_r1
, False)
293 for addr_type
in ADDR_TYPES
:
296 "r3": {"static_routes": [{"network": [NETWORK3_3
[addr_type
]]}]}
299 intf_r3_r1
= topo
["routers"]["r3"]["links"]["r1-link1"]["interface"]
300 intf_r4_r1
= topo
["routers"]["r4"]["links"]["r1-link1"]["interface"]
302 if addr_type
== "ipv6" and "link_local" in PREFERRED_NEXT_HOP
:
303 nh_r3_r1
= get_frr_ipv6_linklocal(tgen
, "r3", intf
=intf_r3_r1
)
304 nh_r4_r1
= get_frr_ipv6_linklocal(tgen
, "r4", intf
=intf_r4_r1
)
306 nh_r3_r1
= topo
["routers"]["r3"]["links"]["r1-link1"][addr_type
].split("/")[
309 nh_r4_r1
= topo
["routers"]["r4"]["links"]["r1-link1"][addr_type
].split("/")[
313 step("Verify next-hop is changed")
314 result
= verify_bgp_rib(
315 tgen
, addr_type
, "r1", input_routes_r3
, next_hop
=[nh_r3_r1
]
317 assert result
is True, "Testcase {} : Failed \n Error {}".format(
321 step("Bringup interface connected to r1 from r4:")
322 shutdown_bringup_interface(tgen
, "r4", intf_r4_r1
, True)
324 for addr_type
in ADDR_TYPES
:
327 "r3": {"static_routes": [{"network": [NETWORK3_3
[addr_type
]]}]}
330 intf_r3_r1
= topo
["routers"]["r3"]["links"]["r1-link1"]["interface"]
331 intf_r4_r1
= topo
["routers"]["r4"]["links"]["r1-link1"]["interface"]
333 if addr_type
== "ipv6" and "link_local" in PREFERRED_NEXT_HOP
:
334 nh_r3_r1
= get_frr_ipv6_linklocal(tgen
, "r3", intf
=intf_r3_r1
)
335 nh_r4_r1
= get_frr_ipv6_linklocal(tgen
, "r4", intf
=intf_r4_r1
)
337 nh_r3_r1
= topo
["routers"]["r3"]["links"]["r1-link1"][addr_type
].split("/")[
340 nh_r4_r1
= topo
["routers"]["r4"]["links"]["r1-link1"][addr_type
].split("/")[
344 step("Verify next-hop is not chnaged aftr shutdown:")
345 result
= verify_bgp_rib(
346 tgen
, addr_type
, "r1", input_routes_r3
, next_hop
=[nh_r3_r1
]
348 assert result
is True, "Testcase {} : Failed \n Error {}".format(
352 step("Active-Standby scenario(as-path prepend and Local pref)")
354 for addr_type
in ADDR_TYPES
:
356 step("Create prefix-list")
362 "pf_ls_{}".format(addr_type
): [
365 "network": NETWORK3_4
[addr_type
],
373 result
= create_prefix_lists(tgen
, input_dict_pf
)
374 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
378 for addr_type
in ADDR_TYPES
:
380 step("Create route-map to match prefix-list and set localpref 500")
385 "rmap_PATH1_{}".format(addr_type
): [
391 "prefix_lists": "pf_ls_{}".format(addr_type
)
394 "set": {"locPrf": 500},
401 result
= create_route_maps(tgen
, input_dict_rm
)
402 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
406 step("Create route-map to match prefix-list and set localpref 600")
411 "rmap_PATH2_{}".format(addr_type
): [
417 "prefix_lists": "pf_ls_{}".format(addr_type
)
420 "set": {"locPrf": 600},
427 result
= create_route_maps(tgen
, input_dict_rm
)
428 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
446 "name": "rmap_PATH1_{}".format(
460 "name": "rmap_PATH2_{}".format(
478 result
= create_router_bgp(tgen
, topo
, input_dict_rma
)
479 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
486 for addr_type
in ADDR_TYPES
:
488 step("Verify bestpath is installed as per highest localpref")
493 {"network": [NETWORK3_3
[addr_type
], NETWORK3_4
[addr_type
]]}
498 result
= verify_best_path_as_per_bgp_attribute(
499 tgen
, addr_type
, dut
, input_routes_r3
, attribute
501 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
505 for addr_type
in ADDR_TYPES
:
507 step("Create route-map to match prefix-list and set localpref 700")
512 "rmap_PATH1_{}".format(addr_type
): [
518 "prefix_lists": "pf_ls_{}".format(addr_type
)
521 "set": {"locPrf": 700},
528 result
= create_route_maps(tgen
, input_dict_rm
)
529 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
533 for addr_type
in ADDR_TYPES
:
535 step("Verify bestpath is changed as per highest localpref")
540 {"network": [NETWORK3_3
[addr_type
], NETWORK3_4
[addr_type
]]}
545 result
= verify_best_path_as_per_bgp_attribute(
546 tgen
, addr_type
, dut
, input_routes_r3
, attribute
548 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
552 for addr_type
in ADDR_TYPES
:
554 step("Create route-map to match prefix-list and set as-path prepend")
559 "rmap_PATH2_{}".format(addr_type
): [
565 "prefix_lists": "pf_ls_{}".format(addr_type
)
570 "path": {"as_num": "111", "as_action": "prepend"},
578 result
= create_route_maps(tgen
, input_dict_rm
)
579 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
585 for addr_type
in ADDR_TYPES
:
587 step("Verify bestpath is changed as per shortest as-path")
592 {"network": [NETWORK3_3
[addr_type
], NETWORK3_4
[addr_type
]]}
597 result
= verify_best_path_as_per_bgp_attribute(
598 tgen
, addr_type
, dut
, input_routes_r3
, attribute
600 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
604 write_test_footer(tc_name
)
607 def test_modify_route_map_match_set_clauses_p1(request
):
610 1.5.13. Verify that Changing route-map configurations(match/set clauses) on
611 the fly it takes immediate effect.
615 tc_name
= request
.node
.name
616 write_test_header(tc_name
)
617 build_config_from_json(tgen
, topo
)
619 if tgen
.routers_have_failure():
620 check_router_status(tgen
)
622 for addr_type
in ADDR_TYPES
:
625 "Configure route-map to set community attribute for a specific"
626 "prefix on R1 in vrf ISR"
633 "pflist_ABC_{}".format(addr_type
): [
636 "network": NETWORK1_1
[addr_type
],
644 result
= create_prefix_lists(tgen
, input_dict_pf
)
645 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
651 "bgp_community_lists": [
653 "community_type": "expanded",
661 result
= create_bgp_community_lists(tgen
, input_dict_cl
)
662 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
664 for addr_type
in ADDR_TYPES
:
668 "rmap_XYZ_{}".format(addr_type
): [
673 "prefix_lists": "pflist_ABC_{}".format(addr_type
)
676 "set": {"community": {"num": "100:100"}},
682 result
= create_route_maps(tgen
, input_dict_rm
)
683 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
687 for addr_type
in ADDR_TYPES
:
690 "Apply this route-map on R1 to vrf ISR while redistributing the"
699 for dut
, vrf
, as_num
in zip(DUT
, VRFS
, AS_NUM
):
700 temp
= {dut
: {"bgp": []}}
701 input_dict_1
.update(temp
)
703 temp
[dut
]["bgp"].append(
712 "redist_type": "static",
714 "route-map": "rmap_XYZ_{}".format(addr_type
)
724 result
= create_router_bgp(tgen
, topo
, input_dict_1
)
725 assert result
is True, "Testcase {} :Failed \n Error: {}".format(
729 for addr_type
in ADDR_TYPES
:
732 "Configure another route-map for filtering the prefixes based on"
733 " community attribute while importing into default vrf"
739 "rmap_IMP_{}".format(addr_type
): [
743 "match": {"community_list": {"id": "COMM"}},
744 "set": {"community": {"num": "none"}},
750 result
= create_route_maps(tgen
, input_dict_rm
)
751 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
755 for addr_type
in ADDR_TYPES
:
758 "Apply the route-map while Importing vrf ISR's prefixes into "
759 "default vrf on router R1:"
767 for dut
, vrf
, as_num
in zip(DUT
, VRFS
, AS_NUM
):
768 temp
= {dut
: {"bgp": []}}
769 input_dict_isr
.update(temp
)
771 temp
[dut
]["bgp"].append(
776 addr_type
: {"unicast": {"import": {"vrf": "ISR"}}}
781 temp
[dut
]["bgp"].append(
789 "vrf": "route-map rmap_IMP_{}".format(addr_type
)
797 result
= create_router_bgp(tgen
, topo
, input_dict_isr
)
798 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
802 for addr_type
in ADDR_TYPES
:
805 "Verify on R1 that only prefixes with community value 100:100"
806 "in vrf ISR are imported to vrf default. While importing, the"
807 " community value has been stripped off:"
813 {"network": [NETWORK1_1
[addr_type
]], "vrf": "default"}
818 result
= verify_bgp_rib(tgen
, addr_type
, "r1", input_routes_r1
)
819 assert result
is True, "Testcase {} : Failed \n Error {}".format(
823 for addr_type
in ADDR_TYPES
:
825 step("Add set clause in route-map IMP:")
830 "rmap_IMP_{}".format(addr_type
): [
834 "match": {"community_list": {"id": "COMM"}},
836 "large_community": {"num": "100:100:100"},
838 "path": {"as_num": "100 100", "as_action": "prepend"},
845 result
= create_route_maps(tgen
, input_dict_rm
)
846 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
850 for addr_type
in ADDR_TYPES
:
853 "Verify that as we continue adding different attributes "
854 "step-by-step in route-map IMP those attributes gets "
855 "attached to prefixes:"
861 {"network": [NETWORK1_1
[addr_type
]], "vrf": "default"}
866 input_dict_comm
= {"largeCommunity": "100:100:100"}
868 result
= verify_bgp_community(
869 tgen
, addr_type
, dut
, [NETWORK1_1
[addr_type
]], input_dict_comm
871 assert result
is True, "Testcase {} : Failed \n Error {}".format(
878 "rmap_IMP_{}".format(addr_type
): [{"set": {"locPrf": 500}}]
883 result
= verify_bgp_attributes(
887 [NETWORK1_1
[addr_type
]],
888 rmap_name
="rmap_IMP_{}".format(addr_type
),
889 input_dict
=input_rmap
,
891 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
895 step("Change community-list to match a different value then " "100:100.")
899 "bgp_community_lists": [
901 "community_type": "expanded",
910 result
= create_bgp_community_lists(tgen
, input_dict_cl
)
911 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
913 for addr_type
in ADDR_TYPES
:
918 {"network": [NETWORK1_1
[addr_type
]], "vrf": "default"}
923 result
= verify_bgp_rib(tgen
, addr_type
, "r1", input_routes_r1
, expected
=False)
926 ), "Testcase {} : Failed \n Error : Routes are still " "present {}".format(
930 write_test_footer(tc_name
)
933 if __name__
== "__main__":
934 args
= ["-s"] + sys
.argv
[1:]
935 sys
.exit(pytest
.main(args
))