4 # Copyright (c) 2019 by VMware, Inc. ("VMware")
5 # Used Copyright (c) 2018 by Network Device Education Foundation, Inc.
6 # ("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
25 Following tests are covered to test ecmp functionality on EBGP.
26 1. Verify routes installed as per maximum-paths configuration (8/16/32)
27 2. Disable/Shut selected paths nexthops and verify other next are installed in
28 the RIB of DUT. Enable interfaces and verify RIB count.
29 3. Verify BGP table and RIB in DUT after clear BGP routes and neighbors.
30 4. Verify routes are cleared from BGP and RIB table of DUT when
31 redistribute static configuration is removed.
32 5. Shut BGP neighbors one by one and verify BGP and routing table updated
34 6. Delete static routes and verify routers are cleared from BGP table and RIB
36 7. Verify routes are cleared from BGP and RIB table of DUT when advertise
37 network configuration is removed.
44 # Save the Current Working Directory to find configuration files.
45 CWD
= os
.path
.dirname(os
.path
.realpath(__file__
))
46 sys
.path
.append(os
.path
.join(CWD
, "../"))
47 sys
.path
.append(os
.path
.join(CWD
, "../../"))
49 # pylint: disable=C0413
50 # Import topogen and topotest helpers
51 from lib
.topogen
import Topogen
, get_topogen
53 from lib
.common_config
import (
61 reset_config_on_routers
,
62 required_linux_kernel_version
,
64 from lib
.topolog
import logger
65 from lib
.bgp
import verify_bgp_convergence
, create_router_bgp
, clear_bgp
66 from lib
.topojson
import build_config_from_json
69 pytestmark
= [pytest
.mark
.bgpd
, pytest
.mark
.staticd
]
73 NEXT_HOPS
= {"ipv4": [], "ipv6": []}
76 NETWORK
= {"ipv4": "11.0.20.1/32", "ipv6": "1::/64"}
77 NEXT_HOP_IP
= {"ipv4": "10.0.0.1", "ipv6": "fd00::1"}
78 BGP_CONVERGENCE
= False
81 def setup_module(mod
):
83 Sets up the pytest environment.
87 global NEXT_HOPS
, INTF_LIST_R3
, INTF_LIST_R2
, TEST_STATIC
90 # Required linux kernel version for this suite to run.
91 result
= required_linux_kernel_version("4.15")
92 if result
is not True:
93 pytest
.skip("Kernel requirements are not met")
95 testsuite_run_time
= time
.asctime(time
.localtime(time
.time()))
96 logger
.info("Testsuite start time: {}".format(testsuite_run_time
))
99 logger
.info("Running setup_module to create topology")
101 # This function initiates the topology build with Topogen...
102 json_file
= "{}/ebgp_ecmp_topo2.json".format(CWD
)
103 tgen
= Topogen(json_file
, mod
.__name
__)
105 topo
= tgen
.json_topo
107 # Starting topology, create tmp files which are loaded to routers
108 # to start daemons and then start routers
111 # Creating configuration from JSON
112 build_config_from_json(tgen
, topo
)
114 # Don't run this test if we have any failure.
115 if tgen
.routers_have_failure():
116 pytest
.skip(tgen
.errors
)
119 # Api call verify whether BGP is converged
120 ADDR_TYPES
= check_address_types()
122 BGP_CONVERGENCE
= verify_bgp_convergence(tgen
, topo
)
123 assert BGP_CONVERGENCE
is True, "setup_module :Failed \n Error:" " {}".format(
128 val
for links
, val
in topo
["routers"]["r2"]["links"].items() if "r3" in links
130 for adt
in ADDR_TYPES
:
131 NEXT_HOPS
[adt
] = [val
[adt
].split("/")[0] for val
in link_data
]
133 NEXT_HOPS
[adt
] = sorted(NEXT_HOPS
[adt
], key
=lambda x
: int(x
.split(".")[2]))
135 NEXT_HOPS
[adt
] = sorted(
136 NEXT_HOPS
[adt
], key
=lambda x
: int(x
.split(":")[-3], 16)
139 INTF_LIST_R2
= [val
["interface"].split("/")[0] for val
in link_data
]
140 INTF_LIST_R2
= sorted(INTF_LIST_R2
, key
=lambda x
: int(x
.split("eth")[1]))
143 val
for links
, val
in topo
["routers"]["r3"]["links"].items() if "r2" in links
145 INTF_LIST_R3
= [val
["interface"].split("/")[0] for val
in link_data
]
146 INTF_LIST_R3
= sorted(INTF_LIST_R3
, key
=lambda x
: int(x
.split("eth")[1]))
148 # STATIC_ROUTE = True
149 logger
.info("Running setup_module() done")
152 def teardown_module():
154 Teardown the pytest environment.
159 logger
.info("Running teardown_module to delete topology")
163 # Stop toplogy and Remove tmp files
167 def static_or_nw(tgen
, topo
, tc_name
, test_type
, dut
):
169 if test_type
== "redist_static":
170 input_dict_static
= {
173 {"network": NETWORK
["ipv4"], "next_hop": NEXT_HOP_IP
["ipv4"]},
174 {"network": NETWORK
["ipv6"], "next_hop": NEXT_HOP_IP
["ipv6"]},
178 logger
.info("Configuring static route on router %s", dut
)
179 result
= create_static_routes(tgen
, input_dict_static
)
180 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
189 "unicast": {"redistribute": [{"redist_type": "static"}]}
192 "unicast": {"redistribute": [{"redist_type": "static"}]}
199 logger
.info("Configuring redistribute static route on router %s", dut
)
200 result
= create_router_bgp(tgen
, topo
, input_dict_2
)
201 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
205 elif test_type
== "advertise_nw":
212 "advertise_networks": [{"network": NETWORK
["ipv4"]}]
217 "advertise_networks": [{"network": NETWORK
["ipv6"]}]
226 "Advertising networks %s %s from router %s",
231 result
= create_router_bgp(tgen
, topo
, input_dict_nw
)
232 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
237 @pytest.mark
.parametrize("ecmp_num", ["8", "16", "32"])
238 @pytest.mark
.parametrize("test_type", ["redist_static", "advertise_nw"])
239 def test_modify_ecmp_max_paths(request
, ecmp_num
, test_type
):
241 Verify routes installed as per maximum-paths
242 configuration (8/16/32).
245 tc_name
= request
.node
.name
246 write_test_header(tc_name
)
249 reset_config_on_routers(tgen
)
251 static_or_nw(tgen
, topo
, tc_name
, test_type
, "r2")
276 logger
.info("Configuring bgp maximum-paths %s on router r3", ecmp_num
)
277 result
= create_router_bgp(tgen
, topo
, input_dict
)
278 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
280 # Verifying RIB routes
284 for addr_type
in ADDR_TYPES
:
285 input_dict_1
= {"r3": {"static_routes": [{"network": NETWORK
[addr_type
]}]}}
287 logger
.info("Verifying %s routes on r3", addr_type
)
289 # Only test the count of nexthops; the actual nexthop addresses
290 # can vary and are not deterministic.
297 next_hop
=NEXT_HOPS
[addr_type
][: int(ecmp_num
)],
302 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
306 write_test_footer(tc_name
)
309 @pytest.mark
.parametrize("ecmp_num", ["8", "16", "32"])
310 @pytest.mark
.parametrize("test_type", ["redist_static", "advertise_nw"])
311 def test_ecmp_after_clear_bgp(request
, ecmp_num
, test_type
):
312 """Verify BGP table and RIB in DUT after clear BGP routes and neighbors"""
314 tc_name
= request
.node
.name
315 write_test_header(tc_name
)
318 reset_config_on_routers(tgen
)
320 # Verifying RIB routes
324 static_or_nw(tgen
, topo
, tc_name
, test_type
, "r2")
325 for addr_type
in ADDR_TYPES
:
326 input_dict_1
= {"r3": {"static_routes": [{"network": NETWORK
[addr_type
]}]}}
328 logger
.info("Verifying %s routes on r3", addr_type
)
334 next_hop
=NEXT_HOPS
[addr_type
][: int(ecmp_num
)],
337 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
342 for addr_type
in ADDR_TYPES
:
343 clear_bgp(tgen
, addr_type
, dut
)
345 # Verify BGP convergence
346 result
= verify_bgp_convergence(tgen
, topo
)
347 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
349 for addr_type
in ADDR_TYPES
:
350 input_dict_1
= {"r3": {"static_routes": [{"network": NETWORK
[addr_type
]}]}}
351 logger
.info("Verifying %s routes on r3", addr_type
)
357 next_hop
=NEXT_HOPS
[addr_type
][: int(ecmp_num
)],
360 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
364 write_test_footer(tc_name
)
367 def test_ecmp_remove_redistribute_static(request
):
368 """Verify routes are cleared from BGP and RIB table of DUT when
369 redistribute static configuration is removed."""
371 tc_name
= request
.node
.name
372 write_test_header(tc_name
)
375 reset_config_on_routers(tgen
)
376 static_or_nw(tgen
, topo
, tc_name
, "redist_static", "r2")
377 for addr_type
in ADDR_TYPES
:
379 # Verifying RIB routes
382 input_dict_1
= {"r3": {"static_routes": [{"network": NETWORK
[addr_type
]}]}}
384 logger
.info("Verifying %s routes on r3", addr_type
)
390 next_hop
=NEXT_HOPS
[addr_type
],
393 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
403 "redistribute": [{"redist_type": "static", "delete": True}]
408 "redistribute": [{"redist_type": "static", "delete": True}]
416 logger
.info("Remove redistribute static")
417 result
= create_router_bgp(tgen
, topo
, input_dict_2
)
418 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
420 for addr_type
in ADDR_TYPES
:
422 # Verifying RIB routes
425 input_dict_1
= {"r3": {"static_routes": [{"network": NETWORK
[addr_type
]}]}}
427 logger
.info("Verifying %s routes on r3 are deleted", addr_type
)
439 ), "Testcase {} : Failed \n Routes still" " present in RIB".format(tc_name
)
441 logger
.info("Enable redistribute static")
446 "ipv4": {"unicast": {"redistribute": [{"redist_type": "static"}]}},
447 "ipv6": {"unicast": {"redistribute": [{"redist_type": "static"}]}},
452 result
= create_router_bgp(tgen
, topo
, input_dict_2
)
453 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
455 for addr_type
in ADDR_TYPES
:
456 # Verifying RIB routes
459 input_dict_1
= {"r3": {"static_routes": [{"network": NETWORK
[addr_type
]}]}}
460 logger
.info("Verifying %s routes on r3", addr_type
)
466 next_hop
=NEXT_HOPS
[addr_type
],
469 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
473 write_test_footer(tc_name
)
476 @pytest.mark
.parametrize("test_type", ["redist_static", "advertise_nw"])
477 def test_ecmp_shut_bgp_neighbor(request
, test_type
):
478 """Shut BGP neighbors one by one and verify BGP and routing table updated
479 accordingly in DUT"""
481 tc_name
= request
.node
.name
482 write_test_header(tc_name
)
485 logger
.info(INTF_LIST_R2
)
486 # Verifying RIB routes
490 reset_config_on_routers(tgen
)
491 static_or_nw(tgen
, topo
, tc_name
, test_type
, "r2")
493 for addr_type
in ADDR_TYPES
:
494 input_dict
= {"r3": {"static_routes": [{"network": NETWORK
[addr_type
]}]}}
496 logger
.info("Verifying %s routes on r3", addr_type
)
502 next_hop
=NEXT_HOPS
[addr_type
],
505 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
509 for intf_num
in range(len(INTF_LIST_R2
) + 1, 16):
510 intf_val
= INTF_LIST_R2
[intf_num
: intf_num
+ 16]
512 input_dict_1
= {"r2": {"interface_list": [intf_val
], "status": "down"}}
513 logger
.info("Shutting down neighbor interface {} on r2".format(intf_val
))
514 result
= interface_status(tgen
, topo
, input_dict_1
)
515 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
519 for addr_type
in ADDR_TYPES
:
520 if intf_num
+ 16 < 32:
521 check_hops
= NEXT_HOPS
[addr_type
]
525 input_dict
= {"r3": {"static_routes": [{"network": NETWORK
[addr_type
]}]}}
526 logger
.info("Verifying %s routes on r3", addr_type
)
528 tgen
, addr_type
, dut
, input_dict
, next_hop
=check_hops
, protocol
=protocol
530 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
534 input_dict_1
= {"r2": {"interface_list": INTF_LIST_R2
, "status": "up"}}
536 logger
.info("Enabling all neighbor interface {} on r2")
537 result
= interface_status(tgen
, topo
, input_dict_1
)
538 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
540 static_or_nw(tgen
, topo
, tc_name
, test_type
, "r2")
541 for addr_type
in ADDR_TYPES
:
542 input_dict
= {"r3": {"static_routes": [{"network": NETWORK
[addr_type
]}]}}
544 logger
.info("Verifying %s routes on r3", addr_type
)
550 next_hop
=NEXT_HOPS
[addr_type
],
553 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
557 write_test_footer(tc_name
)
560 def test_ecmp_remove_static_route(request
):
562 Delete static routes and verify routers are cleared from BGP table,
566 tc_name
= request
.node
.name
567 write_test_header(tc_name
)
570 # Verifying RIB routes
574 reset_config_on_routers(tgen
)
576 static_or_nw(tgen
, topo
, tc_name
, "redist_static", "r2")
577 for addr_type
in ADDR_TYPES
:
578 input_dict_1
= {"r3": {"static_routes": [{"network": NETWORK
[addr_type
]}]}}
580 logger
.info("Verifying %s routes on r3", addr_type
)
586 next_hop
=NEXT_HOPS
[addr_type
],
589 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
593 for addr_type
in ADDR_TYPES
:
598 "network": NETWORK
[addr_type
],
599 "next_hop": NEXT_HOP_IP
[addr_type
],
606 logger
.info("Remove static routes")
607 result
= create_static_routes(tgen
, input_dict_2
)
608 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
612 logger
.info("Verifying %s routes on r3 are removed", addr_type
)
624 ), "Testcase {} : Failed \n Routes still" " present in RIB".format(tc_name
)
626 for addr_type
in ADDR_TYPES
:
627 # Enable static routes
631 {"network": NETWORK
[addr_type
], "next_hop": NEXT_HOP_IP
[addr_type
]}
636 logger
.info("Enable static route")
637 result
= create_static_routes(tgen
, input_dict_4
)
638 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
642 logger
.info("Verifying %s routes on r3", addr_type
)
648 next_hop
=NEXT_HOPS
[addr_type
],
651 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
656 def test_ecmp_remove_nw_advertise(request
):
658 Verify routes are cleared from BGP and RIB table of DUT,
659 when advertise network configuration is removed
662 tc_name
= request
.node
.name
663 write_test_header(tc_name
)
666 # Verifying RIB routes
670 reset_config_on_routers(tgen
)
671 static_or_nw(tgen
, topo
, tc_name
, "advertise_nw", "r2")
672 for addr_type
in ADDR_TYPES
:
673 input_dict
= {"r3": {"static_routes": [{"network": NETWORK
[addr_type
]}]}}
675 logger
.info("Verifying %s routes on r3", addr_type
)
681 next_hop
=NEXT_HOPS
[addr_type
],
684 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
694 "advertise_networks": [
695 {"network": NETWORK
["ipv4"], "delete": True}
701 "advertise_networks": [
702 {"network": NETWORK
["ipv6"], "delete": True}
711 logger
.info("Withdraw advertised networks")
712 result
= create_router_bgp(tgen
, topo
, input_dict_3
)
713 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
715 for addr_type
in ADDR_TYPES
:
716 input_dict
= {"r3": {"static_routes": [{"network": NETWORK
[addr_type
]}]}}
718 logger
.info("Verifying %s routes on r3", addr_type
)
730 ), "Testcase {} : Failed \n Routes still" " present in RIB".format(tc_name
)
732 static_or_nw(tgen
, topo
, tc_name
, "advertise_nw", "r2")
733 for addr_type
in ADDR_TYPES
:
734 input_dict
= {"r3": {"static_routes": [{"network": NETWORK
[addr_type
]}]}}
735 logger
.info("Verifying %s routes on r3", addr_type
)
741 next_hop
=NEXT_HOPS
[addr_type
],
744 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
748 write_test_footer(tc_name
)
751 if __name__
== "__main__":
752 args
= ["-s"] + sys
.argv
[1:]
753 sys
.exit(pytest
.main(args
))