]>
Commit | Line | Data |
---|---|---|
1 | #!/usr/bin/python | |
2 | # SPDX-License-Identifier: ISC | |
3 | ||
4 | # | |
5 | # Copyright (c) 2020 by VMware, Inc. ("VMware") | |
6 | # Used Copyright (c) 2018 by Network Device Education Foundation, Inc. | |
7 | # ("NetDEF") in this file. | |
8 | # | |
9 | ||
10 | ||
11 | """OSPF Basic Functionality Automation.""" | |
12 | import os | |
13 | import sys | |
14 | import time | |
15 | import pytest | |
16 | ||
17 | # Save the Current Working Directory to find configuration files. | |
18 | CWD = os.path.dirname(os.path.realpath(__file__)) | |
19 | sys.path.append(os.path.join(CWD, "../")) | |
20 | sys.path.append(os.path.join(CWD, "../lib/")) | |
21 | ||
22 | # pylint: disable=C0413 | |
23 | # Import topogen and topotest helpers | |
24 | from lib.topogen import Topogen, get_topogen | |
25 | ||
26 | # Import topoJson from lib, to create topology and initial configuration | |
27 | from lib.common_config import ( | |
28 | start_topology, | |
29 | write_test_header, | |
30 | write_test_footer, | |
31 | reset_config_on_routers, | |
32 | verify_rib, | |
33 | create_static_routes, | |
34 | step, | |
35 | ) | |
36 | from lib.topolog import logger | |
37 | from lib.topojson import build_config_from_json | |
38 | ||
39 | from lib.ospf import ( | |
40 | verify_ospf_neighbor, | |
41 | clear_ospf, | |
42 | verify_ospf_rib, | |
43 | redistribute_ospf, | |
44 | ) | |
45 | ||
46 | pytestmark = [pytest.mark.ospfd, pytest.mark.staticd] | |
47 | ||
48 | ||
49 | # Global variables | |
50 | topo = None | |
51 | # Reading the data from JSON File for topology creation | |
52 | NETWORK = { | |
53 | "ipv4": [ | |
54 | "11.0.20.1/32", | |
55 | "11.0.20.2/32", | |
56 | "11.0.20.3/32", | |
57 | "11.0.20.4/32", | |
58 | "11.0.20.5/32", | |
59 | ], | |
60 | "ipv6": ["1::1/128", "1::2/128", "1::3/128", "1::4/128", "1::5/128"], | |
61 | } | |
62 | MASK = {"ipv4": "32", "ipv6": "128"} | |
63 | NEXT_HOP = { | |
64 | "ipv4": ["10.0.0.1", "10.0.1.1", "10.0.2.1", "10.0.3.1", "10.0.4.1"], | |
65 | "ipv6": ["Null0", "Null0", "Null0", "Null0", "Null0"], | |
66 | } | |
67 | """ | |
68 | TOPOOLOGY = | |
69 | Please view in a fixed-width font such as Courier. | |
70 | Topo : Broadcast Networks | |
71 | +---+ +---+ +---+ +---+ | |
72 | |R0 + +R1 + +R2 + +R3 | | |
73 | +-+-+ +-+-+ +-+-+ +-+-+ | |
74 | | | | | | |
75 | | | | | | |
76 | --+-----------+--------------+---------------+----- | |
77 | Ethernet Segment | |
78 | ||
79 | TESTCASES = | |
80 | 1. Verify OSPF ECMP with max path configured as 8 | |
81 | (Edge having 1 uplink port as broadcast network, | |
82 | connect to 8 TORs - LAN case) | |
83 | ||
84 | """ | |
85 | ||
86 | ||
87 | def setup_module(mod): | |
88 | """ | |
89 | Sets up the pytest environment | |
90 | ||
91 | * `mod`: module name | |
92 | """ | |
93 | testsuite_run_time = time.asctime(time.localtime(time.time())) | |
94 | logger.info("Testsuite start time: {}".format(testsuite_run_time)) | |
95 | logger.info("=" * 40) | |
96 | ||
97 | logger.info("Running setup_module to create topology") | |
98 | ||
99 | # This function initiates the topology build with Topogen... | |
100 | json_file = "{}/ospf_ecmp_lan.json".format(CWD) | |
101 | tgen = Topogen(json_file, mod.__name__) | |
102 | global topo | |
103 | topo = tgen.json_topo | |
104 | # ... and here it calls Mininet initialization functions. | |
105 | ||
106 | # Starting topology, create tmp files which are loaded to routers | |
107 | # to start daemons and then start routers | |
108 | start_topology(tgen) | |
109 | ||
110 | # Creating configuration from JSON | |
111 | build_config_from_json(tgen, topo) | |
112 | ||
113 | # Don't run this test if we have any failure. | |
114 | if tgen.routers_have_failure(): | |
115 | pytest.skip(tgen.errors) | |
116 | # Api call verify whether OSPF is converged | |
117 | ospf_covergence = verify_ospf_neighbor(tgen, topo, lan=True) | |
118 | assert ospf_covergence is True, "setup_module :Failed \n Error:" " {}".format( | |
119 | ospf_covergence | |
120 | ) | |
121 | ||
122 | logger.info("Running setup_module() done") | |
123 | ||
124 | ||
125 | def teardown_module(): | |
126 | """Teardown the pytest environment""" | |
127 | ||
128 | logger.info("Running teardown_module to delete topology") | |
129 | ||
130 | tgen = get_topogen() | |
131 | ||
132 | try: | |
133 | # Stop toplogy and Remove tmp files | |
134 | tgen.stop_topology() | |
135 | ||
136 | except OSError: | |
137 | # OSError exception is raised when mininet tries to stop switch | |
138 | # though switch is stopped once but mininet tries to stop same | |
139 | # switch again, where it ended up with exception | |
140 | pass | |
141 | ||
142 | ||
143 | # ################################## | |
144 | # Test cases start here. | |
145 | # ################################## | |
146 | ||
147 | ||
148 | def test_ospf_lan_ecmp_tc18_p0(request): | |
149 | """ | |
150 | OSPF ECMP. | |
151 | ||
152 | Verify OSPF ECMP with max path configured as 8 | |
153 | (Edge having 1 uplink port as broadcast network, | |
154 | connect to 8 TORs - LAN case) | |
155 | ||
156 | """ | |
157 | tc_name = request.node.name | |
158 | write_test_header(tc_name) | |
159 | tgen = get_topogen() | |
160 | ||
161 | # Don't run this test if we have any failure. | |
162 | if tgen.routers_have_failure(): | |
163 | pytest.skip(tgen.errors) | |
164 | ||
165 | global topo | |
166 | step("Bring up the base config as per the topology") | |
167 | step(". Configure ospf in all the routers on LAN interface.") | |
168 | reset_config_on_routers(tgen) | |
169 | step("Verify that OSPF is up with 8 neighborship sessions.") | |
170 | ||
171 | ospf_covergence = verify_ospf_neighbor(tgen, topo, lan=True) | |
172 | assert ospf_covergence is True, "setup_module :Failed \n Error:" " {}".format( | |
173 | ospf_covergence | |
174 | ) | |
175 | ||
176 | step( | |
177 | "Configure a static route in all the routes and " | |
178 | "redistribute static/connected in OSPF." | |
179 | ) | |
180 | ||
181 | for rtr in topo["routers"]: | |
182 | input_dict = { | |
183 | rtr: { | |
184 | "static_routes": [ | |
185 | {"network": NETWORK["ipv4"][0], "no_of_ip": 5, "next_hop": "Null0"} | |
186 | ] | |
187 | } | |
188 | } | |
189 | result = create_static_routes(tgen, input_dict) | |
190 | assert result is True, "Testcase {} : Failed \n Error: {}".format( | |
191 | tc_name, result | |
192 | ) | |
193 | ||
194 | dut = rtr | |
195 | redistribute_ospf(tgen, topo, dut, "static") | |
196 | ||
197 | step( | |
198 | "Verify that route in R0 in stalled with 8 hops. " | |
199 | "Verify ospf route table and ip route table." | |
200 | ) | |
201 | ||
202 | nh = [] | |
203 | for rtr in topo["routers"]: | |
204 | nh.append(topo["routers"][rtr]["links"]["s1"]["ipv4"].split("/")[0]) | |
205 | nh.remove(topo["routers"]["r1"]["links"]["s1"]["ipv4"].split("/")[0]) | |
206 | dut = "r1" | |
207 | result = verify_ospf_rib(tgen, dut, input_dict, next_hop=nh) | |
208 | assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result) | |
209 | ||
210 | protocol = "ospf" | |
211 | result = verify_rib(tgen, "ipv4", dut, input_dict, protocol=protocol, next_hop=nh) | |
212 | assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result) | |
213 | ||
214 | step(" clear ip ospf interface on DUT(r0)") | |
215 | clear_ospf(tgen, "r0") | |
216 | ||
217 | step( | |
218 | "Verify that after clearing the ospf interface all the " | |
219 | "neighbours are up and routes are installed with 8 next hop " | |
220 | "in ospf and ip route tables on R0" | |
221 | ) | |
222 | ||
223 | dut = "r0" | |
224 | ospf_covergence = verify_ospf_neighbor(tgen, topo, dut=dut, lan=True) | |
225 | assert ospf_covergence is True, "setup_module :Failed \n Error:" " {}".format( | |
226 | ospf_covergence | |
227 | ) | |
228 | ||
229 | step(" clear ip ospf interface on R2") | |
230 | clear_ospf(tgen, "r2") | |
231 | ||
232 | dut = "r2" | |
233 | ospf_covergence = verify_ospf_neighbor(tgen, topo, dut=dut, lan=True) | |
234 | assert ospf_covergence is True, "setup_module :Failed \n Error:" " {}".format( | |
235 | ospf_covergence | |
236 | ) | |
237 | ||
238 | step("Delete static/connected cmd in ospf in all the routes one by one.") | |
239 | for rtr in topo["routers"]: | |
240 | input_dict = { | |
241 | rtr: { | |
242 | "static_routes": [ | |
243 | { | |
244 | "network": NETWORK["ipv4"][0], | |
245 | "no_of_ip": 5, | |
246 | "next_hop": "Null0", | |
247 | "delete": True, | |
248 | } | |
249 | ] | |
250 | } | |
251 | } | |
252 | result = create_static_routes(tgen, input_dict) | |
253 | assert result is True, "Testcase {} : Failed \n Error: {}".format( | |
254 | tc_name, result | |
255 | ) | |
256 | ||
257 | step("Verify that all the routes are withdrawn from R0") | |
258 | dut = "r1" | |
259 | result = verify_ospf_rib( | |
260 | tgen, dut, input_dict, next_hop=nh, retry_timeout=10, expected=False | |
261 | ) | |
262 | assert ( | |
263 | result is not True | |
264 | ), "Testcase {} : Failed \n " "r1: OSPF routes are present \n Error: {}".format( | |
265 | tc_name, result | |
266 | ) | |
267 | ||
268 | protocol = "ospf" | |
269 | result = verify_rib( | |
270 | tgen, | |
271 | "ipv4", | |
272 | dut, | |
273 | input_dict, | |
274 | protocol=protocol, | |
275 | next_hop=nh, | |
276 | retry_timeout=10, | |
277 | expected=False, | |
278 | ) | |
279 | assert ( | |
280 | result is not True | |
281 | ), "Testcase {} : Failed \n " "r1: routes are still present \n Error: {}".format( | |
282 | tc_name, result | |
283 | ) | |
284 | ||
285 | write_test_footer(tc_name) | |
286 | ||
287 | ||
288 | if __name__ == "__main__": | |
289 | args = ["-s"] + sys.argv[1:] | |
290 | sys.exit(pytest.main(args)) |