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 time
import sleep
18 # Save the Current Working Directory to find configuration files.
19 CWD
= os
.path
.dirname(os
.path
.realpath(__file__
))
20 sys
.path
.append(os
.path
.join(CWD
, "../"))
21 sys
.path
.append(os
.path
.join(CWD
, "../lib/"))
23 # pylint: disable=C0413
24 # Import topogen and topotest helpers
25 from lib
.topogen
import Topogen
, get_topogen
27 # Import topoJson from lib, to create topology and initial configuration
28 from lib
.common_config
import (
32 reset_config_on_routers
,
36 shutdown_bringup_interface
,
38 from lib
.topolog
import logger
40 from lib
.topojson
import build_config_from_json
41 from lib
.ospf
import (
43 config_ospf_interface
,
48 pytestmark
= [pytest
.mark
.ospfd
, pytest
.mark
.staticd
]
66 Please view in a fixed-width font such as Courier.
68 +R1 +------------+R2 |
77 +R0 +-------------+R3 |
81 1. Verify OSPF ECMP with max path configured as 8 (ECMPconfigured at FRR level)
82 2. Verify OSPF ECMP with max path configured as 2 (Edge having 2 uplink ports)
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_ecmp.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
)
117 assert ospf_covergence
is True, "setup_module :Failed \n Error {}".format(
121 logger
.info("Running setup_module() done")
124 def teardown_module(mod
):
126 Teardown the pytest environment.
131 logger
.info("Running teardown_module to delete topology")
135 # Stop toplogy and Remove tmp files
139 "Testsuite end time: {}".format(time
.asctime(time
.localtime(time
.time())))
141 logger
.info("=" * 40)
144 # ##################################
145 # Test cases start here.
146 # ##################################
149 def test_ospf_ecmp_tc16_p0(request
):
153 Verify OSPF ECMP with max path configured as 8 (ECMP
154 configured at FRR level)
156 tc_name
= request
.node
.name
157 write_test_header(tc_name
)
160 # Don't run this test if we have any failure.
161 if tgen
.routers_have_failure():
162 pytest
.skip(tgen
.errors
)
165 step("Bring up the base config as per the topology")
166 step("Configure 8 interfaces between R1 and R2 and enable ospf in area 0.")
167 reset_config_on_routers(tgen
)
168 step("Verify that OSPF is up with 8 neighborship sessions.")
170 ospf_covergence
= verify_ospf_neighbor(tgen
, topo
, dut
=dut
)
171 assert ospf_covergence
is True, "Testcase Failed \n Error {}".format(
175 step("Configure a static route in R0 and redistribute in OSPF.")
181 "network": NETWORK
["ipv4"][0],
188 result
= create_static_routes(tgen
, input_dict
)
189 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
192 redistribute_ospf(tgen
, topo
, dut
, "static")
194 step("Verify that route in R2 in stalled with 8 next hops.")
196 for item
in range(1, 7):
197 nh
.append(topo
["routers"]["r0"]["links"]["r1-link1"]["ipv4"].split("/")[0])
199 nh2
= topo
["routers"]["r0"]["links"]["r1"]["ipv4"].split("/")[0]
204 result
= verify_ospf_rib(tgen
, dut
, input_dict
, next_hop
=nh
)
205 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
208 result
= verify_rib(tgen
, "ipv4", dut
, input_dict
, protocol
=protocol
, next_hop
=nh
)
209 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
211 step("shut no shut all the interfaces on the remote router - R2")
213 for intfr
in range(1, 7):
214 intf
= topo
["routers"]["r1"]["links"]["r0-link{}".format(intfr
)]["interface"]
215 shutdown_bringup_interface(tgen
, dut
, intf
, False)
217 result
= verify_ospf_rib(tgen
, dut
, input_dict
, next_hop
=nh
, expected
=False)
220 ), "Testcase {} : Failed \n r1: OSPF routes are present \n Error: {}".format(
226 tgen
, "ipv4", dut
, input_dict
, protocol
=protocol
, next_hop
=nh
, expected
=False
230 ), "Testcase {} : Failed \n r1: routes are still present \n Error: {}".format(
234 for intfr
in range(1, 7):
235 intf
= topo
["routers"]["r1"]["links"]["r0-link{}".format(intfr
)]["interface"]
236 shutdown_bringup_interface(tgen
, dut
, intf
, True)
238 result
= verify_ospf_rib(tgen
, dut
, input_dict
, next_hop
=nh
)
239 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
242 result
= verify_rib(tgen
, "ipv4", dut
, input_dict
, protocol
=protocol
, next_hop
=nh
)
243 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
245 step("shut no shut on all the interfaces on DUT (r1)")
246 for intfr
in range(1, 7):
247 intf
= topo
["routers"]["r1"]["links"]["r0-link{}".format(intfr
)]["interface"]
248 shutdown_bringup_interface(tgen
, dut
, intf
, False)
250 for intfr
in range(1, 7):
251 intf
= topo
["routers"]["r1"]["links"]["r0-link{}".format(intfr
)]["interface"]
252 shutdown_bringup_interface(tgen
, dut
, intf
, True)
255 "Verify that all the neighbours are up and routes are installed"
256 " with 8 next hop in ospf and ip route tables on R1."
259 step("Verify that OSPF is up with 8 neighborship sessions.")
261 ospf_covergence
= verify_ospf_neighbor(tgen
, topo
, dut
=dut
)
262 assert ospf_covergence
is True, "Testcase Failed \n Error {}".format(
267 result
= verify_ospf_rib(tgen
, dut
, input_dict
, next_hop
=nh
)
268 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
271 result
= verify_rib(tgen
, "ipv4", dut
, input_dict
, protocol
=protocol
, next_hop
=nh
)
272 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
274 step(" Un configure static route on R0")
277 redistribute_ospf(tgen
, topo
, dut
, "static", delete
=True)
279 # Wait for R0 to flush external LSAs.
282 step("Verify that route is withdrawn from R2.")
284 result
= verify_ospf_rib(
285 tgen
, dut
, input_dict
, next_hop
=nh
, retry_timeout
=10, expected
=False
289 ), "Testcase {} : Failed \n r1: OSPF routes are present \n Error: {}".format(
306 ), "Testcase {} : Failed \n r1: routes are still present \n Error: {}".format(
310 step("Re configure the static route in R0.")
312 redistribute_ospf(tgen
, topo
, dut
, "static")
315 result
= verify_ospf_rib(tgen
, dut
, input_dict
, next_hop
=nh
)
316 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
319 result
= verify_rib(tgen
, "ipv4", dut
, input_dict
, protocol
=protocol
, next_hop
=nh
)
320 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
322 write_test_footer(tc_name
)
325 def test_ospf_ecmp_tc17_p0(request
):
329 Verify OSPF ECMP with max path configured as 2 (Edge having 2 uplink ports)
331 tc_name
= request
.node
.name
332 write_test_header(tc_name
)
335 # Don't run this test if we have any failure.
336 if tgen
.routers_have_failure():
337 pytest
.skip(tgen
.errors
)
340 step("Bring up the base config as per the topology")
341 step("Configure 2 interfaces between R1 and R2 & enable ospf in area 0.")
342 reset_config_on_routers(tgen
)
343 step("Verify that OSPF is up with 2 neighborship sessions.")
345 ospf_covergence
= verify_ospf_neighbor(tgen
, topo
, dut
=dut
)
346 assert ospf_covergence
is True, "Testcase Failed \n Error {}".format(
350 step("Configure a static route in R0 and redistribute in OSPF.")
356 "network": NETWORK
["ipv4"][0],
363 result
= create_static_routes(tgen
, input_dict
)
364 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
367 redistribute_ospf(tgen
, topo
, dut
, "static")
369 step("Verify that route in R2 in stalled with 2 next hops.")
371 nh1
= topo
["routers"]["r0"]["links"]["r1-link1"]["ipv4"].split("/")[0]
372 nh2
= topo
["routers"]["r0"]["links"]["r1"]["ipv4"].split("/")[0]
376 result
= verify_ospf_rib(tgen
, dut
, input_dict
, next_hop
=nh
)
377 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
380 result
= verify_rib(tgen
, "ipv4", dut
, input_dict
, protocol
=protocol
, next_hop
=nh
)
381 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
383 step(" Un configure static route on R0")
386 redistribute_ospf(tgen
, topo
, dut
, "static", delete
=True)
387 # sleep till the route gets withdrawn
390 step("Verify that route is withdrawn from R2.")
392 result
= verify_ospf_rib(
393 tgen
, dut
, input_dict
, next_hop
=nh
, retry_timeout
=10, expected
=False
397 ), "Testcase {} : Failed \n r1: OSPF routes are present \n Error: {}".format(
414 ), "Testcase {} : Failed \n r1: routes are still present \n Error: {}".format(
418 step("Reconfigure the static route in R0.Change ECMP value to 2.")
420 redistribute_ospf(tgen
, topo
, dut
, "static")
422 step("Configure cost on R0 as 100")
423 r0_ospf_cost
= {"r0": {"links": {"r1": {"ospf": {"cost": 100}}}}}
424 result
= config_ospf_interface(tgen
, topo
, r0_ospf_cost
)
425 assert result
is True, "Testcase {} :Failed \n Error: {}".format(tc_name
, result
)
428 result
= verify_ospf_rib(tgen
, dut
, input_dict
, next_hop
=nh
)
429 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
432 result
= verify_rib(tgen
, "ipv4", dut
, input_dict
, protocol
=protocol
, next_hop
=nh
)
433 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
435 write_test_footer(tc_name
)
438 if __name__
== "__main__":
439 args
= ["-s"] + sys
.argv
[1:]
440 sys
.exit(pytest
.main(args
))