2 # SPDX-License-Identifier: ISC
5 # Copyright (c) 2019 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 EBGP.
13 1. Verify routes installed as per maximum-paths configuration (8/16/32)
14 2. Disable/Shut selected paths nexthops and verify other next are installed in
15 the RIB of DUT. Enable interfaces and verify RIB count.
16 3. Verify BGP table and RIB in DUT after clear BGP routes and neighbors.
17 4. Verify routes are cleared from BGP and RIB table of DUT when
18 redistribute static configuration is removed.
19 5. Shut BGP neighbors one by one and verify BGP and routing table updated
21 6. Delete static routes and verify routers are cleared from BGP table and RIB
23 7. Verify routes are cleared from BGP and RIB table of DUT when advertise
24 network configuration is removed.
31 # Save the Current Working Directory to find configuration files.
32 CWD
= os
.path
.dirname(os
.path
.realpath(__file__
))
33 sys
.path
.append(os
.path
.join(CWD
, "../"))
34 sys
.path
.append(os
.path
.join(CWD
, "../../"))
36 # pylint: disable=C0413
37 # Import topogen and topotest helpers
38 from lib
.topogen
import Topogen
, get_topogen
40 from lib
.common_config
import (
48 reset_config_on_routers
,
49 required_linux_kernel_version
,
51 from lib
.topolog
import logger
52 from lib
.bgp
import verify_bgp_convergence
, create_router_bgp
, clear_bgp
53 from lib
.topojson
import build_config_from_json
56 pytestmark
= [pytest
.mark
.bgpd
, pytest
.mark
.staticd
]
60 NEXT_HOPS
= {"ipv4": [], "ipv6": []}
63 NETWORK
= {"ipv4": "11.0.20.1/32", "ipv6": "1::/64"}
64 NEXT_HOP_IP
= {"ipv4": "10.0.0.1", "ipv6": "fd00::1"}
65 BGP_CONVERGENCE
= False
68 def setup_module(mod
):
70 Sets up the pytest environment.
74 global NEXT_HOPS
, INTF_LIST_R3
, INTF_LIST_R2
, TEST_STATIC
77 # Required linux kernel version for this suite to run.
78 result
= required_linux_kernel_version("4.15")
79 if result
is not True:
80 pytest
.skip("Kernel requirements are not met, kernel version should be >=4.15")
82 testsuite_run_time
= time
.asctime(time
.localtime(time
.time()))
83 logger
.info("Testsuite start time: {}".format(testsuite_run_time
))
86 logger
.info("Running setup_module to create topology")
88 # This function initiates the topology build with Topogen...
89 json_file
= "{}/ibgp_ecmp_topo2.json".format(CWD
)
90 tgen
= Topogen(json_file
, mod
.__name
__)
94 # Starting topology, create tmp files which are loaded to routers
95 # to start daemons and then start routers
98 # Creating configuration from JSON
99 build_config_from_json(tgen
, topo
)
101 # Don't run this test if we have any failure.
102 if tgen
.routers_have_failure():
103 pytest
.skip(tgen
.errors
)
106 # Api call verify whether BGP is converged
107 ADDR_TYPES
= check_address_types()
109 for addr_type
in ADDR_TYPES
:
110 BGP_CONVERGENCE
= verify_bgp_convergence(tgen
, topo
)
111 assert BGP_CONVERGENCE
is True, "setup_module :Failed \n Error:" " {}".format(
116 val
for links
, val
in topo
["routers"]["r2"]["links"].items() if "r3" in links
118 for adt
in ADDR_TYPES
:
119 NEXT_HOPS
[adt
] = [val
[adt
].split("/")[0] for val
in link_data
]
121 NEXT_HOPS
[adt
] = sorted(NEXT_HOPS
[adt
], key
=lambda x
: int(x
.split(".")[2]))
123 NEXT_HOPS
[adt
] = sorted(
124 NEXT_HOPS
[adt
], key
=lambda x
: int(x
.split(":")[-3], 16)
127 INTF_LIST_R2
= [val
["interface"].split("/")[0] for val
in link_data
]
128 INTF_LIST_R2
= sorted(INTF_LIST_R2
, key
=lambda x
: int(x
.split("eth")[1]))
131 val
for links
, val
in topo
["routers"]["r3"]["links"].items() if "r2" in links
133 INTF_LIST_R3
= [val
["interface"].split("/")[0] for val
in link_data
]
134 INTF_LIST_R3
= sorted(INTF_LIST_R3
, key
=lambda x
: int(x
.split("eth")[1]))
136 # STATIC_ROUTE = True
137 logger
.info("Running setup_module() done")
140 def teardown_module():
142 Teardown the pytest environment.
147 logger
.info("Running teardown_module to delete topology")
151 # Stop toplogy and Remove tmp files
155 def static_or_nw(tgen
, topo
, tc_name
, test_type
, dut
):
157 if test_type
== "redist_static":
158 input_dict_static
= {
161 {"network": NETWORK
["ipv4"], "next_hop": NEXT_HOP_IP
["ipv4"]},
162 {"network": NETWORK
["ipv6"], "next_hop": NEXT_HOP_IP
["ipv6"]},
166 logger
.info("Configuring static route on router %s", dut
)
167 result
= create_static_routes(tgen
, input_dict_static
)
168 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
177 "unicast": {"redistribute": [{"redist_type": "static"}]}
180 "unicast": {"redistribute": [{"redist_type": "static"}]}
187 logger
.info("Configuring redistribute static route on router %s", dut
)
188 result
= create_router_bgp(tgen
, topo
, input_dict_2
)
189 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
193 elif test_type
== "advertise_nw":
200 "advertise_networks": [{"network": NETWORK
["ipv4"]}]
205 "advertise_networks": [{"network": NETWORK
["ipv6"]}]
214 "Advertising networks %s %s from router %s",
219 result
= create_router_bgp(tgen
, topo
, input_dict_nw
)
220 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
225 @pytest.mark
.parametrize("ecmp_num", ["8", "16", "32"])
226 @pytest.mark
.parametrize("test_type", ["redist_static", "advertise_nw"])
227 def test_modify_ecmp_max_paths(request
, ecmp_num
, test_type
):
229 Verify routes installed as per maximum-paths
230 configuration (8/16/32).
233 tc_name
= request
.node
.name
234 write_test_header(tc_name
)
237 reset_config_on_routers(tgen
)
239 static_or_nw(tgen
, topo
, tc_name
, test_type
, "r2")
264 logger
.info("Configuring bgp maximum-paths %s on router r3", ecmp_num
)
265 result
= create_router_bgp(tgen
, topo
, input_dict
)
266 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
268 # Verifying RIB routes
272 for addr_type
in ADDR_TYPES
:
273 input_dict_1
= {"r3": {"static_routes": [{"network": NETWORK
[addr_type
]}]}}
275 logger
.info("Verifying %s routes on r3", addr_type
)
277 # Test only the count of nexthops, not the specific nexthop addresses -
278 # they're not deterministic
285 next_hop
=NEXT_HOPS
[addr_type
][: int(ecmp_num
)],
290 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
294 write_test_footer(tc_name
)
297 @pytest.mark
.parametrize("ecmp_num", ["8", "16", "32"])
298 @pytest.mark
.parametrize("test_type", ["redist_static", "advertise_nw"])
299 def test_ecmp_after_clear_bgp(request
, ecmp_num
, test_type
):
300 """Verify BGP table and RIB in DUT after clear BGP routes and neighbors"""
302 tc_name
= request
.node
.name
303 write_test_header(tc_name
)
306 reset_config_on_routers(tgen
)
308 # Verifying RIB routes
312 static_or_nw(tgen
, topo
, tc_name
, test_type
, "r2")
313 for addr_type
in ADDR_TYPES
:
314 input_dict_1
= {"r3": {"static_routes": [{"network": NETWORK
[addr_type
]}]}}
316 logger
.info("Verifying %s routes on r3", addr_type
)
322 next_hop
=NEXT_HOPS
[addr_type
][: int(ecmp_num
)],
325 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
330 for addr_type
in ADDR_TYPES
:
331 clear_bgp(tgen
, addr_type
, dut
)
333 # Verify BGP convergence
334 result
= verify_bgp_convergence(tgen
, topo
)
335 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
337 for addr_type
in ADDR_TYPES
:
338 input_dict_1
= {"r3": {"static_routes": [{"network": NETWORK
[addr_type
]}]}}
339 logger
.info("Verifying %s routes on r3", addr_type
)
345 next_hop
=NEXT_HOPS
[addr_type
][: int(ecmp_num
)],
348 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
352 write_test_footer(tc_name
)
355 def test_ecmp_remove_redistribute_static(request
):
356 """Verify routes are cleared from BGP and RIB table of DUT when
357 redistribute static configuration is removed."""
359 tc_name
= request
.node
.name
360 write_test_header(tc_name
)
363 reset_config_on_routers(tgen
)
364 static_or_nw(tgen
, topo
, tc_name
, "redist_static", "r2")
365 for addr_type
in ADDR_TYPES
:
367 # Verifying RIB routes
370 input_dict_1
= {"r3": {"static_routes": [{"network": NETWORK
[addr_type
]}]}}
372 logger
.info("Verifying %s routes on r3", addr_type
)
378 next_hop
=NEXT_HOPS
[addr_type
],
381 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
391 "redistribute": [{"redist_type": "static", "delete": True}]
396 "redistribute": [{"redist_type": "static", "delete": True}]
404 logger
.info("Remove redistribute static")
405 result
= create_router_bgp(tgen
, topo
, input_dict_2
)
406 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
408 for addr_type
in ADDR_TYPES
:
410 # Verifying RIB routes
413 input_dict_1
= {"r3": {"static_routes": [{"network": NETWORK
[addr_type
]}]}}
415 logger
.info("Verifying %s routes on r3 are deleted", addr_type
)
427 ), "Testcase {} : Failed \n Expected: Routes still present in {} RIB. Found: {}".format(
431 logger
.info("Enable redistribute static")
436 "ipv4": {"unicast": {"redistribute": [{"redist_type": "static"}]}},
437 "ipv6": {"unicast": {"redistribute": [{"redist_type": "static"}]}},
442 result
= create_router_bgp(tgen
, topo
, input_dict_2
)
443 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
445 for addr_type
in ADDR_TYPES
:
446 # Verifying RIB routes
449 input_dict_1
= {"r3": {"static_routes": [{"network": NETWORK
[addr_type
]}]}}
450 logger
.info("Verifying %s routes on r3", addr_type
)
456 next_hop
=NEXT_HOPS
[addr_type
],
459 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
463 write_test_footer(tc_name
)
466 @pytest.mark
.parametrize("test_type", ["redist_static", "advertise_nw"])
467 def test_ecmp_shut_bgp_neighbor(request
, test_type
):
468 """Shut BGP neighbors one by one and verify BGP and routing table updated
469 accordingly in DUT"""
471 tc_name
= request
.node
.name
472 write_test_header(tc_name
)
475 logger
.info(INTF_LIST_R2
)
476 # Verifying RIB routes
480 reset_config_on_routers(tgen
)
481 static_or_nw(tgen
, topo
, tc_name
, test_type
, "r2")
483 for addr_type
in ADDR_TYPES
:
484 input_dict
= {"r3": {"static_routes": [{"network": NETWORK
[addr_type
]}]}}
486 logger
.info("Verifying %s routes on r3", addr_type
)
492 next_hop
=NEXT_HOPS
[addr_type
],
495 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
499 for intf_num
in range(len(INTF_LIST_R2
) + 1, 16):
500 intf_val
= INTF_LIST_R2
[intf_num
: intf_num
+ 16]
502 input_dict_1
= {"r2": {"interface_list": [intf_val
], "status": "down"}}
503 logger
.info("Shutting down neighbor interface {} on r2".format(intf_val
))
504 result
= interface_status(tgen
, topo
, input_dict_1
)
505 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
509 for addr_type
in ADDR_TYPES
:
510 if intf_num
+ 16 < 32:
511 check_hops
= NEXT_HOPS
[addr_type
]
515 input_dict
= {"r3": {"static_routes": [{"network": NETWORK
[addr_type
]}]}}
516 logger
.info("Verifying %s routes on r3", addr_type
)
518 tgen
, addr_type
, dut
, input_dict
, next_hop
=check_hops
, protocol
=protocol
520 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
524 input_dict_1
= {"r2": {"interface_list": INTF_LIST_R2
, "status": "up"}}
526 logger
.info("Enabling all neighbor interface {} on r2")
527 result
= interface_status(tgen
, topo
, input_dict_1
)
528 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
530 static_or_nw(tgen
, topo
, tc_name
, test_type
, "r2")
531 for addr_type
in ADDR_TYPES
:
532 input_dict
= {"r3": {"static_routes": [{"network": NETWORK
[addr_type
]}]}}
534 logger
.info("Verifying %s routes on r3", addr_type
)
540 next_hop
=NEXT_HOPS
[addr_type
],
543 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
547 write_test_footer(tc_name
)
550 def test_ecmp_remove_static_route(request
):
552 Delete static routes and verify routers are cleared from BGP table,
556 tc_name
= request
.node
.name
557 write_test_header(tc_name
)
560 # Verifying RIB routes
564 reset_config_on_routers(tgen
)
566 static_or_nw(tgen
, topo
, tc_name
, "redist_static", "r2")
567 for addr_type
in ADDR_TYPES
:
568 input_dict_1
= {"r3": {"static_routes": [{"network": NETWORK
[addr_type
]}]}}
570 logger
.info("Verifying %s routes on r3", addr_type
)
576 next_hop
=NEXT_HOPS
[addr_type
],
579 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
583 for addr_type
in ADDR_TYPES
:
588 "network": NETWORK
[addr_type
],
589 "next_hop": NEXT_HOP_IP
[addr_type
],
596 logger
.info("Remove static routes")
597 result
= create_static_routes(tgen
, input_dict_2
)
598 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
602 logger
.info("Verifying %s routes on r3 are removed", addr_type
)
614 ), "Testcase {} : Failed \n Expected: Routes still present in {} RIB. Found: {}".format(
618 for addr_type
in ADDR_TYPES
:
619 # Enable static routes
623 {"network": NETWORK
[addr_type
], "next_hop": NEXT_HOP_IP
[addr_type
]}
628 logger
.info("Enable static route")
629 result
= create_static_routes(tgen
, input_dict_4
)
630 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
634 logger
.info("Verifying %s routes on r3", addr_type
)
640 next_hop
=NEXT_HOPS
[addr_type
],
643 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
647 write_test_footer(tc_name
)
650 def test_ecmp_remove_nw_advertise(request
):
652 Verify routes are cleared from BGP and RIB table of DUT,
653 when advertise network configuration is removed
656 tc_name
= request
.node
.name
657 write_test_header(tc_name
)
660 # Verifying RIB routes
664 reset_config_on_routers(tgen
)
665 static_or_nw(tgen
, topo
, tc_name
, "advertise_nw", "r2")
666 for addr_type
in ADDR_TYPES
:
667 input_dict
= {"r3": {"static_routes": [{"network": NETWORK
[addr_type
]}]}}
669 logger
.info("Verifying %s routes on r3", addr_type
)
675 next_hop
=NEXT_HOPS
[addr_type
],
678 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
688 "advertise_networks": [
689 {"network": NETWORK
["ipv4"], "delete": True}
695 "advertise_networks": [
696 {"network": NETWORK
["ipv6"], "delete": True}
705 logger
.info("Withdraw advertised networks")
706 result
= create_router_bgp(tgen
, topo
, input_dict_3
)
707 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
709 for addr_type
in ADDR_TYPES
:
710 input_dict
= {"r3": {"static_routes": [{"network": NETWORK
[addr_type
]}]}}
712 logger
.info("Verifying %s routes on r3", addr_type
)
724 ), "Testcase {} : Failed \n Expected: Routes still present in {} RIB. Found: {}".format(
728 static_or_nw(tgen
, topo
, tc_name
, "advertise_nw", "r2")
729 for addr_type
in ADDR_TYPES
:
730 input_dict
= {"r3": {"static_routes": [{"network": NETWORK
[addr_type
]}]}}
731 logger
.info("Verifying %s routes on r3", addr_type
)
737 next_hop
=NEXT_HOPS
[addr_type
],
740 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
745 if __name__
== "__main__":
746 args
= ["-s"] + sys
.argv
[1:]
747 sys
.exit(pytest
.main(args
))