]>
git.proxmox.com Git - mirror_frr.git/blob - tests/topotests/ospf_basic_functionality/test_ospf_p2mp.py
2 # SPDX-License-Identifier: ISC
5 # Copyright (c) 2020 by VMware, Inc. ("VMware")
6 # Used Copyright (c) 2018 by Network Device Education Foundation, Inc.
7 # ("NetDEF") in this file.
11 """OSPF Basic Functionality Automation."""
16 from copy
import deepcopy
17 from ipaddress
import IPv4Address
19 # Save the Current Working Directory to find configuration files.
20 CWD
= os
.path
.dirname(os
.path
.realpath(__file__
))
21 sys
.path
.append(os
.path
.join(CWD
, "../"))
22 sys
.path
.append(os
.path
.join(CWD
, "../lib/"))
24 # pylint: disable=C0413
25 # Import topogen and topotest helpers
26 from lib
.topogen
import Topogen
, get_topogen
28 # Import topoJson from lib, to create topology and initial configuration
29 from lib
.common_config
import (
33 reset_config_on_routers
,
35 create_interfaces_cfg
,
39 from lib
.topolog
import logger
40 from lib
.topojson
import build_config_from_json
41 from lib
.topotest
import frr_unicode
, json_cmp
43 from lib
.ospf
import (
44 verify_ospf_interface
,
48 pytestmark
= [pytest
.mark
.ospfd
, pytest
.mark
.staticd
]
56 Please view in a fixed-width font such as Courier.
58 +R1 +------------+R2 |
67 +R0 +-------------+R3 |
71 1. OSPF P2MP -Verify state change events on p2mp network.
75 def setup_module(mod
):
77 Sets up the pytest environment
81 testsuite_run_time
= time
.asctime(time
.localtime(time
.time()))
82 logger
.info("Testsuite start time: {}".format(testsuite_run_time
))
85 logger
.info("Running setup_module to create topology")
87 # This function initiates the topology build with Topogen...
88 json_file
= "{}/ospf_p2mp.json".format(CWD
)
89 tgen
= Topogen(json_file
, mod
.__name
__)
92 # ... and here it calls Mininet initialization functions.
94 # Starting topology, create tmp files which are loaded to routers
95 # to start daemons and then start routers
98 # Creating configuration from JSON
99 build_config_from_json(tgen
, topo
)
101 # Don't run this test if we have any failure.
102 if tgen
.routers_have_failure():
103 pytest
.skip(tgen
.errors
)
105 logger
.info("Running setup_module() done")
108 def teardown_module(mod
):
110 Teardown the pytest environment.
115 logger
.info("Running teardown_module to delete topology")
119 # Stop toplogy and Remove tmp files
123 "Testsuite end time: {}".format(time
.asctime(time
.localtime(time
.time())))
125 logger
.info("=" * 40)
128 # ##################################
129 # Test cases start here.
130 # ##################################
133 def test_ospf_p2mp_tc1_p0(request
):
134 """OSPF IFSM -Verify state change events on p2mp network."""
135 tc_name
= request
.node
.name
136 write_test_header(tc_name
)
139 # Don't run this test if we have any failure.
140 if tgen
.routers_have_failure():
141 pytest
.skip(tgen
.errors
)
144 step("Bring up the base config as per the topology")
145 reset_config_on_routers(tgen
)
147 "Verify that OSPF is subscribed to multi cast services "
148 "(All SPF, all DR Routers)."
150 step("Verify that interface is enabled in ospf.")
151 step("Verify that config is successful.")
156 "r3": {"ospf": {"mcastMemberOspfAllRouters": True, "ospfEnabled": True}}
160 result
= verify_ospf_interface(tgen
, topo
, dut
=dut
, input_dict
=input_dict
)
161 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
163 step("Delete the ip address")
168 "ipv4": topo
["routers"]["r0"]["links"]["r3"]["ipv4"],
169 "interface": topo
["routers"]["r0"]["links"]["r3"]["interface"],
176 result
= create_interfaces_cfg(tgen
, topo1
)
177 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
179 step("Change the ip on the R0 interface")
181 topo_modify_change_ip
= deepcopy(topo
)
182 intf_ip
= topo_modify_change_ip
["routers"]["r0"]["links"]["r3"]["ipv4"]
183 topo_modify_change_ip
["routers"]["r0"]["links"]["r3"]["ipv4"] = str(
184 IPv4Address(frr_unicode(intf_ip
.split("/")[0])) + 3
185 ) + "/{}".format(intf_ip
.split("/")[1])
187 build_config_from_json(tgen
, topo_modify_change_ip
, save_bkup
=False)
188 step("Verify that interface is enabled in ospf.")
195 "ipAddress": topo_modify_change_ip
["routers"]["r0"]["links"][
197 ]["ipv4"].split("/")[0],
198 "ipAddressPrefixlen": int(
199 topo_modify_change_ip
["routers"]["r0"]["links"]["r3"][
208 result
= verify_ospf_interface(tgen
, topo
, dut
=dut
, input_dict
=input_dict
)
209 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
211 step("Modify the mask on the R0 interface")
212 ip_addr
= topo_modify_change_ip
["routers"]["r0"]["links"]["r3"]["ipv4"]
213 mask
= topo_modify_change_ip
["routers"]["r0"]["links"]["r3"]["ipv4"]
214 step("Delete the ip address")
220 "interface": topo
["routers"]["r0"]["links"]["r3"]["interface"],
227 result
= create_interfaces_cfg(tgen
, topo1
)
228 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
230 step("Change the ip on the R0 interface")
232 topo_modify_change_ip
= deepcopy(topo
)
233 intf_ip
= topo_modify_change_ip
["routers"]["r0"]["links"]["r3"]["ipv4"]
234 topo_modify_change_ip
["routers"]["r0"]["links"]["r3"]["ipv4"] = str(
235 IPv4Address(frr_unicode(intf_ip
.split("/")[0])) + 3
236 ) + "/{}".format(int(intf_ip
.split("/")[1]) + 1)
238 build_config_from_json(tgen
, topo_modify_change_ip
, save_bkup
=False)
239 step("Verify that interface is enabled in ospf.")
246 "ipAddress": topo_modify_change_ip
["routers"]["r0"]["links"][
248 ]["ipv4"].split("/")[0],
249 "ipAddressPrefixlen": int(
250 topo_modify_change_ip
["routers"]["r0"]["links"]["r3"][
259 result
= verify_ospf_interface(tgen
, topo
, dut
=dut
, input_dict
=input_dict
)
260 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
266 "ipv4": topo_modify_change_ip
["routers"]["r0"]["links"]["r3"][
269 "interface": topo_modify_change_ip
["routers"]["r0"]["links"]["r3"][
278 result
= create_interfaces_cfg(tgen
, topo1
)
279 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
281 build_config_from_json(tgen
, topo
, save_bkup
=False)
283 step("Change the area id on the interface")
288 "interface": topo
["routers"]["r0"]["links"]["r3"]["interface"],
289 "ospf": {"area": "0.0.0.0"},
296 result
= create_interfaces_cfg(tgen
, input_dict
)
297 assert result
is True, "Testcase {} :Failed \n Error: {}".format(tc_name
, result
)
303 "interface": topo
["routers"]["r0"]["links"]["r3"]["interface"],
304 "ospf": {"area": "0.0.0.1"},
310 result
= create_interfaces_cfg(tgen
, input_dict
)
311 assert result
is True, "Testcase {} :Failed \n Error: {}".format(tc_name
, result
)
312 step("Verify that interface is enabled in ospf.")
315 "r0": {"links": {"r3": {"ospf": {"area": "0.0.0.1", "ospfEnabled": True}}}}
317 result
= verify_ospf_interface(tgen
, topo
, dut
=dut
, input_dict
=input_dict
)
318 assert result
is True, "Testcase {} : Failed \n Error: {}".format(tc_name
, result
)
324 "interface": topo
["routers"]["r0"]["links"]["r3"]["interface"],
325 "ospf": {"area": "0.0.0.1"},
332 result
= create_interfaces_cfg(tgen
, input_dict
)
333 assert result
is True, "Testcase {} :Failed \n Error: {}".format(tc_name
, result
)
339 "interface": topo
["routers"]["r0"]["links"]["r3"]["interface"],
340 "ospf": {"area": "0.0.0.0"},
346 result
= create_interfaces_cfg(tgen
, input_dict
)
347 assert result
is True, "Testcase {} :Failed \n Error: {}".format(tc_name
, result
)
349 step("Verify if interface is enabled with network type P2MP")
354 "interface": topo
["routers"]["r0"]["links"]["r3"]["interface"],
355 "ospf": {"area": "0.0.0.0", "networkType": "POINTOMULTIPOINT"},
360 result
= create_interfaces_cfg(tgen
, input_dict
)
361 assert result
is True, "Testcase {} :Failed \n Error: {}".format(tc_name
, result
)
363 write_test_footer(tc_name
)
366 @retry(retry_timeout
=30)
367 def verify_ospf_json(tgen
, dut
, input_dict
, cmd
="show ip ospf database json"):
369 show_ospf_json
= run_frr_cmd(dut
, cmd
, isjson
=True)
370 if not bool(show_ospf_json
):
371 return "ospf is not running"
372 result
= json_cmp(show_ospf_json
, input_dict
)
373 return str(result
) if result
else None
376 @pytest.mark
.parametrize("tgen", [2], indirect
=True)
377 def test_ospf_nbrs(tgen
):
381 "routerLinkStates": [
384 "advertisedRouter": "100.1.1.0",
385 "numOfRouterLinks": 6,
389 "advertisedRouter": "100.1.1.1",
390 "numOfRouterLinks": 6,
394 "advertisedRouter": "100.1.1.2",
395 "numOfRouterLinks": 6,
399 "advertisedRouter": "100.1.1.3",
400 "numOfRouterLinks": 7,
409 "show ip ospf n json",
414 "state": "Full/DROther",
419 "state": "Full/DROther",
424 "state": "Full/DROther",
432 "show ip ospf n json",
437 "state": "Full/DROther",
442 "state": "Full/DROther",
447 "state": "Full/DROther",
455 "show ip ospf n json",
460 "state": "Full/DROther",
465 "state": "Full/DROther",
470 "state": "Full/DROther",
478 "show ip ospf n json",
483 "state": "Full/DROther",
488 "state": "Full/DROther",
493 "state": "Full/DROther",
499 ["r0", "show ip ospf database json", db_full
],
500 ["r1", "show ip ospf database json", db_full
],
501 ["r2", "show ip ospf database json", db_full
],
502 ["r3", "show ip ospf database json", db_full
],
503 ["r0", "show ip ospf database json", db_full
],
504 ["r0", "show ip ospf database router json", {}],
505 ["r0", "show ip ospf interface traffic json", {}],
506 ["r1", "show ip ospf interface traffic json", {}],
507 ["r2", "show ip ospf interface traffic json", {}],
508 ["r3", "show ip ospf interface traffic json", {}],
510 for cmd_set
in input:
511 step("test_ospf: %s - %s" % (cmd_set
[0], cmd_set
[1]))
513 verify_ospf_json(tgen
, tgen
.gears
[cmd_set
[0]], cmd_set
[2], cmd_set
[1])
518 if __name__
== "__main__":
519 args
= ["-s"] + sys
.argv
[1:]
520 sys
.exit(pytest
.main(args
))