]> git.proxmox.com Git - mirror_frr.git/blob - tests/topotests/ospf_basic_functionality/test_ospf_lan.py
Merge pull request #9610 from iqras23/best_path
[mirror_frr.git] / tests / topotests / ospf_basic_functionality / test_ospf_lan.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 from copy import deepcopy
30 from lib.topotest import frr_unicode
31
32 # Save the Current Working Directory to find configuration files.
33 CWD = os.path.dirname(os.path.realpath(__file__))
34 sys.path.append(os.path.join(CWD, "../"))
35 sys.path.append(os.path.join(CWD, "../lib/"))
36
37 # pylint: disable=C0413
38 # Import topogen and topotest helpers
39 from lib.topogen import Topogen, get_topogen
40
41 # Import topoJson from lib, to create topology and initial configuration
42 from lib.common_config import (
43 start_topology,
44 write_test_header,
45 create_interfaces_cfg,
46 write_test_footer,
47 reset_config_on_routers,
48 step,
49 shutdown_bringup_interface,
50 stop_router,
51 start_router,
52 topo_daemons,
53 )
54 from lib.topolog import logger
55 from lib.topojson import build_config_from_json
56 from lib.ospf import (
57 verify_ospf_neighbor,
58 clear_ospf,
59 create_router_ospf,
60 verify_ospf_interface,
61 )
62 from ipaddress import IPv4Address
63
64 pytestmark = [pytest.mark.ospfd, pytest.mark.staticd]
65
66
67 # Global variables
68 topo = None
69
70 NETWORK = {
71 "ipv4": [
72 "11.0.20.1/32",
73 "11.0.20.2/32",
74 "11.0.20.3/32",
75 "11.0.20.4/32",
76 "11.0.20.5/32",
77 ]
78 }
79
80 """
81 Topology:
82
83 Please view in a fixed-width font such as Courier.
84 Topo : Broadcast Networks
85 +---+ +---+ +---+ +---+
86 |R0 + +R1 + +R2 + +R3 |
87 +-+-+ +-+-+ +-+-+ +-+-+
88 | | | |
89 | | | |
90 --+-----------+--------------+---------------+-----
91 Ethernet Segment
92
93 Testcases:
94 1. OSPF Hello protocol - Verify DR BDR Elections
95 2. OSPF IFSM -Verify state change events on DR / BDR / DR Other
96
97 """
98
99
100 def setup_module(mod):
101 """
102 Sets up the pytest environment
103
104 * `mod`: module name
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 json_file = "{}/ospf_lan.json".format(CWD)
114 tgen = Topogen(json_file, mod.__name__)
115 global topo
116 topo = tgen.json_topo
117 # ... and here it calls Mininet initialization functions.
118
119 # get list of daemons needs to be started for this suite.
120 daemons = topo_daemons(tgen, topo)
121
122 # Starting topology, create tmp files which are loaded to routers
123 # to start deamons and then start routers
124 start_topology(tgen, daemons)
125
126 # Creating configuration from JSON
127 build_config_from_json(tgen, topo)
128
129 # Don't run this test if we have any failure.
130 if tgen.routers_have_failure():
131 pytest.skip(tgen.errors)
132 # Api call verify whether OSPF is converged
133 ospf_covergence = verify_ospf_neighbor(tgen, topo, lan=True)
134 assert ospf_covergence is True, "setup_module :Failed \n Error:" " {}".format(
135 ospf_covergence
136 )
137
138 logger.info("Running setup_module() done")
139
140
141 def teardown_module():
142 """Teardown the pytest environment"""
143
144 logger.info("Running teardown_module to delete topology")
145
146 tgen = get_topogen()
147
148 try:
149 # Stop toplogy and Remove tmp files
150 tgen.stop_topology
151
152 except OSError:
153 # OSError exception is raised when mininet tries to stop switch
154 # though switch is stopped once but mininet tries to stop same
155 # switch again, where it ended up with exception
156 pass
157
158
159 # ##################################
160 # Test cases start here.
161 # ##################################
162
163
164 def test_ospf_lan_tc1_p0(request):
165 """
166 OSPF Hello protocol - Verify DR BDR Elections
167
168 """
169 tc_name = request.node.name
170 write_test_header(tc_name)
171 tgen = get_topogen()
172
173 # Don't run this test if we have any failure.
174 if tgen.routers_have_failure():
175 pytest.skip(tgen.errors)
176
177 global topo
178 step("Bring up the base config as per the topology")
179 reset_config_on_routers(tgen)
180 step("Verify that DR BDR DRother are elected in the LAN.")
181 input_dict = {
182 "r0": {
183 "ospf": {
184 "neighbors": {
185 "r1": {"state": "Full", "role": "DR"},
186 "r2": {"state": "Full", "role": "DROther"},
187 "r3": {"state": "Full", "role": "DROther"},
188 }
189 }
190 }
191 }
192 dut = "r0"
193 result = verify_ospf_neighbor(tgen, topo, dut, input_dict, lan=True)
194 assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
195
196 step(
197 "Verify that all the routers are in FULL state with DR and BDR "
198 "in the topology"
199 )
200
201 input_dict = {
202 "r1": {
203 "ospf": {
204 "neighbors": {
205 "r0": {"state": "Full", "role": "Backup"},
206 "r2": {"state": "Full", "role": "DROther"},
207 "r3": {"state": "Full", "role": "DROther"},
208 }
209 }
210 }
211 }
212 dut = "r1"
213 result = verify_ospf_neighbor(tgen, topo, dut, input_dict, lan=True)
214 assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
215
216 step(
217 "Configure DR pririty 100 on R0 and clear ospf neighbors " "on all the routers."
218 )
219
220 input_dict = {
221 "r0": {
222 "links": {
223 "s1": {
224 "interface": topo["routers"]["r0"]["links"]["s1"]["interface"],
225 "ospf": {"priority": 100},
226 }
227 }
228 }
229 }
230
231 result = create_interfaces_cfg(tgen, input_dict)
232 assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
233
234 step("Clear ospf neighbours in all routers")
235 for rtr in ["r0", "r1", "r2", "r3"]:
236 clear_ospf(tgen, rtr)
237
238 step("Verify that DR election is triggered and R0 is elected as DR")
239 input_dict = {
240 "r0": {
241 "ospf": {
242 "neighbors": {
243 "r1": {"state": "Full", "role": "Backup"},
244 "r2": {"state": "Full", "role": "DROther"},
245 "r3": {"state": "Full", "role": "DROther"},
246 }
247 }
248 }
249 }
250 dut = "r0"
251 result = verify_ospf_neighbor(tgen, topo, dut, input_dict, lan=True)
252 assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
253
254 step(
255 "Configure DR pririty 150 on R0 and clear ospf neighbors " "on all the routers."
256 )
257
258 input_dict = {
259 "r0": {
260 "links": {
261 "s1": {
262 "interface": topo["routers"]["r0"]["links"]["s1"]["interface"],
263 "ospf": {"priority": 150},
264 }
265 }
266 }
267 }
268
269 result = create_interfaces_cfg(tgen, input_dict)
270 assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
271
272 step("Clear ospf neighbours in all routers")
273 for rtr in ["r0", "r1"]:
274 clear_ospf(tgen, rtr)
275
276 step("Verify that DR election is triggered and R0 is elected as DR")
277 input_dict = {
278 "r0": {
279 "ospf": {
280 "neighbors": {
281 "r1": {"state": "Full", "role": "Backup"},
282 "r2": {"state": "Full", "role": "DROther"},
283 "r3": {"state": "Full", "role": "DROther"},
284 }
285 }
286 }
287 }
288 dut = "r0"
289 result = verify_ospf_neighbor(tgen, topo, dut, input_dict, lan=True)
290 assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
291
292 step("Configure DR priority 0 on R0 & Clear ospf nbrs on all the routers")
293
294 input_dict = {
295 "r0": {
296 "links": {
297 "s1": {
298 "interface": topo["routers"]["r0"]["links"]["s1"]["interface"],
299 "ospf": {"priority": 0},
300 }
301 }
302 }
303 }
304
305 result = create_interfaces_cfg(tgen, input_dict)
306 assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
307
308 step("Clear ospf neighbours in all routers")
309 for rtr in ["r1"]:
310 clear_ospf(tgen, rtr)
311
312 step("Verify that DR election is triggered and R0 is elected as DRother")
313 input_dict = {
314 "r0": {
315 "ospf": {
316 "neighbors": {
317 "r1": {"state": "Full", "role": "DR"},
318 "r2": {"state": "2-Way", "role": "DROther"},
319 "r3": {"state": "2-Way", "role": "DROther"},
320 }
321 }
322 }
323 }
324 dut = "r0"
325 result = verify_ospf_neighbor(tgen, topo, dut, input_dict, lan=True)
326 assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
327
328 step(
329 "Configure DR priority to default on R0 and Clear ospf neighbors"
330 " on all the routers"
331 )
332
333 input_dict = {
334 "r0": {
335 "links": {
336 "s1": {
337 "interface": topo["routers"]["r0"]["links"]["s1"]["interface"],
338 "ospf": {"priority": 100},
339 }
340 }
341 }
342 }
343
344 result = create_interfaces_cfg(tgen, input_dict)
345 assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
346
347 step("Clear ospf neighbours in all routers")
348 for rtr in ["r0", "r1"]:
349 clear_ospf(tgen, rtr)
350
351 step("Verify that DR election is triggered and R0 is elected as DR")
352 input_dict = {
353 "r0": {
354 "ospf": {
355 "neighbors": {
356 "r1": {"state": "Full", "role": "Backup"},
357 "r2": {"state": "Full", "role": "DROther"},
358 "r3": {"state": "Full", "role": "DROther"},
359 }
360 }
361 }
362 }
363 dut = "r0"
364 result = verify_ospf_neighbor(tgen, topo, dut, input_dict, lan=True)
365 assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
366
367 step("Shut interface on R0")
368 dut = "r0"
369 intf = topo["routers"]["r0"]["links"]["s1"]["interface"]
370 shutdown_bringup_interface(tgen, dut, intf, False)
371
372 result = verify_ospf_neighbor(tgen, topo, dut, lan=True, expected=False)
373 assert (
374 result is not True
375 ), "Testcase {} : Failed \n " "r0: OSPF neighbors-hip is up \n Error: {}".format(
376 tc_name, result
377 )
378
379 step("No Shut interface on R0")
380 dut = "r0"
381 intf = topo["routers"]["r0"]["links"]["s1"]["interface"]
382 shutdown_bringup_interface(tgen, dut, intf, True)
383
384 input_dict = {
385 "r0": {
386 "ospf": {
387 "neighbors": {
388 "r1": {"state": "Full", "role": "DR"},
389 "r2": {"state": "Full", "role": "DROther"},
390 "r3": {"state": "Full", "role": "DROther"},
391 }
392 }
393 }
394 }
395 step("Verify that after no shut ospf neighbours are full on R0.")
396 result = verify_ospf_neighbor(tgen, topo, dut, input_dict, lan=True)
397 assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
398
399 step("Clear ospf on DR router in the topology.")
400 clear_ospf(tgen, "r0")
401
402 step("Verify that BDR is getting promoted to DR after clear.")
403 step("Verify that all the nbrs are in FULL state with the elected DR.")
404 result = verify_ospf_neighbor(tgen, topo, dut, input_dict, lan=True)
405 assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
406
407 step("Change the ip on LAN intf on R0 to other ip from the same subnet.")
408 topo_modify_change_ip = deepcopy(topo)
409 intf_ip = topo_modify_change_ip["routers"]["r0"]["links"]["s1"]["ipv4"]
410 topo_modify_change_ip["routers"]["r0"]["links"]["s1"]["ipv4"] = str(
411 IPv4Address(frr_unicode(intf_ip.split("/")[0])) + 4
412 ) + "/{}".format(intf_ip.split("/")[1])
413
414 build_config_from_json(tgen, topo_modify_change_ip, save_bkup=False)
415
416 step(
417 "Verify that OSPF is in FULL state with other routers with "
418 "newly configured IP."
419 )
420 result = verify_ospf_neighbor(tgen, topo, dut, input_dict, lan=True)
421 assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
422
423 step("Change the ospf router id on the R0 and clear ip ospf interface.")
424 change_rid = {"r0": {"ospf": {"router_id": "100.1.1.100"}}}
425
426 result = create_router_ospf(tgen, topo, change_rid)
427 assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
428 topo["routers"]["r0"]["ospf"]["router_id"] = "100.1.1.100"
429 step("Reload the FRR router")
430
431 stop_router(tgen, "r0")
432 start_router(tgen, "r0")
433
434 step(
435 "Verify that OSPF is in FULL state with other routers with"
436 " newly configured router id."
437 )
438 input_dict = {
439 "r1": {
440 "ospf": {
441 "neighbors": {
442 "r0": {"state": "Full", "role": "Backup"},
443 "r2": {"state": "Full", "role": "DROther"},
444 "r3": {"state": "Full", "role": "DROther"},
445 }
446 }
447 }
448 }
449 dut = "r1"
450 result = verify_ospf_neighbor(tgen, topo, dut, input_dict, lan=True)
451 assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
452
453 step("Reconfigure the original router id and clear ip ospf interface.")
454 change_rid = {"r0": {"ospf": {"router_id": "100.1.1.0"}}}
455 result = create_router_ospf(tgen, topo, change_rid)
456 assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
457 topo["routers"]["r0"]["ospf"]["router_id"] = "100.1.1.0"
458 step("Reload the FRR router")
459 # stop/start -> restart FRR router and verify
460 stop_router(tgen, "r0")
461 start_router(tgen, "r0")
462
463 step("Verify that OSPF is enabled with router id previously configured.")
464 input_dict = {
465 "r1": {
466 "ospf": {
467 "neighbors": {
468 "r0": {"state": "Full", "role": "Backup"},
469 "r2": {"state": "Full", "role": "DROther"},
470 "r3": {"state": "Full", "role": "DROther"},
471 }
472 }
473 }
474 }
475 dut = "r1"
476 result = verify_ospf_neighbor(tgen, topo, dut, input_dict, lan=True)
477 assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
478
479 write_test_footer(tc_name)
480
481
482 def test_ospf_lan_tc2_p0(request):
483 """
484 OSPF IFSM -Verify state change events on DR / BDR / DR Other
485
486 """
487 tc_name = request.node.name
488 write_test_header(tc_name)
489 tgen = get_topogen()
490
491 # Don't run this test if we have any failure.
492 if tgen.routers_have_failure():
493 pytest.skip(tgen.errors)
494
495 global topo
496 step("Bring up the base config as per the topology")
497 reset_config_on_routers(tgen)
498 step(
499 "Verify that OSPF is subscribed to multi cast services "
500 "(All SPF, all DR Routers)."
501 )
502 step("Verify that interface is enabled in ospf.")
503 dut = "r0"
504 input_dict = {
505 "r0": {
506 "links": {
507 "s1": {
508 "ospf": {
509 "priority": 98,
510 "timerDeadSecs": 4,
511 "area": "0.0.0.3",
512 "mcastMemberOspfDesignatedRouters": True,
513 "mcastMemberOspfAllRouters": True,
514 "ospfEnabled": True,
515 }
516 }
517 }
518 }
519 }
520 result = verify_ospf_interface(tgen, topo, dut=dut, input_dict=input_dict)
521 assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
522
523 step("Delete the ip address")
524 topo1 = {
525 "r0": {
526 "links": {
527 "r3": {
528 "ipv4": topo["routers"]["r0"]["links"]["s1"]["ipv4"],
529 "interface": topo["routers"]["r0"]["links"]["s1"]["interface"],
530 "delete": True,
531 }
532 }
533 }
534 }
535
536 result = create_interfaces_cfg(tgen, topo1)
537 assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
538
539 step("Change the ip on the R0 interface")
540
541 topo_modify_change_ip = deepcopy(topo)
542 intf_ip = topo_modify_change_ip["routers"]["r0"]["links"]["s1"]["ipv4"]
543 topo_modify_change_ip["routers"]["r0"]["links"]["s1"]["ipv4"] = str(
544 IPv4Address(frr_unicode(intf_ip.split("/")[0])) + 3
545 ) + "/{}".format(intf_ip.split("/")[1])
546
547 build_config_from_json(tgen, topo_modify_change_ip, save_bkup=False)
548 step("Verify that interface is enabled in ospf.")
549 dut = "r0"
550 input_dict = {
551 "r0": {
552 "links": {
553 "s1": {
554 "ospf": {
555 "ipAddress": topo_modify_change_ip["routers"]["r0"]["links"][
556 "s1"
557 ]["ipv4"].split("/")[0],
558 "ipAddressPrefixlen": int(
559 topo_modify_change_ip["routers"]["r0"]["links"]["s1"][
560 "ipv4"
561 ].split("/")[1]
562 ),
563 }
564 }
565 }
566 }
567 }
568 result = verify_ospf_interface(tgen, topo, dut=dut, input_dict=input_dict)
569 assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
570
571 step("Modify the mask on the R0 interface")
572 ip_addr = topo_modify_change_ip["routers"]["r0"]["links"]["s1"]["ipv4"]
573 mask = topo_modify_change_ip["routers"]["r0"]["links"]["s1"]["ipv4"]
574 step("Delete the ip address")
575 topo1 = {
576 "r0": {
577 "links": {
578 "r3": {
579 "ipv4": ip_addr,
580 "interface": topo["routers"]["r0"]["links"]["s1"]["interface"],
581 "delete": True,
582 }
583 }
584 }
585 }
586
587 result = create_interfaces_cfg(tgen, topo1)
588 assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
589
590 step("Change the ip on the R0 interface")
591
592 topo_modify_change_ip = deepcopy(topo)
593 intf_ip = topo_modify_change_ip["routers"]["r0"]["links"]["s1"]["ipv4"]
594 topo_modify_change_ip["routers"]["r0"]["links"]["s1"]["ipv4"] = str(
595 IPv4Address(frr_unicode(intf_ip.split("/")[0])) + 3
596 ) + "/{}".format(int(intf_ip.split("/")[1]) + 1)
597
598 build_config_from_json(tgen, topo_modify_change_ip, save_bkup=False)
599 step("Verify that interface is enabled in ospf.")
600 dut = "r0"
601 input_dict = {
602 "r0": {
603 "links": {
604 "s1": {
605 "ospf": {
606 "ipAddress": topo_modify_change_ip["routers"]["r0"]["links"][
607 "s1"
608 ]["ipv4"].split("/")[0],
609 "ipAddressPrefixlen": int(
610 topo_modify_change_ip["routers"]["r0"]["links"]["s1"][
611 "ipv4"
612 ].split("/")[1]
613 ),
614 }
615 }
616 }
617 }
618 }
619 result = verify_ospf_interface(tgen, topo, dut=dut, input_dict=input_dict)
620 assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
621
622 step("Change the area id on the interface")
623 input_dict = {
624 "r0": {
625 "links": {
626 "s1": {
627 "interface": topo["routers"]["r0"]["links"]["s1"]["interface"],
628 "ospf": {"area": "0.0.0.3"},
629 "delete": True,
630 }
631 }
632 }
633 }
634
635 result = create_interfaces_cfg(tgen, input_dict)
636 assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
637
638 input_dict = {
639 "r0": {
640 "links": {
641 "s1": {
642 "interface": topo["routers"]["r0"]["links"]["s1"]["interface"],
643 "ospf": {"area": "0.0.0.2"},
644 }
645 }
646 }
647 }
648
649 result = create_interfaces_cfg(tgen, input_dict)
650 assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
651 step("Verify that interface is enabled in ospf.")
652 dut = "r0"
653 input_dict = {
654 "r0": {"links": {"s1": {"ospf": {"area": "0.0.0.2", "ospfEnabled": True}}}}
655 }
656 result = verify_ospf_interface(tgen, topo, dut=dut, input_dict=input_dict)
657 assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
658
659 write_test_footer(tc_name)
660
661
662 if __name__ == "__main__":
663 args = ["-s"] + sys.argv[1:]
664 sys.exit(pytest.main(args))