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