4 # Copyright (c) 2021 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."""
30 # Save the Current Working Directory to find configuration files.
31 CWD
= os
.path
.dirname(os
.path
.realpath(__file__
))
32 sys
.path
.append(os
.path
.join(CWD
, "../"))
33 sys
.path
.append(os
.path
.join(CWD
, "../lib/"))
35 # pylint: disable=C0413
36 # Import topogen and topotest helpers
37 from lib
.topogen
import Topogen
, get_topogen
39 # Import topoJson from lib, to create topology and initial configuration
40 from lib
.common_config
import (
44 reset_config_on_routers
,
48 shutdown_bringup_interface
,
50 get_frr_ipv6_linklocal
,
52 from lib
.topolog
import logger
53 from lib
.topojson
import build_config_from_json
55 from lib
.ospf
import (
56 verify_ospf6_neighbor
,
59 config_ospf6_interface
,
63 pytestmark
= [pytest
.mark
.ospfd
, pytest
.mark
.staticd
]
77 "ipv6": ["2::1/128", "2::2/128", "2::3/128", "2::4/128", "2::5/128"],
81 Please view in a fixed-width font such as Courier.
83 +R1 +------------+R2 |
92 +R0 +-------------+R3 |
96 1. Verify OSPF ECMP with max path configured as 8 (ECMPconfigured at FRR level)
97 2. Verify OSPF ECMP with max path configured as 2 (Edge having 2 uplink ports)
101 def setup_module(mod
):
103 Sets up the pytest environment
107 testsuite_run_time
= time
.asctime(time
.localtime(time
.time()))
108 logger
.info("Testsuite start time: {}".format(testsuite_run_time
))
109 logger
.info("=" * 40)
111 logger
.info("Running setup_module to create topology")
113 # This function initiates the topology build with Topogen...
114 json_file
= "{}/ospfv3_ecmp.json".format(CWD
)
115 tgen
= Topogen(json_file
, mod
.__name
__)
117 topo
= tgen
.json_topo
118 # ... and here it calls Mininet initialization functions.
120 # get list of daemons needs to be started for this suite.
121 daemons
= topo_daemons(tgen
, topo
)
123 # Starting topology, create tmp files which are loaded to routers
124 # to start daemons and then start routers
125 start_topology(tgen
, daemons
)
127 # Creating configuration from JSON
128 build_config_from_json(tgen
, topo
)
130 # Don't run this test if we have any failure.
131 if tgen
.routers_have_failure():
132 pytest
.skip(tgen
.errors
)
134 ospf_covergence
= verify_ospf6_neighbor(tgen
, topo
)
135 assert ospf_covergence
is True, "setup_module :Failed \n Error:" " {}".format(
139 logger
.info("Running setup_module() done")
142 def teardown_module(mod
):
144 Teardown the pytest environment.
149 logger
.info("Running teardown_module to delete topology")
153 # Stop toplogy and Remove tmp files
157 "Testsuite end time: {}".format(time
.asctime(time
.localtime(time
.time())))
159 logger
.info("=" * 40)
162 def red_static(dut
, config
=True):
163 """Local def for Redstribute static routes inside ospf."""
167 ospf_red
= {dut
: {"ospf6": {"redistribute": [{"redist_type": "static"}]}}}
171 "ospf6": {"redistribute": [{"redist_type": "static", "delete": True}]}
174 result
= create_router_ospf(tgen
, topo
, ospf_red
)
175 assert result
is True, "Testcase : Failed \n Error: {}".format(result
)
178 def red_connected(dut
, config
=True):
179 """Local def for Redstribute connected routes inside ospf."""
183 ospf_red
= {dut
: {"ospf6": {"redistribute": [{"redist_type": "connected"}]}}}
188 "redistribute": [{"redist_type": "connected", "del_action": True}]
192 result
= create_router_ospf(tgen
, topo
, ospf_red
)
193 assert result
is True, "Testcase: Failed \n Error: {}".format(result
)
196 def get_llip(onrouter
, intf
):
198 API to get the link local ipv6 address of a particular interface
202 * `fromnode`: Source node
203 * `tonode` : interface for which link local ip needs to be returned.
207 result = get_llip('r1', 'r2-link0')
211 1) link local ipv6 address from the interface.
212 2) errormsg - when link local ip not found.
215 intf
= topo
["routers"][onrouter
]["links"][intf
]["interface"]
216 llip
= get_frr_ipv6_linklocal(tgen
, onrouter
, intf
)
218 logger
.info("llip ipv6 address to be set as NH is %s", llip
)
223 def get_glipv6(onrouter
, intf
):
225 API to get the global ipv6 address of a particular interface
229 * `onrouter`: Source node
230 * `intf` : interface for which link local ip needs to be returned.
234 result = get_glipv6('r1', 'r2-link0')
238 1) global ipv6 address from the interface.
239 2) errormsg - when link local ip not found.
241 glipv6
= (topo
["routers"][onrouter
]["links"][intf
]["ipv6"]).split("/")[0]
243 logger
.info("Global ipv6 address to be set as NH is %s", glipv6
)
248 # ##################################
249 # Test cases start here.
250 # ##################################
253 def test_ospfv3_ecmp_tc16_p0(request
):
257 Verify OSPF ECMP with max path configured as 8 (ECMP
258 configured at FRR level)
260 tc_name
= request
.node
.name
261 write_test_header(tc_name
)
264 # Don't run this test if we have any failure.
265 if tgen
.routers_have_failure():
266 pytest
.skip(tgen
.errors
)
269 step("Bring up the base config as per the topology")
270 step("Configure 8 interfaces between R1 and R2 and enable ospf in area 0.")
272 reset_config_on_routers(tgen
)
274 step("Verify that OSPF is up with 8 neighborship sessions.")
276 ospf_covergence
= verify_ospf6_neighbor(tgen
, topo
, dut
=dut
)
277 assert ospf_covergence
is True, "setup_module :Failed \n Error:" " {}".format(
281 step("Configure a static route in R0 and redistribute in OSPF.")
287 "network": NETWORK
["ipv6"][0],
294 result
= create_static_routes(tgen
, input_dict
)
295 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
300 llip
= get_llip("r0", "r1-link1")
301 assert llip
is not None, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
303 step("Verify that route in R2 in stalled with 8 next hops.")
305 for item
in range(1, 7):
308 llip
= get_llip("r0", "r1")
309 assert llip
is not None, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
316 result
= verify_ospf6_rib(tgen
, dut
, input_dict
, next_hop
=nh
)
317 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
320 result
= verify_rib(tgen
, "ipv6", dut
, input_dict
, protocol
=protocol
, next_hop
=nh
)
321 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
323 step("shut no shut all the interfaces on the remote router - R2")
325 for intfr
in range(1, 7):
326 intf
= topo
["routers"]["r1"]["links"]["r0-link{}".format(intfr
)]["interface"]
327 shutdown_bringup_interface(tgen
, dut
, intf
, False)
329 result
= verify_ospf6_rib(tgen
, dut
, input_dict
, next_hop
=nh
, expected
=False)
332 ), "Testcase {} : Failed \n Route present in OSPF RIB. Error: {}".format(
338 tgen
, "ipv6", dut
, input_dict
, protocol
=protocol
, next_hop
=nh
, expected
=False
342 ), "Testcase {} : Failed \n Route present in RIB. Error: {}".format(tc_name
, result
)
344 for intfr
in range(1, 7):
345 intf
= topo
["routers"]["r1"]["links"]["r0-link{}".format(intfr
)]["interface"]
346 shutdown_bringup_interface(tgen
, dut
, intf
, True)
348 result
= verify_ospf6_rib(tgen
, dut
, input_dict
, next_hop
=nh
)
349 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
352 result
= verify_rib(tgen
, "ipv6", dut
, input_dict
, protocol
=protocol
, next_hop
=nh
)
353 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
355 step("shut no shut on all the interfaces on DUT (r1)")
356 for intfr
in range(1, 7):
357 intf
= topo
["routers"]["r1"]["links"]["r0-link{}".format(intfr
)]["interface"]
358 shutdown_bringup_interface(tgen
, dut
, intf
, False)
360 for intfr
in range(1, 7):
361 intf
= topo
["routers"]["r1"]["links"]["r0-link{}".format(intfr
)]["interface"]
362 shutdown_bringup_interface(tgen
, dut
, intf
, True)
365 "Verify that all the neighbours are up and routes are installed"
366 " with 8 next hop in ospf and ip route tables on R1."
369 step("Verify that OSPF is up with 8 neighborship sessions.")
371 ospf_covergence
= verify_ospf6_neighbor(tgen
, topo
, dut
=dut
)
372 assert ospf_covergence
is True, "setup_module :Failed \n Error:" " {}".format(
377 result
= verify_ospf6_rib(tgen
, dut
, input_dict
, next_hop
=nh
)
378 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
381 result
= verify_rib(tgen
, "ipv6", dut
, input_dict
, protocol
=protocol
, next_hop
=nh
)
382 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
384 write_test_footer(tc_name
)
387 def test_ospfv3_ecmp_tc17_p0(request
):
391 Verify OSPF ECMP with max path configured as 2 (Edge having 2 uplink ports)
393 tc_name
= request
.node
.name
394 write_test_header(tc_name
)
397 # Don't run this test if we have any failure.
398 if tgen
.routers_have_failure():
399 pytest
.skip(tgen
.errors
)
402 step("Bring up the base config as per the topology")
403 step("Configure 2 interfaces between R1 and R2 & enable ospf in area 0.")
405 reset_config_on_routers(tgen
)
407 step("Verify that OSPF is up with 2 neighborship sessions.")
409 ospf_covergence
= verify_ospf6_neighbor(tgen
, topo
, dut
=dut
)
410 assert ospf_covergence
is True, "setup_module :Failed \n Error:" " {}".format(
414 step("Configure a static route in R0 and redistribute in OSPF.")
420 "network": NETWORK
["ipv6"][0],
427 result
= create_static_routes(tgen
, input_dict
)
428 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
433 step("Verify that route in R2 in stalled with 2 next hops.")
435 llip
= get_llip("r0", "r1-link1")
436 assert llip
is not None, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
440 llip
= get_llip("r0", "r1")
441 assert llip
is not None, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
448 result
= verify_ospf6_rib(tgen
, dut
, input_dict
, next_hop
=nh
)
449 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
452 result
= verify_rib(tgen
, "ipv6", dut
, input_dict
, protocol
=protocol
, next_hop
=nh
)
453 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
455 step("Configure ECMP value as 1.")
456 max_path
= {"r1": {"ospf6": {"maximum-paths": 1}}}
457 result
= create_router_ospf(tgen
, topo
, max_path
)
458 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
460 result
= verify_rib(tgen
, "ipv6", dut
, input_dict
, protocol
=protocol
, next_hop
=nh2
)
461 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
464 max_path
= {"r1": {"ospf6": {"maximum-paths": 2}}}
465 result
= create_router_ospf(tgen
, topo
, max_path
)
466 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
468 result
= verify_rib(tgen
, "ipv6", dut
, input_dict
, protocol
=protocol
, next_hop
=nh
)
469 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
471 step("Configure cost on R0 as 100")
472 r0_ospf_cost
= {"r0": {"links": {"r1": {"ospf6": {"cost": 100}}}}}
473 result
= config_ospf6_interface(tgen
, topo
, r0_ospf_cost
)
474 assert result
is True, "Testcase {} :Failed \n Error: {}".format(tc_name
, result
)
477 result
= verify_ospf6_rib(tgen
, dut
, input_dict
, next_hop
=nh
)
478 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
481 result
= verify_rib(tgen
, "ipv6", dut
, input_dict
, protocol
=protocol
, next_hop
=nh
)
482 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
484 write_test_footer(tc_name
)
487 if __name__
== "__main__":
488 args
= ["-s"] + sys
.argv
[1:]
489 sys
.exit(pytest
.main(args
))