2 # SPDX-License-Identifier: ISC
5 # Copyright (c) 2021 by VMware, Inc. ("VMware")
6 # Used Copyright (c) 2018 by Network Device Education Foundation, Inc.
7 # ("NetDEF") in this file.
12 Following tests are covered to test ecmp functionality on BGP GSHUT.
13 1. Verify graceful-shutdown functionality with eBGP peers
14 2. Verify graceful-shutdown functionality when daemons
15 bgpd/zebra/staticd and frr services are restarted with eBGP peers
23 # Save the Current Working Directory to find configuration files.
24 CWD
= os
.path
.dirname(os
.path
.realpath(__file__
))
25 sys
.path
.append(os
.path
.join(CWD
, "../"))
26 sys
.path
.append(os
.path
.join(CWD
, "../../"))
28 # pylint: disable=C0413
29 # Import topogen and topotest helpers
30 from lib
.topogen
import Topogen
, get_topogen
32 from lib
.common_config
import (
38 reset_config_on_routers
,
40 get_frr_ipv6_linklocal
,
46 create_bgp_community_lists
,
47 required_linux_kernel_version
,
49 from lib
.topolog
import logger
51 verify_bgp_convergence
,
54 verify_bgp_attributes
,
56 from lib
.topojson
import build_config_from_json
59 pytestmark
= [pytest
.mark
.bgpd
, pytest
.mark
.staticd
]
63 NETWORK
= {"ipv4": "100.0.10.1/32", "ipv6": "1::1/128"}
64 NEXT_HOP_IP_1
= {"ipv4": "10.0.2.1", "ipv6": "fd00:0:0:1::1"}
65 NEXT_HOP_IP_2
= {"ipv4": "10.0.4.2", "ipv6": "fd00:0:0:3::2"}
66 PREFERRED_NEXT_HOP
= "link_local"
67 BGP_CONVERGENCE
= False
70 def setup_module(mod
):
72 Sets up the pytest environment
79 # Required linux kernel version for this suite to run.
80 result
= required_linux_kernel_version("4.16")
81 if result
is not True:
82 pytest
.skip("Kernel requirements are not met, kernel version should be >=4.16")
84 testsuite_run_time
= time
.asctime(time
.localtime(time
.time()))
85 logger
.info("Testsuite start time: {}".format(testsuite_run_time
))
88 logger
.info("Running setup_module to create topology")
90 # This function initiates the topology build with Topogen...
91 json_file
= "{}/ebgp_gshut_topo1.json".format(CWD
)
92 tgen
= Topogen(json_file
, mod
.__name
__)
95 # ... and here it calls Mininet initialization functions.
97 # Starting topology, create tmp files which are loaded to routers
98 # to start daemons and then start routers
101 # Creating configuration from JSON
102 build_config_from_json(tgen
, topo
)
104 # Don't run this test if we have any failure.
105 if tgen
.routers_have_failure():
106 pytest
.skip(tgen
.errors
)
108 # Api call verify whether BGP is converged
109 ADDR_TYPES
= check_address_types()
111 BGP_CONVERGENCE
= verify_bgp_convergence(tgen
, topo
)
112 assert BGP_CONVERGENCE
is True, "setup_module :Failed \n Error:" " {}".format(
116 logger
.info("Running setup_module() done")
119 def teardown_module():
121 Teardown the pytest environment.
126 logger
.info("Running teardown_module to delete topology")
130 # Stop toplogy and Remove tmp files
134 ###########################
136 ###########################
139 def next_hop_per_address_family(
140 tgen
, dut
, peer
, addr_type
, next_hop_dict
, preferred_next_hop
=PREFERRED_NEXT_HOP
143 This function returns link_local or global next_hop per address-family
146 intferface
= topo
["routers"][peer
]["links"]["{}".format(dut
)]["interface"]
147 if addr_type
== "ipv6" and "link_local" in preferred_next_hop
:
148 next_hop
= get_frr_ipv6_linklocal(tgen
, peer
, intf
=intferface
)
150 next_hop
= next_hop_dict
[addr_type
]
155 ###########################
157 ###########################
160 def test_verify_graceful_shutdown_functionality_with_eBGP_peers_p0(request
):
162 Verify graceful-shutdown functionality with eBGP peers
165 tc_name
= request
.node
.name
166 write_test_header(tc_name
)
168 reset_config_on_routers(tgen
)
170 step("Done in base config: Configure base config as per the topology")
171 step("Base config should be up, verify using BGP convergence")
172 result
= verify_bgp_convergence(tgen
, topo
)
173 assert result
is True, "Test case {} : Failed \n Error: {}".format(tc_name
, result
)
175 step("Done in base config: Advertise prefixes from R1")
176 step("Verify BGP routes are received at R3 with best path from R3 to R1")
178 for addr_type
in ADDR_TYPES
:
180 next_hop1
= next_hop_per_address_family(
181 tgen
, "r3", "r1", addr_type
, NEXT_HOP_IP_1
183 next_hop2
= next_hop_per_address_family(
184 tgen
, "r3", "r4", addr_type
, NEXT_HOP_IP_2
187 input_topo
= {key
: topo
["routers"][key
] for key
in ["r1"]}
188 result
= verify_bgp_rib(
189 tgen
, addr_type
, dut
, input_topo
, next_hop
=[next_hop1
, next_hop2
]
191 assert result
is True, "Test case {} : Failed \n Error: {}".format(
195 result
= verify_rib(tgen
, addr_type
, dut
, input_topo
, next_hop
=next_hop1
)
196 assert result
is True, "Test case {} : Failed \n Error: {}".format(
200 step("On R1 configure:")
201 step("Create standard bgp community-list to permit graceful-shutdown:")
204 "bgp_community_lists": [
206 "community_type": "standard",
209 "value": "graceful-shutdown",
215 result
= create_bgp_community_lists(tgen
, input_dict_1
)
216 assert result
is True, "Test case {} : Failed \n Error: {}".format(tc_name
, result
)
218 step("Create route-map to set community GSHUT in OUT direction")
227 "set": {"community": {"num": "graceful-shutdown"}},
234 result
= create_route_maps(tgen
, input_dict_2
)
235 assert result
is True, "Test case {} : Failed \n Error: {}".format(tc_name
, result
)
282 result
= create_router_bgp(tgen
, topo
, input_dict_3
)
283 assert result
is True, "Test case {} : Failed \n Error: {}".format(tc_name
, result
)
286 "FRR is setting local-pref to 0 by-default on receiver GSHUT community, "
287 "below step is not needed, but keeping for reference"
290 "On R3, apply route-map IN direction to match GSHUT community "
291 "and set local-preference to 0."
295 "Verify BGP convergence on R3 and ensure all the neighbours state "
299 result
= verify_bgp_convergence(tgen
, topo
)
300 assert result
is True, "Test case {} : Failed \n Error: {}".format(tc_name
, result
)
302 step("Verify BGP routes on R3:")
303 step("local pref for routes coming from R1 is set to 0.")
305 for addr_type
in ADDR_TYPES
:
309 "GSHUT-OUT": [{"set": {"locPrf": 0}}],
314 static_routes
= [NETWORK
[addr_type
]]
315 result
= verify_bgp_attributes(
316 tgen
, addr_type
, dut
, static_routes
, "GSHUT-OUT", rmap_dict
318 assert result
is True, "Test case {} : Failed \n Error: {}".format(
322 step("Ensure that best path is selected from R4 to R3.")
324 for addr_type
in ADDR_TYPES
:
326 next_hop1
= next_hop_per_address_family(
327 tgen
, "r3", "r1", addr_type
, NEXT_HOP_IP_1
329 next_hop2
= next_hop_per_address_family(
330 tgen
, "r3", "r4", addr_type
, NEXT_HOP_IP_2
333 input_topo
= {key
: topo
["routers"][key
] for key
in ["r1"]}
334 result
= verify_bgp_rib(
335 tgen
, addr_type
, dut
, input_topo
, next_hop
=[next_hop1
, next_hop2
]
337 assert result
is True, "Test case {} : Failed \n Error: {}".format(
341 result
= verify_rib(tgen
, addr_type
, dut
, input_topo
, next_hop
=next_hop2
)
342 assert result
is True, "Test case {} : Failed \n Error: {}".format(
346 write_test_footer(tc_name
)
349 def test_verify_restarting_zebra_bgpd_staticd_frr_with_eBGP_peers_p0(request
):
351 Verify graceful-shutdown functionality when daemons bgpd/zebra/staticd and
352 frr services are restarted with eBGP peers
355 tc_name
= request
.node
.name
356 write_test_header(tc_name
)
358 reset_config_on_routers(tgen
)
360 step("Done in base config: Configure base config as per the topology")
361 step("Base config should be up, verify using BGP convergence")
362 result
= verify_bgp_convergence(tgen
, topo
)
363 assert result
is True, "Test case {} : Failed \n Error: {}".format(tc_name
, result
)
365 step("Done in base config: Advertise prefixes from R1")
366 step("Verify BGP routes are received at R3 with best path from R3 to R1")
368 for addr_type
in ADDR_TYPES
:
370 next_hop1
= next_hop_per_address_family(
371 tgen
, "r3", "r1", addr_type
, NEXT_HOP_IP_1
373 next_hop2
= next_hop_per_address_family(
374 tgen
, "r3", "r4", addr_type
, NEXT_HOP_IP_2
377 input_topo
= {key
: topo
["routers"][key
] for key
in ["r1"]}
378 result
= verify_bgp_rib(
379 tgen
, addr_type
, dut
, input_topo
, next_hop
=[next_hop1
, next_hop2
]
381 assert result
is True, "Test case {} : Failed \n Error: {}".format(
385 result
= verify_rib(tgen
, addr_type
, dut
, input_topo
, next_hop
=next_hop1
)
386 assert result
is True, "Test case {} : Failed \n Error: {}".format(
390 step("On R1 configure:")
391 step("Create standard bgp community-list to permit graceful-shutdown:")
394 "bgp_community_lists": [
396 "community_type": "standard",
399 "value": "graceful-shutdown",
405 result
= create_bgp_community_lists(tgen
, input_dict_1
)
406 assert result
is True, "Test case {} : Failed \n Error: {}".format(tc_name
, result
)
408 step("Create route-map to set community GSHUT in OUT direction")
417 "set": {"community": {"num": "graceful-shutdown"}},
424 result
= create_route_maps(tgen
, input_dict_2
)
425 assert result
is True, "Test case {} : Failed \n Error: {}".format(tc_name
, result
)
472 result
= create_router_bgp(tgen
, topo
, input_dict_3
)
473 assert result
is True, "Test case {} : Failed \n Error: {}".format(tc_name
, result
)
476 "FRR is setting local-pref to 0 by-default on receiver GSHUT community, "
477 "below step is not needed, but keeping for reference"
480 "On R3, apply route-map IN direction to match GSHUT community "
481 "and set local-preference to 0."
485 "Verify BGP convergence on R3 and ensure all the neighbours state "
489 result
= verify_bgp_convergence(tgen
, topo
)
490 assert result
is True, "Test case {} : Failed \n Error: {}".format(tc_name
, result
)
492 step("Verify BGP routes on R3:")
493 step("local pref for routes coming from R1 is set to 0.")
495 for addr_type
in ADDR_TYPES
:
496 rmap_dict
= {"r1": {"route_maps": {"GSHUT-OUT": [{"set": {"locPrf": 0}}]}}}
498 static_routes
= [NETWORK
[addr_type
]]
499 result
= verify_bgp_attributes(
500 tgen
, addr_type
, dut
, static_routes
, "GSHUT-OUT", rmap_dict
502 assert result
is True, "Test case {} : Failed \n Error: {}".format(
506 step("Ensure that best path is selected from R4 to R3.")
508 for addr_type
in ADDR_TYPES
:
510 next_hop1
= next_hop_per_address_family(
511 tgen
, "r3", "r1", addr_type
, NEXT_HOP_IP_1
513 next_hop2
= next_hop_per_address_family(
514 tgen
, "r3", "r4", addr_type
, NEXT_HOP_IP_2
517 input_topo
= {key
: topo
["routers"][key
] for key
in ["r1"]}
518 result
= verify_bgp_rib(
519 tgen
, addr_type
, dut
, input_topo
, next_hop
=[next_hop1
, next_hop2
]
521 assert result
is True, "Test case {} : Failed \n Error: {}".format(
525 result
= verify_rib(tgen
, addr_type
, dut
, input_topo
, next_hop
=next_hop2
)
526 assert result
is True, "Test case {} : Failed \n Error: {}".format(
530 step("Restart daemons and frr services")
532 for daemon
in ["bgpd", "zebra", "staticd", "frr"]:
534 kill_router_daemons(tgen
, "r3", ["staticd"])
535 start_router_daemons(tgen
, "r3", ["staticd"])
537 stop_router(tgen
, "r3")
538 start_router(tgen
, "r3")
541 "Verify BGP convergence on R3 and ensure all the neighbours state "
545 result
= verify_bgp_convergence(tgen
, topo
)
546 assert result
is True, "Test case {} : Failed \n Error: {}".format(
550 step("Verify BGP routes on R3:")
551 step("local pref for routes coming from R1 is set to 0.")
553 for addr_type
in ADDR_TYPES
:
554 rmap_dict
= {"r1": {"route_maps": {"GSHUT-OUT": [{"set": {"locPrf": 0}}]}}}
556 static_routes
= [NETWORK
[addr_type
]]
557 result
= verify_bgp_attributes(
558 tgen
, addr_type
, dut
, static_routes
, "GSHUT-OUT", rmap_dict
560 assert result
is True, "Test case {} : Failed \n Error: {}".format(
564 step("Ensure that best path is selected from R4 to R3.")
566 for addr_type
in ADDR_TYPES
:
568 next_hop1
= next_hop_per_address_family(
569 tgen
, "r3", "r1", addr_type
, NEXT_HOP_IP_1
571 next_hop2
= next_hop_per_address_family(
572 tgen
, "r3", "r4", addr_type
, NEXT_HOP_IP_2
575 input_topo
= {key
: topo
["routers"][key
] for key
in ["r1"]}
576 result
= verify_bgp_rib(
577 tgen
, addr_type
, dut
, input_topo
, next_hop
=[next_hop1
, next_hop2
]
579 assert result
is True, "Test case {} : Failed \n Error: {}".format(
583 result
= verify_rib(tgen
, addr_type
, dut
, input_topo
, next_hop
=next_hop2
)
584 assert result
is True, "Test case {} : Failed \n Error: {}".format(
588 write_test_footer(tc_name
)
591 if __name__
== "__main__":
592 args
= ["-s"] + sys
.argv
[1:]
593 sys
.exit(pytest
.main(args
))