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
,
49 get_frr_ipv6_linklocal
,
51 from lib
.topolog
import logger
52 from lib
.topojson
import build_config_from_json
54 from lib
.ospf
import (
55 verify_ospf6_neighbor
,
58 config_ospf6_interface
,
62 pytestmark
= [pytest
.mark
.ospfd
, pytest
.mark
.staticd
]
76 "ipv6": ["2::1/128", "2::2/128", "2::3/128", "2::4/128", "2::5/128"],
80 Please view in a fixed-width font such as Courier.
82 +R1 +------------+R2 |
91 +R0 +-------------+R3 |
95 1. Verify OSPF ECMP with max path configured as 8 (ECMPconfigured at FRR level)
96 2. Verify OSPF ECMP with max path configured as 2 (Edge having 2 uplink ports)
100 def setup_module(mod
):
102 Sets up the pytest environment
106 testsuite_run_time
= time
.asctime(time
.localtime(time
.time()))
107 logger
.info("Testsuite start time: {}".format(testsuite_run_time
))
108 logger
.info("=" * 40)
110 logger
.info("Running setup_module to create topology")
112 # This function initiates the topology build with Topogen...
113 json_file
= "{}/ospfv3_ecmp.json".format(CWD
)
114 tgen
= Topogen(json_file
, mod
.__name
__)
116 topo
= tgen
.json_topo
117 # ... and here it calls Mininet initialization functions.
119 # Starting topology, create tmp files which are loaded to routers
120 # to start daemons and then start routers
123 # Creating configuration from JSON
124 build_config_from_json(tgen
, topo
)
126 # Don't run this test if we have any failure.
127 if tgen
.routers_have_failure():
128 pytest
.skip(tgen
.errors
)
130 ospf_covergence
= verify_ospf6_neighbor(tgen
, topo
)
131 assert ospf_covergence
is True, "setup_module :Failed \n Error:" " {}".format(
135 logger
.info("Running setup_module() done")
138 def teardown_module(mod
):
140 Teardown the pytest environment.
145 logger
.info("Running teardown_module to delete topology")
149 # Stop toplogy and Remove tmp files
153 "Testsuite end time: {}".format(time
.asctime(time
.localtime(time
.time())))
155 logger
.info("=" * 40)
158 def red_static(dut
, config
=True):
159 """Local def for Redstribute static routes inside ospf."""
163 ospf_red
= {dut
: {"ospf6": {"redistribute": [{"redist_type": "static"}]}}}
167 "ospf6": {"redistribute": [{"redist_type": "static", "delete": True}]}
170 result
= create_router_ospf(tgen
, topo
, ospf_red
)
171 assert result
is True, "Testcase : Failed \n Error: {}".format(result
)
174 def red_connected(dut
, config
=True):
175 """Local def for Redstribute connected routes inside ospf."""
179 ospf_red
= {dut
: {"ospf6": {"redistribute": [{"redist_type": "connected"}]}}}
184 "redistribute": [{"redist_type": "connected", "del_action": True}]
188 result
= create_router_ospf(tgen
, topo
, ospf_red
)
189 assert result
is True, "Testcase: Failed \n Error: {}".format(result
)
192 def get_llip(onrouter
, intf
):
194 API to get the link local ipv6 address of a particular interface
198 * `fromnode`: Source node
199 * `tonode` : interface for which link local ip needs to be returned.
203 result = get_llip('r1', 'r2-link0')
207 1) link local ipv6 address from the interface.
208 2) errormsg - when link local ip not found.
211 intf
= topo
["routers"][onrouter
]["links"][intf
]["interface"]
212 llip
= get_frr_ipv6_linklocal(tgen
, onrouter
, intf
)
214 logger
.info("llip ipv6 address to be set as NH is %s", llip
)
219 def get_glipv6(onrouter
, intf
):
221 API to get the global ipv6 address of a particular interface
225 * `onrouter`: Source node
226 * `intf` : interface for which link local ip needs to be returned.
230 result = get_glipv6('r1', 'r2-link0')
234 1) global ipv6 address from the interface.
235 2) errormsg - when link local ip not found.
237 glipv6
= (topo
["routers"][onrouter
]["links"][intf
]["ipv6"]).split("/")[0]
239 logger
.info("Global ipv6 address to be set as NH is %s", glipv6
)
244 # ##################################
245 # Test cases start here.
246 # ##################################
249 def test_ospfv3_ecmp_tc16_p0(request
):
253 Verify OSPF ECMP with max path configured as 8 (ECMP
254 configured at FRR level)
256 tc_name
= request
.node
.name
257 write_test_header(tc_name
)
260 # Don't run this test if we have any failure.
261 if tgen
.routers_have_failure():
262 pytest
.skip(tgen
.errors
)
265 step("Bring up the base config as per the topology")
266 step("Configure 8 interfaces between R1 and R2 and enable ospf in area 0.")
268 reset_config_on_routers(tgen
)
270 step("Verify that OSPF is up with 8 neighborship sessions.")
272 ospf_covergence
= verify_ospf6_neighbor(tgen
, topo
, dut
=dut
)
273 assert ospf_covergence
is True, "setup_module :Failed \n Error:" " {}".format(
277 step("Configure a static route in R0 and redistribute in OSPF.")
283 "network": NETWORK
["ipv6"][0],
290 result
= create_static_routes(tgen
, input_dict
)
291 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
296 llip
= get_llip("r0", "r1-link1")
297 assert llip
is not None, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
299 step("Verify that route in R2 in stalled with 8 next hops.")
301 for item
in range(1, 7):
304 llip
= get_llip("r0", "r1")
305 assert llip
is not None, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
312 result
= verify_ospf6_rib(tgen
, dut
, input_dict
, next_hop
=nh
)
313 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
316 result
= verify_rib(tgen
, "ipv6", dut
, input_dict
, protocol
=protocol
, next_hop
=nh
)
317 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
319 step("shut no shut all the interfaces on the remote router - R2")
321 for intfr
in range(1, 7):
322 intf
= topo
["routers"]["r1"]["links"]["r0-link{}".format(intfr
)]["interface"]
323 shutdown_bringup_interface(tgen
, dut
, intf
, False)
325 result
= verify_ospf6_rib(tgen
, dut
, input_dict
, next_hop
=nh
, expected
=False)
328 ), "Testcase {} : Failed \n Route present in OSPF RIB. Error: {}".format(
334 tgen
, "ipv6", dut
, input_dict
, protocol
=protocol
, next_hop
=nh
, expected
=False
338 ), "Testcase {} : Failed \n Route present in RIB. Error: {}".format(tc_name
, result
)
340 for intfr
in range(1, 7):
341 intf
= topo
["routers"]["r1"]["links"]["r0-link{}".format(intfr
)]["interface"]
342 shutdown_bringup_interface(tgen
, dut
, intf
, True)
344 result
= verify_ospf6_rib(tgen
, dut
, input_dict
, next_hop
=nh
)
345 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
348 result
= verify_rib(tgen
, "ipv6", dut
, input_dict
, protocol
=protocol
, next_hop
=nh
)
349 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
351 step("shut no shut on all the interfaces on DUT (r1)")
352 for intfr
in range(1, 7):
353 intf
= topo
["routers"]["r1"]["links"]["r0-link{}".format(intfr
)]["interface"]
354 shutdown_bringup_interface(tgen
, dut
, intf
, False)
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
, True)
361 "Verify that all the neighbours are up and routes are installed"
362 " with 8 next hop in ospf and ip route tables on R1."
365 step("Verify that OSPF is up with 8 neighborship sessions.")
367 ospf_covergence
= verify_ospf6_neighbor(tgen
, topo
, dut
=dut
)
368 assert ospf_covergence
is True, "setup_module :Failed \n Error:" " {}".format(
373 result
= verify_ospf6_rib(tgen
, dut
, input_dict
, next_hop
=nh
)
374 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
377 result
= verify_rib(tgen
, "ipv6", dut
, input_dict
, protocol
=protocol
, next_hop
=nh
)
378 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
380 write_test_footer(tc_name
)
383 def test_ospfv3_ecmp_tc17_p0(request
):
387 Verify OSPF ECMP with max path configured as 2 (Edge having 2 uplink ports)
389 tc_name
= request
.node
.name
390 write_test_header(tc_name
)
393 # Don't run this test if we have any failure.
394 if tgen
.routers_have_failure():
395 pytest
.skip(tgen
.errors
)
398 step("Bring up the base config as per the topology")
399 step("Configure 2 interfaces between R1 and R2 & enable ospf in area 0.")
401 reset_config_on_routers(tgen
)
403 step("Verify that OSPF is up with 2 neighborship sessions.")
405 ospf_covergence
= verify_ospf6_neighbor(tgen
, topo
, dut
=dut
)
406 assert ospf_covergence
is True, "setup_module :Failed \n Error:" " {}".format(
410 step("Configure a static route in R0 and redistribute in OSPF.")
416 "network": NETWORK
["ipv6"][0],
423 result
= create_static_routes(tgen
, input_dict
)
424 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
429 step("Verify that route in R2 in stalled with 2 next hops.")
431 llip
= get_llip("r0", "r1-link1")
432 assert llip
is not None, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
436 llip
= get_llip("r0", "r1")
437 assert llip
is not None, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
444 result
= verify_ospf6_rib(tgen
, dut
, input_dict
, next_hop
=nh
)
445 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
448 result
= verify_rib(tgen
, "ipv6", dut
, input_dict
, protocol
=protocol
, next_hop
=nh
)
449 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
451 step("Configure ECMP value as 1.")
452 max_path
= {"r1": {"ospf6": {"maximum-paths": 1}}}
453 result
= create_router_ospf(tgen
, topo
, max_path
)
454 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
456 result
= verify_rib(tgen
, "ipv6", dut
, input_dict
, protocol
=protocol
, next_hop
=nh2
)
457 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
460 max_path
= {"r1": {"ospf6": {"maximum-paths": 2}}}
461 result
= create_router_ospf(tgen
, topo
, max_path
)
462 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
464 result
= verify_rib(tgen
, "ipv6", dut
, input_dict
, protocol
=protocol
, next_hop
=nh
)
465 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
467 step("Configure cost on R0 as 100")
468 r0_ospf_cost
= {"r0": {"links": {"r1": {"ospf6": {"cost": 100}}}}}
469 result
= config_ospf6_interface(tgen
, topo
, r0_ospf_cost
)
470 assert result
is True, "Testcase {} :Failed \n Error: {}".format(tc_name
, result
)
473 result
= verify_ospf6_rib(tgen
, dut
, input_dict
, next_hop
=nh
)
474 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
477 result
= verify_rib(tgen
, "ipv6", dut
, input_dict
, protocol
=protocol
, next_hop
=nh
)
478 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
480 write_test_footer(tc_name
)
483 if __name__
== "__main__":
484 args
= ["-s"] + sys
.argv
[1:]
485 sys
.exit(pytest
.main(args
))