]>
Commit | Line | Data |
---|---|---|
57ea1920 PG |
1 | #!/usr/bin/env python |
2 | ||
3 | # | |
4 | # Copyright (c) 2019 by VMware, Inc. ("VMware") | |
5 | # Used Copyright (c) 2018 by Network Device Education Foundation, Inc. | |
6 | # ("NetDEF") in this file. | |
7 | # | |
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 | |
11 | # in all copies. | |
12 | # | |
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 | |
20 | # OF THIS SOFTWARE. | |
21 | # | |
22 | ||
23 | ||
24 | """ | |
25 | Following tests are covered to test ecmp functionality on iBGP. | |
26 | 1. Verify bgp fast-convergence functionality | |
27 | """ | |
28 | import os | |
29 | import sys | |
30 | import time | |
31 | import json | |
32 | import pytest | |
33 | from time import sleep | |
34 | ||
35 | # Save the Current Working Directory to find configuration files. | |
36 | CWD = os.path.dirname(os.path.realpath(__file__)) | |
37 | sys.path.append(os.path.join(CWD, "../")) | |
38 | sys.path.append(os.path.join(CWD, "../../")) | |
39 | ||
40 | # pylint: disable=C0413 | |
41 | # Import topogen and topotest helpers | |
42 | from lib.topogen import Topogen, get_topogen | |
43 | from mininet.topo import Topo | |
44 | ||
45 | from lib.common_config import ( | |
46 | start_topology, | |
47 | write_test_header, | |
48 | write_test_footer, | |
49 | verify_rib, | |
50 | create_static_routes, | |
51 | check_address_types, | |
52 | interface_status, | |
53 | reset_config_on_routers, | |
54 | required_linux_kernel_version, | |
55 | shutdown_bringup_interface, | |
56 | apply_raw_config, | |
57 | ) | |
58 | from lib.topolog import logger | |
59 | from lib.bgp import verify_bgp_convergence, create_router_bgp, clear_bgp | |
60 | from lib.topojson import build_topo_from_json, build_config_from_json | |
61 | ||
62 | ||
63 | pytestmark = [pytest.mark.bgpd, pytest.mark.staticd] | |
64 | ||
65 | ||
66 | # Reading the data from JSON File for topology and configuration creation | |
67 | jsonFile = "{}/ibgp_ecmp_topo3.json".format(CWD) | |
68 | ||
69 | try: | |
70 | with open(jsonFile, "r") as topoJson: | |
71 | topo = json.load(topoJson) | |
72 | except IOError: | |
73 | assert False, "Could not read file {}".format(jsonFile) | |
74 | ||
75 | # Global variables | |
76 | NEXT_HOPS = {"ipv4": [], "ipv6": []} | |
77 | NETWORK = {"ipv4": "192.168.1.10/32", "ipv6": "fd00:0:0:1::10/128"} | |
78 | NEXT_HOP_IP = {"ipv4": "10.0.0.1", "ipv6": "fd00::1"} | |
79 | BGP_CONVERGENCE = False | |
80 | ||
81 | ||
82 | class CreateTopo(Topo): | |
83 | """ | |
84 | Test topology builder. | |
85 | ||
86 | * `Topo`: Topology object | |
87 | """ | |
88 | ||
89 | def build(self, *_args, **_opts): | |
90 | """Build function.""" | |
91 | tgen = get_topogen(self) | |
92 | ||
93 | # Building topology from json file | |
94 | build_topo_from_json(tgen, topo) | |
95 | ||
96 | ||
97 | def setup_module(mod): | |
98 | """ | |
99 | Sets up the pytest environment. | |
100 | ||
101 | * `mod`: module name | |
102 | """ | |
103 | global NEXT_HOPS, INTF_LIST_R3, INTF_LIST_R2, TEST_STATIC | |
104 | global ADDR_TYPES | |
105 | ||
106 | testsuite_run_time = time.asctime(time.localtime(time.time())) | |
107 | logger.info("Testsuite start time: {}".format(testsuite_run_time)) | |
108 | logger.info("=" * 40) | |
109 | ||
110 | logger.info("Running setup_module to create topology") | |
111 | ||
112 | # This function initiates the topology build with Topogen... | |
113 | tgen = Topogen(CreateTopo, mod.__name__) | |
114 | ||
115 | # Starting topology, create tmp files which are loaded to routers | |
116 | # to start deamons and then start routers | |
117 | start_topology(tgen) | |
118 | ||
119 | # Creating configuration from JSON | |
120 | build_config_from_json(tgen, topo) | |
121 | ||
122 | # Don't run this test if we have any failure. | |
123 | if tgen.routers_have_failure(): | |
124 | pytest.skip(tgen.errors) | |
125 | ||
126 | # Api call verify whether BGP is converged | |
127 | ADDR_TYPES = check_address_types() | |
128 | ||
129 | BGP_CONVERGENCE = verify_bgp_convergence(tgen, topo) | |
130 | assert BGP_CONVERGENCE is True, "setup_module :Failed \n Error:" " {}".format( | |
131 | BGP_CONVERGENCE | |
132 | ) | |
133 | ||
134 | # STATIC_ROUTE = True | |
135 | logger.info("Running setup_module() done") | |
136 | ||
137 | ||
138 | def teardown_module(): | |
139 | """ | |
140 | Teardown the pytest environment. | |
141 | ||
142 | * `mod`: module name | |
143 | """ | |
144 | ||
145 | logger.info("Running teardown_module to delete topology") | |
146 | ||
147 | tgen = get_topogen() | |
148 | ||
149 | # Stop toplogy and Remove tmp files | |
150 | tgen.stop_topology() | |
151 | ||
152 | ||
153 | def static_or_nw(tgen, topo, tc_name, test_type, dut): | |
154 | ||
155 | if test_type == "redist_static": | |
156 | input_dict_static = { | |
157 | dut: { | |
158 | "static_routes": [ | |
159 | {"network": NETWORK["ipv4"], "next_hop": NEXT_HOP_IP["ipv4"]}, | |
160 | {"network": NETWORK["ipv6"], "next_hop": NEXT_HOP_IP["ipv6"]}, | |
161 | ] | |
162 | } | |
163 | } | |
164 | logger.info("Configuring static route on router %s", dut) | |
165 | result = create_static_routes(tgen, input_dict_static) | |
166 | assert result is True, "Testcase {} : Failed \n Error: {}".format( | |
167 | tc_name, result | |
168 | ) | |
169 | ||
170 | input_dict_2 = { | |
171 | dut: { | |
172 | "bgp": { | |
173 | "address_family": { | |
174 | "ipv4": { | |
175 | "unicast": {"redistribute": [{"redist_type": "static"}]} | |
176 | }, | |
177 | "ipv6": { | |
178 | "unicast": {"redistribute": [{"redist_type": "static"}]} | |
179 | }, | |
180 | } | |
181 | } | |
182 | } | |
183 | } | |
184 | ||
185 | logger.info("Configuring redistribute static route on router %s", dut) | |
186 | result = create_router_bgp(tgen, topo, input_dict_2) | |
187 | assert result is True, "Testcase {} : Failed \n Error: {}".format( | |
188 | tc_name, result | |
189 | ) | |
190 | ||
191 | elif test_type == "advertise_nw": | |
192 | input_dict_nw = { | |
193 | dut: { | |
194 | "bgp": { | |
195 | "address_family": { | |
196 | "ipv4": { | |
197 | "unicast": { | |
198 | "advertise_networks": [{"network": NETWORK["ipv4"]}] | |
199 | } | |
200 | }, | |
201 | "ipv6": { | |
202 | "unicast": { | |
203 | "advertise_networks": [{"network": NETWORK["ipv6"]}] | |
204 | } | |
205 | }, | |
206 | } | |
207 | } | |
208 | } | |
209 | } | |
210 | ||
211 | logger.info( | |
212 | "Advertising networks %s %s from router %s", | |
213 | NETWORK["ipv4"], | |
214 | NETWORK["ipv6"], | |
215 | dut, | |
216 | ) | |
217 | result = create_router_bgp(tgen, topo, input_dict_nw) | |
218 | assert result is True, "Testcase {} : Failed \n Error: {}".format( | |
219 | tc_name, result | |
220 | ) | |
221 | ||
222 | ||
223 | @pytest.mark.parametrize("test_type", ["redist_static"]) | |
224 | def test_ecmp_fast_convergence(request, test_type): | |
225 | """This test is to verify bgp fast-convergence cli functionality""" | |
226 | ||
227 | tc_name = request.node.name | |
228 | write_test_header(tc_name) | |
229 | tgen = get_topogen() | |
230 | ||
231 | # Verifying RIB routes | |
232 | dut = "r3" | |
233 | protocol = "bgp" | |
234 | ||
235 | reset_config_on_routers(tgen) | |
236 | static_or_nw(tgen, topo, tc_name, test_type, "r2") | |
237 | ||
238 | for addr_type in ADDR_TYPES: | |
239 | input_dict = {"r3": {"static_routes": [{"network": NETWORK[addr_type]}]}} | |
240 | ||
241 | logger.info("Verifying %s routes on r3", addr_type) | |
242 | result = verify_rib( | |
243 | tgen, | |
244 | addr_type, | |
245 | dut, | |
246 | input_dict, | |
247 | protocol=protocol, | |
248 | ) | |
249 | assert result is True, "Testcase {} : Failed \n Error: {}".format( | |
250 | tc_name, result | |
251 | ) | |
252 | ||
253 | intf1 = topo["routers"]["r2"]["links"]["r3-link1"]["interface"] | |
254 | intf2 = topo["routers"]["r2"]["links"]["r3-link2"]["interface"] | |
255 | ||
256 | logger.info("Shutdown one of the link b/w r2 and r3") | |
257 | shutdown_bringup_interface(tgen, "r2", intf1, False) | |
258 | ||
259 | logger.info("Verify bgp neighbors are still up") | |
260 | result = verify_bgp_convergence(tgen, topo) | |
261 | assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result) | |
262 | ||
263 | logger.info("Shutdown another link b/w r2 and r3") | |
264 | shutdown_bringup_interface(tgen, "r2", intf2, False) | |
265 | ||
266 | logger.info("Wait for 10 sec and make sure bgp neighbors are still up") | |
267 | sleep(10) | |
268 | result = verify_bgp_convergence(tgen, topo) | |
269 | assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result) | |
270 | ||
271 | logger.info("No shut links b/w r2 and r3") | |
272 | shutdown_bringup_interface(tgen, "r2", intf1, True) | |
273 | shutdown_bringup_interface(tgen, "r2", intf2, True) | |
274 | ||
275 | logger.info("Enable bgp fast-convergence cli") | |
276 | raw_config = { | |
277 | "r2": { | |
278 | "raw_config": [ | |
279 | "router bgp {}".format(topo["routers"]["r2"]["bgp"]["local_as"]), | |
280 | "bgp fast-convergence", | |
281 | ] | |
282 | } | |
283 | } | |
284 | result = apply_raw_config(tgen, raw_config) | |
285 | assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result) | |
286 | ||
287 | logger.info("Shutdown one link b/w r2 and r3") | |
288 | shutdown_bringup_interface(tgen, "r2", intf1, False) | |
289 | ||
290 | logger.info("Verify bgp neighbors goes down immediately") | |
291 | result = verify_bgp_convergence(tgen, topo, dut="r2", expected=False) | |
292 | assert result is not True, "Testcase {} : Failed \n Error: {}".format( | |
293 | tc_name, result | |
294 | ) | |
295 | ||
296 | logger.info("Shutdown second link b/w r2 and r3") | |
297 | shutdown_bringup_interface(tgen, "r2", intf2, False) | |
298 | ||
299 | logger.info("Verify bgp neighbors goes down immediately") | |
300 | result = verify_bgp_convergence(tgen, topo, dut="r2", expected=False) | |
301 | assert result is not True, "Testcase {} : Failed \n Error: {}".format( | |
302 | tc_name, result | |
303 | ) | |
304 | ||
305 | write_test_footer(tc_name) | |
306 | ||
307 | ||
308 | if __name__ == "__main__": | |
309 | args = ["-s"] + sys.argv[1:] | |
310 | sys.exit(pytest.main(args)) |