]>
Commit | Line | Data |
---|---|---|
db56171c | 1 | #!/usr/bin/python |
acddc0ed | 2 | # SPDX-License-Identifier: ISC |
db56171c | 3 | |
4 | # | |
5 | # Copyright (c) 2021 by VMware, Inc. ("VMware") | |
6 | # Used Copyright (c) 2018 by Network Device Education Foundation, Inc. | |
7 | # ("NetDEF") in this file. | |
8 | # | |
db56171c | 9 | |
10 | ||
11 | """OSPF Basic Functionality Automation.""" | |
12 | import os | |
13 | import sys | |
14 | import time | |
15 | import pytest | |
db56171c | 16 | |
17 | # Save the Current Working Directory to find configuration files. | |
18 | CWD = os.path.dirname(os.path.realpath(__file__)) | |
19 | sys.path.append(os.path.join(CWD, "../")) | |
20 | sys.path.append(os.path.join(CWD, "../lib/")) | |
21 | ||
22 | # pylint: disable=C0413 | |
23 | # Import topogen and topotest helpers | |
db56171c | 24 | from lib.topogen import Topogen, get_topogen |
db56171c | 25 | |
26 | # Import topoJson from lib, to create topology and initial configuration | |
27 | from lib.common_config import ( | |
28 | start_topology, | |
29 | write_test_header, | |
30 | write_test_footer, | |
31 | reset_config_on_routers, | |
32 | verify_rib, | |
33 | create_static_routes, | |
34 | step, | |
db56171c | 35 | shutdown_bringup_interface, |
db56171c | 36 | get_frr_ipv6_linklocal, |
37 | ) | |
38 | from lib.topolog import logger | |
4953ca97 | 39 | from lib.topojson import build_config_from_json |
db56171c | 40 | |
41 | from lib.ospf import ( | |
42 | verify_ospf6_neighbor, | |
db56171c | 43 | verify_ospf6_rib, |
44 | create_router_ospf, | |
db56171c | 45 | config_ospf6_interface, |
46 | ) | |
47 | ||
db56171c | 48 | |
6ff492b1 DS |
49 | pytestmark = [pytest.mark.ospfd, pytest.mark.staticd] |
50 | ||
51 | ||
db56171c | 52 | # Global variables |
53 | topo = None | |
54 | ||
db56171c | 55 | NETWORK = { |
56 | "ipv4": [ | |
57 | "11.0.20.1/32", | |
58 | "11.0.20.2/32", | |
59 | "11.0.20.3/32", | |
60 | "11.0.20.4/32", | |
61 | "11.0.20.5/32", | |
62 | ], | |
63 | "ipv6": ["2::1/128", "2::2/128", "2::3/128", "2::4/128", "2::5/128"], | |
64 | } | |
65 | """ | |
66 | TOPOLOGY : | |
67 | Please view in a fixed-width font such as Courier. | |
68 | +---+ A1 +---+ | |
69 | +R1 +------------+R2 | | |
70 | +-+-+- +--++ | |
71 | | -- -- | | |
72 | | -- A0 -- | | |
73 | A0| ---- | | |
74 | | ---- | A2 | |
75 | | -- -- | | |
76 | | -- -- | | |
77 | +-+-+- +-+-+ | |
78 | +R0 +-------------+R3 | | |
79 | +---+ A3 +---+ | |
80 | ||
81 | TESTCASES : | |
82 | 1. Verify OSPF ECMP with max path configured as 8 (ECMPconfigured at FRR level) | |
83 | 2. Verify OSPF ECMP with max path configured as 2 (Edge having 2 uplink ports) | |
84 | """ | |
85 | ||
86 | ||
db56171c | 87 | def setup_module(mod): |
88 | """ | |
89 | Sets up the pytest environment | |
90 | ||
91 | * `mod`: module name | |
92 | """ | |
db56171c | 93 | testsuite_run_time = time.asctime(time.localtime(time.time())) |
94 | logger.info("Testsuite start time: {}".format(testsuite_run_time)) | |
95 | logger.info("=" * 40) | |
96 | ||
97 | logger.info("Running setup_module to create topology") | |
98 | ||
99 | # This function initiates the topology build with Topogen... | |
e82b531d CH |
100 | json_file = "{}/ospfv3_ecmp.json".format(CWD) |
101 | tgen = Topogen(json_file, mod.__name__) | |
102 | global topo | |
103 | topo = tgen.json_topo | |
db56171c | 104 | # ... and here it calls Mininet initialization functions. |
105 | ||
db56171c | 106 | # Starting topology, create tmp files which are loaded to routers |
d60a3f0e | 107 | # to start daemons and then start routers |
991a971f | 108 | start_topology(tgen) |
db56171c | 109 | |
110 | # Creating configuration from JSON | |
111 | build_config_from_json(tgen, topo) | |
112 | ||
113 | # Don't run this test if we have any failure. | |
114 | if tgen.routers_have_failure(): | |
115 | pytest.skip(tgen.errors) | |
116 | ||
117 | ospf_covergence = verify_ospf6_neighbor(tgen, topo) | |
118 | assert ospf_covergence is True, "setup_module :Failed \n Error:" " {}".format( | |
119 | ospf_covergence | |
120 | ) | |
121 | ||
122 | logger.info("Running setup_module() done") | |
123 | ||
124 | ||
125 | def teardown_module(mod): | |
126 | """ | |
127 | Teardown the pytest environment. | |
128 | ||
129 | * `mod`: module name | |
130 | """ | |
131 | ||
132 | logger.info("Running teardown_module to delete topology") | |
133 | ||
134 | tgen = get_topogen() | |
135 | ||
136 | # Stop toplogy and Remove tmp files | |
137 | tgen.stop_topology() | |
138 | ||
139 | logger.info( | |
140 | "Testsuite end time: {}".format(time.asctime(time.localtime(time.time()))) | |
141 | ) | |
142 | logger.info("=" * 40) | |
143 | ||
144 | ||
145 | def red_static(dut, config=True): | |
146 | """Local def for Redstribute static routes inside ospf.""" | |
147 | global topo | |
148 | tgen = get_topogen() | |
149 | if config: | |
150 | ospf_red = {dut: {"ospf6": {"redistribute": [{"redist_type": "static"}]}}} | |
151 | else: | |
152 | ospf_red = { | |
153 | dut: { | |
154 | "ospf6": {"redistribute": [{"redist_type": "static", "delete": True}]} | |
155 | } | |
156 | } | |
157 | result = create_router_ospf(tgen, topo, ospf_red) | |
158 | assert result is True, "Testcase : Failed \n Error: {}".format(result) | |
159 | ||
160 | ||
161 | def red_connected(dut, config=True): | |
162 | """Local def for Redstribute connected routes inside ospf.""" | |
163 | global topo | |
164 | tgen = get_topogen() | |
165 | if config: | |
166 | ospf_red = {dut: {"ospf6": {"redistribute": [{"redist_type": "connected"}]}}} | |
167 | else: | |
168 | ospf_red = { | |
169 | dut: { | |
170 | "ospf6": { | |
171 | "redistribute": [{"redist_type": "connected", "del_action": True}] | |
172 | } | |
173 | } | |
174 | } | |
175 | result = create_router_ospf(tgen, topo, ospf_red) | |
176 | assert result is True, "Testcase: Failed \n Error: {}".format(result) | |
177 | ||
178 | ||
179 | def get_llip(onrouter, intf): | |
180 | """ | |
dea6dce3 | 181 | API to get the link local ipv6 address of a particular interface |
db56171c | 182 | |
183 | Parameters | |
184 | ---------- | |
185 | * `fromnode`: Source node | |
186 | * `tonode` : interface for which link local ip needs to be returned. | |
187 | ||
188 | Usage | |
189 | ----- | |
190 | result = get_llip('r1', 'r2-link0') | |
191 | ||
192 | Returns | |
193 | ------- | |
194 | 1) link local ipv6 address from the interface. | |
195 | 2) errormsg - when link local ip not found. | |
196 | """ | |
197 | tgen = get_topogen() | |
198 | intf = topo["routers"][onrouter]["links"][intf]["interface"] | |
199 | llip = get_frr_ipv6_linklocal(tgen, onrouter, intf) | |
200 | if llip: | |
201 | logger.info("llip ipv6 address to be set as NH is %s", llip) | |
202 | return llip | |
203 | return None | |
204 | ||
205 | ||
206 | def get_glipv6(onrouter, intf): | |
207 | """ | |
dea6dce3 | 208 | API to get the global ipv6 address of a particular interface |
db56171c | 209 | |
210 | Parameters | |
211 | ---------- | |
212 | * `onrouter`: Source node | |
213 | * `intf` : interface for which link local ip needs to be returned. | |
214 | ||
215 | Usage | |
216 | ----- | |
217 | result = get_glipv6('r1', 'r2-link0') | |
218 | ||
219 | Returns | |
220 | ------- | |
221 | 1) global ipv6 address from the interface. | |
222 | 2) errormsg - when link local ip not found. | |
223 | """ | |
224 | glipv6 = (topo["routers"][onrouter]["links"][intf]["ipv6"]).split("/")[0] | |
225 | if glipv6: | |
226 | logger.info("Global ipv6 address to be set as NH is %s", glipv6) | |
227 | return glipv6 | |
228 | return None | |
229 | ||
230 | ||
231 | # ################################## | |
232 | # Test cases start here. | |
233 | # ################################## | |
234 | ||
235 | ||
236 | def test_ospfv3_ecmp_tc16_p0(request): | |
237 | """ | |
238 | Verify OSPF ECMP. | |
239 | ||
240 | Verify OSPF ECMP with max path configured as 8 (ECMP | |
241 | configured at FRR level) | |
242 | """ | |
243 | tc_name = request.node.name | |
244 | write_test_header(tc_name) | |
245 | tgen = get_topogen() | |
246 | ||
247 | # Don't run this test if we have any failure. | |
248 | if tgen.routers_have_failure(): | |
249 | pytest.skip(tgen.errors) | |
250 | ||
251 | global topo | |
252 | step("Bring up the base config as per the topology") | |
253 | step("Configure 8 interfaces between R1 and R2 and enable ospf in area 0.") | |
254 | ||
255 | reset_config_on_routers(tgen) | |
256 | ||
257 | step("Verify that OSPF is up with 8 neighborship sessions.") | |
258 | dut = "r1" | |
259 | ospf_covergence = verify_ospf6_neighbor(tgen, topo, dut=dut) | |
260 | assert ospf_covergence is True, "setup_module :Failed \n Error:" " {}".format( | |
261 | ospf_covergence | |
262 | ) | |
263 | ||
264 | step("Configure a static route in R0 and redistribute in OSPF.") | |
265 | ||
266 | input_dict = { | |
267 | "r0": { | |
268 | "static_routes": [ | |
269 | { | |
270 | "network": NETWORK["ipv6"][0], | |
271 | "no_of_ip": 5, | |
272 | "next_hop": "Null0", | |
273 | } | |
274 | ] | |
275 | } | |
276 | } | |
277 | result = create_static_routes(tgen, input_dict) | |
278 | assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result) | |
279 | ||
280 | dut = "r0" | |
281 | red_static(dut) | |
282 | ||
283 | llip = get_llip("r0", "r1-link1") | |
284 | assert llip is not None, "Testcase {} : Failed \n Error: {}".format(tc_name, result) | |
285 | ||
286 | step("Verify that route in R2 in stalled with 8 next hops.") | |
287 | nh = [] | |
288 | for item in range(1, 7): | |
289 | nh.append(llip) | |
290 | ||
291 | llip = get_llip("r0", "r1") | |
292 | assert llip is not None, "Testcase {} : Failed \n Error: {}".format(tc_name, result) | |
293 | ||
294 | nh2 = llip | |
295 | ||
296 | nh.append(nh2) | |
297 | ||
298 | dut = "r1" | |
299 | result = verify_ospf6_rib(tgen, dut, input_dict, next_hop=nh) | |
300 | assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result) | |
301 | ||
302 | protocol = "ospf" | |
303 | result = verify_rib(tgen, "ipv6", dut, input_dict, protocol=protocol, next_hop=nh) | |
304 | assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result) | |
305 | ||
306 | step("shut no shut all the interfaces on the remote router - R2") | |
307 | dut = "r1" | |
308 | for intfr in range(1, 7): | |
309 | intf = topo["routers"]["r1"]["links"]["r0-link{}".format(intfr)]["interface"] | |
310 | shutdown_bringup_interface(tgen, dut, intf, False) | |
311 | ||
312 | result = verify_ospf6_rib(tgen, dut, input_dict, next_hop=nh, expected=False) | |
313 | assert ( | |
314 | result is not True | |
315 | ), "Testcase {} : Failed \n Route present in OSPF RIB. Error: {}".format( | |
316 | tc_name, result | |
317 | ) | |
318 | ||
319 | protocol = "ospf" | |
320 | result = verify_rib( | |
321 | tgen, "ipv6", dut, input_dict, protocol=protocol, next_hop=nh, expected=False | |
322 | ) | |
323 | assert ( | |
324 | result is not True | |
325 | ), "Testcase {} : Failed \n Route present in RIB. Error: {}".format(tc_name, result) | |
326 | ||
327 | for intfr in range(1, 7): | |
328 | intf = topo["routers"]["r1"]["links"]["r0-link{}".format(intfr)]["interface"] | |
329 | shutdown_bringup_interface(tgen, dut, intf, True) | |
330 | ||
331 | result = verify_ospf6_rib(tgen, dut, input_dict, next_hop=nh) | |
332 | assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result) | |
333 | ||
334 | protocol = "ospf" | |
335 | result = verify_rib(tgen, "ipv6", dut, input_dict, protocol=protocol, next_hop=nh) | |
336 | assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result) | |
337 | ||
338 | step("shut no shut on all the interfaces on DUT (r1)") | |
339 | for intfr in range(1, 7): | |
340 | intf = topo["routers"]["r1"]["links"]["r0-link{}".format(intfr)]["interface"] | |
341 | shutdown_bringup_interface(tgen, dut, intf, False) | |
342 | ||
343 | for intfr in range(1, 7): | |
344 | intf = topo["routers"]["r1"]["links"]["r0-link{}".format(intfr)]["interface"] | |
345 | shutdown_bringup_interface(tgen, dut, intf, True) | |
346 | ||
347 | step( | |
348 | "Verify that all the neighbours are up and routes are installed" | |
349 | " with 8 next hop in ospf and ip route tables on R1." | |
350 | ) | |
351 | ||
352 | step("Verify that OSPF is up with 8 neighborship sessions.") | |
353 | dut = "r1" | |
354 | ospf_covergence = verify_ospf6_neighbor(tgen, topo, dut=dut) | |
355 | assert ospf_covergence is True, "setup_module :Failed \n Error:" " {}".format( | |
356 | ospf_covergence | |
357 | ) | |
358 | ||
359 | dut = "r1" | |
360 | result = verify_ospf6_rib(tgen, dut, input_dict, next_hop=nh) | |
361 | assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result) | |
362 | ||
363 | protocol = "ospf" | |
364 | result = verify_rib(tgen, "ipv6", dut, input_dict, protocol=protocol, next_hop=nh) | |
365 | assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result) | |
366 | ||
367 | write_test_footer(tc_name) | |
368 | ||
369 | ||
370 | def test_ospfv3_ecmp_tc17_p0(request): | |
371 | """ | |
372 | Verify OSPF ECMP. | |
373 | ||
374 | Verify OSPF ECMP with max path configured as 2 (Edge having 2 uplink ports) | |
375 | """ | |
376 | tc_name = request.node.name | |
377 | write_test_header(tc_name) | |
378 | tgen = get_topogen() | |
379 | ||
380 | # Don't run this test if we have any failure. | |
381 | if tgen.routers_have_failure(): | |
382 | pytest.skip(tgen.errors) | |
383 | ||
384 | global topo | |
385 | step("Bring up the base config as per the topology") | |
386 | step("Configure 2 interfaces between R1 and R2 & enable ospf in area 0.") | |
387 | ||
388 | reset_config_on_routers(tgen) | |
389 | ||
390 | step("Verify that OSPF is up with 2 neighborship sessions.") | |
391 | dut = "r1" | |
392 | ospf_covergence = verify_ospf6_neighbor(tgen, topo, dut=dut) | |
393 | assert ospf_covergence is True, "setup_module :Failed \n Error:" " {}".format( | |
394 | ospf_covergence | |
395 | ) | |
396 | ||
397 | step("Configure a static route in R0 and redistribute in OSPF.") | |
398 | ||
399 | input_dict = { | |
400 | "r0": { | |
401 | "static_routes": [ | |
402 | { | |
403 | "network": NETWORK["ipv6"][0], | |
404 | "no_of_ip": 5, | |
405 | "next_hop": "Null0", | |
406 | } | |
407 | ] | |
408 | } | |
409 | } | |
410 | result = create_static_routes(tgen, input_dict) | |
411 | assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result) | |
412 | ||
413 | dut = "r0" | |
414 | red_static(dut) | |
415 | ||
416 | step("Verify that route in R2 in stalled with 2 next hops.") | |
417 | ||
418 | llip = get_llip("r0", "r1-link1") | |
419 | assert llip is not None, "Testcase {} : Failed \n Error: {}".format(tc_name, result) | |
420 | ||
421 | nh1 = llip | |
422 | ||
423 | llip = get_llip("r0", "r1") | |
424 | assert llip is not None, "Testcase {} : Failed \n Error: {}".format(tc_name, result) | |
425 | ||
426 | nh2 = llip | |
427 | ||
428 | nh = [nh1, nh2] | |
429 | ||
430 | dut = "r1" | |
431 | result = verify_ospf6_rib(tgen, dut, input_dict, next_hop=nh) | |
432 | assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result) | |
433 | ||
434 | protocol = "ospf" | |
435 | result = verify_rib(tgen, "ipv6", dut, input_dict, protocol=protocol, next_hop=nh) | |
436 | assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result) | |
437 | ||
438 | step("Configure ECMP value as 1.") | |
439 | max_path = {"r1": {"ospf6": {"maximum-paths": 1}}} | |
440 | result = create_router_ospf(tgen, topo, max_path) | |
441 | assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result) | |
442 | ||
443 | result = verify_rib(tgen, "ipv6", dut, input_dict, protocol=protocol, next_hop=nh2) | |
444 | assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result) | |
445 | ||
446 | dut = "r1" | |
447 | max_path = {"r1": {"ospf6": {"maximum-paths": 2}}} | |
448 | result = create_router_ospf(tgen, topo, max_path) | |
449 | assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result) | |
450 | ||
451 | result = verify_rib(tgen, "ipv6", dut, input_dict, protocol=protocol, next_hop=nh) | |
452 | assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result) | |
453 | ||
454 | step("Configure cost on R0 as 100") | |
455 | r0_ospf_cost = {"r0": {"links": {"r1": {"ospf6": {"cost": 100}}}}} | |
456 | result = config_ospf6_interface(tgen, topo, r0_ospf_cost) | |
457 | assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) | |
458 | ||
459 | dut = "r1" | |
460 | result = verify_ospf6_rib(tgen, dut, input_dict, next_hop=nh) | |
461 | assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result) | |
462 | ||
463 | protocol = "ospf" | |
464 | result = verify_rib(tgen, "ipv6", dut, input_dict, protocol=protocol, next_hop=nh) | |
465 | assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result) | |
466 | ||
467 | write_test_footer(tc_name) | |
468 | ||
469 | ||
470 | if __name__ == "__main__": | |
471 | args = ["-s"] + sys.argv[1:] | |
472 | sys.exit(pytest.main(args)) |