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, kernel version should be >=4.15")
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 Expected: Routes still present in {} RIB. Found: {}".format(
443 logger
.info("Enable redistribute static")
448 "ipv4": {"unicast": {"redistribute": [{"redist_type": "static"}]}},
449 "ipv6": {"unicast": {"redistribute": [{"redist_type": "static"}]}},
454 result
= create_router_bgp(tgen
, topo
, input_dict_2
)
455 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
457 for addr_type
in ADDR_TYPES
:
458 # Verifying RIB routes
461 input_dict_1
= {"r3": {"static_routes": [{"network": NETWORK
[addr_type
]}]}}
462 logger
.info("Verifying %s routes on r3", addr_type
)
468 next_hop
=NEXT_HOPS
[addr_type
],
471 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
475 write_test_footer(tc_name
)
478 @pytest.mark
.parametrize("test_type", ["redist_static", "advertise_nw"])
479 def test_ecmp_shut_bgp_neighbor(request
, test_type
):
480 """Shut BGP neighbors one by one and verify BGP and routing table updated
481 accordingly in DUT"""
483 tc_name
= request
.node
.name
484 write_test_header(tc_name
)
487 logger
.info(INTF_LIST_R2
)
488 # Verifying RIB routes
492 reset_config_on_routers(tgen
)
493 static_or_nw(tgen
, topo
, tc_name
, test_type
, "r2")
495 for addr_type
in ADDR_TYPES
:
496 input_dict
= {"r3": {"static_routes": [{"network": NETWORK
[addr_type
]}]}}
498 logger
.info("Verifying %s routes on r3", addr_type
)
504 next_hop
=NEXT_HOPS
[addr_type
],
507 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
511 for intf_num
in range(len(INTF_LIST_R2
) + 1, 16):
512 intf_val
= INTF_LIST_R2
[intf_num
: intf_num
+ 16]
514 input_dict_1
= {"r2": {"interface_list": [intf_val
], "status": "down"}}
515 logger
.info("Shutting down neighbor interface {} on r2".format(intf_val
))
516 result
= interface_status(tgen
, topo
, input_dict_1
)
517 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
521 for addr_type
in ADDR_TYPES
:
522 if intf_num
+ 16 < 32:
523 check_hops
= NEXT_HOPS
[addr_type
]
527 input_dict
= {"r3": {"static_routes": [{"network": NETWORK
[addr_type
]}]}}
528 logger
.info("Verifying %s routes on r3", addr_type
)
530 tgen
, addr_type
, dut
, input_dict
, next_hop
=check_hops
, protocol
=protocol
532 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
536 input_dict_1
= {"r2": {"interface_list": INTF_LIST_R2
, "status": "up"}}
538 logger
.info("Enabling all neighbor interface {} on r2")
539 result
= interface_status(tgen
, topo
, input_dict_1
)
540 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
542 static_or_nw(tgen
, topo
, tc_name
, test_type
, "r2")
543 for addr_type
in ADDR_TYPES
:
544 input_dict
= {"r3": {"static_routes": [{"network": NETWORK
[addr_type
]}]}}
546 logger
.info("Verifying %s routes on r3", addr_type
)
552 next_hop
=NEXT_HOPS
[addr_type
],
555 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
559 write_test_footer(tc_name
)
562 def test_ecmp_remove_static_route(request
):
564 Delete static routes and verify routers are cleared from BGP table,
568 tc_name
= request
.node
.name
569 write_test_header(tc_name
)
572 # Verifying RIB routes
576 reset_config_on_routers(tgen
)
578 static_or_nw(tgen
, topo
, tc_name
, "redist_static", "r2")
579 for addr_type
in ADDR_TYPES
:
580 input_dict_1
= {"r3": {"static_routes": [{"network": NETWORK
[addr_type
]}]}}
582 logger
.info("Verifying %s routes on r3", addr_type
)
588 next_hop
=NEXT_HOPS
[addr_type
],
591 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
595 for addr_type
in ADDR_TYPES
:
600 "network": NETWORK
[addr_type
],
601 "next_hop": NEXT_HOP_IP
[addr_type
],
608 logger
.info("Remove static routes")
609 result
= create_static_routes(tgen
, input_dict_2
)
610 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
614 logger
.info("Verifying %s routes on r3 are removed", addr_type
)
626 ), "Testcase {} : Failed \n Expected: Routes still present in {} RIB. Found: {}".format(
630 for addr_type
in ADDR_TYPES
:
631 # Enable static routes
635 {"network": NETWORK
[addr_type
], "next_hop": NEXT_HOP_IP
[addr_type
]}
640 logger
.info("Enable static route")
641 result
= create_static_routes(tgen
, input_dict_4
)
642 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
646 logger
.info("Verifying %s routes on r3", addr_type
)
652 next_hop
=NEXT_HOPS
[addr_type
],
655 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
660 def test_ecmp_remove_nw_advertise(request
):
662 Verify routes are cleared from BGP and RIB table of DUT,
663 when advertise network configuration is removed
666 tc_name
= request
.node
.name
667 write_test_header(tc_name
)
670 # Verifying RIB routes
674 reset_config_on_routers(tgen
)
675 static_or_nw(tgen
, topo
, tc_name
, "advertise_nw", "r2")
676 for addr_type
in ADDR_TYPES
:
677 input_dict
= {"r3": {"static_routes": [{"network": NETWORK
[addr_type
]}]}}
679 logger
.info("Verifying %s routes on r3", addr_type
)
685 next_hop
=NEXT_HOPS
[addr_type
],
688 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
698 "advertise_networks": [
699 {"network": NETWORK
["ipv4"], "delete": True}
705 "advertise_networks": [
706 {"network": NETWORK
["ipv6"], "delete": True}
715 logger
.info("Withdraw advertised networks")
716 result
= create_router_bgp(tgen
, topo
, input_dict_3
)
717 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
719 for addr_type
in ADDR_TYPES
:
720 input_dict
= {"r3": {"static_routes": [{"network": NETWORK
[addr_type
]}]}}
722 logger
.info("Verifying %s routes on r3", addr_type
)
734 ), "Testcase {} : Failed \n Expected: Routes still present in {} RIB. Found: {}".format(
738 static_or_nw(tgen
, topo
, tc_name
, "advertise_nw", "r2")
739 for addr_type
in ADDR_TYPES
:
740 input_dict
= {"r3": {"static_routes": [{"network": NETWORK
[addr_type
]}]}}
741 logger
.info("Verifying %s routes on r3", addr_type
)
747 next_hop
=NEXT_HOPS
[addr_type
],
750 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
754 write_test_footer(tc_name
)
757 if __name__
== "__main__":
758 args
= ["-s"] + sys
.argv
[1:]
759 sys
.exit(pytest
.main(args
))