Total Complexity | 76 |
Total Lines | 2166 |
Duplicated Lines | 11.91 % |
Changes | 0 |
Duplicate code is one of the most pungent code smells. A rule that is often used is to re-structure code once it is duplicated in three or more places.
Common duplication problems, and corresponding solutions are:
Complex classes like build.tests.unit.test_main often do a lot of different things. To break such a class down, we need to identify a cohesive component within that class. A common approach to find such a component is to look for fields/methods that share the same prefixes, or suffixes.
Once you have determined the fields that belong together, you can apply the Extract Class refactoring. If the component makes sense as a sub-class, Extract Subclass is also a candidate, and is often faster.
1 | """Module to test the main napp file.""" |
||
2 | import json |
||
3 | from unittest import TestCase |
||
4 | from unittest.mock import MagicMock, PropertyMock, call, create_autospec, patch |
||
5 | |||
6 | from kytos.core.events import KytosEvent |
||
7 | from kytos.core.interface import UNI, Interface |
||
8 | from kytos.lib.helpers import get_controller_mock |
||
9 | from napps.kytos.mef_eline.exceptions import InvalidPath |
||
10 | from napps.kytos.mef_eline.models import EVC |
||
11 | from napps.kytos.mef_eline.tests.helpers import get_uni_mocked |
||
12 | |||
13 | |||
14 | # pylint: disable=too-many-public-methods, too-many-lines |
||
15 | class TestMain(TestCase): |
||
16 | """Test the Main class.""" |
||
17 | |||
18 | def setUp(self): |
||
19 | """Execute steps before each tests. |
||
20 | |||
21 | Set the server_name_url_url from kytos/mef_eline |
||
22 | """ |
||
23 | self.server_name_url = "http://localhost:8181/api/kytos/mef_eline" |
||
24 | |||
25 | # The decorator run_on_thread is patched, so methods that listen |
||
26 | # for events do not run on threads while tested. |
||
27 | # Decorators have to be patched before the methods that are |
||
28 | # decorated with them are imported. |
||
29 | patch("kytos.core.helpers.run_on_thread", lambda x: x).start() |
||
30 | # pylint: disable=import-outside-toplevel |
||
31 | from napps.kytos.mef_eline.main import Main |
||
32 | Main.get_eline_controller = MagicMock() |
||
33 | self.addCleanup(patch.stopall) |
||
34 | self.napp = Main(get_controller_mock()) |
||
35 | |||
36 | def test_get_event_listeners(self): |
||
37 | """Verify all event listeners registered.""" |
||
38 | expected_events = [ |
||
39 | "kytos/core.shutdown", |
||
40 | "kytos/core.shutdown.kytos/mef_eline", |
||
41 | "kytos/topology.link_up", |
||
42 | "kytos/topology.link_down", |
||
43 | ] |
||
44 | actual_events = self.napp.listeners() |
||
45 | |||
46 | for _event in expected_events: |
||
47 | self.assertIn(_event, actual_events, _event) |
||
48 | |||
49 | def test_verify_api_urls(self): |
||
50 | """Verify all APIs registered.""" |
||
51 | expected_urls = [ |
||
52 | ({}, {"POST", "OPTIONS"}, "/api/kytos/mef_eline/v2/evc/"), |
||
53 | ({}, {"OPTIONS", "HEAD", "GET"}, "/api/kytos/mef_eline/v2/evc/"), |
||
54 | ( |
||
55 | {"circuit_id": "[circuit_id]"}, |
||
56 | {"OPTIONS", "DELETE"}, |
||
57 | "/api/kytos/mef_eline/v2/evc/<circuit_id>", |
||
58 | ), |
||
59 | ( |
||
60 | {"circuit_id": "[circuit_id]"}, |
||
61 | {"OPTIONS", "HEAD", "GET"}, |
||
62 | "/api/kytos/mef_eline/v2/evc/<circuit_id>", |
||
63 | ), |
||
64 | ( |
||
65 | {"circuit_id": "[circuit_id]"}, |
||
66 | {"OPTIONS", "PATCH"}, |
||
67 | "/api/kytos/mef_eline/v2/evc/<circuit_id>", |
||
68 | ), |
||
69 | ( |
||
70 | {"circuit_id": "[circuit_id]"}, |
||
71 | {"OPTIONS", "HEAD", "GET"}, |
||
72 | "/api/kytos/mef_eline/v2/evc/<circuit_id>/metadata", |
||
73 | ), |
||
74 | ( |
||
75 | {"circuit_id": "[circuit_id]"}, |
||
76 | {"OPTIONS", "POST"}, |
||
77 | "/api/kytos/mef_eline/v2/evc/<circuit_id>/metadata", |
||
78 | ), |
||
79 | ( |
||
80 | {"circuit_id": "[circuit_id]", "key": "[key]"}, |
||
81 | {"OPTIONS", "DELETE"}, |
||
82 | "/api/kytos/mef_eline/v2/evc/<circuit_id>/metadata/<key>", |
||
83 | ), |
||
84 | ( |
||
85 | {"circuit_id": "[circuit_id]"}, |
||
86 | {"OPTIONS", "PATCH"}, |
||
87 | "/api/kytos/mef_eline/v2/evc/<circuit_id>/redeploy", |
||
88 | ), |
||
89 | ( |
||
90 | {}, |
||
91 | {"OPTIONS", "GET", "HEAD"}, |
||
92 | "/api/kytos/mef_eline/v2/evc/schedule", |
||
93 | ), |
||
94 | ({}, {"POST", "OPTIONS"}, "/api/kytos/mef_eline/v2/evc/schedule/"), |
||
95 | ( |
||
96 | {"schedule_id": "[schedule_id]"}, |
||
97 | {"OPTIONS", "DELETE"}, |
||
98 | "/api/kytos/mef_eline/v2/evc/schedule/<schedule_id>", |
||
99 | ), |
||
100 | ( |
||
101 | {"schedule_id": "[schedule_id]"}, |
||
102 | {"OPTIONS", "PATCH"}, |
||
103 | "/api/kytos/mef_eline/v2/evc/schedule/<schedule_id>", |
||
104 | ), |
||
105 | ] |
||
106 | urls = self.get_napp_urls(self.napp) |
||
107 | self.assertEqual(len(expected_urls), len(urls)) |
||
108 | |||
109 | @patch('napps.kytos.mef_eline.main.log') |
||
110 | @patch('napps.kytos.mef_eline.main.Main.execute_consistency') |
||
111 | def test_execute(self, mock_execute_consistency, mock_log): |
||
112 | """Test execute.""" |
||
113 | self.napp.execution_rounds = 0 |
||
114 | self.napp.execute() |
||
115 | mock_execute_consistency.assert_called() |
||
116 | self.assertEqual(mock_log.debug.call_count, 2) |
||
117 | |||
118 | # Test locked should return |
||
119 | mock_execute_consistency.call_count = 0 |
||
120 | mock_log.info.call_count = 0 |
||
121 | # pylint: disable=protected-access |
||
122 | self.napp._lock = MagicMock() |
||
123 | self.napp._lock.locked.return_value = True |
||
124 | # pylint: enable=protected-access |
||
125 | self.napp.execute() |
||
126 | mock_execute_consistency.assert_not_called() |
||
127 | mock_log.info.assert_not_called() |
||
128 | |||
129 | @patch('napps.kytos.mef_eline.main.settings') |
||
130 | @patch('napps.kytos.mef_eline.main.Main._load_evc') |
||
131 | @patch("napps.kytos.mef_eline.controllers.ELineController.upsert_evc") |
||
132 | @patch("napps.kytos.mef_eline.models.evc.EVCDeploy.check_list_traces") |
||
133 | def test_execute_consistency(self, mock_check_list_traces, *args): |
||
134 | """Test execute_consistency.""" |
||
135 | (mongo_controller_upsert_mock, mock_load_evc, mock_settings) = args |
||
136 | |||
137 | stored_circuits = {'1': {'name': 'circuit_1'}, |
||
138 | '2': {'name': 'circuit_2'}, |
||
139 | '3': {'name': 'circuit_3'}} |
||
140 | mongo_controller_upsert_mock.return_value = True |
||
141 | self.napp.mongo_controller.get_circuits.return_value = { |
||
142 | "circuits": stored_circuits |
||
143 | } |
||
144 | |||
145 | mock_settings.WAIT_FOR_OLD_PATH = -1 |
||
146 | evc1 = MagicMock(id=1, service_level=0, creation_time=1) |
||
147 | evc1.is_enabled.return_value = True |
||
148 | evc1.is_active.return_value = False |
||
149 | evc1.lock.locked.return_value = False |
||
150 | evc1.has_recent_removed_flow.return_value = False |
||
151 | evc1.is_recent_updated.return_value = False |
||
152 | evc1.execution_rounds = 0 |
||
153 | evc2 = MagicMock(id=2, service_level=7, creation_time=1) |
||
154 | evc2.is_enabled.return_value = True |
||
155 | evc2.is_active.return_value = False |
||
156 | evc2.lock.locked.return_value = False |
||
157 | evc2.has_recent_removed_flow.return_value = False |
||
158 | evc2.is_recent_updated.return_value = False |
||
159 | evc2.execution_rounds = 0 |
||
160 | self.napp.circuits = {'1': evc1, '2': evc2} |
||
161 | assert self.napp.get_evcs_by_svc_level() == [evc2, evc1] |
||
162 | |||
163 | mock_check_list_traces.return_value = { |
||
164 | 1: True, |
||
165 | 2: False |
||
166 | } |
||
167 | |||
168 | self.napp.execute_consistency() |
||
169 | self.assertEqual(evc1.activate.call_count, 1) |
||
170 | self.assertEqual(evc1.sync.call_count, 1) |
||
171 | self.assertEqual(evc2.deploy.call_count, 1) |
||
172 | mock_load_evc.assert_called_with(stored_circuits['3']) |
||
173 | |||
174 | @patch('napps.kytos.mef_eline.main.settings') |
||
175 | @patch('napps.kytos.mef_eline.main.Main._load_evc') |
||
176 | @patch("napps.kytos.mef_eline.controllers.ELineController.upsert_evc") |
||
177 | @patch("napps.kytos.mef_eline.models.evc.EVCDeploy.check_list_traces") |
||
178 | def test_execute_consistency_wait_for(self, mock_check_list_traces, *args): |
||
179 | """Test execute and wait for setting.""" |
||
180 | (mongo_controller_upsert_mock, _, mock_settings) = args |
||
181 | |||
182 | stored_circuits = {'1': {'name': 'circuit_1'}} |
||
183 | mongo_controller_upsert_mock.return_value = True |
||
184 | self.napp.mongo_controller.get_circuits.return_value = { |
||
185 | "circuits": stored_circuits |
||
186 | } |
||
187 | |||
188 | mock_settings.WAIT_FOR_OLD_PATH = -1 |
||
189 | evc1 = MagicMock(id=1, service_level=0, creation_time=1) |
||
190 | evc1.is_enabled.return_value = True |
||
191 | evc1.is_active.return_value = False |
||
192 | evc1.lock.locked.return_value = False |
||
193 | evc1.has_recent_removed_flow.return_value = False |
||
194 | evc1.is_recent_updated.return_value = False |
||
195 | evc1.execution_rounds = 0 |
||
196 | evc1.deploy.call_count = 0 |
||
197 | self.napp.circuits = {'1': evc1} |
||
198 | assert self.napp.get_evcs_by_svc_level() == [evc1] |
||
199 | mock_settings.WAIT_FOR_OLD_PATH = 1 |
||
200 | |||
201 | mock_check_list_traces.return_value = {1: False} |
||
202 | |||
203 | self.napp.execute_consistency() |
||
204 | self.assertEqual(evc1.deploy.call_count, 0) |
||
205 | self.napp.execute_consistency() |
||
206 | self.assertEqual(evc1.deploy.call_count, 1) |
||
207 | |||
208 | @patch('napps.kytos.mef_eline.main.Main._uni_from_dict') |
||
209 | @patch('napps.kytos.mef_eline.models.evc.EVCBase._validate') |
||
210 | def test_evc_from_dict(self, _validate_mock, uni_from_dict_mock): |
||
211 | """ |
||
212 | Test the helper method that create an EVN from dict. |
||
213 | |||
214 | Verify object creation with circuit data and schedule data. |
||
215 | """ |
||
216 | _validate_mock.return_value = True |
||
217 | uni_from_dict_mock.side_effect = ["uni_a", "uni_z"] |
||
218 | payload = { |
||
219 | "name": "my evc1", |
||
220 | "uni_a": { |
||
221 | "interface_id": "00:00:00:00:00:00:00:01:1", |
||
222 | "tag": {"tag_type": 1, "value": 80}, |
||
223 | }, |
||
224 | "uni_z": { |
||
225 | "interface_id": "00:00:00:00:00:00:00:02:2", |
||
226 | "tag": {"tag_type": 1, "value": 1}, |
||
227 | }, |
||
228 | "circuit_scheduler": [ |
||
229 | {"frequency": "* * * * *", "action": "create"} |
||
230 | ], |
||
231 | "queue_id": 5, |
||
232 | } |
||
233 | # pylint: disable=protected-access |
||
234 | evc_response = self.napp._evc_from_dict(payload) |
||
235 | self.assertIsNotNone(evc_response) |
||
236 | self.assertIsNotNone(evc_response.uni_a) |
||
237 | self.assertIsNotNone(evc_response.uni_z) |
||
238 | self.assertIsNotNone(evc_response.circuit_scheduler) |
||
239 | self.assertIsNotNone(evc_response.name) |
||
240 | self.assertIsNotNone(evc_response.queue_id) |
||
241 | |||
242 | View Code Duplication | @patch("napps.kytos.mef_eline.main.Main._uni_from_dict") |
|
|
|||
243 | @patch("napps.kytos.mef_eline.models.evc.EVCBase._validate") |
||
244 | @patch("kytos.core.Controller.get_interface_by_id") |
||
245 | def test_evc_from_dict_paths( |
||
246 | self, _get_interface_by_id_mock, _validate_mock, uni_from_dict_mock |
||
247 | ): |
||
248 | """ |
||
249 | Test the helper method that create an EVN from dict. |
||
250 | |||
251 | Verify object creation with circuit data and schedule data. |
||
252 | """ |
||
253 | |||
254 | _get_interface_by_id_mock.return_value = get_uni_mocked().interface |
||
255 | _validate_mock.return_value = True |
||
256 | uni_from_dict_mock.side_effect = ["uni_a", "uni_z"] |
||
257 | payload = { |
||
258 | "name": "my evc1", |
||
259 | "uni_a": { |
||
260 | "interface_id": "00:00:00:00:00:00:00:01:1", |
||
261 | "tag": {"tag_type": 1, "value": 80}, |
||
262 | }, |
||
263 | "uni_z": { |
||
264 | "interface_id": "00:00:00:00:00:00:00:02:2", |
||
265 | "tag": {"tag_type": 1, "value": 1}, |
||
266 | }, |
||
267 | "current_path": [], |
||
268 | "primary_path": [ |
||
269 | { |
||
270 | "endpoint_a": { |
||
271 | "interface_id": "00:00:00:00:00:00:00:01:1" |
||
272 | }, |
||
273 | "endpoint_b": { |
||
274 | "interface_id": "00:00:00:00:00:00:00:02:2" |
||
275 | }, |
||
276 | } |
||
277 | ], |
||
278 | "backup_path": [], |
||
279 | } |
||
280 | |||
281 | # pylint: disable=protected-access |
||
282 | evc_response = self.napp._evc_from_dict(payload) |
||
283 | self.assertIsNotNone(evc_response) |
||
284 | self.assertIsNotNone(evc_response.uni_a) |
||
285 | self.assertIsNotNone(evc_response.uni_z) |
||
286 | self.assertIsNotNone(evc_response.circuit_scheduler) |
||
287 | self.assertIsNotNone(evc_response.name) |
||
288 | self.assertEqual(len(evc_response.current_path), 0) |
||
289 | self.assertEqual(len(evc_response.backup_path), 0) |
||
290 | self.assertEqual(len(evc_response.primary_path), 1) |
||
291 | |||
292 | View Code Duplication | @patch("napps.kytos.mef_eline.main.Main._uni_from_dict") |
|
293 | @patch("napps.kytos.mef_eline.models.evc.EVCBase._validate") |
||
294 | @patch("kytos.core.Controller.get_interface_by_id") |
||
295 | def test_evc_from_dict_links( |
||
296 | self, _get_interface_by_id_mock, _validate_mock, uni_from_dict_mock |
||
297 | ): |
||
298 | """ |
||
299 | Test the helper method that create an EVN from dict. |
||
300 | |||
301 | Verify object creation with circuit data and schedule data. |
||
302 | """ |
||
303 | _get_interface_by_id_mock.return_value = get_uni_mocked().interface |
||
304 | _validate_mock.return_value = True |
||
305 | uni_from_dict_mock.side_effect = ["uni_a", "uni_z"] |
||
306 | payload = { |
||
307 | "name": "my evc1", |
||
308 | "uni_a": { |
||
309 | "interface_id": "00:00:00:00:00:00:00:01:1", |
||
310 | "tag": {"tag_type": 1, "value": 80}, |
||
311 | }, |
||
312 | "uni_z": { |
||
313 | "interface_id": "00:00:00:00:00:00:00:02:2", |
||
314 | "tag": {"tag_type": 1, "value": 1}, |
||
315 | }, |
||
316 | "primary_links": [ |
||
317 | { |
||
318 | "endpoint_a": { |
||
319 | "interface_id": "00:00:00:00:00:00:00:01:1" |
||
320 | }, |
||
321 | "endpoint_b": { |
||
322 | "interface_id": "00:00:00:00:00:00:00:02:2" |
||
323 | }, |
||
324 | } |
||
325 | ], |
||
326 | "backup_links": [], |
||
327 | } |
||
328 | |||
329 | # pylint: disable=protected-access |
||
330 | evc_response = self.napp._evc_from_dict(payload) |
||
331 | self.assertIsNotNone(evc_response) |
||
332 | self.assertIsNotNone(evc_response.uni_a) |
||
333 | self.assertIsNotNone(evc_response.uni_z) |
||
334 | self.assertIsNotNone(evc_response.circuit_scheduler) |
||
335 | self.assertIsNotNone(evc_response.name) |
||
336 | self.assertEqual(len(evc_response.current_links_cache), 0) |
||
337 | self.assertEqual(len(evc_response.backup_links), 0) |
||
338 | self.assertEqual(len(evc_response.primary_links), 1) |
||
339 | |||
340 | def test_list_without_circuits(self): |
||
341 | """Test if list circuits return 'no circuit stored.'.""" |
||
342 | circuits = {"circuits": {}} |
||
343 | self.napp.mongo_controller.get_circuits.return_value = circuits |
||
344 | api = self.get_app_test_client(self.napp) |
||
345 | url = f"{self.server_name_url}/v2/evc/" |
||
346 | response = api.get(url) |
||
347 | self.assertEqual(response.status_code, 200, response.data) |
||
348 | self.assertEqual(json.loads(response.data.decode()), {}) |
||
349 | |||
350 | def test_list_no_circuits_stored(self): |
||
351 | """Test if list circuits return all circuits stored.""" |
||
352 | circuits = {"circuits": {}} |
||
353 | self.napp.mongo_controller.get_circuits.return_value = circuits |
||
354 | |||
355 | api = self.get_app_test_client(self.napp) |
||
356 | url = f"{self.server_name_url}/v2/evc/" |
||
357 | |||
358 | response = api.get(url) |
||
359 | expected_result = circuits["circuits"] |
||
360 | self.assertEqual(json.loads(response.data), expected_result) |
||
361 | |||
362 | View Code Duplication | def test_list_with_circuits_stored(self): |
|
363 | """Test if list circuits return all circuits stored.""" |
||
364 | circuits = { |
||
365 | 'circuits': |
||
366 | {"1": {"name": "circuit_1"}, "2": {"name": "circuit_2"}} |
||
367 | } |
||
368 | get_circuits = self.napp.mongo_controller.get_circuits |
||
369 | get_circuits.return_value = circuits |
||
370 | |||
371 | api = self.get_app_test_client(self.napp) |
||
372 | url = f"{self.server_name_url}/v2/evc/" |
||
373 | |||
374 | response = api.get(url) |
||
375 | expected_result = circuits["circuits"] |
||
376 | get_circuits.assert_called_with(archived=False) |
||
377 | self.assertEqual(json.loads(response.data), expected_result) |
||
378 | |||
379 | View Code Duplication | def test_list_with_archived_circuits_archived(self): |
|
380 | """Test if list circuits only archived circuits.""" |
||
381 | circuits = { |
||
382 | 'circuits': |
||
383 | { |
||
384 | "1": {"name": "circuit_1", "archived": True}, |
||
385 | } |
||
386 | } |
||
387 | get_circuits = self.napp.mongo_controller.get_circuits |
||
388 | get_circuits.return_value = circuits |
||
389 | |||
390 | api = self.get_app_test_client(self.napp) |
||
391 | url = f"{self.server_name_url}/v2/evc/?archived=true" |
||
392 | |||
393 | response = api.get(url) |
||
394 | get_circuits.assert_called_with(archived=True) |
||
395 | expected_result = {"1": circuits["circuits"]["1"]} |
||
396 | self.assertEqual(json.loads(response.data), expected_result) |
||
397 | |||
398 | def test_list_with_archived_circuits_all(self): |
||
399 | """Test if list circuits return all circuits.""" |
||
400 | circuits = { |
||
401 | 'circuits': { |
||
402 | "1": {"name": "circuit_1"}, |
||
403 | "2": {"name": "circuit_2", "archived": True}, |
||
404 | } |
||
405 | } |
||
406 | self.napp.mongo_controller.get_circuits.return_value = circuits |
||
407 | |||
408 | api = self.get_app_test_client(self.napp) |
||
409 | url = f"{self.server_name_url}/v2/evc/?archived=null" |
||
410 | |||
411 | response = api.get(url) |
||
412 | expected_result = circuits["circuits"] |
||
413 | self.assertEqual(json.loads(response.data), expected_result) |
||
414 | |||
415 | def test_circuit_with_valid_id(self): |
||
416 | """Test if get_circuit return the circuit attributes.""" |
||
417 | circuit = {"name": "circuit_1"} |
||
418 | self.napp.mongo_controller.get_circuit.return_value = circuit |
||
419 | |||
420 | api = self.get_app_test_client(self.napp) |
||
421 | url = f"{self.server_name_url}/v2/evc/1" |
||
422 | response = api.get(url) |
||
423 | expected_result = circuit |
||
424 | self.assertEqual(json.loads(response.data), expected_result) |
||
425 | |||
426 | def test_circuit_with_invalid_id(self): |
||
427 | """Test if get_circuit return invalid circuit_id.""" |
||
428 | self.napp.mongo_controller.get_circuit.return_value = None |
||
429 | api = self.get_app_test_client(self.napp) |
||
430 | url = f"{self.server_name_url}/v2/evc/3" |
||
431 | response = api.get(url) |
||
432 | expected_result = "circuit_id 3 not found" |
||
433 | self.assertEqual( |
||
434 | json.loads(response.data)["description"], expected_result |
||
435 | ) |
||
436 | |||
437 | @patch("napps.kytos.mef_eline.models.evc.EVC.deploy") |
||
438 | @patch("napps.kytos.mef_eline.scheduler.Scheduler.add") |
||
439 | @patch("napps.kytos.mef_eline.main.Main._uni_from_dict") |
||
440 | @patch("napps.kytos.mef_eline.controllers.ELineController.upsert_evc") |
||
441 | @patch("napps.kytos.mef_eline.main.EVC.as_dict") |
||
442 | @patch("napps.kytos.mef_eline.models.evc.EVC._validate") |
||
443 | def test_create_a_circuit_case_1(self, *args): |
||
444 | """Test create a new circuit.""" |
||
445 | # pylint: disable=too-many-locals |
||
446 | ( |
||
447 | validate_mock, |
||
448 | evc_as_dict_mock, |
||
449 | mongo_controller_upsert_mock, |
||
450 | uni_from_dict_mock, |
||
451 | sched_add_mock, |
||
452 | evc_deploy_mock, |
||
453 | ) = args |
||
454 | |||
455 | validate_mock.return_value = True |
||
456 | mongo_controller_upsert_mock.return_value = True |
||
457 | evc_deploy_mock.return_value = True |
||
458 | uni1 = create_autospec(UNI) |
||
459 | uni2 = create_autospec(UNI) |
||
460 | uni1.interface = create_autospec(Interface) |
||
461 | uni2.interface = create_autospec(Interface) |
||
462 | uni1.interface.switch = "00:00:00:00:00:00:00:01" |
||
463 | uni2.interface.switch = "00:00:00:00:00:00:00:02" |
||
464 | uni_from_dict_mock.side_effect = [uni1, uni2] |
||
465 | evc_as_dict_mock.return_value = {} |
||
466 | sched_add_mock.return_value = True |
||
467 | self.napp.mongo_controller.get_circuits.return_value = {} |
||
468 | |||
469 | api = self.get_app_test_client(self.napp) |
||
470 | url = f"{self.server_name_url}/v2/evc/" |
||
471 | payload = { |
||
472 | "name": "my evc1", |
||
473 | "frequency": "* * * * *", |
||
474 | "uni_a": { |
||
475 | "interface_id": "00:00:00:00:00:00:00:01:1", |
||
476 | "tag": {"tag_type": 1, "value": 80}, |
||
477 | }, |
||
478 | "uni_z": { |
||
479 | "interface_id": "00:00:00:00:00:00:00:02:2", |
||
480 | "tag": {"tag_type": 1, "value": 1}, |
||
481 | }, |
||
482 | "dynamic_backup_path": True, |
||
483 | "primary_constraints": { |
||
484 | "spf_max_path_cost": 8, |
||
485 | "mandatory_metrics": { |
||
486 | "ownership": "red" |
||
487 | } |
||
488 | }, |
||
489 | "secondary_constraints": { |
||
490 | "mandatory_metrics": { |
||
491 | "ownership": "blue" |
||
492 | } |
||
493 | } |
||
494 | } |
||
495 | |||
496 | response = api.post( |
||
497 | url, data=json.dumps(payload), content_type="application/json" |
||
498 | ) |
||
499 | current_data = json.loads(response.data) |
||
500 | |||
501 | # verify expected result from request |
||
502 | self.assertEqual(201, response.status_code, response.data) |
||
503 | self.assertIn("circuit_id", current_data) |
||
504 | |||
505 | # verify uni called |
||
506 | uni_from_dict_mock.called_twice() |
||
507 | uni_from_dict_mock.assert_any_call(payload["uni_z"]) |
||
508 | uni_from_dict_mock.assert_any_call(payload["uni_a"]) |
||
509 | |||
510 | # verify validation called |
||
511 | validate_mock.assert_called_once() |
||
512 | validate_mock.assert_called_with( |
||
513 | frequency="* * * * *", |
||
514 | name="my evc1", |
||
515 | uni_a=uni1, |
||
516 | uni_z=uni2, |
||
517 | dynamic_backup_path=True, |
||
518 | primary_constraints=payload["primary_constraints"], |
||
519 | secondary_constraints=payload["secondary_constraints"], |
||
520 | ) |
||
521 | # verify save method is called |
||
522 | mongo_controller_upsert_mock.assert_called_once() |
||
523 | |||
524 | # verify evc as dict is called to save in the box |
||
525 | evc_as_dict_mock.assert_called() |
||
526 | # verify add circuit in sched |
||
527 | sched_add_mock.assert_called_once() |
||
528 | |||
529 | @staticmethod |
||
530 | def get_napp_urls(napp): |
||
531 | """Return the kytos/mef_eline urls. |
||
532 | |||
533 | The urls will be like: |
||
534 | |||
535 | urls = [ |
||
536 | (options, methods, url) |
||
537 | ] |
||
538 | |||
539 | """ |
||
540 | controller = napp.controller |
||
541 | controller.api_server.register_napp_endpoints(napp) |
||
542 | |||
543 | urls = [] |
||
544 | for rule in controller.api_server.app.url_map.iter_rules(): |
||
545 | options = {} |
||
546 | for arg in rule.arguments: |
||
547 | options[arg] = f"[{0}]".format(arg) |
||
548 | |||
549 | if f"{napp.username}/{napp.name}" in str(rule): |
||
550 | urls.append((options, rule.methods, f"{str(rule)}")) |
||
551 | |||
552 | return urls |
||
553 | |||
554 | @staticmethod |
||
555 | def get_app_test_client(napp): |
||
556 | """Return a flask api test client.""" |
||
557 | napp.controller.api_server.register_napp_endpoints(napp) |
||
558 | return napp.controller.api_server.app.test_client() |
||
559 | |||
560 | def test_create_a_circuit_case_2(self): |
||
561 | """Test create a new circuit trying to send request without a json.""" |
||
562 | api = self.get_app_test_client(self.napp) |
||
563 | url = f"{self.server_name_url}/v2/evc/" |
||
564 | |||
565 | response = api.post(url) |
||
566 | current_data = json.loads(response.data) |
||
567 | expected_message = "The request body mimetype is not application/json." |
||
568 | expected_data = expected_message |
||
569 | self.assertEqual(415, response.status_code, response.data) |
||
570 | self.assertEqual(current_data["description"], expected_data) |
||
571 | |||
572 | def test_create_a_circuit_case_3(self): |
||
573 | """Test create a new circuit trying to send request with an |
||
574 | invalid json.""" |
||
575 | api = self.get_app_test_client(self.napp) |
||
576 | url = f"{self.server_name_url}/v2/evc/" |
||
577 | |||
578 | response = api.post( |
||
579 | url, |
||
580 | data="This is an {Invalid:} JSON", |
||
581 | content_type="application/json", |
||
582 | ) |
||
583 | current_data = json.loads(response.data) |
||
584 | expected_data = "The request body is not a well-formed JSON." |
||
585 | |||
586 | self.assertEqual(400, response.status_code, response.data) |
||
587 | self.assertEqual(current_data["description"], expected_data) |
||
588 | |||
589 | @patch("napps.kytos.mef_eline.main.Main._uni_from_dict") |
||
590 | @patch("napps.kytos.mef_eline.controllers.ELineController.upsert_evc") |
||
591 | def test_create_a_circuit_case_4( |
||
592 | self, |
||
593 | mongo_controller_upsert_mock, |
||
594 | uni_from_dict_mock |
||
595 | ): |
||
596 | """Test create a new circuit trying to send request with an |
||
597 | invalid value.""" |
||
598 | # pylint: disable=too-many-locals |
||
599 | uni_from_dict_mock.side_effect = ValueError("Could not instantiate") |
||
600 | mongo_controller_upsert_mock.return_value = True |
||
601 | api = self.get_app_test_client(self.napp) |
||
602 | url = f"{self.server_name_url}/v2/evc/" |
||
603 | |||
604 | payload = { |
||
605 | "name": "my evc1", |
||
606 | "frequency": "* * * * *", |
||
607 | "uni_a": { |
||
608 | "interface_id": "00:00:00:00:00:00:00:01:76", |
||
609 | "tag": {"tag_type": 1, "value": 80}, |
||
610 | }, |
||
611 | "uni_z": { |
||
612 | "interface_id": "00:00:00:00:00:00:00:02:2", |
||
613 | "tag": {"tag_type": 1, "value": 1}, |
||
614 | }, |
||
615 | } |
||
616 | |||
617 | response = api.post( |
||
618 | url, data=json.dumps(payload), content_type="application/json" |
||
619 | ) |
||
620 | current_data = json.loads(response.data) |
||
621 | expected_data = "Error creating UNI: Invalid value" |
||
622 | self.assertEqual(400, response.status_code, response.data) |
||
623 | self.assertEqual(current_data["description"], expected_data) |
||
624 | |||
625 | payload["name"] = 1 |
||
626 | response = api.post( |
||
627 | url, data=json.dumps(payload), content_type="application/json" |
||
628 | ) |
||
629 | self.assertEqual(400, response.status_code, response.data) |
||
630 | |||
631 | def test_create_a_circuit_invalid_queue_id(self): |
||
632 | """Test create a new circuit with invalid queue_id.""" |
||
633 | api = self.get_app_test_client(self.napp) |
||
634 | url = f"{self.server_name_url}/v2/evc/" |
||
635 | |||
636 | payload = { |
||
637 | "name": "my evc1", |
||
638 | "queue_id": 8, |
||
639 | "uni_a": { |
||
640 | "interface_id": "00:00:00:00:00:00:00:01:76", |
||
641 | "tag": {"tag_type": 1, "value": 80}, |
||
642 | }, |
||
643 | "uni_z": { |
||
644 | "interface_id": "00:00:00:00:00:00:00:02:2", |
||
645 | "tag": {"tag_type": 1, "value": 1}, |
||
646 | }, |
||
647 | } |
||
648 | response = api.post( |
||
649 | url, |
||
650 | data=json.dumps(payload), |
||
651 | content_type="application/json", |
||
652 | ) |
||
653 | current_data = json.loads(response.data) |
||
654 | expected_data = "8 is greater than the maximum of 7" |
||
655 | |||
656 | assert response.status_code == 400 |
||
657 | assert expected_data in current_data["description"], expected_data |
||
658 | |||
659 | @patch("napps.kytos.mef_eline.models.evc.EVC.deploy") |
||
660 | @patch("napps.kytos.mef_eline.scheduler.Scheduler.add") |
||
661 | @patch("napps.kytos.mef_eline.main.Main._uni_from_dict") |
||
662 | @patch("napps.kytos.mef_eline.controllers.ELineController.upsert_evc") |
||
663 | @patch("napps.kytos.mef_eline.models.evc.EVC._validate") |
||
664 | @patch("napps.kytos.mef_eline.main.EVC.as_dict") |
||
665 | def test_create_circuit_already_enabled(self, *args): |
||
666 | """Test create an already created circuit.""" |
||
667 | # pylint: disable=too-many-locals |
||
668 | ( |
||
669 | evc_as_dict_mock, |
||
670 | validate_mock, |
||
671 | mongo_controller_upsert_mock, |
||
672 | uni_from_dict_mock, |
||
673 | sched_add_mock, |
||
674 | evc_deploy_mock, |
||
675 | ) = args |
||
676 | |||
677 | validate_mock.return_value = True |
||
678 | mongo_controller_upsert_mock.return_value = True |
||
679 | sched_add_mock.return_value = True |
||
680 | evc_deploy_mock.return_value = True |
||
681 | uni1 = create_autospec(UNI) |
||
682 | uni2 = create_autospec(UNI) |
||
683 | uni1.interface = create_autospec(Interface) |
||
684 | uni2.interface = create_autospec(Interface) |
||
685 | uni1.interface.switch = "00:00:00:00:00:00:00:01" |
||
686 | uni2.interface.switch = "00:00:00:00:00:00:00:02" |
||
687 | uni_from_dict_mock.side_effect = [uni1, uni2, uni1, uni2] |
||
688 | |||
689 | api = self.get_app_test_client(self.napp) |
||
690 | payload = { |
||
691 | "name": "my evc1", |
||
692 | "uni_a": { |
||
693 | "interface_id": "00:00:00:00:00:00:00:01:1", |
||
694 | "tag": {"tag_type": 1, "value": 80}, |
||
695 | }, |
||
696 | "uni_z": { |
||
697 | "interface_id": "00:00:00:00:00:00:00:02:2", |
||
698 | "tag": {"tag_type": 1, "value": 1}, |
||
699 | }, |
||
700 | "dynamic_backup_path": True, |
||
701 | } |
||
702 | |||
703 | evc_as_dict_mock.return_value = payload |
||
704 | response = api.post( |
||
705 | f"{self.server_name_url}/v2/evc/", |
||
706 | data=json.dumps(payload), |
||
707 | content_type="application/json", |
||
708 | ) |
||
709 | self.assertEqual(201, response.status_code) |
||
710 | |||
711 | response = api.post( |
||
712 | f"{self.server_name_url}/v2/evc/", |
||
713 | data=json.dumps(payload), |
||
714 | content_type="application/json", |
||
715 | ) |
||
716 | current_data = json.loads(response.data) |
||
717 | expected_data = "The EVC already exists." |
||
718 | self.assertEqual(current_data["description"], expected_data) |
||
719 | self.assertEqual(409, response.status_code) |
||
720 | |||
721 | @patch("napps.kytos.mef_eline.main.Main._uni_from_dict") |
||
722 | def test_create_circuit_case_5(self, uni_from_dict_mock): |
||
723 | """Test when neither primary path nor dynamic_backup_path is set.""" |
||
724 | api = self.get_app_test_client(self.napp) |
||
725 | url = f"{self.server_name_url}/v2/evc/" |
||
726 | uni1 = create_autospec(UNI) |
||
727 | uni2 = create_autospec(UNI) |
||
728 | uni1.interface = create_autospec(Interface) |
||
729 | uni2.interface = create_autospec(Interface) |
||
730 | uni1.interface.switch = "00:00:00:00:00:00:00:01" |
||
731 | uni2.interface.switch = "00:00:00:00:00:00:00:02" |
||
732 | uni_from_dict_mock.side_effect = [uni1, uni2, uni1, uni2] |
||
733 | |||
734 | payload = { |
||
735 | "name": "my evc1", |
||
736 | "frequency": "* * * * *", |
||
737 | "uni_a": { |
||
738 | "interface_id": "00:00:00:00:00:00:00:01:1", |
||
739 | "tag": {"tag_type": 1, "value": 80}, |
||
740 | }, |
||
741 | "uni_z": { |
||
742 | "interface_id": "00:00:00:00:00:00:00:02:2", |
||
743 | "tag": {"tag_type": 1, "value": 1}, |
||
744 | }, |
||
745 | } |
||
746 | |||
747 | response = api.post( |
||
748 | url, data=json.dumps(payload), content_type="application/json" |
||
749 | ) |
||
750 | current_data = json.loads(response.data) |
||
751 | expected_data = "The EVC must have a primary path " |
||
752 | expected_data += "or allow dynamic paths." |
||
753 | self.assertEqual(400, response.status_code, response.data) |
||
754 | self.assertEqual(current_data["description"], expected_data) |
||
755 | |||
756 | def test_redeploy_evc(self): |
||
757 | """Test endpoint to redeploy an EVC.""" |
||
758 | evc1 = MagicMock() |
||
759 | evc1.is_enabled.return_value = True |
||
760 | self.napp.circuits = {"1": evc1, "2": MagicMock()} |
||
761 | api = self.get_app_test_client(self.napp) |
||
762 | url = f"{self.server_name_url}/v2/evc/1/redeploy" |
||
763 | response = api.patch(url) |
||
764 | self.assertEqual(response.status_code, 202, response.data) |
||
765 | |||
766 | def test_redeploy_evc_disabled(self): |
||
767 | """Test endpoint to redeploy an EVC.""" |
||
768 | evc1 = MagicMock() |
||
769 | evc1.is_enabled.return_value = False |
||
770 | self.napp.circuits = {"1": evc1, "2": MagicMock()} |
||
771 | api = self.get_app_test_client(self.napp) |
||
772 | url = f"{self.server_name_url}/v2/evc/1/redeploy" |
||
773 | response = api.patch(url) |
||
774 | self.assertEqual(response.status_code, 409, response.data) |
||
775 | |||
776 | def test_redeploy_evc_deleted(self): |
||
777 | """Test endpoint to redeploy an EVC.""" |
||
778 | evc1 = MagicMock() |
||
779 | evc1.is_enabled.return_value = True |
||
780 | self.napp.circuits = {"1": evc1, "2": MagicMock()} |
||
781 | api = self.get_app_test_client(self.napp) |
||
782 | url = f"{self.server_name_url}/v2/evc/3/redeploy" |
||
783 | response = api.patch(url) |
||
784 | self.assertEqual(response.status_code, 404, response.data) |
||
785 | |||
786 | def test_list_schedules__no_data_stored(self): |
||
787 | """Test if list circuits return all circuits stored.""" |
||
788 | self.napp.mongo_controller.get_circuits.return_value = {"circuits": {}} |
||
789 | |||
790 | api = self.get_app_test_client(self.napp) |
||
791 | url = f"{self.server_name_url}/v2/evc/schedule" |
||
792 | |||
793 | response = api.get(url) |
||
794 | expected_result = {} |
||
795 | |||
796 | self.assertEqual(response.status_code, 200, response.data) |
||
797 | self.assertEqual(json.loads(response.data), expected_result) |
||
798 | |||
799 | def _add_mongodb_schedule_data(self, data_mock): |
||
800 | """Add schedule data to mongodb mock object.""" |
||
801 | circuits = {"circuits": {}} |
||
802 | payload_1 = { |
||
803 | "id": "aa:aa:aa", |
||
804 | "name": "my evc1", |
||
805 | "uni_a": { |
||
806 | "interface_id": "00:00:00:00:00:00:00:01:1", |
||
807 | "tag": {"tag_type": 1, "value": 80}, |
||
808 | }, |
||
809 | "uni_z": { |
||
810 | "interface_id": "00:00:00:00:00:00:00:02:2", |
||
811 | "tag": {"tag_type": 1, "value": 1}, |
||
812 | }, |
||
813 | "circuit_scheduler": [ |
||
814 | {"id": "1", "frequency": "* * * * *", "action": "create"}, |
||
815 | {"id": "2", "frequency": "1 * * * *", "action": "remove"}, |
||
816 | ], |
||
817 | } |
||
818 | circuits["circuits"].update({"aa:aa:aa": payload_1}) |
||
819 | payload_2 = { |
||
820 | "id": "bb:bb:bb", |
||
821 | "name": "my second evc2", |
||
822 | "uni_a": { |
||
823 | "interface_id": "00:00:00:00:00:00:00:01:2", |
||
824 | "tag": {"tag_type": 1, "value": 90}, |
||
825 | }, |
||
826 | "uni_z": { |
||
827 | "interface_id": "00:00:00:00:00:00:00:03:2", |
||
828 | "tag": {"tag_type": 1, "value": 100}, |
||
829 | }, |
||
830 | "circuit_scheduler": [ |
||
831 | {"id": "3", "frequency": "1 * * * *", "action": "create"}, |
||
832 | {"id": "4", "frequency": "2 * * * *", "action": "remove"}, |
||
833 | ], |
||
834 | } |
||
835 | circuits["circuits"].update({"bb:bb:bb": payload_2}) |
||
836 | payload_3 = { |
||
837 | "id": "cc:cc:cc", |
||
838 | "name": "my third evc3", |
||
839 | "uni_a": { |
||
840 | "interface_id": "00:00:00:00:00:00:00:03:1", |
||
841 | "tag": {"tag_type": 1, "value": 90}, |
||
842 | }, |
||
843 | "uni_z": { |
||
844 | "interface_id": "00:00:00:00:00:00:00:04:2", |
||
845 | "tag": {"tag_type": 1, "value": 100}, |
||
846 | }, |
||
847 | } |
||
848 | circuits["circuits"].update({"cc:cc:cc": payload_3}) |
||
849 | # Add one circuit to the mongodb. |
||
850 | data_mock.return_value = circuits |
||
851 | |||
852 | def test_list_schedules_from_mongodb(self): |
||
853 | """Test if list circuits return specific circuits stored.""" |
||
854 | self._add_mongodb_schedule_data( |
||
855 | self.napp.mongo_controller.get_circuits |
||
856 | ) |
||
857 | |||
858 | api = self.get_app_test_client(self.napp) |
||
859 | url = f"{self.server_name_url}/v2/evc/schedule" |
||
860 | |||
861 | # Call URL |
||
862 | response = api.get(url) |
||
863 | # Expected JSON data from response |
||
864 | expected = [ |
||
865 | { |
||
866 | "circuit_id": "aa:aa:aa", |
||
867 | "schedule": { |
||
868 | "action": "create", |
||
869 | "frequency": "* * * * *", |
||
870 | "id": "1", |
||
871 | }, |
||
872 | "schedule_id": "1", |
||
873 | }, |
||
874 | { |
||
875 | "circuit_id": "aa:aa:aa", |
||
876 | "schedule": { |
||
877 | "action": "remove", |
||
878 | "frequency": "1 * * * *", |
||
879 | "id": "2", |
||
880 | }, |
||
881 | "schedule_id": "2", |
||
882 | }, |
||
883 | { |
||
884 | "circuit_id": "bb:bb:bb", |
||
885 | "schedule": { |
||
886 | "action": "create", |
||
887 | "frequency": "1 * * * *", |
||
888 | "id": "3", |
||
889 | }, |
||
890 | "schedule_id": "3", |
||
891 | }, |
||
892 | { |
||
893 | "circuit_id": "bb:bb:bb", |
||
894 | "schedule": { |
||
895 | "action": "remove", |
||
896 | "frequency": "2 * * * *", |
||
897 | "id": "4", |
||
898 | }, |
||
899 | "schedule_id": "4", |
||
900 | }, |
||
901 | ] |
||
902 | |||
903 | self.assertEqual(response.status_code, 200, response.data) |
||
904 | self.assertEqual(expected, json.loads(response.data)) |
||
905 | |||
906 | def test_get_specific_schedule_from_mongodb(self): |
||
907 | """Test get schedules from a circuit.""" |
||
908 | self._add_mongodb_schedule_data( |
||
909 | self.napp.mongo_controller.get_circuits |
||
910 | ) |
||
911 | |||
912 | requested_circuit_id = "bb:bb:bb" |
||
913 | evc = self.napp.mongo_controller.get_circuits() |
||
914 | evc = evc["circuits"][requested_circuit_id] |
||
915 | self.napp.mongo_controller.get_circuit.return_value = evc |
||
916 | api = self.get_app_test_client(self.napp) |
||
917 | url = f"{self.server_name_url}/v2/evc/{requested_circuit_id}" |
||
918 | |||
919 | # Call URL |
||
920 | response = api.get(url) |
||
921 | |||
922 | # Expected JSON data from response |
||
923 | expected = [ |
||
924 | {"action": "create", "frequency": "1 * * * *", "id": "3"}, |
||
925 | {"action": "remove", "frequency": "2 * * * *", "id": "4"}, |
||
926 | ] |
||
927 | |||
928 | self.assertEqual(response.status_code, 200) |
||
929 | self.assertEqual( |
||
930 | expected, json.loads(response.data)["circuit_scheduler"] |
||
931 | ) |
||
932 | |||
933 | def test_get_specific_schedules_from_mongodb_not_found(self): |
||
934 | """Test get specific schedule ID that does not exist.""" |
||
935 | requested_id = "blah" |
||
936 | api = self.get_app_test_client(self.napp) |
||
937 | self.napp.mongo_controller.get_circuit.return_value = None |
||
938 | url = f"{self.server_name_url}/v2/evc/{requested_id}" |
||
939 | |||
940 | # Call URL |
||
941 | response = api.get(url) |
||
942 | |||
943 | expected = "circuit_id blah not found" |
||
944 | # Assert response not found |
||
945 | self.assertEqual(response.status_code, 404, response.data) |
||
946 | self.assertEqual(expected, json.loads(response.data)["description"]) |
||
947 | |||
948 | def _uni_from_dict_side_effect(self, uni_dict): |
||
949 | interface_id = uni_dict.get("interface_id") |
||
950 | tag_dict = uni_dict.get("tag") |
||
951 | interface = Interface(interface_id, "0", MagicMock(id="1")) |
||
952 | return UNI(interface, tag_dict) |
||
953 | |||
954 | @patch("apscheduler.schedulers.background.BackgroundScheduler.add_job") |
||
955 | @patch("napps.kytos.mef_eline.scheduler.Scheduler.add") |
||
956 | @patch("napps.kytos.mef_eline.main.Main._uni_from_dict") |
||
957 | @patch("napps.kytos.mef_eline.controllers.ELineController.upsert_evc") |
||
958 | @patch("napps.kytos.mef_eline.main.EVC.as_dict") |
||
959 | @patch("napps.kytos.mef_eline.models.evc.EVC._validate") |
||
960 | def test_create_schedule(self, *args): # pylint: disable=too-many-locals |
||
961 | """Test create a circuit schedule.""" |
||
962 | ( |
||
963 | validate_mock, |
||
964 | evc_as_dict_mock, |
||
965 | mongo_controller_upsert_mock, |
||
966 | uni_from_dict_mock, |
||
967 | sched_add_mock, |
||
968 | scheduler_add_job_mock, |
||
969 | ) = args |
||
970 | |||
971 | validate_mock.return_value = True |
||
972 | mongo_controller_upsert_mock.return_value = True |
||
973 | uni_from_dict_mock.side_effect = self._uni_from_dict_side_effect |
||
974 | evc_as_dict_mock.return_value = {} |
||
975 | sched_add_mock.return_value = True |
||
976 | |||
977 | self._add_mongodb_schedule_data( |
||
978 | self.napp.mongo_controller.get_circuits |
||
979 | ) |
||
980 | |||
981 | requested_id = "bb:bb:bb" |
||
982 | api = self.get_app_test_client(self.napp) |
||
983 | url = f"{self.server_name_url}/v2/evc/schedule/" |
||
984 | |||
985 | payload = { |
||
986 | "circuit_id": requested_id, |
||
987 | "schedule": {"frequency": "1 * * * *", "action": "create"}, |
||
988 | } |
||
989 | |||
990 | # Call URL |
||
991 | response = api.post( |
||
992 | url, data=json.dumps(payload), content_type="application/json" |
||
993 | ) |
||
994 | |||
995 | response_json = json.loads(response.data) |
||
996 | |||
997 | self.assertEqual(response.status_code, 201, response.data) |
||
998 | scheduler_add_job_mock.assert_called_once() |
||
999 | mongo_controller_upsert_mock.assert_called_once() |
||
1000 | self.assertEqual( |
||
1001 | payload["schedule"]["frequency"], response_json["frequency"] |
||
1002 | ) |
||
1003 | self.assertEqual( |
||
1004 | payload["schedule"]["action"], response_json["action"] |
||
1005 | ) |
||
1006 | self.assertIsNotNone(response_json["id"]) |
||
1007 | |||
1008 | # Case 2: there is no schedule |
||
1009 | payload = { |
||
1010 | "circuit_id": "cc:cc:cc", |
||
1011 | "schedule": { |
||
1012 | "frequency": "1 * * * *", |
||
1013 | "action": "create" |
||
1014 | } |
||
1015 | } |
||
1016 | response = api.post(url, data=json.dumps(payload), |
||
1017 | content_type='application/json') |
||
1018 | self.assertEqual(response.status_code, 201) |
||
1019 | |||
1020 | def test_create_schedule_invalid_request(self): |
||
1021 | """Test create schedule API with invalid request.""" |
||
1022 | evc1 = MagicMock() |
||
1023 | self.napp.circuits = {'bb:bb:bb': evc1} |
||
1024 | api = self.get_app_test_client(self.napp) |
||
1025 | url = f'{self.server_name_url}/v2/evc/schedule/' |
||
1026 | |||
1027 | # case 1: empty post |
||
1028 | response = api.post(url, data="") |
||
1029 | self.assertEqual(response.status_code, 415) |
||
1030 | |||
1031 | # case 2: content-type not specified |
||
1032 | payload = { |
||
1033 | "circuit_id": "bb:bb:bb", |
||
1034 | "schedule": { |
||
1035 | "frequency": "1 * * * *", |
||
1036 | "action": "create" |
||
1037 | } |
||
1038 | } |
||
1039 | response = api.post(url, data=json.dumps(payload)) |
||
1040 | self.assertEqual(response.status_code, 415) |
||
1041 | |||
1042 | # case 3: not a dictionary |
||
1043 | payload = [] |
||
1044 | response = api.post(url, data=json.dumps(payload), |
||
1045 | content_type='application/json') |
||
1046 | self.assertEqual(response.status_code, 400) |
||
1047 | |||
1048 | # case 4: missing circuit id |
||
1049 | payload = { |
||
1050 | "schedule": { |
||
1051 | "frequency": "1 * * * *", |
||
1052 | "action": "create" |
||
1053 | } |
||
1054 | } |
||
1055 | response = api.post(url, data=json.dumps(payload), |
||
1056 | content_type='application/json') |
||
1057 | self.assertEqual(response.status_code, 400) |
||
1058 | |||
1059 | # case 5: missing schedule |
||
1060 | payload = { |
||
1061 | "circuit_id": "bb:bb:bb" |
||
1062 | } |
||
1063 | response = api.post(url, data=json.dumps(payload), |
||
1064 | content_type='application/json') |
||
1065 | self.assertEqual(response.status_code, 400) |
||
1066 | |||
1067 | # case 6: invalid circuit |
||
1068 | payload = { |
||
1069 | "circuit_id": "xx:xx:xx", |
||
1070 | "schedule": { |
||
1071 | "frequency": "1 * * * *", |
||
1072 | "action": "create" |
||
1073 | } |
||
1074 | } |
||
1075 | response = api.post(url, data=json.dumps(payload), |
||
1076 | content_type='application/json') |
||
1077 | self.assertEqual(response.status_code, 404) |
||
1078 | |||
1079 | # case 7: archived or deleted evc |
||
1080 | evc1.archived.return_value = True |
||
1081 | payload = { |
||
1082 | "circuit_id": "bb:bb:bb", |
||
1083 | "schedule": { |
||
1084 | "frequency": "1 * * * *", |
||
1085 | "action": "create" |
||
1086 | } |
||
1087 | } |
||
1088 | response = api.post(url, data=json.dumps(payload), |
||
1089 | content_type='application/json') |
||
1090 | self.assertEqual(response.status_code, 403) |
||
1091 | |||
1092 | @patch('apscheduler.schedulers.background.BackgroundScheduler.remove_job') |
||
1093 | @patch('napps.kytos.mef_eline.scheduler.Scheduler.add') |
||
1094 | @patch('napps.kytos.mef_eline.main.Main._uni_from_dict') |
||
1095 | @patch("napps.kytos.mef_eline.controllers.ELineController.upsert_evc") |
||
1096 | @patch('napps.kytos.mef_eline.main.EVC.as_dict') |
||
1097 | @patch('napps.kytos.mef_eline.models.evc.EVC._validate') |
||
1098 | def test_update_schedule(self, *args): # pylint: disable=too-many-locals |
||
1099 | """Test create a circuit schedule.""" |
||
1100 | ( |
||
1101 | validate_mock, |
||
1102 | evc_as_dict_mock, |
||
1103 | mongo_controller_upsert_mock, |
||
1104 | uni_from_dict_mock, |
||
1105 | sched_add_mock, |
||
1106 | scheduler_remove_job_mock, |
||
1107 | ) = args |
||
1108 | |||
1109 | mongo_payload_1 = { |
||
1110 | "circuits": { |
||
1111 | "aa:aa:aa": { |
||
1112 | "id": "aa:aa:aa", |
||
1113 | "name": "my evc1", |
||
1114 | "uni_a": { |
||
1115 | "interface_id": "00:00:00:00:00:00:00:01:1", |
||
1116 | "tag": {"tag_type": 1, "value": 80}, |
||
1117 | }, |
||
1118 | "uni_z": { |
||
1119 | "interface_id": "00:00:00:00:00:00:00:02:2", |
||
1120 | "tag": {"tag_type": 1, "value": 1}, |
||
1121 | }, |
||
1122 | "circuit_scheduler": [ |
||
1123 | { |
||
1124 | "id": "1", |
||
1125 | "frequency": "* * * * *", |
||
1126 | "action": "create" |
||
1127 | } |
||
1128 | ], |
||
1129 | } |
||
1130 | } |
||
1131 | } |
||
1132 | |||
1133 | validate_mock.return_value = True |
||
1134 | mongo_controller_upsert_mock.return_value = True |
||
1135 | sched_add_mock.return_value = True |
||
1136 | uni_from_dict_mock.side_effect = ["uni_a", "uni_z"] |
||
1137 | evc_as_dict_mock.return_value = {} |
||
1138 | self.napp.mongo_controller.get_circuits.return_value = mongo_payload_1 |
||
1139 | scheduler_remove_job_mock.return_value = True |
||
1140 | |||
1141 | requested_schedule_id = "1" |
||
1142 | api = self.get_app_test_client(self.napp) |
||
1143 | url = f"{self.server_name_url}/v2/evc/schedule/{requested_schedule_id}" |
||
1144 | |||
1145 | payload = {"frequency": "*/1 * * * *", "action": "create"} |
||
1146 | |||
1147 | # Call URL |
||
1148 | response = api.patch( |
||
1149 | url, data=json.dumps(payload), content_type="application/json" |
||
1150 | ) |
||
1151 | |||
1152 | response_json = json.loads(response.data) |
||
1153 | |||
1154 | self.assertEqual(response.status_code, 200, response.data) |
||
1155 | scheduler_remove_job_mock.assert_called_once() |
||
1156 | mongo_controller_upsert_mock.assert_called_once() |
||
1157 | self.assertEqual(payload["frequency"], response_json["frequency"]) |
||
1158 | self.assertEqual(payload["action"], response_json["action"]) |
||
1159 | self.assertIsNotNone(response_json["id"]) |
||
1160 | |||
1161 | @patch('napps.kytos.mef_eline.main.Main._find_evc_by_schedule_id') |
||
1162 | def test_update_no_schedule(self, find_evc_by_schedule_id_mock): |
||
1163 | """Test update a circuit schedule.""" |
||
1164 | api = self.get_app_test_client(self.napp) |
||
1165 | url = f"{self.server_name_url}/v2/evc/schedule/1" |
||
1166 | payload = {"frequency": "*/1 * * * *", "action": "create"} |
||
1167 | |||
1168 | find_evc_by_schedule_id_mock.return_value = None, None |
||
1169 | |||
1170 | response = api.patch( |
||
1171 | url, data=json.dumps(payload), content_type="application/json" |
||
1172 | ) |
||
1173 | |||
1174 | self.assertEqual(response.status_code, 404) |
||
1175 | |||
1176 | @patch("napps.kytos.mef_eline.scheduler.Scheduler.add") |
||
1177 | @patch("napps.kytos.mef_eline.main.Main._uni_from_dict") |
||
1178 | @patch("napps.kytos.mef_eline.main.EVC.as_dict") |
||
1179 | @patch("napps.kytos.mef_eline.models.evc.EVC._validate") |
||
1180 | def test_update_schedule_archived(self, *args): |
||
1181 | """Test create a circuit schedule.""" |
||
1182 | # pylint: disable=too-many-locals |
||
1183 | ( |
||
1184 | validate_mock, |
||
1185 | evc_as_dict_mock, |
||
1186 | uni_from_dict_mock, |
||
1187 | sched_add_mock, |
||
1188 | ) = args |
||
1189 | |||
1190 | mongo_payload_1 = { |
||
1191 | "circuits": { |
||
1192 | "aa:aa:aa": { |
||
1193 | "id": "aa:aa:aa", |
||
1194 | "name": "my evc1", |
||
1195 | "archived": True, |
||
1196 | "circuit_scheduler": [ |
||
1197 | { |
||
1198 | "id": "1", |
||
1199 | "frequency": "* * * * *", |
||
1200 | "action": "create" |
||
1201 | } |
||
1202 | ], |
||
1203 | } |
||
1204 | } |
||
1205 | } |
||
1206 | |||
1207 | validate_mock.return_value = True |
||
1208 | sched_add_mock.return_value = True |
||
1209 | uni_from_dict_mock.side_effect = ["uni_a", "uni_z"] |
||
1210 | evc_as_dict_mock.return_value = {} |
||
1211 | self.napp.mongo_controller.get_circuits.return_value = mongo_payload_1 |
||
1212 | |||
1213 | requested_schedule_id = "1" |
||
1214 | api = self.get_app_test_client(self.napp) |
||
1215 | url = f"{self.server_name_url}/v2/evc/schedule/{requested_schedule_id}" |
||
1216 | |||
1217 | payload = {"frequency": "*/1 * * * *", "action": "create"} |
||
1218 | |||
1219 | # Call URL |
||
1220 | response = api.patch( |
||
1221 | url, data=json.dumps(payload), content_type="application/json" |
||
1222 | ) |
||
1223 | |||
1224 | self.assertEqual(response.status_code, 403, response.data) |
||
1225 | |||
1226 | @patch("apscheduler.schedulers.background.BackgroundScheduler.remove_job") |
||
1227 | @patch("napps.kytos.mef_eline.main.Main._uni_from_dict") |
||
1228 | @patch("napps.kytos.mef_eline.controllers.ELineController.upsert_evc") |
||
1229 | @patch("napps.kytos.mef_eline.main.EVC.as_dict") |
||
1230 | @patch("napps.kytos.mef_eline.models.evc.EVC._validate") |
||
1231 | def test_delete_schedule(self, *args): |
||
1232 | """Test create a circuit schedule.""" |
||
1233 | ( |
||
1234 | validate_mock, |
||
1235 | evc_as_dict_mock, |
||
1236 | mongo_controller_upsert_mock, |
||
1237 | uni_from_dict_mock, |
||
1238 | scheduler_remove_job_mock, |
||
1239 | ) = args |
||
1240 | |||
1241 | mongo_payload_1 = { |
||
1242 | "circuits": { |
||
1243 | "2": { |
||
1244 | "id": "2", |
||
1245 | "name": "my evc1", |
||
1246 | "uni_a": { |
||
1247 | "interface_id": "00:00:00:00:00:00:00:01:1", |
||
1248 | "tag": {"tag_type": 1, "value": 80}, |
||
1249 | }, |
||
1250 | "uni_z": { |
||
1251 | "interface_id": "00:00:00:00:00:00:00:02:2", |
||
1252 | "tag": {"tag_type": 1, "value": 1}, |
||
1253 | }, |
||
1254 | "circuit_scheduler": [ |
||
1255 | { |
||
1256 | "id": "1", |
||
1257 | "frequency": "* * * * *", |
||
1258 | "action": "create" |
||
1259 | } |
||
1260 | ], |
||
1261 | } |
||
1262 | } |
||
1263 | } |
||
1264 | validate_mock.return_value = True |
||
1265 | mongo_controller_upsert_mock.return_value = True |
||
1266 | uni_from_dict_mock.side_effect = ["uni_a", "uni_z"] |
||
1267 | evc_as_dict_mock.return_value = {} |
||
1268 | self.napp.mongo_controller.get_circuits.return_value = mongo_payload_1 |
||
1269 | scheduler_remove_job_mock.return_value = True |
||
1270 | |||
1271 | requested_schedule_id = "1" |
||
1272 | api = self.get_app_test_client(self.napp) |
||
1273 | url = f"{self.server_name_url}/v2/evc/schedule/{requested_schedule_id}" |
||
1274 | |||
1275 | # Call URL |
||
1276 | response = api.delete(url) |
||
1277 | |||
1278 | self.assertEqual(response.status_code, 200, response.data) |
||
1279 | scheduler_remove_job_mock.assert_called_once() |
||
1280 | mongo_controller_upsert_mock.assert_called_once() |
||
1281 | self.assertIn("Schedule removed", f"{response.data}") |
||
1282 | |||
1283 | @patch("napps.kytos.mef_eline.main.Main._uni_from_dict") |
||
1284 | @patch("napps.kytos.mef_eline.main.EVC.as_dict") |
||
1285 | @patch("napps.kytos.mef_eline.models.evc.EVC._validate") |
||
1286 | def test_delete_schedule_archived(self, *args): |
||
1287 | """Test create a circuit schedule.""" |
||
1288 | ( |
||
1289 | validate_mock, |
||
1290 | evc_as_dict_mock, |
||
1291 | uni_from_dict_mock, |
||
1292 | ) = args |
||
1293 | |||
1294 | mongo_payload_1 = { |
||
1295 | "circuits": { |
||
1296 | "2": { |
||
1297 | "id": "2", |
||
1298 | "name": "my evc1", |
||
1299 | "archived": True, |
||
1300 | "circuit_scheduler": [ |
||
1301 | { |
||
1302 | "id": "1", |
||
1303 | "frequency": "* * * * *", |
||
1304 | "action": "create" |
||
1305 | } |
||
1306 | ], |
||
1307 | } |
||
1308 | } |
||
1309 | } |
||
1310 | |||
1311 | validate_mock.return_value = True |
||
1312 | uni_from_dict_mock.side_effect = ["uni_a", "uni_z"] |
||
1313 | evc_as_dict_mock.return_value = {} |
||
1314 | self.napp.mongo_controller.get_circuits.return_value = mongo_payload_1 |
||
1315 | |||
1316 | requested_schedule_id = "1" |
||
1317 | api = self.get_app_test_client(self.napp) |
||
1318 | url = f"{self.server_name_url}/v2/evc/schedule/{requested_schedule_id}" |
||
1319 | |||
1320 | # Call URL |
||
1321 | response = api.delete(url) |
||
1322 | |||
1323 | self.assertEqual(response.status_code, 403, response.data) |
||
1324 | |||
1325 | @patch('napps.kytos.mef_eline.main.Main._find_evc_by_schedule_id') |
||
1326 | def test_delete_schedule_not_found(self, mock_find_evc_by_sched): |
||
1327 | """Test delete a circuit schedule - unexisting.""" |
||
1328 | mock_find_evc_by_sched.return_value = (None, False) |
||
1329 | api = self.get_app_test_client(self.napp) |
||
1330 | url = f'{self.server_name_url}/v2/evc/schedule/1' |
||
1331 | response = api.delete(url) |
||
1332 | self.assertEqual(response.status_code, 404) |
||
1333 | |||
1334 | def test_get_evcs_by_svc_level(self) -> None: |
||
1335 | """Test get_evcs_by_svc_level.""" |
||
1336 | levels = [1, 2, 4, 2, 7] |
||
1337 | evcs = {i: MagicMock(service_level=v, creation_time=1) |
||
1338 | for i, v in enumerate(levels)} |
||
1339 | self.napp.circuits = evcs |
||
1340 | expected_levels = sorted(levels, reverse=True) |
||
1341 | evcs_by_level = self.napp.get_evcs_by_svc_level() |
||
1342 | assert evcs_by_level |
||
1343 | |||
1344 | for evc, exp_level in zip(evcs_by_level, expected_levels): |
||
1345 | assert evc.service_level == exp_level |
||
1346 | |||
1347 | evcs = {i: MagicMock(service_level=1, creation_time=i) |
||
1348 | for i in reversed(range(2))} |
||
1349 | self.napp.circuits = evcs |
||
1350 | evcs_by_level = self.napp.get_evcs_by_svc_level() |
||
1351 | for i in range(2): |
||
1352 | assert evcs_by_level[i].creation_time == i |
||
1353 | |||
1354 | def test_get_circuit_not_found(self): |
||
1355 | """Test /v2/evc/<circuit_id> 404.""" |
||
1356 | self.napp.mongo_controller.get_circuit.return_value = None |
||
1357 | api = self.get_app_test_client(self.napp) |
||
1358 | url = f'{self.server_name_url}/v2/evc/1234' |
||
1359 | response = api.get(url) |
||
1360 | self.assertEqual(response.status_code, 404) |
||
1361 | |||
1362 | @patch('requests.post') |
||
1363 | @patch('napps.kytos.mef_eline.scheduler.Scheduler.add') |
||
1364 | @patch("napps.kytos.mef_eline.controllers.ELineController.upsert_evc") |
||
1365 | @patch('napps.kytos.mef_eline.models.evc.EVC._validate') |
||
1366 | @patch('kytos.core.Controller.get_interface_by_id') |
||
1367 | @patch('napps.kytos.mef_eline.models.path.Path.is_valid') |
||
1368 | @patch('napps.kytos.mef_eline.models.evc.EVCDeploy.deploy') |
||
1369 | @patch('napps.kytos.mef_eline.main.Main._uni_from_dict') |
||
1370 | @patch('napps.kytos.mef_eline.main.EVC.as_dict') |
||
1371 | def test_update_circuit(self, *args): |
||
1372 | """Test update a circuit circuit.""" |
||
1373 | # pylint: disable=too-many-locals,duplicate-code |
||
1374 | ( |
||
1375 | evc_as_dict_mock, |
||
1376 | uni_from_dict_mock, |
||
1377 | evc_deploy, |
||
1378 | _, |
||
1379 | interface_by_id_mock, |
||
1380 | _, |
||
1381 | _, |
||
1382 | _, |
||
1383 | requests_mock, |
||
1384 | ) = args |
||
1385 | |||
1386 | interface_by_id_mock.return_value = get_uni_mocked().interface |
||
1387 | unis = [ |
||
1388 | get_uni_mocked(switch_dpid="00:00:00:00:00:00:00:01"), |
||
1389 | get_uni_mocked(switch_dpid="00:00:00:00:00:00:00:02"), |
||
1390 | ] |
||
1391 | uni_from_dict_mock.side_effect = 2 * unis |
||
1392 | |||
1393 | response = MagicMock() |
||
1394 | response.status_code = 201 |
||
1395 | requests_mock.return_value = response |
||
1396 | |||
1397 | api = self.get_app_test_client(self.napp) |
||
1398 | payloads = [ |
||
1399 | { |
||
1400 | "name": "my evc1", |
||
1401 | "uni_a": { |
||
1402 | "interface_id": "00:00:00:00:00:00:00:01:1", |
||
1403 | "tag": {"tag_type": 1, "value": 80}, |
||
1404 | }, |
||
1405 | "uni_z": { |
||
1406 | "interface_id": "00:00:00:00:00:00:00:02:2", |
||
1407 | "tag": {"tag_type": 1, "value": 1}, |
||
1408 | }, |
||
1409 | "dynamic_backup_path": True, |
||
1410 | }, |
||
1411 | { |
||
1412 | "primary_path": [ |
||
1413 | { |
||
1414 | "endpoint_a": {"id": "00:00:00:00:00:00:00:01:1"}, |
||
1415 | "endpoint_b": {"id": "00:00:00:00:00:00:00:02:2"}, |
||
1416 | } |
||
1417 | ] |
||
1418 | }, |
||
1419 | { |
||
1420 | "sb_priority": 3 |
||
1421 | }, |
||
1422 | { |
||
1423 | # It works only with 'enable' and not with 'enabled' |
||
1424 | "enable": True |
||
1425 | }, |
||
1426 | { |
||
1427 | "name": "my evc1", |
||
1428 | "active": True, |
||
1429 | "enable": True, |
||
1430 | "uni_a": { |
||
1431 | "interface_id": "00:00:00:00:00:00:00:01:1", |
||
1432 | "tag": { |
||
1433 | "tag_type": 1, |
||
1434 | "value": 80 |
||
1435 | } |
||
1436 | }, |
||
1437 | "uni_z": { |
||
1438 | "interface_id": "00:00:00:00:00:00:00:02:2", |
||
1439 | "tag": { |
||
1440 | "tag_type": 1, |
||
1441 | "value": 1 |
||
1442 | } |
||
1443 | }, |
||
1444 | "priority": 3, |
||
1445 | "bandwidth": 1000, |
||
1446 | "dynamic_backup_path": True |
||
1447 | } |
||
1448 | ] |
||
1449 | |||
1450 | evc_as_dict_mock.return_value = payloads[0] |
||
1451 | response = api.post( |
||
1452 | f"{self.server_name_url}/v2/evc/", |
||
1453 | data=json.dumps(payloads[0]), |
||
1454 | content_type="application/json", |
||
1455 | ) |
||
1456 | self.assertEqual(201, response.status_code) |
||
1457 | |||
1458 | evc_deploy.reset_mock() |
||
1459 | evc_as_dict_mock.return_value = payloads[1] |
||
1460 | current_data = json.loads(response.data) |
||
1461 | circuit_id = current_data["circuit_id"] |
||
1462 | response = api.patch( |
||
1463 | f"{self.server_name_url}/v2/evc/{circuit_id}", |
||
1464 | data=json.dumps(payloads[1]), |
||
1465 | content_type="application/json", |
||
1466 | ) |
||
1467 | # evc_deploy.assert_called_once() |
||
1468 | self.assertEqual(200, response.status_code) |
||
1469 | |||
1470 | evc_deploy.reset_mock() |
||
1471 | evc_as_dict_mock.return_value = payloads[2] |
||
1472 | response = api.patch( |
||
1473 | f"{self.server_name_url}/v2/evc/{circuit_id}", |
||
1474 | data=json.dumps(payloads[2]), |
||
1475 | content_type="application/json", |
||
1476 | ) |
||
1477 | evc_deploy.assert_not_called() |
||
1478 | self.assertEqual(200, response.status_code) |
||
1479 | |||
1480 | evc_deploy.reset_mock() |
||
1481 | evc_as_dict_mock.return_value = payloads[3] |
||
1482 | response = api.patch(f'{self.server_name_url}/v2/evc/{circuit_id}', |
||
1483 | data=json.dumps(payloads[3]), |
||
1484 | content_type='application/json') |
||
1485 | evc_deploy.assert_called_once() |
||
1486 | self.assertEqual(200, response.status_code) |
||
1487 | |||
1488 | evc_deploy.reset_mock() |
||
1489 | response = api.patch(f'{self.server_name_url}/v2/evc/{circuit_id}', |
||
1490 | data='{"priority":5,}', |
||
1491 | content_type='application/json') |
||
1492 | evc_deploy.assert_not_called() |
||
1493 | self.assertEqual(400, response.status_code) |
||
1494 | |||
1495 | evc_deploy.reset_mock() |
||
1496 | response = api.patch( |
||
1497 | f"{self.server_name_url}/v2/evc/{circuit_id}", |
||
1498 | data=json.dumps(payloads[3]), |
||
1499 | content_type="application/json", |
||
1500 | ) |
||
1501 | evc_deploy.assert_called_once() |
||
1502 | self.assertEqual(200, response.status_code) |
||
1503 | |||
1504 | response = api.patch( |
||
1505 | f"{self.server_name_url}/v2/evc/1234", |
||
1506 | data=json.dumps(payloads[1]), |
||
1507 | content_type="application/json", |
||
1508 | ) |
||
1509 | current_data = json.loads(response.data) |
||
1510 | expected_data = "circuit_id 1234 not found" |
||
1511 | self.assertEqual(current_data["description"], expected_data) |
||
1512 | self.assertEqual(404, response.status_code) |
||
1513 | |||
1514 | api.delete(f"{self.server_name_url}/v2/evc/{circuit_id}") |
||
1515 | evc_deploy.reset_mock() |
||
1516 | response = api.patch( |
||
1517 | f"{self.server_name_url}/v2/evc/{circuit_id}", |
||
1518 | data=json.dumps(payloads[1]), |
||
1519 | content_type="application/json", |
||
1520 | ) |
||
1521 | evc_deploy.assert_not_called() |
||
1522 | self.assertEqual(405, response.status_code) |
||
1523 | |||
1524 | View Code Duplication | @patch("napps.kytos.mef_eline.models.evc.EVC.deploy") |
|
1525 | @patch("napps.kytos.mef_eline.scheduler.Scheduler.add") |
||
1526 | @patch("napps.kytos.mef_eline.main.Main._uni_from_dict") |
||
1527 | @patch("napps.kytos.mef_eline.controllers.ELineController.upsert_evc") |
||
1528 | @patch("napps.kytos.mef_eline.models.evc.EVC._validate") |
||
1529 | @patch("napps.kytos.mef_eline.main.EVC.as_dict") |
||
1530 | def test_update_circuit_invalid_json(self, *args): |
||
1531 | """Test update a circuit circuit.""" |
||
1532 | # pylint: disable=too-many-locals |
||
1533 | ( |
||
1534 | evc_as_dict_mock, |
||
1535 | validate_mock, |
||
1536 | mongo_controller_upsert_mock, |
||
1537 | uni_from_dict_mock, |
||
1538 | sched_add_mock, |
||
1539 | evc_deploy_mock, |
||
1540 | ) = args |
||
1541 | |||
1542 | validate_mock.return_value = True |
||
1543 | mongo_controller_upsert_mock.return_value = True |
||
1544 | sched_add_mock.return_value = True |
||
1545 | evc_deploy_mock.return_value = True |
||
1546 | uni1 = create_autospec(UNI) |
||
1547 | uni2 = create_autospec(UNI) |
||
1548 | uni1.interface = create_autospec(Interface) |
||
1549 | uni2.interface = create_autospec(Interface) |
||
1550 | uni1.interface.switch = "00:00:00:00:00:00:00:01" |
||
1551 | uni2.interface.switch = "00:00:00:00:00:00:00:02" |
||
1552 | uni_from_dict_mock.side_effect = [uni1, uni2, uni1, uni2] |
||
1553 | |||
1554 | api = self.get_app_test_client(self.napp) |
||
1555 | payload1 = { |
||
1556 | "name": "my evc1", |
||
1557 | "uni_a": { |
||
1558 | "interface_id": "00:00:00:00:00:00:00:01:1", |
||
1559 | "tag": {"tag_type": 1, "value": 80}, |
||
1560 | }, |
||
1561 | "uni_z": { |
||
1562 | "interface_id": "00:00:00:00:00:00:00:02:2", |
||
1563 | "tag": {"tag_type": 1, "value": 1}, |
||
1564 | }, |
||
1565 | "dynamic_backup_path": True, |
||
1566 | } |
||
1567 | |||
1568 | payload2 = { |
||
1569 | "dynamic_backup_path": False, |
||
1570 | } |
||
1571 | |||
1572 | evc_as_dict_mock.return_value = payload1 |
||
1573 | response = api.post( |
||
1574 | f"{self.server_name_url}/v2/evc/", |
||
1575 | data=json.dumps(payload1), |
||
1576 | content_type="application/json", |
||
1577 | ) |
||
1578 | self.assertEqual(201, response.status_code) |
||
1579 | |||
1580 | evc_as_dict_mock.return_value = payload2 |
||
1581 | current_data = json.loads(response.data) |
||
1582 | circuit_id = current_data["circuit_id"] |
||
1583 | response = api.patch( |
||
1584 | f"{self.server_name_url}/v2/evc/{circuit_id}", |
||
1585 | data=payload2, |
||
1586 | content_type="application/json", |
||
1587 | ) |
||
1588 | current_data = json.loads(response.data) |
||
1589 | expected_data = "The request body is not a well-formed JSON." |
||
1590 | self.assertEqual(current_data["description"], expected_data) |
||
1591 | self.assertEqual(400, response.status_code) |
||
1592 | |||
1593 | @patch("napps.kytos.mef_eline.models.evc.EVC.deploy") |
||
1594 | @patch("napps.kytos.mef_eline.scheduler.Scheduler.add") |
||
1595 | @patch("napps.kytos.mef_eline.main.Main._uni_from_dict") |
||
1596 | @patch("napps.kytos.mef_eline.main.Main._link_from_dict") |
||
1597 | @patch("napps.kytos.mef_eline.controllers.ELineController.upsert_evc") |
||
1598 | @patch("napps.kytos.mef_eline.models.evc.EVC._validate") |
||
1599 | @patch("napps.kytos.mef_eline.main.EVC.as_dict") |
||
1600 | @patch("napps.kytos.mef_eline.models.path.Path.is_valid") |
||
1601 | def test_update_circuit_invalid_path(self, *args): |
||
1602 | """Test update a circuit circuit.""" |
||
1603 | # pylint: disable=too-many-locals |
||
1604 | ( |
||
1605 | is_valid_mock, |
||
1606 | evc_as_dict_mock, |
||
1607 | validate_mock, |
||
1608 | mongo_controller_upsert_mock, |
||
1609 | link_from_dict_mock, |
||
1610 | uni_from_dict_mock, |
||
1611 | sched_add_mock, |
||
1612 | evc_deploy_mock, |
||
1613 | ) = args |
||
1614 | |||
1615 | is_valid_mock.side_effect = InvalidPath("error") |
||
1616 | validate_mock.return_value = True |
||
1617 | mongo_controller_upsert_mock.return_value = True |
||
1618 | sched_add_mock.return_value = True |
||
1619 | evc_deploy_mock.return_value = True |
||
1620 | link_from_dict_mock.return_value = 1 |
||
1621 | uni1 = create_autospec(UNI) |
||
1622 | uni2 = create_autospec(UNI) |
||
1623 | uni1.interface = create_autospec(Interface) |
||
1624 | uni2.interface = create_autospec(Interface) |
||
1625 | uni1.interface.switch = "00:00:00:00:00:00:00:01" |
||
1626 | uni2.interface.switch = "00:00:00:00:00:00:00:02" |
||
1627 | uni_from_dict_mock.side_effect = [uni1, uni2, uni1, uni2] |
||
1628 | |||
1629 | api = self.get_app_test_client(self.napp) |
||
1630 | payload1 = { |
||
1631 | "name": "my evc1", |
||
1632 | "uni_a": { |
||
1633 | "interface_id": "00:00:00:00:00:00:00:01:1", |
||
1634 | "tag": {"tag_type": 1, "value": 80}, |
||
1635 | }, |
||
1636 | "uni_z": { |
||
1637 | "interface_id": "00:00:00:00:00:00:00:02:2", |
||
1638 | "tag": {"tag_type": 1, "value": 1}, |
||
1639 | }, |
||
1640 | "dynamic_backup_path": True, |
||
1641 | } |
||
1642 | |||
1643 | payload2 = { |
||
1644 | "primary_path": [ |
||
1645 | { |
||
1646 | "endpoint_a": {"id": "00:00:00:00:00:00:00:01:1"}, |
||
1647 | "endpoint_b": {"id": "00:00:00:00:00:00:00:02:2"}, |
||
1648 | } |
||
1649 | ] |
||
1650 | } |
||
1651 | |||
1652 | evc_as_dict_mock.return_value = payload1 |
||
1653 | response = api.post( |
||
1654 | f"{self.server_name_url}/v2/evc/", |
||
1655 | data=json.dumps(payload1), |
||
1656 | content_type="application/json", |
||
1657 | ) |
||
1658 | self.assertEqual(201, response.status_code) |
||
1659 | |||
1660 | evc_as_dict_mock.return_value = payload2 |
||
1661 | current_data = json.loads(response.data) |
||
1662 | circuit_id = current_data["circuit_id"] |
||
1663 | response = api.patch( |
||
1664 | f"{self.server_name_url}/v2/evc/{circuit_id}", |
||
1665 | data=json.dumps(payload2), |
||
1666 | content_type="application/json", |
||
1667 | ) |
||
1668 | current_data = json.loads(response.data) |
||
1669 | expected_data = "primary_path is not a valid path: error" |
||
1670 | self.assertEqual(400, response.status_code) |
||
1671 | self.assertEqual(current_data["description"], expected_data) |
||
1672 | |||
1673 | def test_link_from_dict_non_existent_intf(self): |
||
1674 | """Test _link_from_dict non existent intf.""" |
||
1675 | self.napp.controller.get_interface_by_id = MagicMock(return_value=None) |
||
1676 | link_dict = { |
||
1677 | "endpoint_a": {"id": "a"}, |
||
1678 | "endpoint_b": {"id": "b"} |
||
1679 | } |
||
1680 | with self.assertRaises(ValueError): |
||
1681 | self.napp._uni_from_dict(link_dict) |
||
1682 | |||
1683 | View Code Duplication | @patch("napps.kytos.mef_eline.models.evc.EVC.deploy") |
|
1684 | @patch("napps.kytos.mef_eline.scheduler.Scheduler.add") |
||
1685 | @patch("napps.kytos.mef_eline.main.Main._uni_from_dict") |
||
1686 | @patch("napps.kytos.mef_eline.models.evc.EVC._validate") |
||
1687 | @patch("napps.kytos.mef_eline.controllers.ELineController.upsert_evc") |
||
1688 | def test_update_evc_no_json_mime(self, *args): |
||
1689 | """Test update a circuit with wrong mimetype.""" |
||
1690 | # pylint: disable=too-many-locals |
||
1691 | ( |
||
1692 | mongo_controller_upsert_mock, |
||
1693 | validate_mock, |
||
1694 | uni_from_dict_mock, |
||
1695 | sched_add_mock, |
||
1696 | evc_deploy_mock, |
||
1697 | ) = args |
||
1698 | |||
1699 | validate_mock.return_value = True |
||
1700 | sched_add_mock.return_value = True |
||
1701 | evc_deploy_mock.return_value = True |
||
1702 | uni1 = create_autospec(UNI) |
||
1703 | uni2 = create_autospec(UNI) |
||
1704 | uni1.interface = create_autospec(Interface) |
||
1705 | uni2.interface = create_autospec(Interface) |
||
1706 | uni1.interface.switch = "00:00:00:00:00:00:00:01" |
||
1707 | uni2.interface.switch = "00:00:00:00:00:00:00:02" |
||
1708 | uni_from_dict_mock.side_effect = [uni1, uni2, uni1, uni2] |
||
1709 | mongo_controller_upsert_mock.return_value = True |
||
1710 | |||
1711 | api = self.get_app_test_client(self.napp) |
||
1712 | payload1 = { |
||
1713 | "name": "my evc1", |
||
1714 | "uni_a": { |
||
1715 | "interface_id": "00:00:00:00:00:00:00:01:1", |
||
1716 | "tag": {"tag_type": 1, "value": 80}, |
||
1717 | }, |
||
1718 | "uni_z": { |
||
1719 | "interface_id": "00:00:00:00:00:00:00:02:2", |
||
1720 | "tag": {"tag_type": 1, "value": 1}, |
||
1721 | }, |
||
1722 | "dynamic_backup_path": True, |
||
1723 | } |
||
1724 | |||
1725 | payload2 = {"dynamic_backup_path": False} |
||
1726 | |||
1727 | response = api.post( |
||
1728 | f"{self.server_name_url}/v2/evc/", |
||
1729 | data=json.dumps(payload1), |
||
1730 | content_type="application/json", |
||
1731 | ) |
||
1732 | self.assertEqual(201, response.status_code) |
||
1733 | |||
1734 | current_data = json.loads(response.data) |
||
1735 | circuit_id = current_data["circuit_id"] |
||
1736 | response = api.patch( |
||
1737 | f"{self.server_name_url}/v2/evc/{circuit_id}", data=payload2 |
||
1738 | ) |
||
1739 | current_data = json.loads(response.data) |
||
1740 | expected_data = "The request body mimetype is not application/json." |
||
1741 | self.assertEqual(current_data["description"], expected_data) |
||
1742 | self.assertEqual(415, response.status_code) |
||
1743 | |||
1744 | def test_delete_no_evc(self): |
||
1745 | """Test delete when EVC does not exist.""" |
||
1746 | api = self.get_app_test_client(self.napp) |
||
1747 | response = api.delete(f"{self.server_name_url}/v2/evc/123") |
||
1748 | current_data = json.loads(response.data) |
||
1749 | expected_data = "circuit_id 123 not found" |
||
1750 | self.assertEqual(current_data["description"], expected_data) |
||
1751 | self.assertEqual(404, response.status_code) |
||
1752 | |||
1753 | @patch("napps.kytos.mef_eline.models.evc.EVC.remove_current_flows") |
||
1754 | @patch("napps.kytos.mef_eline.models.evc.EVC.deploy") |
||
1755 | @patch("napps.kytos.mef_eline.scheduler.Scheduler.add") |
||
1756 | @patch("napps.kytos.mef_eline.main.Main._uni_from_dict") |
||
1757 | @patch("napps.kytos.mef_eline.controllers.ELineController.upsert_evc") |
||
1758 | @patch("napps.kytos.mef_eline.models.evc.EVC._validate") |
||
1759 | @patch("napps.kytos.mef_eline.main.EVC.as_dict") |
||
1760 | def test_delete_archived_evc(self, *args): |
||
1761 | """Try to delete an archived EVC""" |
||
1762 | # pylint: disable=too-many-locals |
||
1763 | ( |
||
1764 | evc_as_dict_mock, |
||
1765 | validate_mock, |
||
1766 | mongo_controller_upsert_mock, |
||
1767 | uni_from_dict_mock, |
||
1768 | sched_add_mock, |
||
1769 | evc_deploy_mock, |
||
1770 | remove_current_flows_mock |
||
1771 | ) = args |
||
1772 | |||
1773 | validate_mock.return_value = True |
||
1774 | mongo_controller_upsert_mock.return_value = True |
||
1775 | sched_add_mock.return_value = True |
||
1776 | evc_deploy_mock.return_value = True |
||
1777 | remove_current_flows_mock.return_value = True |
||
1778 | uni1 = create_autospec(UNI) |
||
1779 | uni2 = create_autospec(UNI) |
||
1780 | uni1.interface = create_autospec(Interface) |
||
1781 | uni2.interface = create_autospec(Interface) |
||
1782 | uni1.interface.switch = "00:00:00:00:00:00:00:01" |
||
1783 | uni2.interface.switch = "00:00:00:00:00:00:00:02" |
||
1784 | uni_from_dict_mock.side_effect = [uni1, uni2, uni1, uni2] |
||
1785 | |||
1786 | api = self.get_app_test_client(self.napp) |
||
1787 | payload1 = { |
||
1788 | "name": "my evc1", |
||
1789 | "uni_a": { |
||
1790 | "interface_id": "00:00:00:00:00:00:00:01:1", |
||
1791 | "tag": {"tag_type": 1, "value": 80}, |
||
1792 | }, |
||
1793 | "uni_z": { |
||
1794 | "interface_id": "00:00:00:00:00:00:00:02:2", |
||
1795 | "tag": {"tag_type": 1, "value": 1}, |
||
1796 | }, |
||
1797 | "dynamic_backup_path": True, |
||
1798 | } |
||
1799 | |||
1800 | evc_as_dict_mock.return_value = payload1 |
||
1801 | response = api.post( |
||
1802 | f"{self.server_name_url}/v2/evc/", |
||
1803 | data=json.dumps(payload1), |
||
1804 | content_type="application/json", |
||
1805 | ) |
||
1806 | self.assertEqual(201, response.status_code) |
||
1807 | |||
1808 | current_data = json.loads(response.data) |
||
1809 | circuit_id = current_data["circuit_id"] |
||
1810 | response = api.delete( |
||
1811 | f"{self.server_name_url}/v2/evc/{circuit_id}" |
||
1812 | ) |
||
1813 | self.assertEqual(200, response.status_code) |
||
1814 | |||
1815 | response = api.delete( |
||
1816 | f"{self.server_name_url}/v2/evc/{circuit_id}" |
||
1817 | ) |
||
1818 | current_data = json.loads(response.data) |
||
1819 | expected_data = f"Circuit {circuit_id} already removed" |
||
1820 | self.assertEqual(current_data["description"], expected_data) |
||
1821 | self.assertEqual(404, response.status_code) |
||
1822 | |||
1823 | def test_handle_link_up(self): |
||
1824 | """Test handle_link_up method.""" |
||
1825 | evc_mock = create_autospec(EVC) |
||
1826 | evc_mock.service_level, evc_mock.creation_time = 0, 1 |
||
1827 | evc_mock.is_enabled = MagicMock(side_effect=[True, False, True]) |
||
1828 | evc_mock.lock = MagicMock() |
||
1829 | type(evc_mock).archived = PropertyMock( |
||
1830 | side_effect=[True, False, False] |
||
1831 | ) |
||
1832 | evcs = [evc_mock, evc_mock, evc_mock] |
||
1833 | event = KytosEvent(name="test", content={"link": "abc"}) |
||
1834 | self.napp.circuits = dict(zip(["1", "2", "3"], evcs)) |
||
1835 | self.napp.handle_link_up(event) |
||
1836 | evc_mock.handle_link_up.assert_called_once_with("abc") |
||
1837 | |||
1838 | @patch("time.sleep", return_value=None) |
||
1839 | @patch("napps.kytos.mef_eline.main.settings") |
||
1840 | @patch("napps.kytos.mef_eline.main.emit_event") |
||
1841 | def test_handle_link_down(self, emit_event_mock, settings_mock, _): |
||
1842 | """Test handle_link_down method.""" |
||
1843 | evc1 = MagicMock(id="1", service_level=0, creation_time=1) |
||
1844 | evc1.is_affected_by_link.return_value = True |
||
1845 | evc1.handle_link_down.return_value = True |
||
1846 | evc1.failover_path = None |
||
1847 | evc2 = MagicMock(id="2", service_level=6, creation_time=1) |
||
1848 | evc2.is_affected_by_link.return_value = False |
||
1849 | evc3 = MagicMock(id="3", service_level=5, creation_time=1) |
||
1850 | evc3.is_affected_by_link.return_value = True |
||
1851 | evc3.handle_link_down.return_value = True |
||
1852 | evc3.failover_path = None |
||
1853 | evc4 = MagicMock(id="4", service_level=4, creation_time=1) |
||
1854 | evc4.is_affected_by_link.return_value = True |
||
1855 | evc4.is_failover_path_affected_by_link.return_value = False |
||
1856 | evc4.failover_path = ["2"] |
||
1857 | evc4.get_failover_flows.return_value = { |
||
1858 | "2": ["flow1", "flow2"], |
||
1859 | "3": ["flow3", "flow4", "flow5", "flow6"], |
||
1860 | } |
||
1861 | evc5 = MagicMock(id="5", service_level=7, creation_time=1) |
||
1862 | evc5.is_affected_by_link.return_value = True |
||
1863 | evc5.is_failover_path_affected_by_link.return_value = False |
||
1864 | evc5.failover_path = ["3"] |
||
1865 | evc5.get_failover_flows.return_value = { |
||
1866 | "4": ["flow7", "flow8"], |
||
1867 | "5": ["flow9", "flow10"], |
||
1868 | } |
||
1869 | link = MagicMock(id="123") |
||
1870 | event = KytosEvent(name="test", content={"link": link}) |
||
1871 | self.napp.circuits = {"1": evc1, "2": evc2, "3": evc3, "4": evc4, |
||
1872 | "5": evc5} |
||
1873 | settings_mock.BATCH_SIZE = 2 |
||
1874 | self.napp.handle_link_down(event) |
||
1875 | |||
1876 | assert evc5.service_level > evc4.service_level |
||
1877 | # evc5 batched flows should be sent first |
||
1878 | emit_event_mock.assert_has_calls([ |
||
1879 | call( |
||
1880 | self.napp.controller, |
||
1881 | context="kytos.flow_manager", |
||
1882 | name="flows.install", |
||
1883 | dpid="4", |
||
1884 | flow_dict={"flows": ["flow7", "flow8"]}, |
||
1885 | ), |
||
1886 | call( |
||
1887 | self.napp.controller, |
||
1888 | context="kytos.flow_manager", |
||
1889 | name="flows.install", |
||
1890 | dpid="5", |
||
1891 | flow_dict={"flows": ["flow9", "flow10"]}, |
||
1892 | ), |
||
1893 | call( |
||
1894 | self.napp.controller, |
||
1895 | context="kytos.flow_manager", |
||
1896 | name="flows.install", |
||
1897 | dpid="2", |
||
1898 | flow_dict={"flows": ["flow1", "flow2"]}, |
||
1899 | ), |
||
1900 | call( |
||
1901 | self.napp.controller, |
||
1902 | context="kytos.flow_manager", |
||
1903 | name="flows.install", |
||
1904 | dpid="3", |
||
1905 | flow_dict={"flows": ["flow3", "flow4"]}, |
||
1906 | ), |
||
1907 | call( |
||
1908 | self.napp.controller, |
||
1909 | context="kytos.flow_manager", |
||
1910 | name="flows.install", |
||
1911 | dpid="3", |
||
1912 | flow_dict={"flows": ["flow5", "flow6"]}, |
||
1913 | ), |
||
1914 | ]) |
||
1915 | event_name = "evc_affected_by_link_down" |
||
1916 | assert evc3.service_level > evc1.service_level |
||
1917 | # evc3 should be handled before evc1 |
||
1918 | emit_event_mock.assert_has_calls([ |
||
1919 | call(self.napp.controller, event_name, evc_id="3", link_id="123"), |
||
1920 | call(self.napp.controller, event_name, evc_id="1", link_id="123"), |
||
1921 | ]) |
||
1922 | evc4.sync.assert_called_once() |
||
1923 | event_name = "redeployed_link_down" |
||
1924 | emit_event_mock.assert_has_calls([ |
||
1925 | call(self.napp.controller, event_name, evc_id="4"), |
||
1926 | ]) |
||
1927 | |||
1928 | @patch("napps.kytos.mef_eline.main.emit_event") |
||
1929 | def test_handle_evc_affected_by_link_down(self, emit_event_mock): |
||
1930 | """Test handle_evc_affected_by_link_down method.""" |
||
1931 | evc1 = MagicMock(id="1") |
||
1932 | evc1.handle_link_down.return_value = True |
||
1933 | evc2 = MagicMock(id="2") |
||
1934 | evc2.handle_link_down.return_value = False |
||
1935 | self.napp.circuits = {"1": evc1, "2": evc2} |
||
1936 | |||
1937 | event = KytosEvent(name="e1", content={"evc_id": "3", "link_id": "1"}) |
||
1938 | self.napp.handle_evc_affected_by_link_down(event) |
||
1939 | emit_event_mock.assert_not_called() |
||
1940 | |||
1941 | event.content["evc_id"] = "1" |
||
1942 | self.napp.handle_evc_affected_by_link_down(event) |
||
1943 | emit_event_mock.assert_called_with( |
||
1944 | self.napp.controller, "redeployed_link_down", evc_id="1" |
||
1945 | ) |
||
1946 | |||
1947 | event.content["evc_id"] = "2" |
||
1948 | self.napp.handle_evc_affected_by_link_down(event) |
||
1949 | emit_event_mock.assert_called_with( |
||
1950 | self.napp.controller, "error_redeploy_link_down", evc_id="2" |
||
1951 | ) |
||
1952 | |||
1953 | def test_handle_evc_deployed(self): |
||
1954 | """Test handle_evc_deployed method.""" |
||
1955 | evc = create_autospec(EVC, id="1") |
||
1956 | evc.lock = MagicMock() |
||
1957 | self.napp.circuits = {"1": evc} |
||
1958 | |||
1959 | event = KytosEvent(name="e1", content={"evc_id": "2"}) |
||
1960 | self.napp.handle_evc_deployed(event) |
||
1961 | evc.setup_failover_path.assert_not_called() |
||
1962 | |||
1963 | event.content["evc_id"] = "1" |
||
1964 | self.napp.handle_evc_deployed(event) |
||
1965 | evc.setup_failover_path.assert_called() |
||
1966 | |||
1967 | def test_add_metadata(self): |
||
1968 | """Test method to add metadata""" |
||
1969 | evc_mock = create_autospec(EVC) |
||
1970 | evc_mock.metadata = {} |
||
1971 | evc_mock.id = 1234 |
||
1972 | self.napp.circuits = {"1234": evc_mock} |
||
1973 | |||
1974 | api = self.get_app_test_client(self.napp) |
||
1975 | payload = {"metadata1": 1, "metadata2": 2} |
||
1976 | response = api.post( |
||
1977 | f"{self.server_name_url}/v2/evc/1234/metadata", |
||
1978 | data=json.dumps(payload), |
||
1979 | content_type="application/json", |
||
1980 | ) |
||
1981 | |||
1982 | self.assertEqual(response.status_code, 201) |
||
1983 | evc_mock.extend_metadata.assert_called_with(payload) |
||
1984 | |||
1985 | def test_add_metadata_malformed_json(self): |
||
1986 | """Test method to add metadata with a malformed json""" |
||
1987 | api = self.get_app_test_client(self.napp) |
||
1988 | payload = '{"metadata1": 1, "metadata2": 2,}' |
||
1989 | response = api.post( |
||
1990 | f"{self.server_name_url}/v2/evc/1234/metadata", |
||
1991 | data=payload, |
||
1992 | content_type="application/json" |
||
1993 | ) |
||
1994 | |||
1995 | self.assertEqual(response.status_code, 400) |
||
1996 | self.assertEqual( |
||
1997 | response.json["description"], |
||
1998 | "The request body is not a well-formed JSON." |
||
1999 | ) |
||
2000 | |||
2001 | def test_add_metadata_no_body(self): |
||
2002 | """Test method to add metadata with no body""" |
||
2003 | api = self.get_app_test_client(self.napp) |
||
2004 | response = api.post( |
||
2005 | f"{self.server_name_url}/v2/evc/1234/metadata" |
||
2006 | ) |
||
2007 | self.assertEqual(response.status_code, 400) |
||
2008 | self.assertEqual( |
||
2009 | response.json["description"], |
||
2010 | "The request body is empty." |
||
2011 | ) |
||
2012 | |||
2013 | def test_add_metadata_no_evc(self): |
||
2014 | """Test method to add metadata with no evc""" |
||
2015 | api = self.get_app_test_client(self.napp) |
||
2016 | payload = {"metadata1": 1, "metadata2": 2} |
||
2017 | response = api.post( |
||
2018 | f"{self.server_name_url}/v2/evc/1234/metadata", |
||
2019 | data=json.dumps(payload), |
||
2020 | content_type="application/json", |
||
2021 | ) |
||
2022 | self.assertEqual(response.status_code, 404) |
||
2023 | self.assertEqual( |
||
2024 | response.json["description"], |
||
2025 | "circuit_id 1234 not found." |
||
2026 | ) |
||
2027 | |||
2028 | def test_add_metadata_wrong_content_type(self): |
||
2029 | """Test method to add metadata with wrong content type""" |
||
2030 | api = self.get_app_test_client(self.napp) |
||
2031 | payload = {"metadata1": 1, "metadata2": 2} |
||
2032 | response = api.post( |
||
2033 | f"{self.server_name_url}/v2/evc/1234/metadata", |
||
2034 | data=json.dumps(payload), |
||
2035 | content_type="application/xml", |
||
2036 | ) |
||
2037 | self.assertEqual(response.status_code, 415) |
||
2038 | self.assertEqual( |
||
2039 | response.json["description"], |
||
2040 | "The content type must be application/json " |
||
2041 | "(received application/xml)." |
||
2042 | ) |
||
2043 | |||
2044 | def test_get_metadata(self): |
||
2045 | """Test method to get metadata""" |
||
2046 | evc_mock = create_autospec(EVC) |
||
2047 | evc_mock.metadata = {'metadata1': 1, 'metadata2': 2} |
||
2048 | evc_mock.id = 1234 |
||
2049 | self.napp.circuits = {"1234": evc_mock} |
||
2050 | |||
2051 | api = self.get_app_test_client(self.napp) |
||
2052 | response = api.get( |
||
2053 | f"{self.server_name_url}/v2/evc/1234/metadata", |
||
2054 | ) |
||
2055 | self.assertEqual(response.status_code, 200) |
||
2056 | self.assertEqual(response.json, {"metadata": evc_mock.metadata}) |
||
2057 | |||
2058 | def test_delete_metadata(self): |
||
2059 | """Test method to delete metadata""" |
||
2060 | evc_mock = create_autospec(EVC) |
||
2061 | evc_mock.metadata = {'metadata1': 1, 'metadata2': 2} |
||
2062 | evc_mock.id = 1234 |
||
2063 | self.napp.circuits = {"1234": evc_mock} |
||
2064 | |||
2065 | api = self.get_app_test_client(self.napp) |
||
2066 | response = api.delete( |
||
2067 | f"{self.server_name_url}/v2/evc/1234/metadata/metadata1", |
||
2068 | ) |
||
2069 | self.assertEqual(response.status_code, 200) |
||
2070 | |||
2071 | def test_delete_metadata_no_evc(self): |
||
2072 | """Test method to delete metadata with no evc""" |
||
2073 | api = self.get_app_test_client(self.napp) |
||
2074 | response = api.delete( |
||
2075 | f"{self.server_name_url}/v2/evc/1234/metadata/metadata1", |
||
2076 | ) |
||
2077 | self.assertEqual(response.status_code, 404) |
||
2078 | self.assertEqual( |
||
2079 | response.json["description"], |
||
2080 | "circuit_id 1234 not found." |
||
2081 | ) |
||
2082 | |||
2083 | @patch('napps.kytos.mef_eline.main.Main._load_evc') |
||
2084 | def test_load_all_evcs(self, load_evc_mock): |
||
2085 | """Test load_evcs method""" |
||
2086 | mock_circuits = { |
||
2087 | 'circuits': { |
||
2088 | 1: 'circuit_1', |
||
2089 | 2: 'circuit_2', |
||
2090 | 3: 'circuit_3', |
||
2091 | 4: 'circuit_4' |
||
2092 | } |
||
2093 | } |
||
2094 | self.napp.mongo_controller.get_circuits.return_value = mock_circuits |
||
2095 | self.napp.circuits = {2: 'circuit_2', 3: 'circuit_3'} |
||
2096 | self.napp.load_all_evcs() |
||
2097 | load_evc_mock.assert_has_calls([call('circuit_1'), call('circuit_4')]) |
||
2098 | |||
2099 | @patch('napps.kytos.mef_eline.main.Main._evc_from_dict') |
||
2100 | def test_load_evc(self, evc_from_dict_mock): |
||
2101 | """Test _load_evc method""" |
||
2102 | # pylint: disable=protected-access |
||
2103 | # case 1: early return with ValueError exception |
||
2104 | evc_from_dict_mock.side_effect = ValueError("err") |
||
2105 | evc_dict = MagicMock() |
||
2106 | self.assertEqual(self.napp._load_evc(evc_dict), None) |
||
2107 | |||
2108 | # case2: archived evc |
||
2109 | evc = MagicMock() |
||
2110 | evc.archived = True |
||
2111 | evc_from_dict_mock.side_effect = None |
||
2112 | evc_from_dict_mock.return_value = evc |
||
2113 | self.assertEqual(self.napp._load_evc(evc_dict), None) |
||
2114 | |||
2115 | # case3: success creating |
||
2116 | evc.archived = False |
||
2117 | evc.id = 1 |
||
2118 | self.napp.sched = MagicMock() |
||
2119 | |||
2120 | result = self.napp._load_evc(evc_dict) |
||
2121 | self.assertEqual(result, evc) |
||
2122 | evc.deactivate.assert_called() |
||
2123 | evc.sync.assert_called() |
||
2124 | self.napp.sched.add.assert_called_with(evc) |
||
2125 | self.assertEqual(self.napp.circuits[1], evc) |
||
2126 | |||
2127 | def test_handle_flow_mod_error(self): |
||
2128 | """Test handle_flow_mod_error method""" |
||
2129 | flow = MagicMock() |
||
2130 | flow.cookie = 0xaa00000000000011 |
||
2131 | event = MagicMock() |
||
2132 | event.content = {'flow': flow, 'error_command': 'add'} |
||
2133 | evc = create_autospec(EVC) |
||
2134 | evc.remove_current_flows = MagicMock() |
||
2135 | self.napp.circuits = {"00000000000011": evc} |
||
2136 | self.napp.handle_flow_mod_error(event) |
||
2137 | evc.remove_current_flows.assert_called_once() |
||
2138 | |||
2139 | @patch("kytos.core.Controller.get_interface_by_id") |
||
2140 | def test_uni_from_dict(self, _get_interface_by_id_mock): |
||
2141 | """Test _uni_from_dict method.""" |
||
2142 | # pylint: disable=protected-access |
||
2143 | # case1: early return on empty dict |
||
2144 | self.assertEqual(self.napp._uni_from_dict(None), False) |
||
2145 | |||
2146 | # case2: invalid interface raises ValueError |
||
2147 | _get_interface_by_id_mock.return_value = None |
||
2148 | uni_dict = { |
||
2149 | "interface_id": "00:01:1", |
||
2150 | "tag": {"tag_type": 1, "value": 81}, |
||
2151 | } |
||
2152 | with self.assertRaises(ValueError): |
||
2153 | self.napp._uni_from_dict(uni_dict) |
||
2154 | |||
2155 | # case3: success creation |
||
2156 | uni_mock = get_uni_mocked(switch_id="00:01") |
||
2157 | _get_interface_by_id_mock.return_value = uni_mock.interface |
||
2158 | uni = self.napp._uni_from_dict(uni_dict) |
||
2159 | self.assertEqual(uni, uni_mock) |
||
2160 | |||
2161 | # case4: success creation without tag |
||
2162 | uni_mock.user_tag = None |
||
2163 | del uni_dict["tag"] |
||
2164 | uni = self.napp._uni_from_dict(uni_dict) |
||
2165 | self.assertEqual(uni, uni_mock) |
||
2166 |