]> git.proxmox.com Git - mirror_frr.git/blame - tests/topotests/ospf_basic_functionality/test_ospf_p2mp.py
tests: fix pylint test errors
[mirror_frr.git] / tests / topotests / ospf_basic_functionality / test_ospf_p2mp.py
CommitLineData
dc5298d7 1#!/usr/bin/python
2
3#
4# Copyright (c) 2020 by VMware, Inc. ("VMware")
5# Used Copyright (c) 2018 by Network Device Education Foundation, Inc.
6# ("NetDEF") in this file.
7#
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
11# in all copies.
12#
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
20# OF THIS SOFTWARE.
21#
22
23
24"""OSPF Basic Functionality Automation."""
25import os
26import sys
27import time
28import pytest
29import json
30from copy import deepcopy
31from ipaddress import IPv4Address
32
33# Save the Current Working Directory to find configuration files.
34CWD = os.path.dirname(os.path.realpath(__file__))
35sys.path.append(os.path.join(CWD, "../"))
36sys.path.append(os.path.join(CWD, "../lib/"))
37
38# pylint: disable=C0413
39# Import topogen and topotest helpers
8db751b8 40from lib.micronet_compat import Topo
dc5298d7 41from lib.topogen import Topogen, get_topogen
42import ipaddress
43
44# Import topoJson from lib, to create topology and initial configuration
45from lib.common_config import (
46 start_topology,
47 write_test_header,
48 write_test_footer,
49 reset_config_on_routers,
50 verify_rib,
51 create_static_routes,
52 step,
53 create_route_maps,
54 shutdown_bringup_interface,
55 create_interfaces_cfg,
56 topo_daemons,
57)
58from lib.topolog import logger
59from lib.topojson import build_topo_from_json, build_config_from_json
8db751b8 60from lib.topotest import frr_unicode
dc5298d7 61
62from lib.ospf import (
63 verify_ospf_neighbor,
64 config_ospf_interface,
65 clear_ospf,
66 verify_ospf_rib,
67 create_router_ospf,
68 verify_ospf_interface,
69 verify_ospf_database,
70)
71
6ff492b1
DS
72pytestmark = [pytest.mark.ospfd, pytest.mark.staticd]
73
74
dc5298d7 75# Global variables
76topo = None
77
78# Reading the data from JSON File for topology creation
79jsonFile = "{}/ospf_p2mp.json".format(CWD)
80try:
81 with open(jsonFile, "r") as topoJson:
82 topo = json.load(topoJson)
83except IOError:
84 assert False, "Could not read file {}".format(jsonFile)
85
86"""
87TOPOOLOGY =
88 Please view in a fixed-width font such as Courier.
89 +---+ A1 +---+
90 +R1 +------------+R2 |
91 +-+-+- +--++
92 | -- -- |
93 | -- A0 -- |
94 A0| ---- |
95 | ---- | A2
96 | -- -- |
97 | -- -- |
98 +-+-+- +-+-+
99 +R0 +-------------+R3 |
100 +---+ A3 +---+
101
102TESTCASES =
1031. OSPF P2MP -Verify state change events on p2mp network.
104 """
105
106
107class CreateTopo(Topo):
108 """
109 Test topology builder.
110
111 * `Topo`: Topology object
112 """
113
114 def build(self, *_args, **_opts):
115 """Build function."""
116 tgen = get_topogen(self)
117
118 # Building topology from json file
119 build_topo_from_json(tgen, topo)
120
121
122def setup_module(mod):
123 """
124 Sets up the pytest environment
125
126 * `mod`: module name
127 """
128 global topo
129 testsuite_run_time = time.asctime(time.localtime(time.time()))
130 logger.info("Testsuite start time: {}".format(testsuite_run_time))
131 logger.info("=" * 40)
132
133 logger.info("Running setup_module to create topology")
134
135 # This function initiates the topology build with Topogen...
136 tgen = Topogen(CreateTopo, mod.__name__)
137 # ... and here it calls Mininet initialization functions.
138
139 # get list of daemons needs to be started for this suite.
140 daemons = topo_daemons(tgen, topo)
141
142 # Starting topology, create tmp files which are loaded to routers
143 # to start deamons and then start routers
144 start_topology(tgen, daemons)
145
146 # Creating configuration from JSON
147 build_config_from_json(tgen, topo)
148
149 # Don't run this test if we have any failure.
150 if tgen.routers_have_failure():
151 pytest.skip(tgen.errors)
152
dc5298d7 153 logger.info("Running setup_module() done")
154
155
156def teardown_module(mod):
157 """
158 Teardown the pytest environment.
159
160 * `mod`: module name
161 """
162
163 logger.info("Running teardown_module to delete topology")
164
165 tgen = get_topogen()
166
167 # Stop toplogy and Remove tmp files
168 tgen.stop_topology()
169
170 logger.info(
171 "Testsuite end time: {}".format(time.asctime(time.localtime(time.time())))
172 )
173 logger.info("=" * 40)
174
175
176# ##################################
177# Test cases start here.
178# ##################################
179
180
181def test_ospf_p2mp_tc1_p0(request):
182 """OSPF IFSM -Verify state change events on p2mp network."""
183 tc_name = request.node.name
184 write_test_header(tc_name)
185 tgen = get_topogen()
186
187 # Don't run this test if we have any failure.
188 if tgen.routers_have_failure():
189 pytest.skip(tgen.errors)
190
191 global topo
192 step("Bring up the base config as per the topology")
193 reset_config_on_routers(tgen)
194 step(
195 "Verify that OSPF is subscribed to multi cast services "
196 "(All SPF, all DR Routers)."
197 )
198 step("Verify that interface is enabled in ospf.")
199 step("Verify that config is successful.")
200 dut = "r0"
201 input_dict = {
202 "r0": {
203 "links": {
204 "r3": {"ospf": {"mcastMemberOspfAllRouters": True, "ospfEnabled": True}}
205 }
206 }
207 }
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)
210
211 step("Delete the ip address")
212 topo1 = {
213 "r0": {
214 "links": {
215 "r3": {
216 "ipv4": topo["routers"]["r0"]["links"]["r3"]["ipv4"],
217 "interface": topo["routers"]["r0"]["links"]["r3"]["interface"],
218 "delete": True,
219 }
220 }
221 }
222 }
223
224 result = create_interfaces_cfg(tgen, topo1)
225 assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
226
227 step("Change the ip on the R0 interface")
228
229 topo_modify_change_ip = deepcopy(topo)
230 intf_ip = topo_modify_change_ip["routers"]["r0"]["links"]["r3"]["ipv4"]
231 topo_modify_change_ip["routers"]["r0"]["links"]["r3"]["ipv4"] = str(
d7d21c3a 232 IPv4Address(frr_unicode(intf_ip.split("/")[0])) + 3
dc5298d7 233 ) + "/{}".format(intf_ip.split("/")[1])
234
235 build_config_from_json(tgen, topo_modify_change_ip, save_bkup=False)
236 step("Verify that interface is enabled in ospf.")
237 dut = "r0"
238 input_dict = {
239 "r0": {
240 "links": {
241 "r3": {
242 "ospf": {
243 "ipAddress": topo_modify_change_ip["routers"]["r0"]["links"][
244 "r3"
245 ]["ipv4"].split("/")[0],
246 "ipAddressPrefixlen": int(
247 topo_modify_change_ip["routers"]["r0"]["links"]["r3"][
248 "ipv4"
249 ].split("/")[1]
250 ),
251 }
252 }
253 }
254 }
255 }
256 result = verify_ospf_interface(tgen, topo, dut=dut, input_dict=input_dict)
257 assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
258
259 step("Modify the mask on the R0 interface")
260 ip_addr = topo_modify_change_ip["routers"]["r0"]["links"]["r3"]["ipv4"]
261 mask = topo_modify_change_ip["routers"]["r0"]["links"]["r3"]["ipv4"]
262 step("Delete the ip address")
263 topo1 = {
264 "r0": {
265 "links": {
266 "r3": {
267 "ipv4": ip_addr,
268 "interface": topo["routers"]["r0"]["links"]["r3"]["interface"],
269 "delete": True,
270 }
271 }
272 }
273 }
274
275 result = create_interfaces_cfg(tgen, topo1)
276 assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
277
278 step("Change the ip on the R0 interface")
279
280 topo_modify_change_ip = deepcopy(topo)
281 intf_ip = topo_modify_change_ip["routers"]["r0"]["links"]["r3"]["ipv4"]
282 topo_modify_change_ip["routers"]["r0"]["links"]["r3"]["ipv4"] = str(
d7d21c3a 283 IPv4Address(frr_unicode(intf_ip.split("/")[0])) + 3
dc5298d7 284 ) + "/{}".format(int(intf_ip.split("/")[1]) + 1)
285
286 build_config_from_json(tgen, topo_modify_change_ip, save_bkup=False)
287 step("Verify that interface is enabled in ospf.")
288 dut = "r0"
289 input_dict = {
290 "r0": {
291 "links": {
292 "r3": {
293 "ospf": {
294 "ipAddress": topo_modify_change_ip["routers"]["r0"]["links"][
295 "r3"
296 ]["ipv4"].split("/")[0],
297 "ipAddressPrefixlen": int(
298 topo_modify_change_ip["routers"]["r0"]["links"]["r3"][
299 "ipv4"
300 ].split("/")[1]
301 ),
302 }
303 }
304 }
305 }
306 }
307 result = verify_ospf_interface(tgen, topo, dut=dut, input_dict=input_dict)
308 assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
309
310 topo1 = {
311 "r0": {
312 "links": {
313 "r3": {
314 "ipv4": topo_modify_change_ip["routers"]["r0"]["links"]["r3"][
315 "ipv4"
316 ],
317 "interface": topo_modify_change_ip["routers"]["r0"]["links"]["r3"][
318 "interface"
319 ],
320 "delete": True,
321 }
322 }
323 }
324 }
325
326 result = create_interfaces_cfg(tgen, topo1)
327 assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
328
329 build_config_from_json(tgen, topo, save_bkup=False)
330
331 step("Change the area id on the interface")
332 input_dict = {
333 "r0": {
334 "links": {
335 "r3": {
336 "interface": topo["routers"]["r0"]["links"]["r3"]["interface"],
337 "ospf": {"area": "0.0.0.0"},
338 "delete": True,
339 }
340 }
341 }
342 }
343
344 result = create_interfaces_cfg(tgen, input_dict)
345 assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
346
347 input_dict = {
348 "r0": {
349 "links": {
350 "r3": {
351 "interface": topo["routers"]["r0"]["links"]["r3"]["interface"],
352 "ospf": {"area": "0.0.0.1"},
353 }
354 }
355 }
356 }
357
358 result = create_interfaces_cfg(tgen, input_dict)
359 assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
360 step("Verify that interface is enabled in ospf.")
361 dut = "r0"
362 input_dict = {
363 "r0": {"links": {"r3": {"ospf": {"area": "0.0.0.1", "ospfEnabled": True}}}}
364 }
365 result = verify_ospf_interface(tgen, topo, dut=dut, input_dict=input_dict)
366 assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
367
368 input_dict = {
369 "r0": {
370 "links": {
371 "r3": {
372 "interface": topo["routers"]["r0"]["links"]["r3"]["interface"],
373 "ospf": {"area": "0.0.0.1"},
374 "delete": True,
375 }
376 }
377 }
378 }
379
380 result = create_interfaces_cfg(tgen, input_dict)
381 assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
382
383 input_dict = {
384 "r0": {
385 "links": {
386 "r3": {
387 "interface": topo["routers"]["r0"]["links"]["r3"]["interface"],
388 "ospf": {"area": "0.0.0.0"},
389 }
390 }
391 }
392 }
393
394 result = create_interfaces_cfg(tgen, input_dict)
395 assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
396
397 step("Verify if interface is enabled with network type P2MP")
398 input_dict = {
399 "r0": {
400 "links": {
401 "r3": {
402 "interface": topo["routers"]["r0"]["links"]["r3"]["interface"],
5980ad0a 403 "ospf": {"area": "0.0.0.0", "networkType": "POINTOMULTIPOINT"},
dc5298d7 404 }
405 }
406 }
407 }
408 result = create_interfaces_cfg(tgen, input_dict)
409 assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
410
411 write_test_footer(tc_name)
412
413
414if __name__ == "__main__":
415 args = ["-s"] + sys.argv[1:]
416 sys.exit(pytest.main(args))