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