]> git.proxmox.com Git - mirror_frr.git/blob - tests/topotests/ospf_basic_functionality/test_ospf_rte_calc.py
*: Add camelCase JSON keys in addition to PascalCase
[mirror_frr.git] / tests / topotests / ospf_basic_functionality / test_ospf_rte_calc.py
1 #!/usr/bin/python
2
3 #
4 # Copyright (c) 2020 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 """OSPF Basic Functionality Automation."""
25 import os
26 import sys
27 import time
28 import pytest
29 import ipaddress
30
31 # Save the Current Working Directory to find configuration files.
32 CWD = os.path.dirname(os.path.realpath(__file__))
33 sys.path.append(os.path.join(CWD, "../"))
34 sys.path.append(os.path.join(CWD, "../lib/"))
35
36 # pylint: disable=C0413
37 # Import topogen and topotest helpers
38 from lib.topogen import Topogen, get_topogen
39
40 # Import topoJson from lib, to create topology and initial configuration
41 from lib.common_config import (
42 start_topology,
43 write_test_header,
44 create_interfaces_cfg,
45 write_test_footer,
46 reset_config_on_routers,
47 verify_rib,
48 create_static_routes,
49 step,
50 shutdown_bringup_interface,
51 topo_daemons,
52 )
53 from lib.bgp import verify_bgp_convergence, create_router_bgp
54 from lib.topolog import logger
55 from lib.topojson import build_config_from_json
56
57 from lib.ospf import (
58 verify_ospf_neighbor,
59 clear_ospf,
60 verify_ospf_rib,
61 redistribute_ospf,
62 config_ospf_interface,
63 verify_ospf_interface,
64 )
65
66 pytestmark = [pytest.mark.ospfd, pytest.mark.staticd]
67
68
69 # Global variables
70 topo = None
71
72 # number of retries.
73 nretry = 5
74
75 NETWORK = {
76 "ipv4": [
77 "11.0.20.1/32",
78 "11.0.20.2/32",
79 "11.0.20.3/32",
80 "11.0.20.4/32",
81 "11.0.20.5/32",
82 ]
83 }
84 TOPOOLOGY = """
85 Please view in a fixed-width font such as Courier.
86 +---+ A1 +---+
87 +R1 +------------+R2 |
88 +-+-+- +--++
89 | -- -- |
90 | -- A0 -- |
91 A0| ---- |
92 | ---- | A2
93 | -- -- |
94 | -- -- |
95 +-+-+- +-+-+
96 +R0 +-------------+R3 |
97 +---+ A3 +---+
98 """
99
100 TESTCASES = """
101 1. Test OSPF intra area route calculations.
102 2. Test OSPF inter area route calculations.
103 3. Test OSPF redistribution of connected routes.
104 """
105
106
107 def setup_module(mod):
108 """
109 Sets up the pytest environment
110
111 * `mod`: module name
112 """
113 testsuite_run_time = time.asctime(time.localtime(time.time()))
114 logger.info("Testsuite start time: {}".format(testsuite_run_time))
115 logger.info("=" * 40)
116
117 logger.info("Running setup_module to create topology")
118
119 # This function initiates the topology build with Topogen...
120 json_file = "{}/ospf_rte_calc.json".format(CWD)
121 tgen = Topogen(json_file, mod.__name__)
122 global topo
123 topo = tgen.json_topo
124 # ... and here it calls Mininet initialization functions.
125
126 # get list of daemons needs to be started for this suite.
127 daemons = topo_daemons(tgen, topo)
128
129 # Starting topology, create tmp files which are loaded to routers
130 # to start deamons and then start routers
131 start_topology(tgen, daemons)
132
133 # Creating configuration from JSON
134 build_config_from_json(tgen, topo)
135
136 # Don't run this test if we have any failure.
137 if tgen.routers_have_failure():
138 pytest.skip(tgen.errors)
139 # Api call verify whether OSPF is converged
140 ospf_covergence = verify_ospf_neighbor(tgen, topo)
141 assert ospf_covergence is True, "setup_module :Failed \n Error:" " {}".format(
142 ospf_covergence
143 )
144
145 logger.info("Running setup_module() done")
146
147
148 def teardown_module(mod):
149 """
150 Teardown the pytest environment.
151
152 * `mod`: module name
153 """
154
155 logger.info("Running teardown_module to delete topology")
156
157 tgen = get_topogen()
158
159 # Stop toplogy and Remove tmp files
160 tgen.stop_topology()
161
162 logger.info(
163 "Testsuite end time: {}".format(time.asctime(time.localtime(time.time())))
164 )
165 logger.info("=" * 40)
166
167
168 # ##################################
169 # Test cases start here.
170 # ##################################
171
172
173 def test_ospf_redistribution_tc5_p0(request):
174 """Test OSPF intra area route calculations."""
175 tc_name = request.node.name
176 write_test_header(tc_name)
177 tgen = get_topogen()
178
179 # Don't run this test if we have any failure.
180 if tgen.routers_have_failure():
181 pytest.skip(tgen.errors)
182
183 global topo
184 step("Bring up the base config.")
185 reset_config_on_routers(tgen)
186
187 step("Verify that OSPF neighbors are FULL.")
188 ospf_covergence = verify_ospf_neighbor(tgen, topo)
189 assert ospf_covergence is True, "setup_module :Failed \n Error:" " {}".format(
190 ospf_covergence
191 )
192
193 step("verify intra area route is calculated for r0-r3 interface ip in R1")
194 ip = topo["routers"]["r0"]["links"]["r3"]["ipv4"]
195 ip_net = str(ipaddress.ip_interface(u"{}".format(ip)).network)
196 nh = topo["routers"]["r0"]["links"]["r1"]["ipv4"].split("/")[0]
197 input_dict = {
198 "r1": {"static_routes": [{"network": ip_net, "no_of_ip": 1, "routeType": "N"}]}
199 }
200
201 dut = "r1"
202 result = verify_ospf_rib(tgen, dut, input_dict, next_hop=nh)
203 assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
204
205 protocol = "ospf"
206 result = verify_rib(tgen, "ipv4", dut, input_dict, protocol=protocol, next_hop=nh)
207 assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
208
209 step("Delete the ip address on newly configured interface of R0")
210 topo1 = {
211 "r0": {
212 "links": {
213 "r3": {
214 "ipv4": topo["routers"]["r0"]["links"]["r3"]["ipv4"],
215 "interface": topo["routers"]["r0"]["links"]["r3"]["interface"],
216 "delete": True,
217 }
218 }
219 }
220 }
221
222 result = create_interfaces_cfg(tgen, topo1)
223 assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
224
225 dut = "r1"
226 for num in range(0, nretry):
227 result = verify_ospf_rib(tgen, dut, input_dict, next_hop=nh, expected=False)
228 if result is not True:
229 break
230
231 assert result is not True, (
232 "Testcase {} : Failed \n "
233 "r1: OSPF routes are present after deleting ip address of newly "
234 "configured interface of R0 \n Error: {}".format(tc_name, result)
235 )
236
237 protocol = "ospf"
238 result = verify_rib(
239 tgen,
240 "ipv4",
241 dut,
242 input_dict,
243 protocol=protocol,
244 next_hop=nh,
245 retry_timeout=10,
246 expected=False,
247 )
248 assert result is not True, (
249 "Testcase {} : Failed \n "
250 "r1: OSPF routes are present in fib after deleting ip address of newly "
251 "configured interface of R0 \n Error: {}".format(tc_name, result)
252 )
253
254 step("Add back the deleted ip address on newly configured interface of R0")
255 topo1 = {
256 "r0": {
257 "links": {
258 "r3": {
259 "ipv4": topo["routers"]["r0"]["links"]["r3"]["ipv4"],
260 "interface": topo["routers"]["r0"]["links"]["r3"]["interface"],
261 }
262 }
263 }
264 }
265
266 result = create_interfaces_cfg(tgen, topo1)
267 assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
268
269 dut = "r1"
270 result = verify_ospf_rib(tgen, dut, input_dict, next_hop=nh)
271 assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
272
273 protocol = "ospf"
274 result = verify_rib(tgen, "ipv4", dut, input_dict, protocol=protocol, next_hop=nh)
275 assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
276
277 step("Shut no shut interface on R0")
278 dut = "r0"
279 intf = topo["routers"]["r0"]["links"]["r3"]["interface"]
280 shutdown_bringup_interface(tgen, dut, intf, False)
281
282 step("un shut the OSPF interface on R0")
283 dut = "r0"
284 shutdown_bringup_interface(tgen, dut, intf, True)
285
286 dut = "r1"
287 result = verify_ospf_rib(tgen, dut, input_dict)
288 assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
289
290 protocol = "ospf"
291 result = verify_rib(tgen, "ipv4", dut, input_dict, protocol=protocol, next_hop=nh)
292 assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
293
294 write_test_footer(tc_name)
295
296
297 def test_ospf_redistribution_tc6_p0(request):
298 """Test OSPF inter area route calculations."""
299 tc_name = request.node.name
300 write_test_header(tc_name)
301 tgen = get_topogen()
302
303 # Don't run this test if we have any failure.
304 if tgen.routers_have_failure():
305 pytest.skip(tgen.errors)
306
307 global topo
308 step("Bring up the base config.")
309 reset_config_on_routers(tgen)
310
311 step("Verify that OSPF neighbors are FULL.")
312 ospf_covergence = verify_ospf_neighbor(tgen, topo)
313 assert ospf_covergence is True, "setup_module :Failed \n Error:" " {}".format(
314 ospf_covergence
315 )
316
317 step("verify intra area route is calculated for r0-r3 interface ip in R1")
318 ip = topo["routers"]["r0"]["links"]["r3"]["ipv4"]
319 ip_net = str(ipaddress.ip_interface(u"{}".format(ip)).network)
320 nh = topo["routers"]["r0"]["links"]["r1"]["ipv4"].split("/")[0]
321 input_dict = {
322 "r1": {"static_routes": [{"network": ip_net, "no_of_ip": 1, "routeType": "N"}]}
323 }
324
325 dut = "r1"
326 result = verify_ospf_rib(tgen, dut, input_dict, next_hop=nh)
327 assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
328
329 protocol = "ospf"
330 result = verify_rib(tgen, "ipv4", dut, input_dict, protocol=protocol, next_hop=nh)
331 assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
332
333 step("Delete the ip address on newly configured loopback of R0")
334 topo1 = {
335 "r0": {
336 "links": {
337 "r3": {
338 "ipv4": topo["routers"]["r0"]["links"]["r3"]["ipv4"],
339 "interface": topo["routers"]["r0"]["links"]["r3"]["interface"],
340 "delete": True,
341 }
342 }
343 }
344 }
345
346 result = create_interfaces_cfg(tgen, topo1)
347 assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
348
349 dut = "r1"
350 for num in range(0, nretry):
351 result = verify_ospf_rib(tgen, dut, input_dict, next_hop=nh, expected=False)
352 if result is not True:
353 break
354 assert result is not True, (
355 "Testcase {} : Failed \n "
356 "r1: OSPF routes are present after deleting ip address of newly "
357 "configured loopback of R0 \n Error: {}".format(tc_name, result)
358 )
359
360 protocol = "ospf"
361 result = verify_rib(
362 tgen,
363 "ipv4",
364 dut,
365 input_dict,
366 protocol=protocol,
367 next_hop=nh,
368 expected=False,
369 )
370 assert result is not True, (
371 "Testcase {} : Failed \n "
372 "r1: OSPF routes are present in fib after deleting ip address of newly "
373 "configured loopback of R0 \n Error: {}".format(tc_name, result)
374 )
375
376 step("Add back the deleted ip address on newly configured interface of R0")
377 topo1 = {
378 "r0": {
379 "links": {
380 "r3": {
381 "ipv4": topo["routers"]["r0"]["links"]["r3"]["ipv4"],
382 "interface": topo["routers"]["r0"]["links"]["r3"]["interface"],
383 }
384 }
385 }
386 }
387
388 result = create_interfaces_cfg(tgen, topo1)
389 assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
390
391 dut = "r1"
392 result = verify_ospf_rib(tgen, dut, input_dict, next_hop=nh)
393 assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
394
395 protocol = "ospf"
396 result = verify_rib(tgen, "ipv4", dut, input_dict, protocol=protocol, next_hop=nh)
397 assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
398
399 step("Shut no shut interface on R0")
400 dut = "r0"
401 intf = topo["routers"]["r0"]["links"]["r3"]["interface"]
402 shutdown_bringup_interface(tgen, dut, intf, False)
403
404 step("un shut the OSPF interface on R0")
405 dut = "r0"
406 shutdown_bringup_interface(tgen, dut, intf, True)
407
408 dut = "r1"
409 result = verify_ospf_rib(tgen, dut, input_dict)
410 assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
411
412 protocol = "ospf"
413 result = verify_rib(tgen, "ipv4", dut, input_dict, protocol=protocol, next_hop=nh)
414 assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
415
416 write_test_footer(tc_name)
417
418
419 def test_ospf_redistribution_tc8_p1(request):
420 """
421 Test OSPF redistribution of connected routes.
422
423 Verify OSPF redistribution of connected routes when bgp multi hop
424 neighbor is configured using ospf routes
425
426 """
427 tc_name = request.node.name
428 write_test_header(tc_name)
429 tgen = get_topogen()
430 global topo
431 step("Bring up the base config.")
432 step(
433 "Configure loopback interface on all routers, and redistribut"
434 "e connected routes into ospf"
435 )
436 reset_config_on_routers(tgen)
437
438 step(
439 "verify that connected routes -loopback is found in all routers"
440 "advertised/exchaged via ospf"
441 )
442 for rtr in topo["routers"]:
443 redistribute_ospf(tgen, topo, rtr, "static")
444 redistribute_ospf(tgen, topo, rtr, "connected")
445 for node in topo["routers"]:
446 input_dict = {
447 "r0": {
448 "static_routes": [
449 {
450 "network": topo["routers"][node]["links"]["lo"]["ipv4"],
451 "no_of_ip": 1,
452 }
453 ]
454 }
455 }
456 for rtr in topo["routers"]:
457 result = verify_rib(tgen, "ipv4", rtr, input_dict)
458 assert result is True, "Testcase {} : Failed \n Error: {}".format(
459 tc_name, result
460 )
461
462 step("Configure E BGP multi hop using the loopback addresses.")
463 as_num = 100
464 for node in topo["routers"]:
465 as_num += 1
466 topo["routers"][node].update(
467 {
468 "bgp": {
469 "local_as": as_num,
470 "address_family": {"ipv4": {"unicast": {"neighbor": {}}}},
471 }
472 }
473 )
474 for node in topo["routers"]:
475 for rtr in topo["routers"]:
476 if node is not rtr:
477 topo["routers"][node]["bgp"]["address_family"]["ipv4"]["unicast"][
478 "neighbor"
479 ].update(
480 {
481 rtr: {
482 "dest_link": {
483 "lo": {"source_link": "lo", "ebgp_multihop": 2}
484 }
485 }
486 }
487 )
488
489 result = create_router_bgp(tgen, topo, topo["routers"])
490 assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
491
492 step("Verify that BGP neighbor is ESTABLISHED")
493 result = verify_bgp_convergence(tgen, topo)
494 assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
495 step(
496 "Configure couple of static routes in R0 and "
497 "Redistribute static routes in R1 bgp."
498 )
499
500 for rtr in topo["routers"]:
501 redistribute_ospf(tgen, topo, rtr, "static", delete=True)
502
503 input_dict = {
504 "r0": {
505 "static_routes": [
506 {
507 "network": NETWORK["ipv4"][0],
508 "no_of_ip": 5,
509 "next_hop": "Null0",
510 }
511 ]
512 }
513 }
514 result = create_static_routes(tgen, input_dict)
515 assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
516
517 configure_bgp_on_r0 = {
518 "r0": {
519 "bgp": {
520 "address_family": {
521 "ipv4": {"unicast": {"redistribute": [{"redist_type": "static"}]}}
522 }
523 }
524 }
525 }
526 result = create_router_bgp(tgen, topo, configure_bgp_on_r0)
527 assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
528 protocol = "bgp"
529 for rtr in ["r1", "r2", "r3"]:
530 result = verify_rib(tgen, "ipv4", rtr, input_dict, protocol=protocol)
531 assert result is True, "Testcase {} : Failed \n Error: {}".format(
532 tc_name, result
533 )
534
535 step("Clear ospf neighbours in R0")
536 for rtr in topo["routers"]:
537 clear_ospf(tgen, rtr)
538
539 step("Verify that OSPF neighbours are reset and forms new adjacencies.")
540 # Api call verify whether OSPF is converged
541 ospf_covergence = verify_ospf_neighbor(tgen, topo)
542 assert ospf_covergence is True, "setup_module :Failed \n Error:" " {}".format(
543 ospf_covergence
544 )
545
546 step("Verify that BGP neighbours are reset and forms new adjacencies.")
547 result = verify_bgp_convergence(tgen, topo)
548 assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
549
550 protocol = "bgp"
551 for rtr in ["r1", "r2", "r3"]:
552 result = verify_rib(tgen, "ipv4", rtr, input_dict, protocol=protocol)
553 assert result is True, "Testcase {} : Failed \n Error: {}".format(
554 tc_name, result
555 )
556
557 write_test_footer(tc_name)
558
559
560 def test_ospf_cost_tc52_p0(request):
561 """OSPF Cost - verifying ospf interface cost functionality"""
562 tc_name = request.node.name
563 write_test_header(tc_name)
564 tgen = get_topogen()
565 global topo
566 step("Bring up the base config.")
567 reset_config_on_routers(tgen)
568
569 step(
570 "Configure ospf cost as 20 on interface between R0 and R1. "
571 "Configure ospf cost as 30 between interface between R0 and R2."
572 )
573
574 r0_ospf_cost = {
575 "r0": {"links": {"r1": {"ospf": {"cost": 20}}, "r2": {"ospf": {"cost": 30}}}}
576 }
577 result = config_ospf_interface(tgen, topo, r0_ospf_cost)
578 assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
579
580 step(
581 "Verify that cost is updated in the ospf interface between"
582 " r0 and r1 as 30 and r0 and r2 as 20"
583 )
584 dut = "r0"
585 result = verify_ospf_interface(tgen, topo, dut=dut, input_dict=r0_ospf_cost)
586 assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
587
588 step(
589 "Swap the costs between interfaces on r0, between r0 and r1 to 30"
590 ", r0 and r2 to 20"
591 )
592
593 r0_ospf_cost = {
594 "r0": {"links": {"r1": {"ospf": {"cost": 30}}, "r2": {"ospf": {"cost": 20}}}}
595 }
596 result = config_ospf_interface(tgen, topo, r0_ospf_cost)
597 assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
598
599 step(
600 "Verify that cost is updated in the ospf interface between r0 "
601 "and r1 as 30 and r0 and r2 as 20."
602 )
603 result = verify_ospf_interface(tgen, topo, dut=dut, input_dict=r0_ospf_cost)
604 assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
605
606 step(" Un configure cost from the interface r0 - r1.")
607
608 r0_ospf_cost = {"r0": {"links": {"r1": {"ospf": {"cost": 30, "del_action": True}}}}}
609 result = config_ospf_interface(tgen, topo, r0_ospf_cost)
610 assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
611
612 input_dict = {
613 "r0": {"links": {"r1": {"ospf": {"cost": 10}}, "r2": {"ospf": {"cost": 20}}}}
614 }
615 step(
616 "Verify that cost is updated in the ospf interface between r0"
617 " and r1 as 10 and r0 and r2 as 20."
618 )
619
620 result = verify_ospf_interface(tgen, topo, dut=dut, input_dict=input_dict)
621 assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
622
623 step(" Un configure cost from the interface r0 - r2.")
624
625 r0_ospf_cost = {"r0": {"links": {"r2": {"ospf": {"cost": 20, "del_action": True}}}}}
626 result = config_ospf_interface(tgen, topo, r0_ospf_cost)
627 assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
628
629 step(
630 "Verify that cost is updated in the ospf interface between r0"
631 "and r1 as 10 and r0 and r2 as 10"
632 )
633
634 input_dict = {
635 "r0": {"links": {"r1": {"ospf": {"cost": 10}}, "r2": {"ospf": {"cost": 10}}}}
636 }
637 result = verify_ospf_interface(tgen, topo, dut=dut, input_dict=input_dict)
638 assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
639
640 write_test_footer(tc_name)
641
642
643 if __name__ == "__main__":
644 args = ["-s"] + sys.argv[1:]
645 sys.exit(pytest.main(args))