2 # SPDX-License-Identifier: ISC
5 # Copyright (c) 2020 by VMware, Inc. ("VMware")
6 # Used Copyright (c) 2018 by Network Device Education Foundation, Inc.
7 # ("NetDEF") in this file.
11 """OSPF Basic Functionality Automation."""
16 from copy
import deepcopy
17 from lib
.topotest
import frr_unicode
19 # Save the Current Working Directory to find configuration files.
20 CWD
= os
.path
.dirname(os
.path
.realpath(__file__
))
21 sys
.path
.append(os
.path
.join(CWD
, "../"))
22 sys
.path
.append(os
.path
.join(CWD
, "../lib/"))
24 # pylint: disable=C0413
25 # Import topogen and topotest helpers
26 from lib
.topogen
import Topogen
, get_topogen
28 # Import topoJson from lib, to create topology and initial configuration
29 from lib
.common_config
import (
32 create_interfaces_cfg
,
34 reset_config_on_routers
,
36 shutdown_bringup_interface
,
40 from lib
.topolog
import logger
41 from lib
.topojson
import build_config_from_json
42 from lib
.ospf
import (
46 verify_ospf_interface
,
48 from ipaddress
import IPv4Address
50 pytestmark
= [pytest
.mark
.ospfd
, pytest
.mark
.staticd
]
69 Please view in a fixed-width font such as Courier.
70 Topo : Broadcast Networks
71 +---+ +---+ +---+ +---+
72 |R0 + +R1 + +R2 + +R3 |
73 +-+-+ +-+-+ +-+-+ +-+-+
76 --+-----------+--------------+---------------+-----
80 1. OSPF Hello protocol - Verify DR BDR Elections
81 2. OSPF IFSM -Verify state change events on DR / BDR / DR Other
86 def setup_module(mod
):
88 Sets up the pytest environment
92 testsuite_run_time
= time
.asctime(time
.localtime(time
.time()))
93 logger
.info("Testsuite start time: {}".format(testsuite_run_time
))
96 logger
.info("Running setup_module to create topology")
98 # This function initiates the topology build with Topogen...
99 json_file
= "{}/ospf_lan.json".format(CWD
)
100 tgen
= Topogen(json_file
, mod
.__name
__)
102 topo
= tgen
.json_topo
103 # ... and here it calls Mininet initialization functions.
105 # Starting topology, create tmp files which are loaded to routers
106 # to start daemons and then start routers
109 # Creating configuration from JSON
110 build_config_from_json(tgen
, topo
)
112 # Don't run this test if we have any failure.
113 if tgen
.routers_have_failure():
114 pytest
.skip(tgen
.errors
)
115 # Api call verify whether OSPF is converged
116 ospf_covergence
= verify_ospf_neighbor(tgen
, topo
, lan
=True)
117 assert ospf_covergence
is True, "setup_module :Failed \n Error:" " {}".format(
121 logger
.info("Running setup_module() done")
124 def teardown_module():
125 """Teardown the pytest environment"""
127 logger
.info("Running teardown_module to delete topology")
132 # Stop toplogy and Remove tmp files
136 # OSError exception is raised when mininet tries to stop switch
137 # though switch is stopped once but mininet tries to stop same
138 # switch again, where it ended up with exception
142 # ##################################
143 # Test cases start here.
144 # ##################################
147 def test_ospf_lan_tc1_p0(request
):
149 OSPF Hello protocol - Verify DR BDR Elections
152 tc_name
= request
.node
.name
153 write_test_header(tc_name
)
156 # Don't run this test if we have any failure.
157 if tgen
.routers_have_failure():
158 pytest
.skip(tgen
.errors
)
161 step("Bring up the base config as per the topology")
162 reset_config_on_routers(tgen
)
163 step("Verify that DR BDR DRother are elected in the LAN.")
168 "r1": {"state": "Full", "role": "DR"},
169 "r2": {"state": "Full", "role": "DROther"},
170 "r3": {"state": "Full", "role": "DROther"},
176 result
= verify_ospf_neighbor(tgen
, topo
, dut
, input_dict
, lan
=True)
177 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
180 "Verify that all the routers are in FULL state with DR and BDR "
188 "r0": {"state": "Full", "role": "Backup"},
189 "r2": {"state": "Full", "role": "DROther"},
190 "r3": {"state": "Full", "role": "DROther"},
196 result
= verify_ospf_neighbor(tgen
, topo
, dut
, input_dict
, lan
=True)
197 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
200 "Configure DR pririty 100 on R0 and clear ospf neighbors " "on all the routers."
207 "interface": topo
["routers"]["r0"]["links"]["s1"]["interface"],
208 "ospf": {"priority": 100},
214 result
= create_interfaces_cfg(tgen
, input_dict
)
215 assert result
is True, "Testcase {} :Failed \n Error: {}".format(tc_name
, result
)
217 step("Clear ospf neighbours in all routers")
218 for rtr
in ["r0", "r1", "r2", "r3"]:
219 clear_ospf(tgen
, rtr
)
221 step("Verify that DR election is triggered and R0 is elected as DR")
226 "r1": {"state": "Full", "role": "Backup"},
227 "r2": {"state": "Full", "role": "DROther"},
228 "r3": {"state": "Full", "role": "DROther"},
234 result
= verify_ospf_neighbor(tgen
, topo
, dut
, input_dict
, lan
=True)
235 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
238 "Configure DR pririty 150 on R0 and clear ospf neighbors " "on all the routers."
245 "interface": topo
["routers"]["r0"]["links"]["s1"]["interface"],
246 "ospf": {"priority": 150},
252 result
= create_interfaces_cfg(tgen
, input_dict
)
253 assert result
is True, "Testcase {} :Failed \n Error: {}".format(tc_name
, result
)
255 step("Clear ospf neighbours in all routers")
256 for rtr
in ["r0", "r1"]:
257 clear_ospf(tgen
, rtr
)
259 step("Verify that DR election is triggered and R0 is elected as DR")
264 "r1": {"state": "Full", "role": "Backup"},
265 "r2": {"state": "Full", "role": "DROther"},
266 "r3": {"state": "Full", "role": "DROther"},
272 result
= verify_ospf_neighbor(tgen
, topo
, dut
, input_dict
, lan
=True)
273 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
275 step("Configure DR priority 0 on R0 & Clear ospf nbrs on all the routers")
281 "interface": topo
["routers"]["r0"]["links"]["s1"]["interface"],
282 "ospf": {"priority": 0},
288 result
= create_interfaces_cfg(tgen
, input_dict
)
289 assert result
is True, "Testcase {} :Failed \n Error: {}".format(tc_name
, result
)
291 step("Clear ospf neighbours in all routers")
293 clear_ospf(tgen
, rtr
)
295 step("Verify that DR election is triggered and R0 is elected as DRother")
300 "r1": {"state": "Full", "role": "DR"},
301 "r2": {"state": "2-Way", "role": "DROther"},
302 "r3": {"state": "2-Way", "role": "DROther"},
308 result
= verify_ospf_neighbor(tgen
, topo
, dut
, input_dict
, lan
=True)
309 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
312 "Configure DR priority to default on R0 and Clear ospf neighbors"
313 " on all the routers"
320 "interface": topo
["routers"]["r0"]["links"]["s1"]["interface"],
321 "ospf": {"priority": 100},
327 result
= create_interfaces_cfg(tgen
, input_dict
)
328 assert result
is True, "Testcase {} :Failed \n Error: {}".format(tc_name
, result
)
330 step("Clear ospf neighbours in all routers")
331 for rtr
in ["r0", "r1"]:
332 clear_ospf(tgen
, rtr
)
334 step("Verify that DR election is triggered and R0 is elected as DR")
339 "r1": {"state": "Full", "role": "Backup"},
340 "r2": {"state": "Full", "role": "DROther"},
341 "r3": {"state": "Full", "role": "DROther"},
347 result
= verify_ospf_neighbor(tgen
, topo
, dut
, input_dict
, lan
=True)
348 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
350 step("Shut interface on R0")
352 intf
= topo
["routers"]["r0"]["links"]["s1"]["interface"]
353 shutdown_bringup_interface(tgen
, dut
, intf
, False)
355 result
= verify_ospf_neighbor(tgen
, topo
, dut
, lan
=True, expected
=False)
358 ), "Testcase {} : Failed \n " "r0: OSPF neighbors-hip is up \n Error: {}".format(
362 step("No Shut interface on R0")
364 intf
= topo
["routers"]["r0"]["links"]["s1"]["interface"]
365 shutdown_bringup_interface(tgen
, dut
, intf
, True)
371 "r1": {"state": "Full", "role": "DR"},
372 "r2": {"state": "Full", "role": "DROther"},
373 "r3": {"state": "Full", "role": "DROther"},
378 step("Verify that after no shut ospf neighbours are full on R0.")
379 result
= verify_ospf_neighbor(tgen
, topo
, dut
, input_dict
, lan
=True)
380 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
382 step("Clear ospf on DR router in the topology.")
383 clear_ospf(tgen
, "r0")
385 step("Verify that BDR is getting promoted to DR after clear.")
386 step("Verify that all the nbrs are in FULL state with the elected DR.")
387 result
= verify_ospf_neighbor(tgen
, topo
, dut
, input_dict
, lan
=True)
388 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
390 step("Change the ip on LAN intf on R0 to other ip from the same subnet.")
391 topo_modify_change_ip
= deepcopy(topo
)
392 intf_ip
= topo_modify_change_ip
["routers"]["r0"]["links"]["s1"]["ipv4"]
393 topo_modify_change_ip
["routers"]["r0"]["links"]["s1"]["ipv4"] = str(
394 IPv4Address(frr_unicode(intf_ip
.split("/")[0])) + 4
395 ) + "/{}".format(intf_ip
.split("/")[1])
397 build_config_from_json(tgen
, topo_modify_change_ip
, save_bkup
=False)
399 clear_ospf(tgen
, "r0")
401 "Verify that OSPF is in FULL state with other routers with "
402 "newly configured IP."
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
)
407 step("Change the ospf router id on the R0 and clear ip ospf interface.")
408 change_rid
= {"r0": {"ospf": {"router_id": "100.1.1.100"}}}
410 result
= create_router_ospf(tgen
, topo
, change_rid
)
411 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
412 topo
["routers"]["r0"]["ospf"]["router_id"] = "100.1.1.100"
413 step("Reload the FRR router")
415 stop_router(tgen
, "r0")
416 start_router(tgen
, "r0")
419 "Verify that OSPF is in FULL state with other routers with"
420 " newly configured router id."
426 "r0": {"state": "Full", "role": "Backup"},
427 "r2": {"state": "Full", "role": "DROther"},
428 "r3": {"state": "Full", "role": "DROther"},
434 result
= verify_ospf_neighbor(tgen
, topo
, dut
, input_dict
, lan
=True)
435 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
437 step("Reconfigure the original router id and clear ip ospf interface.")
438 change_rid
= {"r0": {"ospf": {"router_id": "100.1.1.0"}}}
439 result
= create_router_ospf(tgen
, topo
, change_rid
)
440 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
441 topo
["routers"]["r0"]["ospf"]["router_id"] = "100.1.1.0"
442 step("Reload the FRR router")
443 # stop/start -> restart FRR router and verify
444 stop_router(tgen
, "r0")
445 start_router(tgen
, "r0")
447 step("Verify that OSPF is enabled with router id previously configured.")
452 "r0": {"state": "Full", "role": "Backup"},
453 "r2": {"state": "Full", "role": "DROther"},
454 "r3": {"state": "Full", "role": "DROther"},
460 result
= verify_ospf_neighbor(tgen
, topo
, dut
, input_dict
, lan
=True)
461 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
463 write_test_footer(tc_name
)
466 def test_ospf_lan_tc2_p0(request
):
468 OSPF IFSM -Verify state change events on DR / BDR / DR Other
471 tc_name
= request
.node
.name
472 write_test_header(tc_name
)
475 # Don't run this test if we have any failure.
476 if tgen
.routers_have_failure():
477 pytest
.skip(tgen
.errors
)
480 step("Bring up the base config as per the topology")
481 reset_config_on_routers(tgen
)
483 "Verify that OSPF is subscribed to multi cast services "
484 "(All SPF, all DR Routers)."
486 step("Verify that interface is enabled in ospf.")
496 "mcastMemberOspfDesignatedRouters": True,
497 "mcastMemberOspfAllRouters": True,
504 result
= verify_ospf_interface(tgen
, topo
, dut
=dut
, input_dict
=input_dict
)
505 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
507 step("Delete the ip address")
512 "ipv4": topo
["routers"]["r0"]["links"]["s1"]["ipv4"],
513 "interface": topo
["routers"]["r0"]["links"]["s1"]["interface"],
520 result
= create_interfaces_cfg(tgen
, topo1
)
521 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
523 step("Change the ip on the R0 interface")
525 topo_modify_change_ip
= deepcopy(topo
)
526 intf_ip
= topo_modify_change_ip
["routers"]["r0"]["links"]["s1"]["ipv4"]
527 topo_modify_change_ip
["routers"]["r0"]["links"]["s1"]["ipv4"] = str(
528 IPv4Address(frr_unicode(intf_ip
.split("/")[0])) + 3
529 ) + "/{}".format(intf_ip
.split("/")[1])
531 build_config_from_json(tgen
, topo_modify_change_ip
, save_bkup
=False)
532 step("Verify that interface is enabled in ospf.")
539 "ipAddress": topo_modify_change_ip
["routers"]["r0"]["links"][
541 ]["ipv4"].split("/")[0],
542 "ipAddressPrefixlen": int(
543 topo_modify_change_ip
["routers"]["r0"]["links"]["s1"][
552 result
= verify_ospf_interface(tgen
, topo
, dut
=dut
, input_dict
=input_dict
)
553 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
555 step("Modify the mask on the R0 interface")
556 ip_addr
= topo_modify_change_ip
["routers"]["r0"]["links"]["s1"]["ipv4"]
557 mask
= topo_modify_change_ip
["routers"]["r0"]["links"]["s1"]["ipv4"]
558 step("Delete the ip address")
564 "interface": topo
["routers"]["r0"]["links"]["s1"]["interface"],
571 result
= create_interfaces_cfg(tgen
, topo1
)
572 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
574 step("Change the ip on the R0 interface")
576 topo_modify_change_ip
= deepcopy(topo
)
577 intf_ip
= topo_modify_change_ip
["routers"]["r0"]["links"]["s1"]["ipv4"]
578 topo_modify_change_ip
["routers"]["r0"]["links"]["s1"]["ipv4"] = str(
579 IPv4Address(frr_unicode(intf_ip
.split("/")[0])) + 3
580 ) + "/{}".format(int(intf_ip
.split("/")[1]) + 1)
582 build_config_from_json(tgen
, topo_modify_change_ip
, save_bkup
=False)
583 step("Verify that interface is enabled in ospf.")
590 "ipAddress": topo_modify_change_ip
["routers"]["r0"]["links"][
592 ]["ipv4"].split("/")[0],
593 "ipAddressPrefixlen": int(
594 topo_modify_change_ip
["routers"]["r0"]["links"]["s1"][
603 result
= verify_ospf_interface(tgen
, topo
, dut
=dut
, input_dict
=input_dict
)
604 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
606 step("Change the area id on the interface")
611 "interface": topo
["routers"]["r0"]["links"]["s1"]["interface"],
612 "ospf": {"area": "0.0.0.3"},
619 result
= create_interfaces_cfg(tgen
, input_dict
)
620 assert result
is True, "Testcase {} :Failed \n Error: {}".format(tc_name
, result
)
626 "interface": topo
["routers"]["r0"]["links"]["s1"]["interface"],
627 "ospf": {"area": "0.0.0.2"},
633 result
= create_interfaces_cfg(tgen
, input_dict
)
634 assert result
is True, "Testcase {} :Failed \n Error: {}".format(tc_name
, result
)
635 step("Verify that interface is enabled in ospf.")
638 "r0": {"links": {"s1": {"ospf": {"area": "0.0.0.2", "ospfEnabled": True}}}}
640 result
= verify_ospf_interface(tgen
, topo
, dut
=dut
, input_dict
=input_dict
)
641 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
643 write_test_footer(tc_name
)
646 if __name__
== "__main__":
647 args
= ["-s"] + sys
.argv
[1:]
648 sys
.exit(pytest
.main(args
))