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 neigbors 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
]
74 NEXT_HOPS
= {"ipv4": [], "ipv6": []}
77 NETWORK
= {"ipv4": "11.0.20.1/32", "ipv6": "1::/64"}
78 NEXT_HOP_IP
= {"ipv4": "10.0.0.1", "ipv6": "fd00::1"}
79 BGP_CONVERGENCE
= False
82 def setup_module(mod
):
84 Sets up the pytest environment.
88 global NEXT_HOPS
, INTF_LIST_R3
, INTF_LIST_R2
, TEST_STATIC
91 # Required linux kernel version for this suite to run.
92 result
= required_linux_kernel_version("4.15")
93 if result
is not True:
94 pytest
.skip("Kernel requirements are not met")
96 testsuite_run_time
= time
.asctime(time
.localtime(time
.time()))
97 logger
.info("Testsuite start time: {}".format(testsuite_run_time
))
100 logger
.info("Running setup_module to create topology")
102 # This function initiates the topology build with Topogen...
103 json_file
= "{}/ebgp_ecmp_topo2.json".format(CWD
)
104 tgen
= Topogen(json_file
, mod
.__name
__)
106 topo
= tgen
.json_topo
108 # Starting topology, create tmp files which are loaded to routers
109 # to start deamons and then start routers
112 # Creating configuration from JSON
113 build_config_from_json(tgen
, topo
)
115 # Don't run this test if we have any failure.
116 if tgen
.routers_have_failure():
117 pytest
.skip(tgen
.errors
)
120 # Api call verify whether BGP is converged
121 ADDR_TYPES
= check_address_types()
123 BGP_CONVERGENCE
= verify_bgp_convergence(tgen
, topo
)
124 assert BGP_CONVERGENCE
is True, "setup_module :Failed \n Error:" " {}".format(
129 val
for links
, val
in topo
["routers"]["r2"]["links"].items() if "r3" in links
131 for adt
in ADDR_TYPES
:
132 NEXT_HOPS
[adt
] = [val
[adt
].split("/")[0] for val
in link_data
]
134 NEXT_HOPS
[adt
] = sorted(NEXT_HOPS
[adt
], key
=lambda x
: int(x
.split(".")[2]))
136 NEXT_HOPS
[adt
] = sorted(
137 NEXT_HOPS
[adt
], key
=lambda x
: int(x
.split(":")[-3], 16)
140 INTF_LIST_R2
= [val
["interface"].split("/")[0] for val
in link_data
]
141 INTF_LIST_R2
= sorted(INTF_LIST_R2
, key
=lambda x
: int(x
.split("eth")[1]))
144 val
for links
, val
in topo
["routers"]["r3"]["links"].items() if "r2" in links
146 INTF_LIST_R3
= [val
["interface"].split("/")[0] for val
in link_data
]
147 INTF_LIST_R3
= sorted(INTF_LIST_R3
, key
=lambda x
: int(x
.split("eth")[1]))
149 # STATIC_ROUTE = True
150 logger
.info("Running setup_module() done")
153 def teardown_module():
155 Teardown the pytest environment.
160 logger
.info("Running teardown_module to delete topology")
164 # Stop toplogy and Remove tmp files
168 def static_or_nw(tgen
, topo
, tc_name
, test_type
, dut
):
170 if test_type
== "redist_static":
171 input_dict_static
= {
174 {"network": NETWORK
["ipv4"], "next_hop": NEXT_HOP_IP
["ipv4"]},
175 {"network": NETWORK
["ipv6"], "next_hop": NEXT_HOP_IP
["ipv6"]},
179 logger
.info("Configuring static route on router %s", dut
)
180 result
= create_static_routes(tgen
, input_dict_static
)
181 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
190 "unicast": {"redistribute": [{"redist_type": "static"}]}
193 "unicast": {"redistribute": [{"redist_type": "static"}]}
200 logger
.info("Configuring redistribute static route on router %s", dut
)
201 result
= create_router_bgp(tgen
, topo
, input_dict_2
)
202 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
206 elif test_type
== "advertise_nw":
213 "advertise_networks": [{"network": NETWORK
["ipv4"]}]
218 "advertise_networks": [{"network": NETWORK
["ipv6"]}]
227 "Advertising networks %s %s from router %s",
232 result
= create_router_bgp(tgen
, topo
, input_dict_nw
)
233 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
238 @pytest.mark
.parametrize("ecmp_num", ["8", "16", "32"])
239 @pytest.mark
.parametrize("test_type", ["redist_static", "advertise_nw"])
240 def test_modify_ecmp_max_paths(request
, ecmp_num
, test_type
):
242 Verify routes installed as per maximum-paths
243 configuration (8/16/32).
246 tc_name
= request
.node
.name
247 write_test_header(tc_name
)
250 reset_config_on_routers(tgen
)
252 static_or_nw(tgen
, topo
, tc_name
, test_type
, "r2")
277 logger
.info("Configuring bgp maximum-paths %s on router r3", ecmp_num
)
278 result
= create_router_bgp(tgen
, topo
, input_dict
)
279 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
281 # Verifying RIB routes
285 for addr_type
in ADDR_TYPES
:
286 input_dict_1
= {"r3": {"static_routes": [{"network": NETWORK
[addr_type
]}]}}
288 logger
.info("Verifying %s routes on r3", addr_type
)
290 # Only test the count of nexthops; the actual nexthop addresses
291 # can vary and are not deterministic.
298 next_hop
=NEXT_HOPS
[addr_type
][: int(ecmp_num
)],
303 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
307 write_test_footer(tc_name
)
310 @pytest.mark
.parametrize("ecmp_num", ["8", "16", "32"])
311 @pytest.mark
.parametrize("test_type", ["redist_static", "advertise_nw"])
312 def test_ecmp_after_clear_bgp(request
, ecmp_num
, test_type
):
313 """ Verify BGP table and RIB in DUT after clear BGP routes and neighbors"""
315 tc_name
= request
.node
.name
316 write_test_header(tc_name
)
319 reset_config_on_routers(tgen
)
321 # Verifying RIB routes
325 static_or_nw(tgen
, topo
, tc_name
, test_type
, "r2")
326 for addr_type
in ADDR_TYPES
:
327 input_dict_1
= {"r3": {"static_routes": [{"network": NETWORK
[addr_type
]}]}}
329 logger
.info("Verifying %s routes on r3", addr_type
)
335 next_hop
=NEXT_HOPS
[addr_type
][: int(ecmp_num
)],
338 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
343 for addr_type
in ADDR_TYPES
:
344 clear_bgp(tgen
, addr_type
, dut
)
346 # Verify BGP convergence
347 result
= verify_bgp_convergence(tgen
, topo
)
348 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
350 for addr_type
in ADDR_TYPES
:
351 input_dict_1
= {"r3": {"static_routes": [{"network": NETWORK
[addr_type
]}]}}
352 logger
.info("Verifying %s routes on r3", addr_type
)
358 next_hop
=NEXT_HOPS
[addr_type
][: int(ecmp_num
)],
361 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
365 write_test_footer(tc_name
)
368 def test_ecmp_remove_redistribute_static(request
):
369 """Verify routes are cleared from BGP and RIB table of DUT when
370 redistribute static configuration is removed."""
372 tc_name
= request
.node
.name
373 write_test_header(tc_name
)
376 reset_config_on_routers(tgen
)
377 static_or_nw(tgen
, topo
, tc_name
, "redist_static", "r2")
378 for addr_type
in ADDR_TYPES
:
380 # Verifying RIB routes
383 input_dict_1
= {"r3": {"static_routes": [{"network": NETWORK
[addr_type
]}]}}
385 logger
.info("Verifying %s routes on r3", addr_type
)
391 next_hop
=NEXT_HOPS
[addr_type
],
394 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
404 "redistribute": [{"redist_type": "static", "delete": True}]
409 "redistribute": [{"redist_type": "static", "delete": True}]
417 logger
.info("Remove redistribute static")
418 result
= create_router_bgp(tgen
, topo
, input_dict_2
)
419 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
421 for addr_type
in ADDR_TYPES
:
423 # Verifying RIB routes
426 input_dict_1
= {"r3": {"static_routes": [{"network": NETWORK
[addr_type
]}]}}
428 logger
.info("Verifying %s routes on r3 are deleted", addr_type
)
440 ), "Testcase {} : Failed \n Routes still" " present in RIB".format(tc_name
)
442 logger
.info("Enable redistribute static")
447 "ipv4": {"unicast": {"redistribute": [{"redist_type": "static"}]}},
448 "ipv6": {"unicast": {"redistribute": [{"redist_type": "static"}]}},
453 result
= create_router_bgp(tgen
, topo
, input_dict_2
)
454 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
456 for addr_type
in ADDR_TYPES
:
457 # Verifying RIB routes
460 input_dict_1
= {"r3": {"static_routes": [{"network": NETWORK
[addr_type
]}]}}
461 logger
.info("Verifying %s routes on r3", addr_type
)
467 next_hop
=NEXT_HOPS
[addr_type
],
470 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
474 write_test_footer(tc_name
)
477 @pytest.mark
.parametrize("test_type", ["redist_static", "advertise_nw"])
478 def test_ecmp_shut_bgp_neighbor(request
, test_type
):
479 """Shut BGP neigbors one by one and verify BGP and routing table updated
480 accordingly in DUT"""
482 tc_name
= request
.node
.name
483 write_test_header(tc_name
)
486 logger
.info(INTF_LIST_R2
)
487 # Verifying RIB routes
491 reset_config_on_routers(tgen
)
492 static_or_nw(tgen
, topo
, tc_name
, test_type
, "r2")
494 for addr_type
in ADDR_TYPES
:
495 input_dict
= {"r3": {"static_routes": [{"network": NETWORK
[addr_type
]}]}}
497 logger
.info("Verifying %s routes on r3", addr_type
)
503 next_hop
=NEXT_HOPS
[addr_type
],
506 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
510 for intf_num
in range(len(INTF_LIST_R2
) + 1, 16):
511 intf_val
= INTF_LIST_R2
[intf_num
: intf_num
+ 16]
513 input_dict_1
= {"r2": {"interface_list": [intf_val
], "status": "down"}}
514 logger
.info("Shutting down neighbor interface {} on r2".format(intf_val
))
515 result
= interface_status(tgen
, topo
, input_dict_1
)
516 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
520 for addr_type
in ADDR_TYPES
:
521 if intf_num
+ 16 < 32:
522 check_hops
= NEXT_HOPS
[addr_type
]
526 input_dict
= {"r3": {"static_routes": [{"network": NETWORK
[addr_type
]}]}}
527 logger
.info("Verifying %s routes on r3", addr_type
)
529 tgen
, addr_type
, dut
, input_dict
, next_hop
=check_hops
, protocol
=protocol
531 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
535 input_dict_1
= {"r2": {"interface_list": INTF_LIST_R2
, "status": "up"}}
537 logger
.info("Enabling all neighbor interface {} on r2")
538 result
= interface_status(tgen
, topo
, input_dict_1
)
539 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
541 static_or_nw(tgen
, topo
, tc_name
, test_type
, "r2")
542 for addr_type
in ADDR_TYPES
:
543 input_dict
= {"r3": {"static_routes": [{"network": NETWORK
[addr_type
]}]}}
545 logger
.info("Verifying %s routes on r3", addr_type
)
551 next_hop
=NEXT_HOPS
[addr_type
],
554 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
558 write_test_footer(tc_name
)
561 def test_ecmp_remove_static_route(request
):
563 Delete static routes and verify routers are cleared from BGP table,
567 tc_name
= request
.node
.name
568 write_test_header(tc_name
)
571 # Verifying RIB routes
575 reset_config_on_routers(tgen
)
577 static_or_nw(tgen
, topo
, tc_name
, "redist_static", "r2")
578 for addr_type
in ADDR_TYPES
:
579 input_dict_1
= {"r3": {"static_routes": [{"network": NETWORK
[addr_type
]}]}}
581 logger
.info("Verifying %s routes on r3", addr_type
)
587 next_hop
=NEXT_HOPS
[addr_type
],
590 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
594 for addr_type
in ADDR_TYPES
:
599 "network": NETWORK
[addr_type
],
600 "next_hop": NEXT_HOP_IP
[addr_type
],
607 logger
.info("Remove static routes")
608 result
= create_static_routes(tgen
, input_dict_2
)
609 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
613 logger
.info("Verifying %s routes on r3 are removed", addr_type
)
625 ), "Testcase {} : Failed \n Routes still" " present in RIB".format(tc_name
)
627 for addr_type
in ADDR_TYPES
:
628 # Enable static routes
632 {"network": NETWORK
[addr_type
], "next_hop": NEXT_HOP_IP
[addr_type
]}
637 logger
.info("Enable static route")
638 result
= create_static_routes(tgen
, input_dict_4
)
639 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
643 logger
.info("Verifying %s routes on r3", addr_type
)
649 next_hop
=NEXT_HOPS
[addr_type
],
652 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
657 def test_ecmp_remove_nw_advertise(request
):
659 Verify routes are cleared from BGP and RIB table of DUT,
660 when advertise network configuration is removed
663 tc_name
= request
.node
.name
664 write_test_header(tc_name
)
667 # Verifying RIB routes
671 reset_config_on_routers(tgen
)
672 static_or_nw(tgen
, topo
, tc_name
, "advertise_nw", "r2")
673 for addr_type
in ADDR_TYPES
:
674 input_dict
= {"r3": {"static_routes": [{"network": NETWORK
[addr_type
]}]}}
676 logger
.info("Verifying %s routes on r3", addr_type
)
682 next_hop
=NEXT_HOPS
[addr_type
],
685 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
695 "advertise_networks": [
696 {"network": NETWORK
["ipv4"], "delete": True}
702 "advertise_networks": [
703 {"network": NETWORK
["ipv6"], "delete": True}
712 logger
.info("Withdraw advertised networks")
713 result
= create_router_bgp(tgen
, topo
, input_dict_3
)
714 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
716 for addr_type
in ADDR_TYPES
:
717 input_dict
= {"r3": {"static_routes": [{"network": NETWORK
[addr_type
]}]}}
719 logger
.info("Verifying %s routes on r3", addr_type
)
731 ), "Testcase {} : Failed \n Routes still" " present in RIB".format(tc_name
)
733 static_or_nw(tgen
, topo
, tc_name
, "advertise_nw", "r2")
734 for addr_type
in ADDR_TYPES
:
735 input_dict
= {"r3": {"static_routes": [{"network": NETWORK
[addr_type
]}]}}
736 logger
.info("Verifying %s routes on r3", addr_type
)
742 next_hop
=NEXT_HOPS
[addr_type
],
745 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
749 write_test_footer(tc_name
)
752 if __name__
== "__main__":
753 args
= ["-s"] + sys
.argv
[1:]
754 sys
.exit(pytest
.main(args
))