4 # Copyright (c) 2020 by VMware, Inc. ("VMware")
5 # Used Copyright (c) 2018 by Network Device Education Foundation, Inc.
6 # ("NetDEF") in this file.
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
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
24 """OSPF Basic Functionality Automation."""
29 from copy
import deepcopy
30 from lib
.topotest
import frr_unicode
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/"))
37 # pylint: disable=C0413
38 # Import topogen and topotest helpers
39 from lib
.topogen
import Topogen
, get_topogen
41 # Import topoJson from lib, to create topology and initial configuration
42 from lib
.common_config
import (
45 create_interfaces_cfg
,
47 reset_config_on_routers
,
49 shutdown_bringup_interface
,
53 from lib
.topolog
import logger
54 from lib
.topojson
import build_config_from_json
55 from lib
.ospf
import (
59 verify_ospf_interface
,
61 from ipaddress
import IPv4Address
63 pytestmark
= [pytest
.mark
.ospfd
, pytest
.mark
.staticd
]
82 Please view in a fixed-width font such as Courier.
83 Topo : Broadcast Networks
84 +---+ +---+ +---+ +---+
85 |R0 + +R1 + +R2 + +R3 |
86 +-+-+ +-+-+ +-+-+ +-+-+
89 --+-----------+--------------+---------------+-----
93 1. OSPF Hello protocol - Verify DR BDR Elections
94 2. OSPF IFSM -Verify state change events on DR / BDR / DR Other
99 def setup_module(mod
):
101 Sets up the pytest environment
105 testsuite_run_time
= time
.asctime(time
.localtime(time
.time()))
106 logger
.info("Testsuite start time: {}".format(testsuite_run_time
))
107 logger
.info("=" * 40)
109 logger
.info("Running setup_module to create topology")
111 # This function initiates the topology build with Topogen...
112 json_file
= "{}/ospf_lan.json".format(CWD
)
113 tgen
= Topogen(json_file
, mod
.__name
__)
115 topo
= tgen
.json_topo
116 # ... and here it calls Mininet initialization functions.
118 # Starting topology, create tmp files which are loaded to routers
119 # to start daemons and then start routers
122 # Creating configuration from JSON
123 build_config_from_json(tgen
, topo
)
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(
134 logger
.info("Running setup_module() done")
137 def teardown_module():
138 """Teardown the pytest environment"""
140 logger
.info("Running teardown_module to delete topology")
145 # Stop toplogy and Remove tmp files
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
155 # ##################################
156 # Test cases start here.
157 # ##################################
160 def test_ospf_lan_tc1_p0(request
):
162 OSPF Hello protocol - Verify DR BDR Elections
165 tc_name
= request
.node
.name
166 write_test_header(tc_name
)
169 # Don't run this test if we have any failure.
170 if tgen
.routers_have_failure():
171 pytest
.skip(tgen
.errors
)
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.")
181 "r1": {"state": "Full", "role": "DR"},
182 "r2": {"state": "Full", "role": "DROther"},
183 "r3": {"state": "Full", "role": "DROther"},
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
)
193 "Verify that all the routers are in FULL state with DR and BDR "
201 "r0": {"state": "Full", "role": "Backup"},
202 "r2": {"state": "Full", "role": "DROther"},
203 "r3": {"state": "Full", "role": "DROther"},
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
)
213 "Configure DR pririty 100 on R0 and clear ospf neighbors " "on all the routers."
220 "interface": topo
["routers"]["r0"]["links"]["s1"]["interface"],
221 "ospf": {"priority": 100},
227 result
= create_interfaces_cfg(tgen
, input_dict
)
228 assert result
is True, "Testcase {} :Failed \n Error: {}".format(tc_name
, result
)
230 step("Clear ospf neighbours in all routers")
231 for rtr
in ["r0", "r1", "r2", "r3"]:
232 clear_ospf(tgen
, rtr
)
234 step("Verify that DR election is triggered and R0 is elected as DR")
239 "r1": {"state": "Full", "role": "Backup"},
240 "r2": {"state": "Full", "role": "DROther"},
241 "r3": {"state": "Full", "role": "DROther"},
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
)
251 "Configure DR pririty 150 on R0 and clear ospf neighbors " "on all the routers."
258 "interface": topo
["routers"]["r0"]["links"]["s1"]["interface"],
259 "ospf": {"priority": 150},
265 result
= create_interfaces_cfg(tgen
, input_dict
)
266 assert result
is True, "Testcase {} :Failed \n Error: {}".format(tc_name
, result
)
268 step("Clear ospf neighbours in all routers")
269 for rtr
in ["r0", "r1"]:
270 clear_ospf(tgen
, rtr
)
272 step("Verify that DR election is triggered and R0 is elected as DR")
277 "r1": {"state": "Full", "role": "Backup"},
278 "r2": {"state": "Full", "role": "DROther"},
279 "r3": {"state": "Full", "role": "DROther"},
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
)
288 step("Configure DR priority 0 on R0 & Clear ospf nbrs on all the routers")
294 "interface": topo
["routers"]["r0"]["links"]["s1"]["interface"],
295 "ospf": {"priority": 0},
301 result
= create_interfaces_cfg(tgen
, input_dict
)
302 assert result
is True, "Testcase {} :Failed \n Error: {}".format(tc_name
, result
)
304 step("Clear ospf neighbours in all routers")
306 clear_ospf(tgen
, rtr
)
308 step("Verify that DR election is triggered and R0 is elected as DRother")
313 "r1": {"state": "Full", "role": "DR"},
314 "r2": {"state": "2-Way", "role": "DROther"},
315 "r3": {"state": "2-Way", "role": "DROther"},
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
)
325 "Configure DR priority to default on R0 and Clear ospf neighbors"
326 " on all the routers"
333 "interface": topo
["routers"]["r0"]["links"]["s1"]["interface"],
334 "ospf": {"priority": 100},
340 result
= create_interfaces_cfg(tgen
, input_dict
)
341 assert result
is True, "Testcase {} :Failed \n Error: {}".format(tc_name
, result
)
343 step("Clear ospf neighbours in all routers")
344 for rtr
in ["r0", "r1"]:
345 clear_ospf(tgen
, rtr
)
347 step("Verify that DR election is triggered and R0 is elected as DR")
352 "r1": {"state": "Full", "role": "Backup"},
353 "r2": {"state": "Full", "role": "DROther"},
354 "r3": {"state": "Full", "role": "DROther"},
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
)
363 step("Shut interface on R0")
365 intf
= topo
["routers"]["r0"]["links"]["s1"]["interface"]
366 shutdown_bringup_interface(tgen
, dut
, intf
, False)
368 result
= verify_ospf_neighbor(tgen
, topo
, dut
, lan
=True, expected
=False)
371 ), "Testcase {} : Failed \n " "r0: OSPF neighbors-hip is up \n Error: {}".format(
375 step("No Shut interface on R0")
377 intf
= topo
["routers"]["r0"]["links"]["s1"]["interface"]
378 shutdown_bringup_interface(tgen
, dut
, intf
, True)
384 "r1": {"state": "Full", "role": "DR"},
385 "r2": {"state": "Full", "role": "DROther"},
386 "r3": {"state": "Full", "role": "DROther"},
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
)
395 step("Clear ospf on DR router in the topology.")
396 clear_ospf(tgen
, "r0")
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
)
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])
410 build_config_from_json(tgen
, topo_modify_change_ip
, save_bkup
=False)
412 clear_ospf(tgen
, "r0")
414 "Verify that OSPF is in FULL state with other routers with "
415 "newly configured IP."
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
)
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"}}}
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")
428 stop_router(tgen
, "r0")
429 start_router(tgen
, "r0")
432 "Verify that OSPF is in FULL state with other routers with"
433 " newly configured router id."
439 "r0": {"state": "Full", "role": "Backup"},
440 "r2": {"state": "Full", "role": "DROther"},
441 "r3": {"state": "Full", "role": "DROther"},
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
)
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")
460 step("Verify that OSPF is enabled with router id previously configured.")
465 "r0": {"state": "Full", "role": "Backup"},
466 "r2": {"state": "Full", "role": "DROther"},
467 "r3": {"state": "Full", "role": "DROther"},
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
)
476 write_test_footer(tc_name
)
479 def test_ospf_lan_tc2_p0(request
):
481 OSPF IFSM -Verify state change events on DR / BDR / DR Other
484 tc_name
= request
.node
.name
485 write_test_header(tc_name
)
488 # Don't run this test if we have any failure.
489 if tgen
.routers_have_failure():
490 pytest
.skip(tgen
.errors
)
493 step("Bring up the base config as per the topology")
494 reset_config_on_routers(tgen
)
496 "Verify that OSPF is subscribed to multi cast services "
497 "(All SPF, all DR Routers)."
499 step("Verify that interface is enabled in ospf.")
509 "mcastMemberOspfDesignatedRouters": True,
510 "mcastMemberOspfAllRouters": True,
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
)
520 step("Delete the ip address")
525 "ipv4": topo
["routers"]["r0"]["links"]["s1"]["ipv4"],
526 "interface": topo
["routers"]["r0"]["links"]["s1"]["interface"],
533 result
= create_interfaces_cfg(tgen
, topo1
)
534 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
536 step("Change the ip on the R0 interface")
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])
544 build_config_from_json(tgen
, topo_modify_change_ip
, save_bkup
=False)
545 step("Verify that interface is enabled in ospf.")
552 "ipAddress": topo_modify_change_ip
["routers"]["r0"]["links"][
554 ]["ipv4"].split("/")[0],
555 "ipAddressPrefixlen": int(
556 topo_modify_change_ip
["routers"]["r0"]["links"]["s1"][
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
)
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")
577 "interface": topo
["routers"]["r0"]["links"]["s1"]["interface"],
584 result
= create_interfaces_cfg(tgen
, topo1
)
585 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
587 step("Change the ip on the R0 interface")
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)
595 build_config_from_json(tgen
, topo_modify_change_ip
, save_bkup
=False)
596 step("Verify that interface is enabled in ospf.")
603 "ipAddress": topo_modify_change_ip
["routers"]["r0"]["links"][
605 ]["ipv4"].split("/")[0],
606 "ipAddressPrefixlen": int(
607 topo_modify_change_ip
["routers"]["r0"]["links"]["s1"][
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
)
619 step("Change the area id on the interface")
624 "interface": topo
["routers"]["r0"]["links"]["s1"]["interface"],
625 "ospf": {"area": "0.0.0.3"},
632 result
= create_interfaces_cfg(tgen
, input_dict
)
633 assert result
is True, "Testcase {} :Failed \n Error: {}".format(tc_name
, result
)
639 "interface": topo
["routers"]["r0"]["links"]["s1"]["interface"],
640 "ospf": {"area": "0.0.0.2"},
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.")
651 "r0": {"links": {"s1": {"ospf": {"area": "0.0.0.2", "ospfEnabled": True}}}}
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
)
656 write_test_footer(tc_name
)
659 if __name__
== "__main__":
660 args
= ["-s"] + sys
.argv
[1:]
661 sys
.exit(pytest
.main(args
))