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 from copy
import deepcopy
31 from ipaddress
import IPv4Address
32 from lib
.topotest
import frr_unicode
34 # Save the Current Working Directory to find configuration files.
35 CWD
= os
.path
.dirname(os
.path
.realpath(__file__
))
36 sys
.path
.append(os
.path
.join(CWD
, "../"))
37 sys
.path
.append(os
.path
.join(CWD
, "../lib/"))
39 # pylint: disable=C0413
40 # Import topogen and topotest helpers
41 from mininet
.topo
import Topo
42 from lib
.topogen
import Topogen
, get_topogen
45 # Import topoJson from lib, to create topology and initial configuration
46 from lib
.common_config
import (
50 reset_config_on_routers
,
55 shutdown_bringup_interface
,
56 create_interfaces_cfg
,
58 get_frr_ipv6_linklocal
,
60 from lib
.topolog
import logger
61 from lib
.topojson
import build_topo_from_json
, build_config_from_json
63 from lib
.ospf
import (
64 verify_ospf6_neighbor
,
65 config_ospf_interface
,
69 verify_ospf6_interface
,
70 verify_ospf6_database
,
71 config_ospf6_interface
,
74 from ipaddress
import IPv6Address
79 # Reading the data from JSON File for topology creation
80 jsonFile
= "{}/ospfv3_ecmp.json".format(CWD
)
82 with
open(jsonFile
, "r") as topoJson
:
83 topo
= json
.load(topoJson
)
85 assert False, "Could not read file {}".format(jsonFile
)
95 "ipv6": ["2::1/128", "2::2/128", "2::3/128", "2::4/128", "2::5/128"],
99 Please view in a fixed-width font such as Courier.
101 +R1 +------------+R2 |
110 +R0 +-------------+R3 |
114 1. Verify OSPF ECMP with max path configured as 8 (ECMPconfigured at FRR level)
115 2. Verify OSPF ECMP with max path configured as 2 (Edge having 2 uplink ports)
119 class CreateTopo(Topo
):
121 Test topology builder.
123 * `Topo`: Topology object
126 def build(self
, *_args
, **_opts
):
127 """Build function."""
128 tgen
= get_topogen(self
)
130 # Building topology from json file
131 build_topo_from_json(tgen
, topo
)
134 def setup_module(mod
):
136 Sets up the pytest environment
141 testsuite_run_time
= time
.asctime(time
.localtime(time
.time()))
142 logger
.info("Testsuite start time: {}".format(testsuite_run_time
))
143 logger
.info("=" * 40)
145 logger
.info("Running setup_module to create topology")
147 # This function initiates the topology build with Topogen...
148 tgen
= Topogen(CreateTopo
, mod
.__name
__)
149 # ... and here it calls Mininet initialization functions.
151 # get list of daemons needs to be started for this suite.
152 daemons
= topo_daemons(tgen
, topo
)
154 # Starting topology, create tmp files which are loaded to routers
155 # to start deamons and then start routers
156 start_topology(tgen
, daemons
)
158 # Creating configuration from JSON
159 build_config_from_json(tgen
, topo
)
161 # Don't run this test if we have any failure.
162 if tgen
.routers_have_failure():
163 pytest
.skip(tgen
.errors
)
165 ospf_covergence
= verify_ospf6_neighbor(tgen
, topo
)
166 assert ospf_covergence
is True, "setup_module :Failed \n Error:" " {}".format(
170 logger
.info("Running setup_module() done")
173 def teardown_module(mod
):
175 Teardown the pytest environment.
180 logger
.info("Running teardown_module to delete topology")
184 # Stop toplogy and Remove tmp files
188 "Testsuite end time: {}".format(time
.asctime(time
.localtime(time
.time())))
190 logger
.info("=" * 40)
193 def red_static(dut
, config
=True):
194 """Local def for Redstribute static routes inside ospf."""
198 ospf_red
= {dut
: {"ospf6": {"redistribute": [{"redist_type": "static"}]}}}
202 "ospf6": {"redistribute": [{"redist_type": "static", "delete": True}]}
205 result
= create_router_ospf(tgen
, topo
, ospf_red
)
206 assert result
is True, "Testcase : Failed \n Error: {}".format(result
)
209 def red_connected(dut
, config
=True):
210 """Local def for Redstribute connected routes inside ospf."""
214 ospf_red
= {dut
: {"ospf6": {"redistribute": [{"redist_type": "connected"}]}}}
219 "redistribute": [{"redist_type": "connected", "del_action": True}]
223 result
= create_router_ospf(tgen
, topo
, ospf_red
)
224 assert result
is True, "Testcase: Failed \n Error: {}".format(result
)
227 def get_llip(onrouter
, intf
):
229 API to get the link local ipv6 address of a perticular interface
233 * `fromnode`: Source node
234 * `tonode` : interface for which link local ip needs to be returned.
238 result = get_llip('r1', 'r2-link0')
242 1) link local ipv6 address from the interface.
243 2) errormsg - when link local ip not found.
246 intf
= topo
["routers"][onrouter
]["links"][intf
]["interface"]
247 llip
= get_frr_ipv6_linklocal(tgen
, onrouter
, intf
)
249 logger
.info("llip ipv6 address to be set as NH is %s", llip
)
254 def get_glipv6(onrouter
, intf
):
256 API to get the global ipv6 address of a perticular interface
260 * `onrouter`: Source node
261 * `intf` : interface for which link local ip needs to be returned.
265 result = get_glipv6('r1', 'r2-link0')
269 1) global ipv6 address from the interface.
270 2) errormsg - when link local ip not found.
272 glipv6
= (topo
["routers"][onrouter
]["links"][intf
]["ipv6"]).split("/")[0]
274 logger
.info("Global ipv6 address to be set as NH is %s", glipv6
)
279 # ##################################
280 # Test cases start here.
281 # ##################################
284 def test_ospfv3_ecmp_tc16_p0(request
):
288 Verify OSPF ECMP with max path configured as 8 (ECMP
289 configured at FRR level)
291 tc_name
= request
.node
.name
292 write_test_header(tc_name
)
295 # Don't run this test if we have any failure.
296 if tgen
.routers_have_failure():
297 pytest
.skip(tgen
.errors
)
300 step("Bring up the base config as per the topology")
301 step("Configure 8 interfaces between R1 and R2 and enable ospf in area 0.")
303 reset_config_on_routers(tgen
)
305 step("Verify that OSPF is up with 8 neighborship sessions.")
307 ospf_covergence
= verify_ospf6_neighbor(tgen
, topo
, dut
=dut
)
308 assert ospf_covergence
is True, "setup_module :Failed \n Error:" " {}".format(
312 step("Configure a static route in R0 and redistribute in OSPF.")
318 "network": NETWORK
["ipv6"][0],
325 result
= create_static_routes(tgen
, input_dict
)
326 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
331 llip
= get_llip("r0", "r1-link1")
332 assert llip
is not None, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
334 step("Verify that route in R2 in stalled with 8 next hops.")
336 for item
in range(1, 7):
339 llip
= get_llip("r0", "r1")
340 assert llip
is not None, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
347 result
= verify_ospf6_rib(tgen
, dut
, input_dict
, next_hop
=nh
)
348 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
351 result
= verify_rib(tgen
, "ipv6", dut
, input_dict
, protocol
=protocol
, next_hop
=nh
)
352 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
354 step("shut no shut all the interfaces on the remote router - R2")
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 result
= verify_ospf6_rib(tgen
, dut
, input_dict
, next_hop
=nh
, expected
=False)
363 ), "Testcase {} : Failed \n Route present in OSPF RIB. Error: {}".format(
369 tgen
, "ipv6", dut
, input_dict
, protocol
=protocol
, next_hop
=nh
, expected
=False
373 ), "Testcase {} : Failed \n Route present in RIB. Error: {}".format(tc_name
, result
)
375 for intfr
in range(1, 7):
376 intf
= topo
["routers"]["r1"]["links"]["r0-link{}".format(intfr
)]["interface"]
377 shutdown_bringup_interface(tgen
, dut
, intf
, True)
379 result
= verify_ospf6_rib(tgen
, dut
, input_dict
, next_hop
=nh
)
380 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
383 result
= verify_rib(tgen
, "ipv6", dut
, input_dict
, protocol
=protocol
, next_hop
=nh
)
384 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
386 step("shut no shut on all the interfaces on DUT (r1)")
387 for intfr
in range(1, 7):
388 intf
= topo
["routers"]["r1"]["links"]["r0-link{}".format(intfr
)]["interface"]
389 shutdown_bringup_interface(tgen
, dut
, intf
, False)
391 for intfr
in range(1, 7):
392 intf
= topo
["routers"]["r1"]["links"]["r0-link{}".format(intfr
)]["interface"]
393 shutdown_bringup_interface(tgen
, dut
, intf
, True)
396 "Verify that all the neighbours are up and routes are installed"
397 " with 8 next hop in ospf and ip route tables on R1."
400 step("Verify that OSPF is up with 8 neighborship sessions.")
402 ospf_covergence
= verify_ospf6_neighbor(tgen
, topo
, dut
=dut
)
403 assert ospf_covergence
is True, "setup_module :Failed \n Error:" " {}".format(
408 result
= verify_ospf6_rib(tgen
, dut
, input_dict
, next_hop
=nh
)
409 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
412 result
= verify_rib(tgen
, "ipv6", dut
, input_dict
, protocol
=protocol
, next_hop
=nh
)
413 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
415 write_test_footer(tc_name
)
418 def test_ospfv3_ecmp_tc17_p0(request
):
422 Verify OSPF ECMP with max path configured as 2 (Edge having 2 uplink ports)
424 tc_name
= request
.node
.name
425 write_test_header(tc_name
)
428 # Don't run this test if we have any failure.
429 if tgen
.routers_have_failure():
430 pytest
.skip(tgen
.errors
)
433 step("Bring up the base config as per the topology")
434 step("Configure 2 interfaces between R1 and R2 & enable ospf in area 0.")
436 reset_config_on_routers(tgen
)
438 step("Verify that OSPF is up with 2 neighborship sessions.")
440 ospf_covergence
= verify_ospf6_neighbor(tgen
, topo
, dut
=dut
)
441 assert ospf_covergence
is True, "setup_module :Failed \n Error:" " {}".format(
445 step("Configure a static route in R0 and redistribute in OSPF.")
451 "network": NETWORK
["ipv6"][0],
458 result
= create_static_routes(tgen
, input_dict
)
459 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
464 step("Verify that route in R2 in stalled with 2 next hops.")
466 llip
= get_llip("r0", "r1-link1")
467 assert llip
is not None, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
471 llip
= get_llip("r0", "r1")
472 assert llip
is not None, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
479 result
= verify_ospf6_rib(tgen
, dut
, input_dict
, next_hop
=nh
)
480 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
483 result
= verify_rib(tgen
, "ipv6", dut
, input_dict
, protocol
=protocol
, next_hop
=nh
)
484 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
486 step("Configure ECMP value as 1.")
487 max_path
= {"r1": {"ospf6": {"maximum-paths": 1}}}
488 result
= create_router_ospf(tgen
, topo
, max_path
)
489 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
491 result
= verify_rib(tgen
, "ipv6", dut
, input_dict
, protocol
=protocol
, next_hop
=nh2
)
492 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
495 max_path
= {"r1": {"ospf6": {"maximum-paths": 2}}}
496 result
= create_router_ospf(tgen
, topo
, max_path
)
497 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
499 result
= verify_rib(tgen
, "ipv6", dut
, input_dict
, protocol
=protocol
, next_hop
=nh
)
500 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
502 step("Configure cost on R0 as 100")
503 r0_ospf_cost
= {"r0": {"links": {"r1": {"ospf6": {"cost": 100}}}}}
504 result
= config_ospf6_interface(tgen
, topo
, r0_ospf_cost
)
505 assert result
is True, "Testcase {} :Failed \n Error: {}".format(tc_name
, result
)
508 result
= verify_ospf6_rib(tgen
, dut
, input_dict
, next_hop
=nh
)
509 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
512 result
= verify_rib(tgen
, "ipv6", dut
, input_dict
, protocol
=protocol
, next_hop
=nh
)
513 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
515 write_test_footer(tc_name
)
518 if __name__
== "__main__":
519 args
= ["-s"] + sys
.argv
[1:]
520 sys
.exit(pytest
.main(args
))