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.
45 # Save the Current Working Directory to find configuration files.
46 CWD
= os
.path
.dirname(os
.path
.realpath(__file__
))
47 sys
.path
.append(os
.path
.join(CWD
, "../"))
48 sys
.path
.append(os
.path
.join(CWD
, "../../"))
50 # pylint: disable=C0413
51 # Import topogen and topotest helpers
52 from lib
.topogen
import Topogen
, get_topogen
53 from mininet
.topo
import Topo
55 from lib
.common_config
import (
63 reset_config_on_routers
,
64 required_linux_kernel_version
,
66 from lib
.topolog
import logger
67 from lib
.bgp
import verify_bgp_convergence
, create_router_bgp
, clear_bgp
68 from lib
.topojson
import build_topo_from_json
, build_config_from_json
71 pytestmark
= [pytest
.mark
.bgpd
, pytest
.mark
.staticd
]
74 # Reading the data from JSON File for topology and configuration creation
75 jsonFile
= "{}/ebgp_ecmp_topo2.json".format(CWD
)
78 with
open(jsonFile
, "r") as topoJson
:
79 topo
= json
.load(topoJson
)
81 assert False, "Could not read file {}".format(jsonFile
)
84 NEXT_HOPS
= {"ipv4": [], "ipv6": []}
87 NETWORK
= {"ipv4": "11.0.20.1/32", "ipv6": "1::/64"}
88 NEXT_HOP_IP
= {"ipv4": "10.0.0.1", "ipv6": "fd00::1"}
89 BGP_CONVERGENCE
= False
92 class CreateTopo(Topo
):
94 Test topology builder.
96 * `Topo`: Topology object
99 def build(self
, *_args
, **_opts
):
100 """Build function."""
101 tgen
= get_topogen(self
)
103 # Building topology from json file
104 build_topo_from_json(tgen
, topo
)
107 def setup_module(mod
):
109 Sets up the pytest environment.
113 global NEXT_HOPS
, INTF_LIST_R3
, INTF_LIST_R2
, TEST_STATIC
116 # Required linux kernel version for this suite to run.
117 result
= required_linux_kernel_version("4.15")
118 if result
is not True:
119 pytest
.skip("Kernel requirements are not met")
121 testsuite_run_time
= time
.asctime(time
.localtime(time
.time()))
122 logger
.info("Testsuite start time: {}".format(testsuite_run_time
))
123 logger
.info("=" * 40)
125 logger
.info("Running setup_module to create topology")
127 # This function initiates the topology build with Topogen...
128 tgen
= Topogen(CreateTopo
, mod
.__name
__)
130 # Starting topology, create tmp files which are loaded to routers
131 # to start deamons and then start routers
134 # Creating configuration from JSON
135 build_config_from_json(tgen
, topo
)
137 # Don't run this test if we have any failure.
138 if tgen
.routers_have_failure():
139 pytest
.skip(tgen
.errors
)
142 # Api call verify whether BGP is converged
143 ADDR_TYPES
= check_address_types()
145 BGP_CONVERGENCE
= verify_bgp_convergence(tgen
, topo
)
146 assert BGP_CONVERGENCE
is True, "setup_module :Failed \n Error:" " {}".format(
151 val
for links
, val
in topo
["routers"]["r2"]["links"].items() if "r3" in links
153 for adt
in ADDR_TYPES
:
154 NEXT_HOPS
[adt
] = [val
[adt
].split("/")[0] for val
in link_data
]
156 NEXT_HOPS
[adt
] = sorted(NEXT_HOPS
[adt
], key
=lambda x
: int(x
.split(".")[2]))
158 NEXT_HOPS
[adt
] = sorted(
159 NEXT_HOPS
[adt
], key
=lambda x
: int(x
.split(":")[-3], 16)
162 INTF_LIST_R2
= [val
["interface"].split("/")[0] for val
in link_data
]
163 INTF_LIST_R2
= sorted(INTF_LIST_R2
, key
=lambda x
: int(x
.split("eth")[1]))
166 val
for links
, val
in topo
["routers"]["r3"]["links"].items() if "r2" in links
168 INTF_LIST_R3
= [val
["interface"].split("/")[0] for val
in link_data
]
169 INTF_LIST_R3
= sorted(INTF_LIST_R3
, key
=lambda x
: int(x
.split("eth")[1]))
171 # STATIC_ROUTE = True
172 logger
.info("Running setup_module() done")
175 def teardown_module():
177 Teardown the pytest environment.
182 logger
.info("Running teardown_module to delete topology")
186 # Stop toplogy and Remove tmp files
190 def static_or_nw(tgen
, topo
, tc_name
, test_type
, dut
):
192 if test_type
== "redist_static":
193 input_dict_static
= {
196 {"network": NETWORK
["ipv4"], "next_hop": NEXT_HOP_IP
["ipv4"]},
197 {"network": NETWORK
["ipv6"], "next_hop": NEXT_HOP_IP
["ipv6"]},
201 logger
.info("Configuring static route on router %s", dut
)
202 result
= create_static_routes(tgen
, input_dict_static
)
203 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
212 "unicast": {"redistribute": [{"redist_type": "static"}]}
215 "unicast": {"redistribute": [{"redist_type": "static"}]}
222 logger
.info("Configuring redistribute static route on router %s", dut
)
223 result
= create_router_bgp(tgen
, topo
, input_dict_2
)
224 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
228 elif test_type
== "advertise_nw":
235 "advertise_networks": [{"network": NETWORK
["ipv4"]}]
240 "advertise_networks": [{"network": NETWORK
["ipv6"]}]
249 "Advertising networks %s %s from router %s",
254 result
= create_router_bgp(tgen
, topo
, input_dict_nw
)
255 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
260 @pytest.mark
.parametrize("ecmp_num", ["8", "16", "32"])
261 @pytest.mark
.parametrize("test_type", ["redist_static", "advertise_nw"])
262 def test_modify_ecmp_max_paths(request
, ecmp_num
, test_type
):
264 Verify routes installed as per maximum-paths
265 configuration (8/16/32).
268 tc_name
= request
.node
.name
269 write_test_header(tc_name
)
272 reset_config_on_routers(tgen
)
274 static_or_nw(tgen
, topo
, tc_name
, test_type
, "r2")
299 logger
.info("Configuring bgp maximum-paths %s on router r3", ecmp_num
)
300 result
= create_router_bgp(tgen
, topo
, input_dict
)
301 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
303 # Verifying RIB routes
307 for addr_type
in ADDR_TYPES
:
308 input_dict_1
= {"r3": {"static_routes": [{"network": NETWORK
[addr_type
]}]}}
310 logger
.info("Verifying %s routes on r3", addr_type
)
312 # Only test the count of nexthops; the actual nexthop addresses
313 # can vary and are not deterministic.
320 next_hop
=NEXT_HOPS
[addr_type
][: int(ecmp_num
)],
325 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
329 write_test_footer(tc_name
)
332 @pytest.mark
.parametrize("ecmp_num", ["8", "16", "32"])
333 @pytest.mark
.parametrize("test_type", ["redist_static", "advertise_nw"])
334 def test_ecmp_after_clear_bgp(request
, ecmp_num
, test_type
):
335 """ Verify BGP table and RIB in DUT after clear BGP routes and neighbors"""
337 tc_name
= request
.node
.name
338 write_test_header(tc_name
)
341 reset_config_on_routers(tgen
)
343 # Verifying RIB routes
347 static_or_nw(tgen
, topo
, tc_name
, test_type
, "r2")
348 for addr_type
in ADDR_TYPES
:
349 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(
365 for addr_type
in ADDR_TYPES
:
366 clear_bgp(tgen
, addr_type
, dut
)
368 # Verify BGP convergence
369 result
= verify_bgp_convergence(tgen
, topo
)
370 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
372 for addr_type
in ADDR_TYPES
:
373 input_dict_1
= {"r3": {"static_routes": [{"network": NETWORK
[addr_type
]}]}}
374 logger
.info("Verifying %s routes on r3", addr_type
)
380 next_hop
=NEXT_HOPS
[addr_type
][: int(ecmp_num
)],
383 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
387 write_test_footer(tc_name
)
390 def test_ecmp_remove_redistribute_static(request
):
391 """Verify routes are cleared from BGP and RIB table of DUT when
392 redistribute static configuration is removed."""
394 tc_name
= request
.node
.name
395 write_test_header(tc_name
)
398 reset_config_on_routers(tgen
)
399 static_or_nw(tgen
, topo
, tc_name
, "redist_static", "r2")
400 for addr_type
in ADDR_TYPES
:
402 # Verifying RIB routes
405 input_dict_1
= {"r3": {"static_routes": [{"network": NETWORK
[addr_type
]}]}}
407 logger
.info("Verifying %s routes on r3", addr_type
)
413 next_hop
=NEXT_HOPS
[addr_type
],
416 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
426 "redistribute": [{"redist_type": "static", "delete": True}]
431 "redistribute": [{"redist_type": "static", "delete": True}]
439 logger
.info("Remove redistribute static")
440 result
= create_router_bgp(tgen
, topo
, input_dict_2
)
441 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
443 for addr_type
in ADDR_TYPES
:
445 # Verifying RIB routes
448 input_dict_1
= {"r3": {"static_routes": [{"network": NETWORK
[addr_type
]}]}}
450 logger
.info("Verifying %s routes on r3 are deleted", addr_type
)
462 ), "Testcase {} : Failed \n Routes still" " present in RIB".format(tc_name
)
464 logger
.info("Enable redistribute static")
469 "ipv4": {"unicast": {"redistribute": [{"redist_type": "static"}]}},
470 "ipv6": {"unicast": {"redistribute": [{"redist_type": "static"}]}},
475 result
= create_router_bgp(tgen
, topo
, input_dict_2
)
476 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
478 for addr_type
in ADDR_TYPES
:
479 # Verifying RIB routes
482 input_dict_1
= {"r3": {"static_routes": [{"network": NETWORK
[addr_type
]}]}}
483 logger
.info("Verifying %s routes on r3", addr_type
)
489 next_hop
=NEXT_HOPS
[addr_type
],
492 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
496 write_test_footer(tc_name
)
499 @pytest.mark
.parametrize("test_type", ["redist_static", "advertise_nw"])
500 def test_ecmp_shut_bgp_neighbor(request
, test_type
):
501 """Shut BGP neigbors one by one and verify BGP and routing table updated
502 accordingly in DUT"""
504 tc_name
= request
.node
.name
505 write_test_header(tc_name
)
508 logger
.info(INTF_LIST_R2
)
509 # Verifying RIB routes
513 reset_config_on_routers(tgen
)
514 static_or_nw(tgen
, topo
, tc_name
, test_type
, "r2")
516 for addr_type
in ADDR_TYPES
:
517 input_dict
= {"r3": {"static_routes": [{"network": NETWORK
[addr_type
]}]}}
519 logger
.info("Verifying %s routes on r3", addr_type
)
525 next_hop
=NEXT_HOPS
[addr_type
],
528 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
532 for intf_num
in range(len(INTF_LIST_R2
) + 1, 16):
533 intf_val
= INTF_LIST_R2
[intf_num
: intf_num
+ 16]
535 input_dict_1
= {"r2": {"interface_list": [intf_val
], "status": "down"}}
536 logger
.info("Shutting down neighbor interface {} on r2".format(intf_val
))
537 result
= interface_status(tgen
, topo
, input_dict_1
)
538 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
542 for addr_type
in ADDR_TYPES
:
543 if intf_num
+ 16 < 32:
544 check_hops
= NEXT_HOPS
[addr_type
]
548 input_dict
= {"r3": {"static_routes": [{"network": NETWORK
[addr_type
]}]}}
549 logger
.info("Verifying %s routes on r3", addr_type
)
551 tgen
, addr_type
, dut
, input_dict
, next_hop
=check_hops
, protocol
=protocol
553 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
557 input_dict_1
= {"r2": {"interface_list": INTF_LIST_R2
, "status": "up"}}
559 logger
.info("Enabling all neighbor interface {} on r2")
560 result
= interface_status(tgen
, topo
, input_dict_1
)
561 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
563 static_or_nw(tgen
, topo
, tc_name
, test_type
, "r2")
564 for addr_type
in ADDR_TYPES
:
565 input_dict
= {"r3": {"static_routes": [{"network": NETWORK
[addr_type
]}]}}
567 logger
.info("Verifying %s routes on r3", addr_type
)
573 next_hop
=NEXT_HOPS
[addr_type
],
576 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
580 write_test_footer(tc_name
)
583 def test_ecmp_remove_static_route(request
):
585 Delete static routes and verify routers are cleared from BGP table,
589 tc_name
= request
.node
.name
590 write_test_header(tc_name
)
593 # Verifying RIB routes
597 reset_config_on_routers(tgen
)
599 static_or_nw(tgen
, topo
, tc_name
, "redist_static", "r2")
600 for addr_type
in ADDR_TYPES
:
601 input_dict_1
= {"r3": {"static_routes": [{"network": NETWORK
[addr_type
]}]}}
603 logger
.info("Verifying %s routes on r3", addr_type
)
609 next_hop
=NEXT_HOPS
[addr_type
],
612 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
616 for addr_type
in ADDR_TYPES
:
621 "network": NETWORK
[addr_type
],
622 "next_hop": NEXT_HOP_IP
[addr_type
],
629 logger
.info("Remove static routes")
630 result
= create_static_routes(tgen
, input_dict_2
)
631 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
635 logger
.info("Verifying %s routes on r3 are removed", addr_type
)
647 ), "Testcase {} : Failed \n Routes still" " present in RIB".format(tc_name
)
649 for addr_type
in ADDR_TYPES
:
650 # Enable static routes
654 {"network": NETWORK
[addr_type
], "next_hop": NEXT_HOP_IP
[addr_type
]}
659 logger
.info("Enable static route")
660 result
= create_static_routes(tgen
, input_dict_4
)
661 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
665 logger
.info("Verifying %s routes on r3", addr_type
)
671 next_hop
=NEXT_HOPS
[addr_type
],
674 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
679 def test_ecmp_remove_nw_advertise(request
):
681 Verify routes are cleared from BGP and RIB table of DUT,
682 when advertise network configuration is removed
685 tc_name
= request
.node
.name
686 write_test_header(tc_name
)
689 # Verifying RIB routes
693 reset_config_on_routers(tgen
)
694 static_or_nw(tgen
, topo
, tc_name
, "advertise_nw", "r2")
695 for addr_type
in ADDR_TYPES
:
696 input_dict
= {"r3": {"static_routes": [{"network": NETWORK
[addr_type
]}]}}
698 logger
.info("Verifying %s routes on r3", addr_type
)
704 next_hop
=NEXT_HOPS
[addr_type
],
707 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
717 "advertise_networks": [
718 {"network": NETWORK
["ipv4"], "delete": True}
724 "advertise_networks": [
725 {"network": NETWORK
["ipv6"], "delete": True}
734 logger
.info("Withdraw advertised networks")
735 result
= create_router_bgp(tgen
, topo
, input_dict_3
)
736 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
738 for addr_type
in ADDR_TYPES
:
739 input_dict
= {"r3": {"static_routes": [{"network": NETWORK
[addr_type
]}]}}
741 logger
.info("Verifying %s routes on r3", addr_type
)
753 ), "Testcase {} : Failed \n Routes still" " present in RIB".format(tc_name
)
755 static_or_nw(tgen
, topo
, tc_name
, "advertise_nw", "r2")
756 for addr_type
in ADDR_TYPES
:
757 input_dict
= {"r3": {"static_routes": [{"network": NETWORK
[addr_type
]}]}}
758 logger
.info("Verifying %s routes on r3", addr_type
)
764 next_hop
=NEXT_HOPS
[addr_type
],
767 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
771 write_test_footer(tc_name
)
774 if __name__
== "__main__":
775 args
= ["-s"] + sys
.argv
[1:]
776 sys
.exit(pytest
.main(args
))