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
,
52 from lib
.bgp
import verify_bgp_convergence
, create_router_bgp
53 from lib
.topolog
import logger
54 from lib
.topojson
import build_config_from_json
56 from lib
.ospf
import (
61 config_ospf_interface
,
62 verify_ospf_interface
,
65 pytestmark
= [pytest
.mark
.ospfd
, pytest
.mark
.staticd
]
84 NETWORK_APP_E
= {"ipv4": ["12.0.0.0/24", "12.0.0.0/16", "12.0.0.0/8"]}
86 Please view in a fixed-width font such as Courier.
88 +R1 +------------+R2 |
97 +R0 +-------------+R3 |
102 1. Test OSPF intra area route calculations.
103 2. Test OSPF inter area route calculations.
104 3. Test OSPF redistribution of connected routes.
108 def setup_module(mod
):
110 Sets up the pytest environment
114 testsuite_run_time
= time
.asctime(time
.localtime(time
.time()))
115 logger
.info("Testsuite start time: {}".format(testsuite_run_time
))
116 logger
.info("=" * 40)
118 logger
.info("Running setup_module to create topology")
120 # This function initiates the topology build with Topogen...
121 json_file
= "{}/ospf_rte_calc.json".format(CWD
)
122 tgen
= Topogen(json_file
, mod
.__name
__)
124 topo
= tgen
.json_topo
125 # ... and here it calls Mininet initialization functions.
127 # Starting topology, create tmp files which are loaded to routers
128 # to start daemons and then start routers
131 # Creating configuration from JSON
132 build_config_from_json(tgen
, topo
)
134 # Don't run this test if we have any failure.
135 if tgen
.routers_have_failure():
136 pytest
.skip(tgen
.errors
)
137 # Api call verify whether OSPF is converged
138 ospf_covergence
= verify_ospf_neighbor(tgen
, topo
)
139 assert ospf_covergence
is True, "setup_module :Failed \n Error:" " {}".format(
143 logger
.info("Running setup_module() done")
146 def teardown_module(mod
):
148 Teardown the pytest environment.
153 logger
.info("Running teardown_module to delete topology")
157 # Stop toplogy and Remove tmp files
161 "Testsuite end time: {}".format(time
.asctime(time
.localtime(time
.time())))
163 logger
.info("=" * 40)
166 # ##################################
167 # Test cases start here.
168 # ##################################
171 def test_ospf_redistribution_tc5_p0(request
):
172 """Test OSPF intra area route calculations."""
173 tc_name
= request
.node
.name
174 write_test_header(tc_name
)
177 # Don't run this test if we have any failure.
178 if tgen
.routers_have_failure():
179 pytest
.skip(tgen
.errors
)
182 step("Bring up the base config.")
183 reset_config_on_routers(tgen
)
185 step("Verify that OSPF neighbors are FULL.")
186 ospf_covergence
= verify_ospf_neighbor(tgen
, topo
)
187 assert ospf_covergence
is True, "setup_module :Failed \n Error:" " {}".format(
191 step("verify intra area route is calculated for r0-r3 interface ip in R1")
192 ip
= topo
["routers"]["r0"]["links"]["r3"]["ipv4"]
193 ip_net
= str(ipaddress
.ip_interface(u
"{}".format(ip
)).network
)
194 nh
= topo
["routers"]["r0"]["links"]["r1"]["ipv4"].split("/")[0]
196 "r1": {"static_routes": [{"network": ip_net
, "no_of_ip": 1, "routeType": "N"}]}
200 result
= verify_ospf_rib(tgen
, dut
, input_dict
, next_hop
=nh
)
201 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
204 result
= verify_rib(tgen
, "ipv4", dut
, input_dict
, protocol
=protocol
, next_hop
=nh
)
205 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
207 step("Delete the ip address on newly configured interface of R0")
212 "ipv4": topo
["routers"]["r0"]["links"]["r3"]["ipv4"],
213 "interface": topo
["routers"]["r0"]["links"]["r3"]["interface"],
220 result
= create_interfaces_cfg(tgen
, topo1
)
221 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
224 for num
in range(0, nretry
):
225 result
= verify_ospf_rib(tgen
, dut
, input_dict
, next_hop
=nh
, expected
=False)
226 if result
is not True:
229 assert result
is not True, (
230 "Testcase {} : Failed \n "
231 "r1: OSPF routes are present after deleting ip address of newly "
232 "configured interface of R0 \n Error: {}".format(tc_name
, result
)
246 assert result
is not True, (
247 "Testcase {} : Failed \n "
248 "r1: OSPF routes are present in fib after deleting ip address of newly "
249 "configured interface of R0 \n Error: {}".format(tc_name
, result
)
252 step("Add back the deleted ip address on newly configured interface of R0")
257 "ipv4": topo
["routers"]["r0"]["links"]["r3"]["ipv4"],
258 "interface": topo
["routers"]["r0"]["links"]["r3"]["interface"],
264 result
= create_interfaces_cfg(tgen
, topo1
)
265 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
268 result
= verify_ospf_rib(tgen
, dut
, input_dict
, next_hop
=nh
)
269 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
272 result
= verify_rib(tgen
, "ipv4", dut
, input_dict
, protocol
=protocol
, next_hop
=nh
)
273 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
275 step("Shut no shut interface on R0")
277 intf
= topo
["routers"]["r0"]["links"]["r3"]["interface"]
278 shutdown_bringup_interface(tgen
, dut
, intf
, False)
280 step("un shut the OSPF interface on R0")
282 shutdown_bringup_interface(tgen
, dut
, intf
, True)
285 result
= verify_ospf_rib(tgen
, dut
, input_dict
)
286 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
289 result
= verify_rib(tgen
, "ipv4", dut
, input_dict
, protocol
=protocol
, next_hop
=nh
)
290 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
292 write_test_footer(tc_name
)
295 def test_ospf_redistribution_tc6_p0(request
):
296 """Test OSPF inter area route calculations."""
297 tc_name
= request
.node
.name
298 write_test_header(tc_name
)
301 # Don't run this test if we have any failure.
302 if tgen
.routers_have_failure():
303 pytest
.skip(tgen
.errors
)
306 step("Bring up the base config.")
307 reset_config_on_routers(tgen
)
309 step("Verify that OSPF neighbors are FULL.")
310 ospf_covergence
= verify_ospf_neighbor(tgen
, topo
)
311 assert ospf_covergence
is True, "setup_module :Failed \n Error:" " {}".format(
315 step("verify intra area route is calculated for r0-r3 interface ip in R1")
316 ip
= topo
["routers"]["r0"]["links"]["r3"]["ipv4"]
317 ip_net
= str(ipaddress
.ip_interface(u
"{}".format(ip
)).network
)
318 nh
= topo
["routers"]["r0"]["links"]["r1"]["ipv4"].split("/")[0]
320 "r1": {"static_routes": [{"network": ip_net
, "no_of_ip": 1, "routeType": "N"}]}
324 result
= verify_ospf_rib(tgen
, dut
, input_dict
, next_hop
=nh
)
325 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
328 result
= verify_rib(tgen
, "ipv4", dut
, input_dict
, protocol
=protocol
, next_hop
=nh
)
329 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
331 step("Delete the ip address on newly configured loopback of R0")
336 "ipv4": topo
["routers"]["r0"]["links"]["r3"]["ipv4"],
337 "interface": topo
["routers"]["r0"]["links"]["r3"]["interface"],
344 result
= create_interfaces_cfg(tgen
, topo1
)
345 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
348 for num
in range(0, nretry
):
349 result
= verify_ospf_rib(tgen
, dut
, input_dict
, next_hop
=nh
, expected
=False)
350 if result
is not True:
352 assert result
is not True, (
353 "Testcase {} : Failed \n "
354 "r1: OSPF routes are present after deleting ip address of newly "
355 "configured loopback of R0 \n Error: {}".format(tc_name
, result
)
368 assert result
is not True, (
369 "Testcase {} : Failed \n "
370 "r1: OSPF routes are present in fib after deleting ip address of newly "
371 "configured loopback of R0 \n Error: {}".format(tc_name
, result
)
374 step("Add back the deleted ip address on newly configured interface of R0")
379 "ipv4": topo
["routers"]["r0"]["links"]["r3"]["ipv4"],
380 "interface": topo
["routers"]["r0"]["links"]["r3"]["interface"],
386 result
= create_interfaces_cfg(tgen
, topo1
)
387 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
390 result
= verify_ospf_rib(tgen
, dut
, input_dict
, next_hop
=nh
)
391 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
394 result
= verify_rib(tgen
, "ipv4", dut
, input_dict
, protocol
=protocol
, next_hop
=nh
)
395 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
397 step("Shut no shut interface on R0")
399 intf
= topo
["routers"]["r0"]["links"]["r3"]["interface"]
400 shutdown_bringup_interface(tgen
, dut
, intf
, False)
402 step("un shut the OSPF interface on R0")
404 shutdown_bringup_interface(tgen
, dut
, intf
, True)
407 result
= verify_ospf_rib(tgen
, dut
, input_dict
)
408 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
411 result
= verify_rib(tgen
, "ipv4", dut
, input_dict
, protocol
=protocol
, next_hop
=nh
)
412 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
414 write_test_footer(tc_name
)
417 def test_ospf_redistribution_tc8_p1(request
):
419 Test OSPF redistribution of connected routes.
421 Verify OSPF redistribution of connected routes when bgp multi hop
422 neighbor is configured using ospf routes
425 tc_name
= request
.node
.name
426 write_test_header(tc_name
)
429 step("Bring up the base config.")
431 "Configure loopback interface on all routers, and redistribut"
432 "e connected routes into ospf"
434 reset_config_on_routers(tgen
)
437 "verify that connected routes -loopback is found in all routers"
438 "advertised/exchaged via ospf"
440 for rtr
in topo
["routers"]:
441 redistribute_ospf(tgen
, topo
, rtr
, "static")
442 redistribute_ospf(tgen
, topo
, rtr
, "connected")
443 for node
in topo
["routers"]:
448 "network": topo
["routers"][node
]["links"]["lo"]["ipv4"],
454 for rtr
in topo
["routers"]:
455 result
= verify_rib(tgen
, "ipv4", rtr
, input_dict
)
456 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
460 step("Configure E BGP multi hop using the loopback addresses.")
462 for node
in topo
["routers"]:
464 topo
["routers"][node
].update(
468 "address_family": {"ipv4": {"unicast": {"neighbor": {}}}},
472 for node
in topo
["routers"]:
473 for rtr
in topo
["routers"]:
475 topo
["routers"][node
]["bgp"]["address_family"]["ipv4"]["unicast"][
481 "lo": {"source_link": "lo", "ebgp_multihop": 2}
487 result
= create_router_bgp(tgen
, topo
, topo
["routers"])
488 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
490 step("Verify that BGP neighbor is ESTABLISHED")
491 result
= verify_bgp_convergence(tgen
, topo
)
492 assert result
is True, "Testcase {} :Failed \n Error: {}".format(tc_name
, result
)
494 "Configure couple of static routes in R0 and "
495 "Redistribute static routes in R1 bgp."
498 for rtr
in topo
["routers"]:
499 redistribute_ospf(tgen
, topo
, rtr
, "static", delete
=True)
505 "network": NETWORK
["ipv4"][0],
512 result
= create_static_routes(tgen
, input_dict
)
513 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
515 configure_bgp_on_r0
= {
519 "ipv4": {"unicast": {"redistribute": [{"redist_type": "static"}]}}
524 result
= create_router_bgp(tgen
, topo
, configure_bgp_on_r0
)
525 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
527 for rtr
in ["r1", "r2", "r3"]:
528 result
= verify_rib(tgen
, "ipv4", rtr
, input_dict
, protocol
=protocol
)
529 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
533 step("Clear ospf neighbours in R0")
534 for rtr
in topo
["routers"]:
535 clear_ospf(tgen
, rtr
)
537 step("Verify that OSPF neighbours are reset and forms new adjacencies.")
538 # Api call verify whether OSPF is converged
539 ospf_covergence
= verify_ospf_neighbor(tgen
, topo
)
540 assert ospf_covergence
is True, "setup_module :Failed \n Error:" " {}".format(
544 step("Verify that BGP neighbours are reset and forms new adjacencies.")
545 result
= verify_bgp_convergence(tgen
, topo
)
546 assert result
is True, "Testcase {} :Failed \n Error: {}".format(tc_name
, result
)
549 for rtr
in ["r1", "r2", "r3"]:
550 result
= verify_rib(tgen
, "ipv4", rtr
, input_dict
, protocol
=protocol
)
551 assert result
is True, "Testcase {} : Failed \n Error: {}".format(
555 write_test_footer(tc_name
)
558 def test_ospf_rfc2328_appendinxE_p0(request
):
560 Test OSPF appendinx E RFC2328.
563 tc_name
= request
.node
.name
564 write_test_header(tc_name
)
567 step("Bring up the base config.")
569 reset_config_on_routers(tgen
)
571 step("Verify that OSPF neighbours are Full.")
572 # Api call verify whether OSPF is converged
573 ospf_covergence
= verify_ospf_neighbor(tgen
, topo
)
574 assert ospf_covergence
is True, "setup_module :Failed \n Error:" " {}".format(
578 redistribute_ospf(tgen
, topo
, "r0", "static")
580 step("Configure static route with prefix 24, 16, 8 to check ")
585 "network": NETWORK_APP_E
["ipv4"][0],
592 result
= create_static_routes(tgen
, input_dict
)
593 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
599 "network": NETWORK_APP_E
["ipv4"][1],
606 result
= create_static_routes(tgen
, input_dict
)
607 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
613 "network": NETWORK_APP_E
["ipv4"][2],
620 result
= create_static_routes(tgen
, input_dict
)
621 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
623 step("Verify that ospf originates routes with mask 24, 16, 8")
624 ip_net
= NETWORK_APP_E
["ipv4"][0]
625 input_dict
= {"r1": {"static_routes": [{"network": ip_net
, "no_of_ip": 1}]}}
628 result
= verify_ospf_rib(tgen
, dut
, input_dict
)
629 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
632 result
= verify_rib(tgen
, "ipv4", dut
, input_dict
, protocol
=protocol
)
633 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
635 ip_net
= NETWORK_APP_E
["ipv4"][1]
636 input_dict
= {"r1": {"static_routes": [{"network": ip_net
, "no_of_ip": 1}]}}
639 result
= verify_ospf_rib(tgen
, dut
, input_dict
)
640 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
643 result
= verify_rib(tgen
, "ipv4", dut
, input_dict
, protocol
=protocol
)
644 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
646 ip_net
= NETWORK_APP_E
["ipv4"][2]
647 input_dict
= {"r1": {"static_routes": [{"network": ip_net
, "no_of_ip": 1}]}}
650 result
= verify_ospf_rib(tgen
, dut
, input_dict
)
651 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
654 result
= verify_rib(tgen
, "ipv4", dut
, input_dict
, protocol
=protocol
)
655 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
657 step("Delete static route with prefix 24, 16, 8 to check ")
662 "network": NETWORK_APP_E
["ipv4"][0],
670 result
= create_static_routes(tgen
, input_dict
)
671 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
677 "network": NETWORK_APP_E
["ipv4"][1],
685 result
= create_static_routes(tgen
, input_dict
)
686 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
692 "network": NETWORK_APP_E
["ipv4"][2],
700 result
= create_static_routes(tgen
, input_dict
)
701 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
703 write_test_footer(tc_name
)
706 def test_ospf_cost_tc52_p0(request
):
707 """OSPF Cost - verifying ospf interface cost functionality"""
708 tc_name
= request
.node
.name
709 write_test_header(tc_name
)
712 step("Bring up the base config.")
713 reset_config_on_routers(tgen
)
716 "Configure ospf cost as 20 on interface between R0 and R1. "
717 "Configure ospf cost as 30 between interface between R0 and R2."
721 "r0": {"links": {"r1": {"ospf": {"cost": 20}}, "r2": {"ospf": {"cost": 30}}}}
723 result
= config_ospf_interface(tgen
, topo
, r0_ospf_cost
)
724 assert result
is True, "Testcase {} :Failed \n Error: {}".format(tc_name
, result
)
727 "Verify that cost is updated in the ospf interface between"
728 " r0 and r1 as 30 and r0 and r2 as 20"
731 result
= verify_ospf_interface(tgen
, topo
, dut
=dut
, input_dict
=r0_ospf_cost
)
732 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
735 "Swap the costs between interfaces on r0, between r0 and r1 to 30"
740 "r0": {"links": {"r1": {"ospf": {"cost": 30}}, "r2": {"ospf": {"cost": 20}}}}
742 result
= config_ospf_interface(tgen
, topo
, r0_ospf_cost
)
743 assert result
is True, "Testcase {} :Failed \n Error: {}".format(tc_name
, result
)
746 "Verify that cost is updated in the ospf interface between r0 "
747 "and r1 as 30 and r0 and r2 as 20."
749 result
= verify_ospf_interface(tgen
, topo
, dut
=dut
, input_dict
=r0_ospf_cost
)
750 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
752 step(" Un configure cost from the interface r0 - r1.")
754 r0_ospf_cost
= {"r0": {"links": {"r1": {"ospf": {"cost": 30, "del_action": True}}}}}
755 result
= config_ospf_interface(tgen
, topo
, r0_ospf_cost
)
756 assert result
is True, "Testcase {} :Failed \n Error: {}".format(tc_name
, result
)
759 "r0": {"links": {"r1": {"ospf": {"cost": 10}}, "r2": {"ospf": {"cost": 20}}}}
762 "Verify that cost is updated in the ospf interface between r0"
763 " and r1 as 10 and r0 and r2 as 20."
766 result
= verify_ospf_interface(tgen
, topo
, dut
=dut
, input_dict
=input_dict
)
767 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
769 step(" Un configure cost from the interface r0 - r2.")
771 r0_ospf_cost
= {"r0": {"links": {"r2": {"ospf": {"cost": 20, "del_action": True}}}}}
772 result
= config_ospf_interface(tgen
, topo
, r0_ospf_cost
)
773 assert result
is True, "Testcase {} :Failed \n Error: {}".format(tc_name
, result
)
776 "Verify that cost is updated in the ospf interface between r0"
777 "and r1 as 10 and r0 and r2 as 10"
781 "r0": {"links": {"r1": {"ospf": {"cost": 10}}, "r2": {"ospf": {"cost": 10}}}}
783 result
= verify_ospf_interface(tgen
, topo
, dut
=dut
, input_dict
=input_dict
)
784 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
786 write_test_footer(tc_name
)
789 if __name__
== "__main__":
790 args
= ["-s"] + sys
.argv
[1:]
791 sys
.exit(pytest
.main(args
))