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."""
31 # Save the Current Working Directory to find configuration files.
32 CWD
= os
.path
.dirname(os
.path
.realpath(__file__
))
33 sys
.path
.append(os
.path
.join(CWD
, "../"))
34 sys
.path
.append(os
.path
.join(CWD
, "../lib/"))
36 # pylint: disable=C0413
37 # Import topogen and topotest helpers
38 from lib
.topogen
import Topogen
, get_topogen
40 # Import topoJson from lib, to create topology and initial configuration
41 from lib
.common_config
import (
44 create_interfaces_cfg
,
46 reset_config_on_routers
,
50 shutdown_bringup_interface
,
53 from lib
.bgp
import verify_bgp_convergence
, create_router_bgp
54 from lib
.topolog
import logger
55 from lib
.topojson
import build_config_from_json
57 from lib
.ospf
import (
62 config_ospf_interface
,
63 verify_ospf_interface
,
66 pytestmark
= [pytest
.mark
.ospfd
, pytest
.mark
.staticd
]
85 Please view in a fixed-width font such as Courier.
87 +R1 +------------+R2 |
96 +R0 +-------------+R3 |
101 1. Test OSPF intra area route calculations.
102 2. Test OSPF inter area route calculations.
103 3. Test OSPF redistribution of connected routes.
107 def setup_module(mod
):
109 Sets up the pytest environment
113 testsuite_run_time
= time
.asctime(time
.localtime(time
.time()))
114 logger
.info("Testsuite start time: {}".format(testsuite_run_time
))
115 logger
.info("=" * 40)
117 logger
.info("Running setup_module to create topology")
119 # This function initiates the topology build with Topogen...
120 json_file
= "{}/ospf_rte_calc.json".format(CWD
)
121 tgen
= Topogen(json_file
, mod
.__name
__)
123 topo
= tgen
.json_topo
124 # ... and here it calls Mininet initialization functions.
126 # get list of daemons needs to be started for this suite.
127 daemons
= topo_daemons(tgen
, topo
)
129 # Starting topology, create tmp files which are loaded to routers
130 # to start deamons and then start routers
131 start_topology(tgen
, daemons
)
133 # Creating configuration from JSON
134 build_config_from_json(tgen
, topo
)
136 # Don't run this test if we have any failure.
137 if tgen
.routers_have_failure():
138 pytest
.skip(tgen
.errors
)
139 # Api call verify whether OSPF is converged
140 ospf_covergence
= verify_ospf_neighbor(tgen
, topo
)
141 assert ospf_covergence
is True, "setup_module :Failed \n Error:" " {}".format(
145 logger
.info("Running setup_module() done")
148 def teardown_module(mod
):
150 Teardown the pytest environment.
155 logger
.info("Running teardown_module to delete topology")
159 # Stop toplogy and Remove tmp files
163 "Testsuite end time: {}".format(time
.asctime(time
.localtime(time
.time())))
165 logger
.info("=" * 40)
168 # ##################################
169 # Test cases start here.
170 # ##################################
173 def test_ospf_redistribution_tc5_p0(request
):
174 """Test OSPF intra area route calculations."""
175 tc_name
= request
.node
.name
176 write_test_header(tc_name
)
179 # Don't run this test if we have any failure.
180 if tgen
.routers_have_failure():
181 pytest
.skip(tgen
.errors
)
184 step("Bring up the base config.")
185 reset_config_on_routers(tgen
)
187 step("Verify that OSPF neighbors are FULL.")
188 ospf_covergence
= verify_ospf_neighbor(tgen
, topo
)
189 assert ospf_covergence
is True, "setup_module :Failed \n Error:" " {}".format(
193 step("verify intra area route is calculated for r0-r3 interface ip in R1")
194 ip
= topo
["routers"]["r0"]["links"]["r3"]["ipv4"]
195 ip_net
= str(ipaddress
.ip_interface(u
"{}".format(ip
)).network
)
196 nh
= topo
["routers"]["r0"]["links"]["r1"]["ipv4"].split("/")[0]
198 "r1": {"static_routes": [{"network": ip_net
, "no_of_ip": 1, "routeType": "N"}]}
202 result
= verify_ospf_rib(tgen
, dut
, input_dict
, next_hop
=nh
)
203 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
206 result
= verify_rib(tgen
, "ipv4", dut
, input_dict
, protocol
=protocol
, next_hop
=nh
)
207 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
209 step("Delete the ip address on newly configured interface of R0")
214 "ipv4": topo
["routers"]["r0"]["links"]["r3"]["ipv4"],
215 "interface": topo
["routers"]["r0"]["links"]["r3"]["interface"],
222 result
= create_interfaces_cfg(tgen
, topo1
)
223 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
226 for num
in range(0, nretry
):
227 result
= verify_ospf_rib(tgen
, dut
, input_dict
, next_hop
=nh
, expected
=False)
228 if result
is not True:
231 assert result
is not True, (
232 "Testcase {} : Failed \n "
233 "r1: OSPF routes are present after deleting ip address of newly "
234 "configured interface of R0 \n Error: {}".format(tc_name
, result
)
248 assert result
is not True, (
249 "Testcase {} : Failed \n "
250 "r1: OSPF routes are present in fib after deleting ip address of newly "
251 "configured interface of R0 \n Error: {}".format(tc_name
, result
)
254 step("Add back the deleted ip address on newly configured interface of R0")
259 "ipv4": topo
["routers"]["r0"]["links"]["r3"]["ipv4"],
260 "interface": topo
["routers"]["r0"]["links"]["r3"]["interface"],
266 result
= create_interfaces_cfg(tgen
, topo1
)
267 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
270 result
= verify_ospf_rib(tgen
, dut
, input_dict
, next_hop
=nh
)
271 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
274 result
= verify_rib(tgen
, "ipv4", dut
, input_dict
, protocol
=protocol
, next_hop
=nh
)
275 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
277 step("Shut no shut interface on R0")
279 intf
= topo
["routers"]["r0"]["links"]["r3"]["interface"]
280 shutdown_bringup_interface(tgen
, dut
, intf
, False)
282 step("un shut the OSPF interface on R0")
284 shutdown_bringup_interface(tgen
, dut
, intf
, True)
287 result
= verify_ospf_rib(tgen
, dut
, input_dict
)
288 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
291 result
= verify_rib(tgen
, "ipv4", dut
, input_dict
, protocol
=protocol
, next_hop
=nh
)
292 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
294 write_test_footer(tc_name
)
297 def test_ospf_redistribution_tc6_p0(request
):
298 """Test OSPF inter area route calculations."""
299 tc_name
= request
.node
.name
300 write_test_header(tc_name
)
303 # Don't run this test if we have any failure.
304 if tgen
.routers_have_failure():
305 pytest
.skip(tgen
.errors
)
308 step("Bring up the base config.")
309 reset_config_on_routers(tgen
)
311 step("Verify that OSPF neighbors are FULL.")
312 ospf_covergence
= verify_ospf_neighbor(tgen
, topo
)
313 assert ospf_covergence
is True, "setup_module :Failed \n Error:" " {}".format(
317 step("verify intra area route is calculated for r0-r3 interface ip in R1")
318 ip
= topo
["routers"]["r0"]["links"]["r3"]["ipv4"]
319 ip_net
= str(ipaddress
.ip_interface(u
"{}".format(ip
)).network
)
320 nh
= topo
["routers"]["r0"]["links"]["r1"]["ipv4"].split("/")[0]
322 "r1": {"static_routes": [{"network": ip_net
, "no_of_ip": 1, "routeType": "N"}]}
326 result
= verify_ospf_rib(tgen
, dut
, input_dict
, next_hop
=nh
)
327 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
330 result
= verify_rib(tgen
, "ipv4", dut
, input_dict
, protocol
=protocol
, next_hop
=nh
)
331 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
333 step("Delete the ip address on newly configured loopback of R0")
338 "ipv4": topo
["routers"]["r0"]["links"]["r3"]["ipv4"],
339 "interface": topo
["routers"]["r0"]["links"]["r3"]["interface"],
346 result
= create_interfaces_cfg(tgen
, topo1
)
347 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
350 for num
in range(0, nretry
):
351 result
= verify_ospf_rib(tgen
, dut
, input_dict
, next_hop
=nh
, expected
=False)
352 if result
is not True:
354 assert result
is not True, (
355 "Testcase {} : Failed \n "
356 "r1: OSPF routes are present after deleting ip address of newly "
357 "configured loopback of R0 \n Error: {}".format(tc_name
, result
)
370 assert result
is not True, (
371 "Testcase {} : Failed \n "
372 "r1: OSPF routes are present in fib after deleting ip address of newly "
373 "configured loopback of R0 \n Error: {}".format(tc_name
, result
)
376 step("Add back the deleted ip address on newly configured interface of R0")
381 "ipv4": topo
["routers"]["r0"]["links"]["r3"]["ipv4"],
382 "interface": topo
["routers"]["r0"]["links"]["r3"]["interface"],
388 result
= create_interfaces_cfg(tgen
, topo1
)
389 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
392 result
= verify_ospf_rib(tgen
, dut
, input_dict
, next_hop
=nh
)
393 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
396 result
= verify_rib(tgen
, "ipv4", dut
, input_dict
, protocol
=protocol
, next_hop
=nh
)
397 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
399 step("Shut no shut interface on R0")
401 intf
= topo
["routers"]["r0"]["links"]["r3"]["interface"]
402 shutdown_bringup_interface(tgen
, dut
, intf
, False)
404 step("un shut the OSPF interface on R0")
406 shutdown_bringup_interface(tgen
, dut
, intf
, True)
409 result
= verify_ospf_rib(tgen
, dut
, input_dict
)
410 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
413 result
= verify_rib(tgen
, "ipv4", dut
, input_dict
, protocol
=protocol
, next_hop
=nh
)
414 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
416 write_test_footer(tc_name
)
419 def test_ospf_redistribution_tc8_p1(request
):
421 Test OSPF redistribution of connected routes.
423 Verify OSPF redistribution of connected routes when bgp multi hop
424 neighbor is configured using ospf routes
427 tc_name
= request
.node
.name
428 write_test_header(tc_name
)
431 step("Bring up the base config.")
433 "Configure loopback interface on all routers, and redistribut"
434 "e connected routes into ospf"
436 reset_config_on_routers(tgen
)
439 "verify that connected routes -loopback is found in all routers"
440 "advertised/exchaged via ospf"
442 for rtr
in topo
["routers"]:
443 redistribute_ospf(tgen
, topo
, rtr
, "static")
444 redistribute_ospf(tgen
, topo
, rtr
, "connected")
445 for node
in topo
["routers"]:
450 "network": topo
["routers"][node
]["links"]["lo"]["ipv4"],
456 for rtr
in topo
["routers"]:
457 result
= verify_rib(tgen
, "ipv4", rtr
, input_dict
)
458 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
462 step("Configure E BGP multi hop using the loopback addresses.")
464 for node
in topo
["routers"]:
466 topo
["routers"][node
].update(
470 "address_family": {"ipv4": {"unicast": {"neighbor": {}}}},
474 for node
in topo
["routers"]:
475 for rtr
in topo
["routers"]:
477 topo
["routers"][node
]["bgp"]["address_family"]["ipv4"]["unicast"][
483 "lo": {"source_link": "lo", "ebgp_multihop": 2}
489 result
= create_router_bgp(tgen
, topo
, topo
["routers"])
490 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
492 step("Verify that BGP neighbor is ESTABLISHED")
493 result
= verify_bgp_convergence(tgen
, topo
)
494 assert result
is True, "Testcase {} :Failed \n Error: {}".format(tc_name
, result
)
496 "Configure couple of static routes in R0 and "
497 "Redistribute static routes in R1 bgp."
500 for rtr
in topo
["routers"]:
501 redistribute_ospf(tgen
, topo
, rtr
, "static", delete
=True)
507 "network": NETWORK
["ipv4"][0],
514 result
= create_static_routes(tgen
, input_dict
)
515 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
517 configure_bgp_on_r0
= {
521 "ipv4": {"unicast": {"redistribute": [{"redist_type": "static"}]}}
526 result
= create_router_bgp(tgen
, topo
, configure_bgp_on_r0
)
527 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
529 for rtr
in ["r1", "r2", "r3"]:
530 result
= verify_rib(tgen
, "ipv4", rtr
, input_dict
, protocol
=protocol
)
531 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
535 step("Clear ospf neighbours in R0")
536 for rtr
in topo
["routers"]:
537 clear_ospf(tgen
, rtr
)
539 step("Verify that OSPF neighbours are reset and forms new adjacencies.")
540 # Api call verify whether OSPF is converged
541 ospf_covergence
= verify_ospf_neighbor(tgen
, topo
)
542 assert ospf_covergence
is True, "setup_module :Failed \n Error:" " {}".format(
546 step("Verify that BGP neighbours are reset and forms new adjacencies.")
547 result
= verify_bgp_convergence(tgen
, topo
)
548 assert result
is True, "Testcase {} :Failed \n Error: {}".format(tc_name
, result
)
551 for rtr
in ["r1", "r2", "r3"]:
552 result
= verify_rib(tgen
, "ipv4", rtr
, input_dict
, protocol
=protocol
)
553 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
557 write_test_footer(tc_name
)
560 def test_ospf_cost_tc52_p0(request
):
561 """OSPF Cost - verifying ospf interface cost functionality"""
562 tc_name
= request
.node
.name
563 write_test_header(tc_name
)
566 step("Bring up the base config.")
567 reset_config_on_routers(tgen
)
570 "Configure ospf cost as 20 on interface between R0 and R1. "
571 "Configure ospf cost as 30 between interface between R0 and R2."
575 "r0": {"links": {"r1": {"ospf": {"cost": 20}}, "r2": {"ospf": {"cost": 30}}}}
577 result
= config_ospf_interface(tgen
, topo
, r0_ospf_cost
)
578 assert result
is True, "Testcase {} :Failed \n Error: {}".format(tc_name
, result
)
581 "Verify that cost is updated in the ospf interface between"
582 " r0 and r1 as 30 and r0 and r2 as 20"
585 result
= verify_ospf_interface(tgen
, topo
, dut
=dut
, input_dict
=r0_ospf_cost
)
586 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
589 "Swap the costs between interfaces on r0, between r0 and r1 to 30"
594 "r0": {"links": {"r1": {"ospf": {"cost": 30}}, "r2": {"ospf": {"cost": 20}}}}
596 result
= config_ospf_interface(tgen
, topo
, r0_ospf_cost
)
597 assert result
is True, "Testcase {} :Failed \n Error: {}".format(tc_name
, result
)
600 "Verify that cost is updated in the ospf interface between r0 "
601 "and r1 as 30 and r0 and r2 as 20."
603 result
= verify_ospf_interface(tgen
, topo
, dut
=dut
, input_dict
=r0_ospf_cost
)
604 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
606 step(" Un configure cost from the interface r0 - r1.")
608 r0_ospf_cost
= {"r0": {"links": {"r1": {"ospf": {"cost": 30, "del_action": True}}}}}
609 result
= config_ospf_interface(tgen
, topo
, r0_ospf_cost
)
610 assert result
is True, "Testcase {} :Failed \n Error: {}".format(tc_name
, result
)
613 "r0": {"links": {"r1": {"ospf": {"cost": 10}}, "r2": {"ospf": {"cost": 20}}}}
616 "Verify that cost is updated in the ospf interface between r0"
617 " and r1 as 10 and r0 and r2 as 20."
620 result
= verify_ospf_interface(tgen
, topo
, dut
=dut
, input_dict
=input_dict
)
621 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
623 step(" Un configure cost from the interface r0 - r2.")
625 r0_ospf_cost
= {"r0": {"links": {"r2": {"ospf": {"cost": 20, "del_action": True}}}}}
626 result
= config_ospf_interface(tgen
, topo
, r0_ospf_cost
)
627 assert result
is True, "Testcase {} :Failed \n Error: {}".format(tc_name
, result
)
630 "Verify that cost is updated in the ospf interface between r0"
631 "and r1 as 10 and r0 and r2 as 10"
635 "r0": {"links": {"r1": {"ospf": {"cost": 10}}, "r2": {"ospf": {"cost": 10}}}}
637 result
= verify_ospf_interface(tgen
, topo
, dut
=dut
, input_dict
=input_dict
)
638 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
640 write_test_footer(tc_name
)
643 if __name__
== "__main__":
644 args
= ["-s"] + sys
.argv
[1:]
645 sys
.exit(pytest
.main(args
))