Total Complexity | 97 |
Total Lines | 2836 |
Duplicated Lines | 17.7 % |
Coverage | 93.13% |
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 | 1 | import asyncio |
|
3 | 1 | from unittest.mock import (AsyncMock, MagicMock, Mock, call, |
|
4 | create_autospec, patch) |
||
5 | |||
6 | 1 | import pytest |
|
7 | 1 | from kytos.lib.helpers import get_controller_mock, get_test_client |
|
8 | 1 | from kytos.core.helpers import now |
|
9 | 1 | from kytos.core.common import EntityStatus |
|
10 | 1 | from kytos.core.events import KytosEvent |
|
11 | 1 | from kytos.core.exceptions import KytosTagError |
|
12 | 1 | from kytos.core.interface import TAGRange, UNI, Interface |
|
13 | 1 | from napps.kytos.mef_eline.exceptions import FlowModException, InvalidPath |
|
14 | 1 | from napps.kytos.mef_eline.models import EVC, Path |
|
15 | 1 | from napps.kytos.mef_eline.tests.helpers import get_uni_mocked |
|
16 | |||
17 | |||
18 | 1 | async def test_on_table_enabled(): |
|
19 | """Test on_table_enabled""" |
||
20 | # pylint: disable=import-outside-toplevel |
||
21 | 1 | from napps.kytos.mef_eline.main import Main |
|
22 | 1 | controller = get_controller_mock() |
|
23 | 1 | controller.buffers.app.aput = AsyncMock() |
|
24 | 1 | Main.get_eline_controller = MagicMock() |
|
25 | 1 | napp = Main(controller) |
|
26 | |||
27 | # Succesfully setting table groups |
||
28 | 1 | content = {"mef_eline": {"epl": 2}} |
|
29 | 1 | event = KytosEvent(name="kytos/of_multi_table.enable_table", |
|
30 | content=content) |
||
31 | 1 | await napp.on_table_enabled(event) |
|
32 | 1 | assert napp.table_group["epl"] == 2 |
|
33 | 1 | assert napp.table_group["evpl"] == 0 |
|
34 | 1 | assert controller.buffers.app.aput.call_count == 1 |
|
35 | |||
36 | # Failure at setting table groups |
||
37 | 1 | content = {"mef_eline": {"unknown": 123}} |
|
38 | 1 | event = KytosEvent(name="kytos/of_multi_table.enable_table", |
|
39 | content=content) |
||
40 | 1 | await napp.on_table_enabled(event) |
|
41 | 1 | assert controller.buffers.app.aput.call_count == 1 |
|
42 | |||
43 | # Failure with early return |
||
44 | 1 | content = {} |
|
45 | 1 | event = KytosEvent(name="kytos/of_multi_table.enable_table", |
|
46 | content=content) |
||
47 | 1 | await napp.on_table_enabled(event) |
|
48 | 1 | assert controller.buffers.app.aput.call_count == 1 |
|
49 | |||
50 | |||
51 | # pylint: disable=too-many-public-methods, too-many-lines |
||
52 | # pylint: disable=too-many-arguments,too-many-locals |
||
53 | 1 | class TestMain: |
|
54 | """Test the Main class.""" |
||
55 | |||
56 | 1 | def setup_method(self): |
|
57 | """Execute steps before each tests. |
||
58 | |||
59 | Set the server_name_url_url from kytos/mef_eline |
||
60 | """ |
||
61 | |||
62 | # The decorator run_on_thread is patched, so methods that listen |
||
63 | # for events do not run on threads while tested. |
||
64 | # Decorators have to be patched before the methods that are |
||
65 | # decorated with them are imported. |
||
66 | 1 | patch("kytos.core.helpers.run_on_thread", lambda x: x).start() |
|
67 | # pylint: disable=import-outside-toplevel |
||
68 | 1 | from napps.kytos.mef_eline.main import Main |
|
69 | 1 | Main.get_eline_controller = MagicMock() |
|
70 | 1 | controller = get_controller_mock() |
|
71 | 1 | self.napp = Main(controller) |
|
72 | 1 | self.api_client = get_test_client(controller, self.napp) |
|
73 | 1 | self.base_endpoint = "kytos/mef_eline" |
|
74 | |||
75 | 1 | def test_get_event_listeners(self): |
|
76 | """Verify all event listeners registered.""" |
||
77 | 1 | expected_events = [ |
|
78 | "kytos/core.shutdown", |
||
79 | "kytos/core.shutdown.kytos/mef_eline", |
||
80 | "kytos/topology.link_up", |
||
81 | "kytos/topology.link_down", |
||
82 | ] |
||
83 | 1 | actual_events = self.napp.listeners() |
|
84 | |||
85 | 1 | for _event in expected_events: |
|
86 | 1 | assert _event in actual_events, _event |
|
87 | |||
88 | 1 | @patch('napps.kytos.mef_eline.main.log') |
|
89 | 1 | @patch('napps.kytos.mef_eline.main.Main.execute_consistency') |
|
90 | 1 | def test_execute(self, mock_execute_consistency, mock_log): |
|
91 | """Test execute.""" |
||
92 | 1 | self.napp.execution_rounds = 0 |
|
93 | 1 | self.napp.execute() |
|
94 | 1 | mock_execute_consistency.assert_called() |
|
95 | 1 | assert mock_log.debug.call_count == 2 |
|
96 | |||
97 | # Test locked should return |
||
98 | 1 | mock_execute_consistency.call_count = 0 |
|
99 | 1 | mock_log.info.call_count = 0 |
|
100 | # pylint: disable=protected-access |
||
101 | 1 | self.napp._lock = MagicMock() |
|
102 | 1 | self.napp._lock.locked.return_value = True |
|
103 | # pylint: enable=protected-access |
||
104 | 1 | self.napp.execute() |
|
105 | 1 | mock_execute_consistency.assert_not_called() |
|
106 | 1 | mock_log.info.assert_not_called() |
|
107 | |||
108 | 1 | @patch('napps.kytos.mef_eline.main.settings') |
|
109 | 1 | @patch("napps.kytos.mef_eline.controllers.ELineController.upsert_evc") |
|
110 | 1 | @patch("napps.kytos.mef_eline.models.evc.EVCDeploy.check_list_traces") |
|
111 | 1 | def test_execute_consistency(self, mock_check_list_traces, *args): |
|
112 | """Test execute_consistency.""" |
||
113 | 1 | (mongo_controller_upsert_mock, mock_settings) = args |
|
114 | |||
115 | 1 | stored_circuits = {'1': {'name': 'circuit_1'}, |
|
116 | '2': {'name': 'circuit_2'}, |
||
117 | '3': {'name': 'circuit_3'}} |
||
118 | 1 | mongo_controller_upsert_mock.return_value = True |
|
119 | 1 | self.napp.mongo_controller.get_circuits.return_value = { |
|
120 | "circuits": stored_circuits |
||
121 | } |
||
122 | |||
123 | 1 | mock_settings.WAIT_FOR_OLD_PATH = -1 |
|
124 | 1 | evc1 = MagicMock(id=1, service_level=0, creation_time=1) |
|
125 | 1 | evc1.is_enabled.return_value = True |
|
126 | 1 | evc1.is_active.return_value = False |
|
127 | 1 | evc1.lock.locked.return_value = False |
|
128 | 1 | evc1.has_recent_removed_flow.return_value = False |
|
129 | 1 | evc1.is_recent_updated.return_value = False |
|
130 | 1 | evc1.execution_rounds = 0 |
|
131 | 1 | evc2 = MagicMock(id=2, service_level=7, creation_time=1) |
|
132 | 1 | evc2.is_enabled.return_value = True |
|
133 | 1 | evc2.is_active.return_value = False |
|
134 | 1 | evc2.lock.locked.return_value = False |
|
135 | 1 | evc2.has_recent_removed_flow.return_value = False |
|
136 | 1 | evc2.is_recent_updated.return_value = False |
|
137 | 1 | evc2.execution_rounds = 0 |
|
138 | 1 | self.napp.circuits = {'1': evc1, '2': evc2} |
|
139 | 1 | assert self.napp.get_evcs_by_svc_level() == [evc2, evc1] |
|
140 | |||
141 | 1 | mock_check_list_traces.return_value = { |
|
142 | 1: True, |
||
143 | 2: False |
||
144 | } |
||
145 | |||
146 | 1 | self.napp.execute_consistency() |
|
147 | 1 | assert evc1.activate.call_count == 1 |
|
148 | 1 | assert evc1.sync.call_count == 1 |
|
149 | 1 | evc1.try_setup_failover_path.assert_called_with(warn_if_not_path=False) |
|
150 | 1 | assert evc2.deploy.call_count == 1 |
|
151 | 1 | evc2.try_setup_failover_path.assert_called_with(warn_if_not_path=False) |
|
152 | |||
153 | 1 | @patch('napps.kytos.mef_eline.main.settings') |
|
154 | 1 | @patch('napps.kytos.mef_eline.main.Main._load_evc') |
|
155 | 1 | @patch("napps.kytos.mef_eline.controllers.ELineController.upsert_evc") |
|
156 | 1 | @patch("napps.kytos.mef_eline.models.evc.EVCDeploy.check_list_traces") |
|
157 | 1 | def test_execute_consistency_wait_for(self, mock_check_list_traces, *args): |
|
158 | """Test execute and wait for setting.""" |
||
159 | 1 | (mongo_controller_upsert_mock, _, mock_settings) = args |
|
160 | |||
161 | 1 | stored_circuits = {'1': {'name': 'circuit_1'}} |
|
162 | 1 | mongo_controller_upsert_mock.return_value = True |
|
163 | 1 | self.napp.mongo_controller.get_circuits.return_value = { |
|
164 | "circuits": stored_circuits |
||
165 | } |
||
166 | |||
167 | 1 | mock_settings.WAIT_FOR_OLD_PATH = -1 |
|
168 | 1 | evc1 = MagicMock(id=1, service_level=0, creation_time=1) |
|
169 | 1 | evc1.is_enabled.return_value = True |
|
170 | 1 | evc1.is_active.return_value = False |
|
171 | 1 | evc1.lock.locked.return_value = False |
|
172 | 1 | evc1.has_recent_removed_flow.return_value = False |
|
173 | 1 | evc1.is_recent_updated.return_value = False |
|
174 | 1 | evc1.execution_rounds = 0 |
|
175 | 1 | evc1.deploy.call_count = 0 |
|
176 | 1 | self.napp.circuits = {'1': evc1} |
|
177 | 1 | assert self.napp.get_evcs_by_svc_level() == [evc1] |
|
178 | 1 | mock_settings.WAIT_FOR_OLD_PATH = 1 |
|
179 | |||
180 | 1 | mock_check_list_traces.return_value = {1: False} |
|
181 | |||
182 | 1 | self.napp.execute_consistency() |
|
183 | 1 | assert evc1.deploy.call_count == 0 |
|
184 | 1 | self.napp.execute_consistency() |
|
185 | 1 | assert evc1.deploy.call_count == 1 |
|
186 | |||
187 | 1 | @patch('napps.kytos.mef_eline.main.Main._uni_from_dict') |
|
188 | 1 | @patch('napps.kytos.mef_eline.models.evc.EVCBase._validate') |
|
189 | 1 | def test_evc_from_dict(self, _validate_mock, uni_from_dict_mock): |
|
190 | """ |
||
191 | Test the helper method that create an EVN from dict. |
||
192 | |||
193 | Verify object creation with circuit data and schedule data. |
||
194 | """ |
||
195 | 1 | _validate_mock.return_value = True |
|
196 | 1 | uni_from_dict_mock.side_effect = ["uni_a", "uni_z"] |
|
197 | 1 | payload = { |
|
198 | "name": "my evc1", |
||
199 | "uni_a": { |
||
200 | "interface_id": "00:00:00:00:00:00:00:01:1", |
||
201 | "tag": {"tag_type": 'vlan', "value": 80}, |
||
202 | }, |
||
203 | "uni_z": { |
||
204 | "interface_id": "00:00:00:00:00:00:00:02:2", |
||
205 | "tag": {"tag_type": 'vlan', "value": 1}, |
||
206 | }, |
||
207 | "circuit_scheduler": [ |
||
208 | {"frequency": "* * * * *", "action": "create"} |
||
209 | ], |
||
210 | "queue_id": 5, |
||
211 | } |
||
212 | # pylint: disable=protected-access |
||
213 | 1 | evc_response = self.napp._evc_from_dict(payload) |
|
214 | 1 | assert evc_response is not None |
|
215 | 1 | assert evc_response.uni_a is not None |
|
216 | 1 | assert evc_response.uni_z is not None |
|
217 | 1 | assert evc_response.circuit_scheduler is not None |
|
218 | 1 | assert evc_response.name is not None |
|
219 | 1 | assert evc_response.queue_id is not None |
|
220 | |||
221 | 1 | @patch("napps.kytos.mef_eline.main.Main._uni_from_dict") |
|
222 | 1 | @patch("napps.kytos.mef_eline.models.evc.EVCBase._validate") |
|
223 | 1 | @patch("kytos.core.Controller.get_interface_by_id") |
|
224 | 1 | def test_evc_from_dict_paths( |
|
225 | self, _get_interface_by_id_mock, _validate_mock, uni_from_dict_mock |
||
226 | ): |
||
227 | """ |
||
228 | Test the helper method that create an EVN from dict. |
||
229 | |||
230 | Verify object creation with circuit data and schedule data. |
||
231 | """ |
||
232 | |||
233 | 1 | _get_interface_by_id_mock.return_value = get_uni_mocked().interface |
|
234 | 1 | _validate_mock.return_value = True |
|
235 | 1 | uni_from_dict_mock.side_effect = ["uni_a", "uni_z"] |
|
236 | 1 | payload = { |
|
237 | "name": "my evc1", |
||
238 | "uni_a": { |
||
239 | "interface_id": "00:00:00:00:00:00:00:01:1", |
||
240 | "tag": {"tag_type": 'vlan', "value": 80}, |
||
241 | }, |
||
242 | "uni_z": { |
||
243 | "interface_id": "00:00:00:00:00:00:00:02:2", |
||
244 | "tag": {"tag_type": 'vlan', "value": 1}, |
||
245 | }, |
||
246 | "current_path": [], |
||
247 | "primary_path": [ |
||
248 | { |
||
249 | "endpoint_a": { |
||
250 | "interface_id": "00:00:00:00:00:00:00:01:1" |
||
251 | }, |
||
252 | "endpoint_b": { |
||
253 | "interface_id": "00:00:00:00:00:00:00:02:2" |
||
254 | }, |
||
255 | } |
||
256 | ], |
||
257 | "backup_path": [], |
||
258 | } |
||
259 | |||
260 | # pylint: disable=protected-access |
||
261 | 1 | evc_response = self.napp._evc_from_dict(payload) |
|
262 | 1 | assert evc_response is not None |
|
263 | 1 | assert evc_response.uni_a is not None |
|
264 | 1 | assert evc_response.uni_z is not None |
|
265 | 1 | assert evc_response.circuit_scheduler is not None |
|
266 | 1 | assert evc_response.name is not None |
|
267 | 1 | assert len(evc_response.current_path) == 0 |
|
268 | 1 | assert len(evc_response.backup_path) == 0 |
|
269 | 1 | assert len(evc_response.primary_path) == 1 |
|
270 | |||
271 | 1 | @patch("napps.kytos.mef_eline.main.Main._uni_from_dict") |
|
272 | 1 | @patch("napps.kytos.mef_eline.models.evc.EVCBase._validate") |
|
273 | 1 | @patch("kytos.core.Controller.get_interface_by_id") |
|
274 | 1 | def test_evc_from_dict_links( |
|
275 | self, _get_interface_by_id_mock, _validate_mock, uni_from_dict_mock |
||
276 | ): |
||
277 | """ |
||
278 | Test the helper method that create an EVN from dict. |
||
279 | |||
280 | Verify object creation with circuit data and schedule data. |
||
281 | """ |
||
282 | 1 | _get_interface_by_id_mock.return_value = get_uni_mocked().interface |
|
283 | 1 | _validate_mock.return_value = True |
|
284 | 1 | uni_from_dict_mock.side_effect = ["uni_a", "uni_z"] |
|
285 | 1 | payload = { |
|
286 | "name": "my evc1", |
||
287 | "uni_a": { |
||
288 | "interface_id": "00:00:00:00:00:00:00:01:1", |
||
289 | "tag": {"tag_type": 'vlan', "value": 80}, |
||
290 | }, |
||
291 | "uni_z": { |
||
292 | "interface_id": "00:00:00:00:00:00:00:02:2", |
||
293 | "tag": {"tag_type": 'vlan', "value": 1}, |
||
294 | }, |
||
295 | "primary_links": [ |
||
296 | { |
||
297 | "endpoint_a": { |
||
298 | "interface_id": "00:00:00:00:00:00:00:01:1" |
||
299 | }, |
||
300 | "endpoint_b": { |
||
301 | "interface_id": "00:00:00:00:00:00:00:02:2" |
||
302 | }, |
||
303 | "metadata": { |
||
304 | "s_vlan": { |
||
305 | "tag_type": 'vlan', |
||
306 | "value": 100 |
||
307 | } |
||
308 | }, |
||
309 | } |
||
310 | ], |
||
311 | "backup_links": [], |
||
312 | } |
||
313 | |||
314 | # pylint: disable=protected-access |
||
315 | 1 | evc_response = self.napp._evc_from_dict(payload) |
|
316 | 1 | assert evc_response is not None |
|
317 | 1 | assert evc_response.uni_a is not None |
|
318 | 1 | assert evc_response.uni_z is not None |
|
319 | 1 | assert evc_response.circuit_scheduler is not None |
|
320 | 1 | assert evc_response.name is not None |
|
321 | 1 | assert len(evc_response.current_links_cache) == 0 |
|
322 | 1 | assert len(evc_response.backup_links) == 0 |
|
323 | 1 | assert len(evc_response.primary_links) == 1 |
|
324 | |||
325 | 1 | async def test_list_without_circuits(self): |
|
326 | """Test if list circuits return 'no circuit stored.'.""" |
||
327 | 1 | circuits = {"circuits": {}} |
|
328 | 1 | self.napp.mongo_controller.get_circuits.return_value = circuits |
|
329 | 1 | url = f"{self.base_endpoint}/v2/evc/" |
|
330 | 1 | response = await self.api_client.get(url) |
|
331 | 1 | assert response.status_code == 200, response.data |
|
332 | 1 | assert not response.json() |
|
333 | |||
334 | 1 | async def test_list_no_circuits_stored(self): |
|
335 | """Test if list circuits return all circuits stored.""" |
||
336 | 1 | circuits = {"circuits": {}} |
|
337 | 1 | self.napp.mongo_controller.get_circuits.return_value = circuits |
|
338 | |||
339 | 1 | url = f"{self.base_endpoint}/v2/evc/" |
|
340 | 1 | response = await self.api_client.get(url) |
|
341 | 1 | expected_result = circuits["circuits"] |
|
342 | 1 | assert response.json() == expected_result |
|
343 | |||
344 | 1 | View Code Duplication | async def test_list_with_circuits_stored(self): |
|
|||
345 | """Test if list circuits return all circuits stored.""" |
||
346 | 1 | circuits = { |
|
347 | 'circuits': |
||
348 | {"1": {"name": "circuit_1"}, "2": {"name": "circuit_2"}} |
||
349 | } |
||
350 | 1 | get_circuits = self.napp.mongo_controller.get_circuits |
|
351 | 1 | get_circuits.return_value = circuits |
|
352 | |||
353 | 1 | url = f"{self.base_endpoint}/v2/evc/" |
|
354 | 1 | response = await self.api_client.get(url) |
|
355 | 1 | expected_result = circuits["circuits"] |
|
356 | 1 | get_circuits.assert_called_with(archived="false", metadata={}) |
|
357 | 1 | assert response.json() == expected_result |
|
358 | |||
359 | 1 | View Code Duplication | async def test_list_with_archived_circuits_archived(self): |
360 | """Test if list circuits only archived circuits.""" |
||
361 | 1 | circuits = { |
|
362 | 'circuits': |
||
363 | { |
||
364 | "1": {"name": "circuit_1", "archived": True}, |
||
365 | } |
||
366 | } |
||
367 | 1 | get_circuits = self.napp.mongo_controller.get_circuits |
|
368 | 1 | get_circuits.return_value = circuits |
|
369 | |||
370 | 1 | url = f"{self.base_endpoint}/v2/evc/?archived=true&metadata.a=1" |
|
371 | 1 | response = await self.api_client.get(url) |
|
372 | 1 | get_circuits.assert_called_with(archived="true", |
|
373 | metadata={"metadata.a": "1"}) |
||
374 | 1 | expected_result = {"1": circuits["circuits"]["1"]} |
|
375 | 1 | assert response.json() == expected_result |
|
376 | |||
377 | 1 | async def test_list_with_archived_circuits_all(self): |
|
378 | """Test if list circuits return all circuits.""" |
||
379 | 1 | circuits = { |
|
380 | 'circuits': { |
||
381 | "1": {"name": "circuit_1"}, |
||
382 | "2": {"name": "circuit_2", "archived": True}, |
||
383 | } |
||
384 | } |
||
385 | 1 | self.napp.mongo_controller.get_circuits.return_value = circuits |
|
386 | |||
387 | 1 | url = f"{self.base_endpoint}/v2/evc/?archived=null" |
|
388 | 1 | response = await self.api_client.get(url) |
|
389 | 1 | expected_result = circuits["circuits"] |
|
390 | 1 | assert response.json() == expected_result |
|
391 | |||
392 | 1 | async def test_circuit_with_valid_id(self): |
|
393 | """Test if get_circuit return the circuit attributes.""" |
||
394 | 1 | circuit = {"name": "circuit_1"} |
|
395 | 1 | self.napp.mongo_controller.get_circuit.return_value = circuit |
|
396 | |||
397 | 1 | url = f"{self.base_endpoint}/v2/evc/1" |
|
398 | 1 | response = await self.api_client.get(url) |
|
399 | 1 | expected_result = circuit |
|
400 | 1 | assert response.json() == expected_result |
|
401 | |||
402 | 1 | async def test_circuit_with_invalid_id(self): |
|
403 | """Test if get_circuit return invalid circuit_id.""" |
||
404 | 1 | self.napp.mongo_controller.get_circuit.return_value = None |
|
405 | 1 | url = f"{self.base_endpoint}/v2/evc/3" |
|
406 | 1 | response = await self.api_client.get(url) |
|
407 | expected_result = "circuit_id 3 not found" |
||
408 | assert response.json()["description"] == expected_result |
||
409 | |||
410 | 1 | @patch("napps.kytos.mef_eline.main.Main._check_no_tag_duplication") |
|
411 | 1 | @patch("napps.kytos.mef_eline.models.evc.EVC._tag_lists_equal") |
|
412 | 1 | @patch("napps.kytos.mef_eline.main.Main._use_uni_tags") |
|
413 | 1 | @patch("napps.kytos.mef_eline.models.evc.EVC.deploy") |
|
414 | 1 | @patch("napps.kytos.mef_eline.scheduler.Scheduler.add") |
|
415 | 1 | @patch("napps.kytos.mef_eline.main.Main._uni_from_dict") |
|
416 | 1 | @patch("napps.kytos.mef_eline.controllers.ELineController.upsert_evc") |
|
417 | 1 | @patch("napps.kytos.mef_eline.main.EVC.as_dict") |
|
418 | 1 | @patch("napps.kytos.mef_eline.models.evc.EVC._validate") |
|
419 | 1 | async def test_create_a_circuit_case_1( |
|
420 | self, |
||
421 | validate_mock, |
||
422 | evc_as_dict_mock, |
||
423 | mongo_controller_upsert_mock, |
||
424 | uni_from_dict_mock, |
||
425 | sched_add_mock, |
||
426 | evc_deploy_mock, |
||
427 | mock_use_uni_tags, |
||
428 | mock_tags_equal, |
||
429 | mock_check_duplicate |
||
430 | ): |
||
431 | """Test create a new circuit.""" |
||
432 | # pylint: disable=too-many-locals |
||
433 | 1 | self.napp.controller.loop = asyncio.get_running_loop() |
|
434 | 1 | validate_mock.return_value = True |
|
435 | 1 | mongo_controller_upsert_mock.return_value = True |
|
436 | 1 | evc_deploy_mock.return_value = True |
|
437 | 1 | mock_use_uni_tags.return_value = True |
|
438 | 1 | mock_tags_equal.return_value = True |
|
439 | 1 | mock_check_duplicate.return_value = True |
|
440 | 1 | uni1 = create_autospec(UNI) |
|
441 | 1 | uni2 = create_autospec(UNI) |
|
442 | 1 | uni1.interface = create_autospec(Interface) |
|
443 | 1 | uni2.interface = create_autospec(Interface) |
|
444 | 1 | uni1.interface.switch = "00:00:00:00:00:00:00:01" |
|
445 | 1 | uni2.interface.switch = "00:00:00:00:00:00:00:02" |
|
446 | 1 | uni_from_dict_mock.side_effect = [uni1, uni2] |
|
447 | 1 | evc_as_dict_mock.return_value = {} |
|
448 | 1 | sched_add_mock.return_value = True |
|
449 | 1 | self.napp.mongo_controller.get_circuits.return_value = {} |
|
450 | |||
451 | 1 | url = f"{self.base_endpoint}/v2/evc/" |
|
452 | 1 | payload = { |
|
453 | "name": "my evc1", |
||
454 | "frequency": "* * * * *", |
||
455 | "uni_a": { |
||
456 | "interface_id": "00:00:00:00:00:00:00:01:1", |
||
457 | "tag": {"tag_type": 'vlan', "value": 80}, |
||
458 | }, |
||
459 | "uni_z": { |
||
460 | "interface_id": "00:00:00:00:00:00:00:02:2", |
||
461 | "tag": {"tag_type": 'vlan', "value": 1}, |
||
462 | }, |
||
463 | "dynamic_backup_path": True, |
||
464 | "primary_constraints": { |
||
465 | "spf_max_path_cost": 8, |
||
466 | "mandatory_metrics": { |
||
467 | "ownership": "red" |
||
468 | } |
||
469 | }, |
||
470 | "secondary_constraints": { |
||
471 | "spf_attribute": "priority", |
||
472 | "mandatory_metrics": { |
||
473 | "ownership": "blue" |
||
474 | } |
||
475 | } |
||
476 | } |
||
477 | |||
478 | 1 | response = await self.api_client.post(url, json=payload) |
|
479 | 1 | current_data = response.json() |
|
480 | |||
481 | # verify expected result from request |
||
482 | 1 | assert 201 == response.status_code |
|
483 | 1 | assert "circuit_id" in current_data |
|
484 | |||
485 | # verify uni called |
||
486 | 1 | uni_from_dict_mock.called_twice() |
|
487 | 1 | uni_from_dict_mock.assert_any_call(payload["uni_z"]) |
|
488 | 1 | uni_from_dict_mock.assert_any_call(payload["uni_a"]) |
|
489 | |||
490 | # verify validation called |
||
491 | 1 | validate_mock.assert_called_once() |
|
492 | 1 | validate_mock.assert_called_with( |
|
493 | table_group={'evpl': 0, 'epl': 0}, |
||
494 | frequency="* * * * *", |
||
495 | name="my evc1", |
||
496 | uni_a=uni1, |
||
497 | uni_z=uni2, |
||
498 | dynamic_backup_path=True, |
||
499 | primary_constraints=payload["primary_constraints"], |
||
500 | secondary_constraints=payload["secondary_constraints"], |
||
501 | ) |
||
502 | # verify save method is called |
||
503 | 1 | mongo_controller_upsert_mock.assert_called_once() |
|
504 | |||
505 | # verify evc as dict is called to save in the box |
||
506 | 1 | evc_as_dict_mock.assert_called() |
|
507 | # verify add circuit in sched |
||
508 | 1 | sched_add_mock.assert_called_once() |
|
509 | |||
510 | 1 | async def test_create_a_circuit_case_2(self): |
|
511 | """Test create a new circuit trying to send request without a json.""" |
||
512 | 1 | self.napp.controller.loop = asyncio.get_running_loop() |
|
513 | 1 | url = f"{self.base_endpoint}/v2/evc/" |
|
514 | |||
515 | 1 | response = await self.api_client.post(url) |
|
516 | current_data = response.json() |
||
517 | assert 400 == response.status_code |
||
518 | assert "Missing required request body" in current_data["description"] |
||
519 | |||
520 | 1 | async def test_create_a_circuit_case_3(self): |
|
521 | """Test create a new circuit trying to send request with an |
||
522 | invalid json.""" |
||
523 | 1 | self.napp.controller.loop = asyncio.get_running_loop() |
|
524 | 1 | url = f"{self.base_endpoint}/v2/evc/" |
|
525 | |||
526 | 1 | response = await self.api_client.post( |
|
527 | url, |
||
528 | json="This is an {Invalid:} JSON", |
||
529 | ) |
||
530 | current_data = response.json() |
||
531 | assert 400 == response.status_code |
||
532 | assert "This is an {Invalid:" in current_data["description"] |
||
533 | |||
534 | 1 | @patch("napps.kytos.mef_eline.main.Main._uni_from_dict") |
|
535 | 1 | @patch("napps.kytos.mef_eline.controllers.ELineController.upsert_evc") |
|
536 | 1 | async def test_create_a_circuit_case_4( |
|
537 | self, |
||
538 | mongo_controller_upsert_mock, |
||
539 | uni_from_dict_mock |
||
540 | ): |
||
541 | """Test create a new circuit trying to send request with an |
||
542 | invalid value.""" |
||
543 | 1 | self.napp.controller.loop = asyncio.get_running_loop() |
|
544 | # pylint: disable=too-many-locals |
||
545 | 1 | uni_from_dict_mock.side_effect = ValueError("Could not instantiate") |
|
546 | 1 | mongo_controller_upsert_mock.return_value = True |
|
547 | 1 | url = f"{self.base_endpoint}/v2/evc/" |
|
548 | |||
549 | 1 | payload = { |
|
550 | "name": "my evc1", |
||
551 | "frequency": "* * * * *", |
||
552 | "uni_a": { |
||
553 | "interface_id": "00:00:00:00:00:00:00:01:76", |
||
554 | "tag": {"tag_type": 'vlan', "value": 80}, |
||
555 | }, |
||
556 | "uni_z": { |
||
557 | "interface_id": "00:00:00:00:00:00:00:02:2", |
||
558 | "tag": {"tag_type": 'vlan', "value": 1}, |
||
559 | }, |
||
560 | } |
||
561 | |||
562 | 1 | response = await self.api_client.post(url, json=payload) |
|
563 | current_data = response.json() |
||
564 | expected_data = "Error creating UNI: Invalid value" |
||
565 | assert 400 == response.status_code |
||
566 | assert current_data["description"] == expected_data |
||
567 | |||
568 | payload["name"] = 1 |
||
569 | response = await self.api_client.post(url, json=payload) |
||
570 | assert 400 == response.status_code, response.data |
||
571 | |||
572 | 1 | View Code Duplication | @patch("napps.kytos.mef_eline.main.Main._check_no_tag_duplication") |
573 | 1 | @patch("napps.kytos.mef_eline.models.evc.EVC._tag_lists_equal") |
|
574 | 1 | @patch("napps.kytos.mef_eline.main.Main._use_uni_tags") |
|
575 | 1 | @patch("napps.kytos.mef_eline.models.evc.EVC.deploy") |
|
576 | 1 | @patch("napps.kytos.mef_eline.scheduler.Scheduler.add") |
|
577 | 1 | @patch("napps.kytos.mef_eline.main.Main._uni_from_dict") |
|
578 | 1 | @patch("napps.kytos.mef_eline.controllers.ELineController.upsert_evc") |
|
579 | 1 | @patch("napps.kytos.mef_eline.main.EVC.as_dict") |
|
580 | 1 | @patch("napps.kytos.mef_eline.models.evc.EVC._validate") |
|
581 | 1 | async def test_create_a_circuit_case_5( |
|
582 | self, |
||
583 | validate_mock, |
||
584 | evc_as_dict_mock, |
||
585 | mongo_controller_upsert_mock, |
||
586 | uni_from_dict_mock, |
||
587 | sched_add_mock, |
||
588 | evc_deploy_mock, |
||
589 | mock_use_uni_tags, |
||
590 | mock_tags_equal, |
||
591 | mock_check_duplicate |
||
592 | ): |
||
593 | """Test create a new intra circuit with a disabled switch""" |
||
594 | # pylint: disable=too-many-locals |
||
595 | 1 | self.napp.controller.loop = asyncio.get_running_loop() |
|
596 | 1 | validate_mock.return_value = True |
|
597 | 1 | mongo_controller_upsert_mock.return_value = True |
|
598 | 1 | evc_deploy_mock.return_value = True |
|
599 | 1 | mock_use_uni_tags.return_value = True |
|
600 | 1 | mock_tags_equal.return_value = True |
|
601 | 1 | mock_check_duplicate.return_value = True |
|
602 | 1 | uni1 = create_autospec(UNI) |
|
603 | 1 | uni2 = create_autospec(UNI) |
|
604 | 1 | uni1.interface = create_autospec(Interface) |
|
605 | 1 | uni2.interface = create_autospec(Interface) |
|
606 | 1 | switch = MagicMock() |
|
607 | 1 | switch.status = EntityStatus.DISABLED |
|
608 | 1 | switch.return_value = "00:00:00:00:00:00:00:01" |
|
609 | 1 | uni1.interface.switch = switch |
|
610 | 1 | uni2.interface.switch = switch |
|
611 | 1 | uni_from_dict_mock.side_effect = [uni1, uni2] |
|
612 | 1 | evc_as_dict_mock.return_value = {} |
|
613 | 1 | sched_add_mock.return_value = True |
|
614 | 1 | self.napp.mongo_controller.get_circuits.return_value = {} |
|
615 | |||
616 | 1 | url = f"{self.base_endpoint}/v2/evc/" |
|
617 | 1 | payload = { |
|
618 | "name": "my evc1", |
||
619 | "frequency": "* * * * *", |
||
620 | "uni_a": { |
||
621 | "interface_id": "00:00:00:00:00:00:00:01:1", |
||
622 | "tag": {"tag_type": 'vlan', "value": 80}, |
||
623 | }, |
||
624 | "uni_z": { |
||
625 | "interface_id": "00:00:00:00:00:00:00:01:2", |
||
626 | "tag": {"tag_type": 'vlan', "value": 1}, |
||
627 | }, |
||
628 | "dynamic_backup_path": True, |
||
629 | "primary_constraints": { |
||
630 | "spf_max_path_cost": 8, |
||
631 | "mandatory_metrics": { |
||
632 | "ownership": "red" |
||
633 | } |
||
634 | }, |
||
635 | "secondary_constraints": { |
||
636 | "spf_attribute": "priority", |
||
637 | "mandatory_metrics": { |
||
638 | "ownership": "blue" |
||
639 | } |
||
640 | } |
||
641 | } |
||
642 | |||
643 | 1 | response = await self.api_client.post(url, json=payload) |
|
644 | 1 | current_data = response.json() |
|
645 | |||
646 | # verify expected result from request |
||
647 | 1 | assert 201 == response.status_code |
|
648 | 1 | assert "circuit_id" in current_data |
|
649 | |||
650 | # verify uni called |
||
651 | 1 | uni_from_dict_mock.called_twice() |
|
652 | 1 | uni_from_dict_mock.assert_any_call(payload["uni_z"]) |
|
653 | 1 | uni_from_dict_mock.assert_any_call(payload["uni_a"]) |
|
654 | |||
655 | # verify validation called |
||
656 | 1 | validate_mock.assert_called_once() |
|
657 | 1 | validate_mock.assert_called_with( |
|
658 | table_group={'evpl': 0, 'epl': 0}, |
||
659 | frequency="* * * * *", |
||
660 | name="my evc1", |
||
661 | uni_a=uni1, |
||
662 | uni_z=uni2, |
||
663 | dynamic_backup_path=True, |
||
664 | primary_constraints=payload["primary_constraints"], |
||
665 | secondary_constraints=payload["secondary_constraints"], |
||
666 | ) |
||
667 | # verify save method is called |
||
668 | 1 | mongo_controller_upsert_mock.assert_called_once() |
|
669 | |||
670 | # verify evc as dict is called to save in the box |
||
671 | 1 | evc_as_dict_mock.assert_called() |
|
672 | # verify add circuit in sched |
||
673 | 1 | sched_add_mock.assert_called_once() |
|
674 | |||
675 | 1 | View Code Duplication | @patch("napps.kytos.mef_eline.main.Main._check_no_tag_duplication") |
676 | 1 | @patch("napps.kytos.mef_eline.models.evc.EVC._tag_lists_equal") |
|
677 | 1 | @patch("napps.kytos.mef_eline.main.Main._use_uni_tags") |
|
678 | 1 | @patch("napps.kytos.mef_eline.models.evc.EVC.deploy") |
|
679 | 1 | @patch("napps.kytos.mef_eline.scheduler.Scheduler.add") |
|
680 | 1 | @patch("napps.kytos.mef_eline.main.Main._uni_from_dict") |
|
681 | 1 | @patch("napps.kytos.mef_eline.controllers.ELineController.upsert_evc") |
|
682 | 1 | @patch("napps.kytos.mef_eline.main.EVC.as_dict") |
|
683 | 1 | @patch("napps.kytos.mef_eline.models.evc.EVC._validate") |
|
684 | 1 | async def test_create_a_circuit_case_6( |
|
685 | self, |
||
686 | validate_mock, |
||
687 | evc_as_dict_mock, |
||
688 | mongo_controller_upsert_mock, |
||
689 | uni_from_dict_mock, |
||
690 | sched_add_mock, |
||
691 | evc_deploy_mock, |
||
692 | mock_use_uni_tags, |
||
693 | mock_tags_equal, |
||
694 | mock_check_duplicate |
||
695 | ): |
||
696 | """Test create a new intra circuit with a disabled interface""" |
||
697 | # pylint: disable=too-many-locals |
||
698 | 1 | self.napp.controller.loop = asyncio.get_running_loop() |
|
699 | 1 | validate_mock.return_value = True |
|
700 | 1 | mongo_controller_upsert_mock.return_value = True |
|
701 | 1 | evc_deploy_mock.return_value = True |
|
702 | 1 | mock_use_uni_tags.return_value = True |
|
703 | 1 | mock_tags_equal.return_value = True |
|
704 | 1 | mock_check_duplicate.return_value = True |
|
705 | 1 | uni1 = create_autospec(UNI) |
|
706 | 1 | uni2 = create_autospec(UNI) |
|
707 | 1 | uni1.interface = create_autospec(Interface) |
|
708 | 1 | uni2.interface = create_autospec(Interface) |
|
709 | 1 | switch = MagicMock() |
|
710 | 1 | switch.status = EntityStatus.UP |
|
711 | 1 | switch.return_value = "00:00:00:00:00:00:00:01" |
|
712 | 1 | uni1.interface.switch = switch |
|
713 | 1 | uni1.interface.status = EntityStatus.DISABLED |
|
714 | 1 | uni2.interface.switch = switch |
|
715 | 1 | uni_from_dict_mock.side_effect = [uni1, uni2] |
|
716 | 1 | evc_as_dict_mock.return_value = {} |
|
717 | 1 | sched_add_mock.return_value = True |
|
718 | 1 | self.napp.mongo_controller.get_circuits.return_value = {} |
|
719 | |||
720 | 1 | url = f"{self.base_endpoint}/v2/evc/" |
|
721 | 1 | payload = { |
|
722 | "name": "my evc1", |
||
723 | "frequency": "* * * * *", |
||
724 | "uni_a": { |
||
725 | "interface_id": "00:00:00:00:00:00:00:01:1", |
||
726 | "tag": {"tag_type": 'vlan', "value": 80}, |
||
727 | }, |
||
728 | "uni_z": { |
||
729 | "interface_id": "00:00:00:00:00:00:00:01:2", |
||
730 | "tag": {"tag_type": 'vlan', "value": 1}, |
||
731 | }, |
||
732 | "dynamic_backup_path": True, |
||
733 | "primary_constraints": { |
||
734 | "spf_max_path_cost": 8, |
||
735 | "mandatory_metrics": { |
||
736 | "ownership": "red" |
||
737 | } |
||
738 | }, |
||
739 | "secondary_constraints": { |
||
740 | "spf_attribute": "priority", |
||
741 | "mandatory_metrics": { |
||
742 | "ownership": "blue" |
||
743 | } |
||
744 | } |
||
745 | } |
||
746 | |||
747 | 1 | response = await self.api_client.post(url, json=payload) |
|
748 | 1 | current_data = response.json() |
|
749 | |||
750 | # verify expected result from request |
||
751 | 1 | assert 201 == response.status_code |
|
752 | 1 | assert "circuit_id" in current_data |
|
753 | |||
754 | # verify uni called |
||
755 | 1 | uni_from_dict_mock.called_twice() |
|
756 | 1 | uni_from_dict_mock.assert_any_call(payload["uni_z"]) |
|
757 | 1 | uni_from_dict_mock.assert_any_call(payload["uni_a"]) |
|
758 | |||
759 | # verify validation called |
||
760 | 1 | validate_mock.assert_called_once() |
|
761 | 1 | validate_mock.assert_called_with( |
|
762 | table_group={'evpl': 0, 'epl': 0}, |
||
763 | frequency="* * * * *", |
||
764 | name="my evc1", |
||
765 | uni_a=uni1, |
||
766 | uni_z=uni2, |
||
767 | dynamic_backup_path=True, |
||
768 | primary_constraints=payload["primary_constraints"], |
||
769 | secondary_constraints=payload["secondary_constraints"], |
||
770 | ) |
||
771 | # verify save method is called |
||
772 | 1 | mongo_controller_upsert_mock.assert_called_once() |
|
773 | |||
774 | # verify evc as dict is called to save in the box |
||
775 | 1 | evc_as_dict_mock.assert_called() |
|
776 | # verify add circuit in sched |
||
777 | 1 | sched_add_mock.assert_called_once() |
|
778 | |||
779 | 1 | async def test_create_a_circuit_invalid_queue_id(self): |
|
780 | """Test create a new circuit with invalid queue_id.""" |
||
781 | 1 | self.napp.controller.loop = asyncio.get_running_loop() |
|
782 | 1 | url = f"{self.base_endpoint}/v2/evc/" |
|
783 | |||
784 | 1 | payload = { |
|
785 | "name": "my evc1", |
||
786 | "queue_id": 8, |
||
787 | "uni_a": { |
||
788 | "interface_id": "00:00:00:00:00:00:00:01:76", |
||
789 | "tag": {"tag_type": 'vlan', "value": 80}, |
||
790 | }, |
||
791 | "uni_z": { |
||
792 | "interface_id": "00:00:00:00:00:00:00:02:2", |
||
793 | "tag": {"tag_type": 'vlan', "value": 1}, |
||
794 | }, |
||
795 | } |
||
796 | 1 | response = await self.api_client.post(url, json=payload) |
|
797 | current_data = response.json() |
||
798 | expected_data = "8 is greater than the maximum of 7" |
||
799 | |||
800 | assert response.status_code == 400 |
||
801 | assert expected_data in current_data["description"] |
||
802 | |||
803 | 1 | @patch("napps.kytos.mef_eline.main.Main._check_no_tag_duplication") |
|
804 | 1 | @patch("napps.kytos.mef_eline.models.evc.EVC._tag_lists_equal") |
|
805 | 1 | @patch("napps.kytos.mef_eline.main.Main._use_uni_tags") |
|
806 | 1 | @patch("napps.kytos.mef_eline.models.evc.EVC.deploy") |
|
807 | 1 | @patch("napps.kytos.mef_eline.scheduler.Scheduler.add") |
|
808 | 1 | @patch("napps.kytos.mef_eline.main.Main._uni_from_dict") |
|
809 | 1 | @patch("napps.kytos.mef_eline.controllers.ELineController.upsert_evc") |
|
810 | 1 | @patch("napps.kytos.mef_eline.models.evc.EVC._validate") |
|
811 | 1 | @patch("napps.kytos.mef_eline.main.EVC.as_dict") |
|
812 | 1 | async def test_create_circuit_already_enabled( |
|
813 | self, |
||
814 | evc_as_dict_mock, |
||
815 | validate_mock, |
||
816 | mongo_controller_upsert_mock, |
||
817 | uni_from_dict_mock, |
||
818 | sched_add_mock, |
||
819 | evc_deploy_mock, |
||
820 | mock_use_uni_tags, |
||
821 | mock_tags_equal, |
||
822 | mock_check_duplicate |
||
823 | ): |
||
824 | """Test create an already created circuit.""" |
||
825 | # pylint: disable=too-many-locals |
||
826 | 1 | self.napp.controller.loop = asyncio.get_running_loop() |
|
827 | 1 | validate_mock.return_value = True |
|
828 | 1 | mongo_controller_upsert_mock.return_value = True |
|
829 | 1 | sched_add_mock.return_value = True |
|
830 | 1 | evc_deploy_mock.return_value = True |
|
831 | 1 | mock_tags_equal.return_value = True |
|
832 | 1 | mock_check_duplicate.return_value = True |
|
833 | 1 | mock_use_uni_tags.side_effect = [ |
|
834 | None, KytosTagError("The EVC already exists.") |
||
835 | ] |
||
836 | 1 | uni1 = create_autospec(UNI) |
|
837 | 1 | uni2 = create_autospec(UNI) |
|
838 | 1 | uni1.interface = create_autospec(Interface) |
|
839 | 1 | uni2.interface = create_autospec(Interface) |
|
840 | 1 | uni1.interface.switch = "00:00:00:00:00:00:00:01" |
|
841 | 1 | uni2.interface.switch = "00:00:00:00:00:00:00:02" |
|
842 | 1 | uni_from_dict_mock.side_effect = [uni1, uni2, uni1, uni2] |
|
843 | |||
844 | 1 | payload = { |
|
845 | "name": "my evc1", |
||
846 | "uni_a": { |
||
847 | "interface_id": "00:00:00:00:00:00:00:01:1", |
||
848 | "tag": {"tag_type": 'vlan', "value": 80}, |
||
849 | }, |
||
850 | "uni_z": { |
||
851 | "interface_id": "00:00:00:00:00:00:00:02:2", |
||
852 | "tag": {"tag_type": 'vlan', "value": 1}, |
||
853 | }, |
||
854 | "dynamic_backup_path": True, |
||
855 | } |
||
856 | |||
857 | 1 | evc_as_dict_mock.return_value = payload |
|
858 | 1 | response = await self.api_client.post( |
|
859 | f"{self.base_endpoint}/v2/evc/", |
||
860 | json=payload |
||
861 | ) |
||
862 | 1 | assert 201 == response.status_code |
|
863 | |||
864 | 1 | response = await self.api_client.post( |
|
865 | f"{self.base_endpoint}/v2/evc/", |
||
866 | json=payload |
||
867 | ) |
||
868 | current_data = response.json() |
||
869 | expected_data = "The EVC already exists." |
||
870 | assert current_data["description"] == expected_data |
||
871 | assert 400 == response.status_code |
||
872 | |||
873 | 1 | @patch("napps.kytos.mef_eline.models.evc.EVC._tag_lists_equal") |
|
874 | 1 | @patch("napps.kytos.mef_eline.main.Main._uni_from_dict") |
|
875 | 1 | async def test_create_circuit_case_5( |
|
876 | self, |
||
877 | uni_from_dict_mock, |
||
878 | mock_tags_equal |
||
879 | ): |
||
880 | """Test when neither primary path nor dynamic_backup_path is set.""" |
||
881 | 1 | self.napp.controller.loop = asyncio.get_running_loop() |
|
882 | 1 | mock_tags_equal.return_value = True |
|
883 | 1 | url = f"{self.base_endpoint}/v2/evc/" |
|
884 | 1 | uni1 = create_autospec(UNI) |
|
885 | 1 | uni2 = create_autospec(UNI) |
|
886 | 1 | uni1.interface = create_autospec(Interface) |
|
887 | 1 | uni2.interface = create_autospec(Interface) |
|
888 | 1 | uni1.interface.switch = "00:00:00:00:00:00:00:01" |
|
889 | 1 | uni2.interface.switch = "00:00:00:00:00:00:00:02" |
|
890 | 1 | uni_from_dict_mock.side_effect = [uni1, uni2, uni1, uni2] |
|
891 | |||
892 | 1 | payload = { |
|
893 | "name": "my evc1", |
||
894 | "frequency": "* * * * *", |
||
895 | "uni_a": { |
||
896 | "interface_id": "00:00:00:00:00:00:00:01:1", |
||
897 | "tag": {"tag_type": 'vlan', "value": 80}, |
||
898 | }, |
||
899 | "uni_z": { |
||
900 | "interface_id": "00:00:00:00:00:00:00:02:2", |
||
901 | "tag": {"tag_type": 'vlan', "value": 1}, |
||
902 | }, |
||
903 | } |
||
904 | |||
905 | 1 | response = await self.api_client.post(url, json=payload) |
|
906 | current_data = response.json() |
||
907 | expected_data = "The EVC must have a primary path " |
||
908 | expected_data += "or allow dynamic paths." |
||
909 | assert 400 == response.status_code, response.data |
||
910 | assert current_data["description"] == expected_data |
||
911 | |||
912 | 1 | @patch("napps.kytos.mef_eline.main.Main._evc_from_dict") |
|
913 | 1 | async def test_create_circuit_case_6(self, mock_evc): |
|
914 | """Test create_circuit with KytosTagError""" |
||
915 | 1 | self.napp.controller.loop = asyncio.get_running_loop() |
|
916 | 1 | url = f"{self.base_endpoint}/v2/evc/" |
|
917 | 1 | mock_evc.side_effect = KytosTagError("") |
|
918 | 1 | payload = { |
|
919 | "name": "my evc1", |
||
920 | "uni_a": { |
||
921 | "interface_id": "00:00:00:00:00:00:00:01:1", |
||
922 | }, |
||
923 | "uni_z": { |
||
924 | "interface_id": "00:00:00:00:00:00:00:02:2", |
||
925 | }, |
||
926 | } |
||
927 | 1 | response = await self.api_client.post(url, json=payload) |
|
928 | assert response.status_code == 400, response.data |
||
929 | |||
930 | 1 | async def test_redeploy_evc(self): |
|
931 | """Test endpoint to redeploy an EVC.""" |
||
932 | 1 | evc1 = MagicMock() |
|
933 | 1 | evc1.is_enabled.return_value = True |
|
934 | 1 | self.napp.circuits = {"1": evc1, "2": MagicMock()} |
|
935 | 1 | url = f"{self.base_endpoint}/v2/evc/1/redeploy" |
|
936 | 1 | response = await self.api_client.patch(url) |
|
937 | 1 | evc1.remove_failover_flows.assert_called() |
|
938 | 1 | evc1.remove_current_flows.assert_called_with( |
|
939 | sync=False, return_path=True |
||
940 | ) |
||
941 | 1 | assert response.status_code == 202, response.data |
|
942 | |||
943 | 1 | url = f"{self.base_endpoint}/v2/evc/1/redeploy" |
|
944 | 1 | url = url + "?try_avoid_same_s_vlan=false" |
|
945 | 1 | response = await self.api_client.patch(url) |
|
946 | 1 | evc1.remove_current_flows.assert_called_with( |
|
947 | sync=False, return_path=False |
||
948 | ) |
||
949 | |||
950 | 1 | url = f"{self.base_endpoint}/v2/evc/1/redeploy" |
|
951 | 1 | url = url + "?try_avoid_same_s_vlan=True" |
|
952 | 1 | response = await self.api_client.patch(url) |
|
953 | 1 | evc1.remove_current_flows.assert_called_with( |
|
954 | sync=False, return_path=True |
||
955 | ) |
||
956 | |||
957 | 1 | async def test_redeploy_evc_disabled(self): |
|
958 | """Test endpoint to redeploy an EVC.""" |
||
959 | 1 | evc1 = MagicMock() |
|
960 | 1 | evc1.is_enabled.return_value = False |
|
961 | 1 | self.napp.circuits = {"1": evc1, "2": MagicMock()} |
|
962 | 1 | url = f"{self.base_endpoint}/v2/evc/1/redeploy" |
|
963 | 1 | response = await self.api_client.patch(url) |
|
964 | 1 | evc1.remove_failover_flows.assert_not_called() |
|
965 | 1 | evc1.remove_current_flows.assert_not_called() |
|
966 | 1 | assert response.status_code == 409, response.data |
|
967 | |||
968 | 1 | async def test_redeploy_evc_deleted(self): |
|
969 | """Test endpoint to redeploy an EVC.""" |
||
970 | 1 | evc1 = MagicMock() |
|
971 | 1 | evc1.is_enabled.return_value = True |
|
972 | 1 | self.napp.circuits = {"1": evc1, "2": MagicMock()} |
|
973 | 1 | url = f"{self.base_endpoint}/v2/evc/3/redeploy" |
|
974 | 1 | response = await self.api_client.patch(url) |
|
975 | assert response.status_code == 404, response.data |
||
976 | |||
977 | 1 | async def test_list_schedules__no_data_stored(self): |
|
978 | """Test if list circuits return all circuits stored.""" |
||
979 | 1 | self.napp.mongo_controller.get_circuits.return_value = {"circuits": {}} |
|
980 | |||
981 | 1 | url = f"{self.base_endpoint}/v2/evc/schedule" |
|
982 | |||
983 | 1 | response = await self.api_client.get(url) |
|
984 | 1 | assert response.status_code == 200 |
|
985 | 1 | assert not response.json() |
|
986 | |||
987 | 1 | def _add_mongodb_schedule_data(self, data_mock): |
|
988 | """Add schedule data to mongodb mock object.""" |
||
989 | 1 | circuits = {"circuits": {}} |
|
990 | 1 | payload_1 = { |
|
991 | "id": "aa:aa:aa", |
||
992 | "name": "my evc1", |
||
993 | "uni_a": { |
||
994 | "interface_id": "00:00:00:00:00:00:00:01:1", |
||
995 | "tag": {"tag_type": 'vlan', "value": 80}, |
||
996 | }, |
||
997 | "uni_z": { |
||
998 | "interface_id": "00:00:00:00:00:00:00:02:2", |
||
999 | "tag": {"tag_type": 'vlan', "value": 1}, |
||
1000 | }, |
||
1001 | "circuit_scheduler": [ |
||
1002 | {"id": "1", "frequency": "* * * * *", "action": "create"}, |
||
1003 | {"id": "2", "frequency": "1 * * * *", "action": "remove"}, |
||
1004 | ], |
||
1005 | } |
||
1006 | 1 | circuits["circuits"].update({"aa:aa:aa": payload_1}) |
|
1007 | 1 | payload_2 = { |
|
1008 | "id": "bb:bb:bb", |
||
1009 | "name": "my second evc2", |
||
1010 | "uni_a": { |
||
1011 | "interface_id": "00:00:00:00:00:00:00:01:2", |
||
1012 | "tag": {"tag_type": 'vlan', "value": 90}, |
||
1013 | }, |
||
1014 | "uni_z": { |
||
1015 | "interface_id": "00:00:00:00:00:00:00:03:2", |
||
1016 | "tag": {"tag_type": 'vlan', "value": 100}, |
||
1017 | }, |
||
1018 | "circuit_scheduler": [ |
||
1019 | {"id": "3", "frequency": "1 * * * *", "action": "create"}, |
||
1020 | {"id": "4", "frequency": "2 * * * *", "action": "remove"}, |
||
1021 | ], |
||
1022 | } |
||
1023 | 1 | circuits["circuits"].update({"bb:bb:bb": payload_2}) |
|
1024 | 1 | payload_3 = { |
|
1025 | "id": "cc:cc:cc", |
||
1026 | "name": "my third evc3", |
||
1027 | "uni_a": { |
||
1028 | "interface_id": "00:00:00:00:00:00:00:03:1", |
||
1029 | "tag": {"tag_type": 'vlan', "value": 90}, |
||
1030 | }, |
||
1031 | "uni_z": { |
||
1032 | "interface_id": "00:00:00:00:00:00:00:04:2", |
||
1033 | "tag": {"tag_type": 'vlan', "value": 100}, |
||
1034 | }, |
||
1035 | } |
||
1036 | 1 | circuits["circuits"].update({"cc:cc:cc": payload_3}) |
|
1037 | # Add one circuit to the mongodb. |
||
1038 | 1 | data_mock.return_value = circuits |
|
1039 | |||
1040 | 1 | async def test_list_schedules_from_mongodb(self): |
|
1041 | """Test if list circuits return specific circuits stored.""" |
||
1042 | 1 | self._add_mongodb_schedule_data( |
|
1043 | self.napp.mongo_controller.get_circuits |
||
1044 | ) |
||
1045 | |||
1046 | 1 | url = f"{self.base_endpoint}/v2/evc/schedule" |
|
1047 | |||
1048 | # Call URL |
||
1049 | 1 | response = await self.api_client.get(url) |
|
1050 | # Expected JSON data from response |
||
1051 | 1 | expected = [ |
|
1052 | { |
||
1053 | "circuit_id": "aa:aa:aa", |
||
1054 | "schedule": { |
||
1055 | "action": "create", |
||
1056 | "frequency": "* * * * *", |
||
1057 | "id": "1", |
||
1058 | }, |
||
1059 | "schedule_id": "1", |
||
1060 | }, |
||
1061 | { |
||
1062 | "circuit_id": "aa:aa:aa", |
||
1063 | "schedule": { |
||
1064 | "action": "remove", |
||
1065 | "frequency": "1 * * * *", |
||
1066 | "id": "2", |
||
1067 | }, |
||
1068 | "schedule_id": "2", |
||
1069 | }, |
||
1070 | { |
||
1071 | "circuit_id": "bb:bb:bb", |
||
1072 | "schedule": { |
||
1073 | "action": "create", |
||
1074 | "frequency": "1 * * * *", |
||
1075 | "id": "3", |
||
1076 | }, |
||
1077 | "schedule_id": "3", |
||
1078 | }, |
||
1079 | { |
||
1080 | "circuit_id": "bb:bb:bb", |
||
1081 | "schedule": { |
||
1082 | "action": "remove", |
||
1083 | "frequency": "2 * * * *", |
||
1084 | "id": "4", |
||
1085 | }, |
||
1086 | "schedule_id": "4", |
||
1087 | }, |
||
1088 | ] |
||
1089 | |||
1090 | 1 | assert response.status_code == 200 |
|
1091 | 1 | assert expected == response.json() |
|
1092 | |||
1093 | 1 | async def test_get_specific_schedule_from_mongodb(self): |
|
1094 | """Test get schedules from a circuit.""" |
||
1095 | 1 | self._add_mongodb_schedule_data( |
|
1096 | self.napp.mongo_controller.get_circuits |
||
1097 | ) |
||
1098 | |||
1099 | 1 | requested_circuit_id = "bb:bb:bb" |
|
1100 | 1 | evc = self.napp.mongo_controller.get_circuits() |
|
1101 | 1 | evc = evc["circuits"][requested_circuit_id] |
|
1102 | 1 | self.napp.mongo_controller.get_circuit.return_value = evc |
|
1103 | 1 | url = f"{self.base_endpoint}/v2/evc/{requested_circuit_id}" |
|
1104 | |||
1105 | # Call URL |
||
1106 | 1 | response = await self.api_client.get(url) |
|
1107 | |||
1108 | # Expected JSON data from response |
||
1109 | 1 | expected = [ |
|
1110 | {"action": "create", "frequency": "1 * * * *", "id": "3"}, |
||
1111 | {"action": "remove", "frequency": "2 * * * *", "id": "4"}, |
||
1112 | ] |
||
1113 | |||
1114 | 1 | assert response.status_code == 200 |
|
1115 | 1 | assert expected == response.json()["circuit_scheduler"] |
|
1116 | |||
1117 | 1 | async def test_get_specific_schedules_from_mongodb_not_found(self): |
|
1118 | """Test get specific schedule ID that does not exist.""" |
||
1119 | 1 | requested_id = "blah" |
|
1120 | 1 | self.napp.mongo_controller.get_circuit.return_value = None |
|
1121 | 1 | url = f"{self.base_endpoint}/v2/evc/{requested_id}" |
|
1122 | |||
1123 | # Call URL |
||
1124 | 1 | response = await self.api_client.get(url) |
|
1125 | |||
1126 | expected = "circuit_id blah not found" |
||
1127 | # Assert response not found |
||
1128 | assert response.status_code == 404 |
||
1129 | assert expected == response.json()["description"] |
||
1130 | |||
1131 | 1 | def _uni_from_dict_side_effect(self, uni_dict): |
|
1132 | 1 | interface_id = uni_dict.get("interface_id") |
|
1133 | 1 | tag_dict = uni_dict.get("tag") |
|
1134 | 1 | interface = Interface(interface_id, "0", MagicMock(id="1")) |
|
1135 | 1 | return UNI(interface, tag_dict) |
|
1136 | |||
1137 | 1 | @patch("apscheduler.schedulers.background.BackgroundScheduler.add_job") |
|
1138 | 1 | @patch("napps.kytos.mef_eline.scheduler.Scheduler.add") |
|
1139 | 1 | @patch("napps.kytos.mef_eline.main.Main._uni_from_dict") |
|
1140 | 1 | @patch("napps.kytos.mef_eline.controllers.ELineController.upsert_evc") |
|
1141 | 1 | @patch("napps.kytos.mef_eline.main.EVC.as_dict") |
|
1142 | 1 | @patch("napps.kytos.mef_eline.models.evc.EVC._validate") |
|
1143 | 1 | async def test_create_schedule( |
|
1144 | self, |
||
1145 | validate_mock, |
||
1146 | evc_as_dict_mock, |
||
1147 | mongo_controller_upsert_mock, |
||
1148 | uni_from_dict_mock, |
||
1149 | sched_add_mock, |
||
1150 | scheduler_add_job_mock |
||
1151 | ): |
||
1152 | """Test create a circuit schedule.""" |
||
1153 | 1 | self.napp.controller.loop = asyncio.get_running_loop() |
|
1154 | 1 | validate_mock.return_value = True |
|
1155 | 1 | mongo_controller_upsert_mock.return_value = True |
|
1156 | 1 | uni_from_dict_mock.side_effect = self._uni_from_dict_side_effect |
|
1157 | 1 | evc_as_dict_mock.return_value = {} |
|
1158 | 1 | sched_add_mock.return_value = True |
|
1159 | |||
1160 | 1 | self._add_mongodb_schedule_data( |
|
1161 | self.napp.mongo_controller.get_circuits |
||
1162 | ) |
||
1163 | |||
1164 | 1 | requested_id = "bb:bb:bb" |
|
1165 | 1 | url = f"{self.base_endpoint}/v2/evc/schedule/" |
|
1166 | |||
1167 | 1 | payload = { |
|
1168 | "circuit_id": requested_id, |
||
1169 | "schedule": {"frequency": "1 * * * *", "action": "create"}, |
||
1170 | "metadata": {"metadata1": "test_data"}, |
||
1171 | } |
||
1172 | |||
1173 | # Call URL |
||
1174 | 1 | response = await self.api_client.post(url, json=payload) |
|
1175 | 1 | response_json = response.json() |
|
1176 | |||
1177 | 1 | assert response.status_code == 201 |
|
1178 | 1 | scheduler_add_job_mock.assert_called_once() |
|
1179 | 1 | mongo_controller_upsert_mock.assert_called_once() |
|
1180 | 1 | assert payload["schedule"]["frequency"] == response_json["frequency"] |
|
1181 | 1 | assert payload["schedule"]["action"] == response_json["action"] |
|
1182 | 1 | assert response_json["id"] is not None |
|
1183 | |||
1184 | # Case 2: there is no schedule |
||
1185 | 1 | payload = { |
|
1186 | "circuit_id": "cc:cc:cc", |
||
1187 | "schedule": { |
||
1188 | "frequency": "1 * * * *", |
||
1189 | "action": "create" |
||
1190 | } |
||
1191 | } |
||
1192 | 1 | response = await self.api_client.post(url, json=payload) |
|
1193 | 1 | assert response.status_code == 201 |
|
1194 | |||
1195 | 1 | async def test_create_schedule_invalid_request(self): |
|
1196 | """Test create schedule API with invalid request.""" |
||
1197 | 1 | self.napp.controller.loop = asyncio.get_running_loop() |
|
1198 | 1 | evc1 = MagicMock() |
|
1199 | 1 | self.napp.circuits = {'bb:bb:bb': evc1} |
|
1200 | 1 | url = f'{self.base_endpoint}/v2/evc/schedule/' |
|
1201 | |||
1202 | # case 1: empty post |
||
1203 | 1 | response = await self.api_client.post(url, json={}) |
|
1204 | assert response.status_code == 400 |
||
1205 | |||
1206 | # case 2: not a dictionary |
||
1207 | payload = [] |
||
1208 | response = await self.api_client.post(url, json=payload) |
||
1209 | assert response.status_code == 400 |
||
1210 | |||
1211 | # case 3: missing circuit id |
||
1212 | payload = { |
||
1213 | "schedule": { |
||
1214 | "frequency": "1 * * * *", |
||
1215 | "action": "create" |
||
1216 | } |
||
1217 | } |
||
1218 | response = await self.api_client.post(url, json=payload) |
||
1219 | assert response.status_code == 400 |
||
1220 | |||
1221 | # case 4: missing schedule |
||
1222 | payload = { |
||
1223 | "circuit_id": "bb:bb:bb" |
||
1224 | } |
||
1225 | response = await self.api_client.post(url, json=payload) |
||
1226 | assert response.status_code == 400 |
||
1227 | |||
1228 | # case 5: invalid circuit |
||
1229 | payload = { |
||
1230 | "circuit_id": "xx:xx:xx", |
||
1231 | "schedule": { |
||
1232 | "frequency": "1 * * * *", |
||
1233 | "action": "create" |
||
1234 | } |
||
1235 | } |
||
1236 | response = await self.api_client.post(url, json=payload) |
||
1237 | assert response.status_code == 404 |
||
1238 | |||
1239 | # case 6: invalid json |
||
1240 | response = await self.api_client.post(url, json="test") |
||
1241 | assert response.status_code == 400 |
||
1242 | |||
1243 | 1 | @patch('apscheduler.schedulers.background.BackgroundScheduler.remove_job') |
|
1244 | 1 | @patch('napps.kytos.mef_eline.scheduler.Scheduler.add') |
|
1245 | 1 | @patch('napps.kytos.mef_eline.main.Main._uni_from_dict') |
|
1246 | 1 | @patch("napps.kytos.mef_eline.controllers.ELineController.upsert_evc") |
|
1247 | 1 | @patch('napps.kytos.mef_eline.main.EVC.as_dict') |
|
1248 | 1 | @patch('napps.kytos.mef_eline.models.evc.EVC._validate') |
|
1249 | 1 | async def test_update_schedule( |
|
1250 | self, |
||
1251 | validate_mock, |
||
1252 | evc_as_dict_mock, |
||
1253 | mongo_controller_upsert_mock, |
||
1254 | uni_from_dict_mock, |
||
1255 | sched_add_mock, |
||
1256 | scheduler_remove_job_mock |
||
1257 | ): |
||
1258 | """Test create a circuit schedule.""" |
||
1259 | 1 | self.napp.controller.loop = asyncio.get_running_loop() |
|
1260 | 1 | mongo_payload_1 = { |
|
1261 | "circuits": { |
||
1262 | "aa:aa:aa": { |
||
1263 | "id": "aa:aa:aa", |
||
1264 | "name": "my evc1", |
||
1265 | "uni_a": { |
||
1266 | "interface_id": "00:00:00:00:00:00:00:01:1", |
||
1267 | "tag": {"tag_type": 'vlan', "value": 80}, |
||
1268 | }, |
||
1269 | "uni_z": { |
||
1270 | "interface_id": "00:00:00:00:00:00:00:02:2", |
||
1271 | "tag": {"tag_type": 'vlan', "value": 1}, |
||
1272 | }, |
||
1273 | "circuit_scheduler": [ |
||
1274 | { |
||
1275 | "id": "1", |
||
1276 | "frequency": "* * * * *", |
||
1277 | "action": "create" |
||
1278 | } |
||
1279 | ], |
||
1280 | } |
||
1281 | } |
||
1282 | } |
||
1283 | |||
1284 | 1 | validate_mock.return_value = True |
|
1285 | 1 | mongo_controller_upsert_mock.return_value = True |
|
1286 | 1 | sched_add_mock.return_value = True |
|
1287 | 1 | uni_from_dict_mock.side_effect = ["uni_a", "uni_z"] |
|
1288 | 1 | evc_as_dict_mock.return_value = {} |
|
1289 | 1 | self.napp.mongo_controller.get_circuits.return_value = mongo_payload_1 |
|
1290 | 1 | scheduler_remove_job_mock.return_value = True |
|
1291 | |||
1292 | 1 | requested_schedule_id = "1" |
|
1293 | 1 | url = f"{self.base_endpoint}/v2/evc/schedule/{requested_schedule_id}" |
|
1294 | |||
1295 | 1 | payload = {"frequency": "*/1 * * * *", "action": "create"} |
|
1296 | |||
1297 | # Call URL |
||
1298 | 1 | response = await self.api_client.patch(url, json=payload) |
|
1299 | 1 | response_json = response.json() |
|
1300 | |||
1301 | 1 | assert response.status_code == 200 |
|
1302 | 1 | scheduler_remove_job_mock.assert_called_once() |
|
1303 | 1 | mongo_controller_upsert_mock.assert_called_once() |
|
1304 | 1 | assert payload["frequency"] == response_json["frequency"] |
|
1305 | 1 | assert payload["action"] == response_json["action"] |
|
1306 | 1 | assert response_json["id"] is not None |
|
1307 | |||
1308 | 1 | @patch('napps.kytos.mef_eline.main.Main._find_evc_by_schedule_id') |
|
1309 | 1 | async def test_update_no_schedule( |
|
1310 | self, find_evc_by_schedule_id_mock |
||
1311 | ): |
||
1312 | """Test update a circuit schedule.""" |
||
1313 | 1 | self.napp.controller.loop = asyncio.get_running_loop() |
|
1314 | 1 | url = f"{self.base_endpoint}/v2/evc/schedule/1" |
|
1315 | 1 | payload = {"frequency": "*/1 * * * *", "action": "create"} |
|
1316 | |||
1317 | 1 | find_evc_by_schedule_id_mock.return_value = None, None |
|
1318 | |||
1319 | 1 | response = await self.api_client.patch(url, json=payload) |
|
1320 | assert response.status_code == 404 |
||
1321 | |||
1322 | 1 | @patch("apscheduler.schedulers.background.BackgroundScheduler.remove_job") |
|
1323 | 1 | @patch("napps.kytos.mef_eline.main.Main._uni_from_dict") |
|
1324 | 1 | @patch("napps.kytos.mef_eline.controllers.ELineController.upsert_evc") |
|
1325 | 1 | @patch("napps.kytos.mef_eline.main.EVC.as_dict") |
|
1326 | 1 | @patch("napps.kytos.mef_eline.models.evc.EVC._validate") |
|
1327 | 1 | async def test_delete_schedule(self, *args): |
|
1328 | """Test create a circuit schedule.""" |
||
1329 | 1 | ( |
|
1330 | validate_mock, |
||
1331 | evc_as_dict_mock, |
||
1332 | mongo_controller_upsert_mock, |
||
1333 | uni_from_dict_mock, |
||
1334 | scheduler_remove_job_mock, |
||
1335 | ) = args |
||
1336 | |||
1337 | 1 | mongo_payload_1 = { |
|
1338 | "circuits": { |
||
1339 | "2": { |
||
1340 | "id": "2", |
||
1341 | "name": "my evc1", |
||
1342 | "uni_a": { |
||
1343 | "interface_id": "00:00:00:00:00:00:00:01:1", |
||
1344 | "tag": {"tag_type": 'vlan', "value": 80}, |
||
1345 | }, |
||
1346 | "uni_z": { |
||
1347 | "interface_id": "00:00:00:00:00:00:00:02:2", |
||
1348 | "tag": {"tag_type": 'vlan', "value": 1}, |
||
1349 | }, |
||
1350 | "circuit_scheduler": [ |
||
1351 | { |
||
1352 | "id": "1", |
||
1353 | "frequency": "* * * * *", |
||
1354 | "action": "create" |
||
1355 | } |
||
1356 | ], |
||
1357 | } |
||
1358 | } |
||
1359 | } |
||
1360 | 1 | validate_mock.return_value = True |
|
1361 | 1 | mongo_controller_upsert_mock.return_value = True |
|
1362 | 1 | uni_from_dict_mock.side_effect = ["uni_a", "uni_z"] |
|
1363 | 1 | evc_as_dict_mock.return_value = {} |
|
1364 | 1 | self.napp.mongo_controller.get_circuits.return_value = mongo_payload_1 |
|
1365 | 1 | scheduler_remove_job_mock.return_value = True |
|
1366 | |||
1367 | 1 | requested_schedule_id = "1" |
|
1368 | 1 | url = f"{self.base_endpoint}/v2/evc/schedule/{requested_schedule_id}" |
|
1369 | |||
1370 | # Call URL |
||
1371 | 1 | response = await self.api_client.delete(url) |
|
1372 | |||
1373 | 1 | assert response.status_code == 200 |
|
1374 | 1 | scheduler_remove_job_mock.assert_called_once() |
|
1375 | 1 | mongo_controller_upsert_mock.assert_called_once() |
|
1376 | 1 | assert "Schedule removed" in f"{response.json()}" |
|
1377 | |||
1378 | 1 | @patch('napps.kytos.mef_eline.main.Main._find_evc_by_schedule_id') |
|
1379 | 1 | async def test_delete_schedule_not_found(self, mock_find_evc_by_sched): |
|
1380 | """Test delete a circuit schedule - unexisting.""" |
||
1381 | 1 | mock_find_evc_by_sched.return_value = (None, False) |
|
1382 | 1 | url = f'{self.base_endpoint}/v2/evc/schedule/1' |
|
1383 | 1 | response = await self.api_client.delete(url) |
|
1384 | assert response.status_code == 404 |
||
1385 | |||
1386 | 1 | def test_get_evcs_by_svc_level(self) -> None: |
|
1387 | """Test get_evcs_by_svc_level.""" |
||
1388 | 1 | levels = [1, 2, 4, 2, 7] |
|
1389 | 1 | evcs = {i: MagicMock(service_level=v, creation_time=1) |
|
1390 | for i, v in enumerate(levels)} |
||
1391 | 1 | self.napp.circuits = evcs |
|
1392 | 1 | expected_levels = sorted(levels, reverse=True) |
|
1393 | 1 | evcs_by_level = self.napp.get_evcs_by_svc_level() |
|
1394 | 1 | assert evcs_by_level |
|
1395 | |||
1396 | 1 | for evc, exp_level in zip(evcs_by_level, expected_levels): |
|
1397 | 1 | assert evc.service_level == exp_level |
|
1398 | |||
1399 | 1 | evcs = {i: MagicMock(service_level=1, creation_time=i) |
|
1400 | for i in reversed(range(2))} |
||
1401 | 1 | self.napp.circuits = evcs |
|
1402 | 1 | evcs_by_level = self.napp.get_evcs_by_svc_level() |
|
1403 | 1 | for i in range(2): |
|
1404 | 1 | assert evcs_by_level[i].creation_time == i |
|
1405 | |||
1406 | 1 | self.napp.circuits[1].is_enabled = lambda: False |
|
1407 | 1 | evcs_by_level = self.napp.get_evcs_by_svc_level() |
|
1408 | 1 | assert len(evcs_by_level) == 1 |
|
1409 | |||
1410 | 1 | self.napp.circuits[1].is_enabled = lambda: False |
|
1411 | 1 | evcs_by_level = self.napp.get_evcs_by_svc_level(enable_filter=False) |
|
1412 | 1 | assert len(evcs_by_level) == 2 |
|
1413 | |||
1414 | 1 | async def test_get_circuit_not_found(self): |
|
1415 | """Test /v2/evc/<circuit_id> 404.""" |
||
1416 | 1 | self.napp.mongo_controller.get_circuit.return_value = None |
|
1417 | 1 | url = f'{self.base_endpoint}/v2/evc/1234' |
|
1418 | 1 | response = await self.api_client.get(url) |
|
1419 | assert response.status_code == 404 |
||
1420 | |||
1421 | 1 | @patch('httpx.post') |
|
1422 | 1 | @patch("napps.kytos.mef_eline.main.Main._use_uni_tags") |
|
1423 | 1 | @patch('napps.kytos.mef_eline.scheduler.Scheduler.add') |
|
1424 | 1 | @patch('napps.kytos.mef_eline.controllers.ELineController.update_evc') |
|
1425 | 1 | @patch("napps.kytos.mef_eline.controllers.ELineController.upsert_evc") |
|
1426 | 1 | @patch('napps.kytos.mef_eline.models.evc.EVC._validate') |
|
1427 | 1 | @patch('kytos.core.Controller.get_interface_by_id') |
|
1428 | 1 | @patch('napps.kytos.mef_eline.models.path.Path.is_valid') |
|
1429 | 1 | @patch('napps.kytos.mef_eline.models.evc.EVCDeploy.deploy') |
|
1430 | 1 | @patch('napps.kytos.mef_eline.main.Main._uni_from_dict') |
|
1431 | 1 | @patch('napps.kytos.mef_eline.main.EVC.as_dict') |
|
1432 | 1 | async def test_update_circuit( |
|
1433 | self, |
||
1434 | evc_as_dict_mock, |
||
1435 | uni_from_dict_mock, |
||
1436 | evc_deploy, |
||
1437 | _is_valid_mock, |
||
1438 | interface_by_id_mock, |
||
1439 | _mock_validate, |
||
1440 | _mongo_controller_upsert_mock, |
||
1441 | _mongo_controller_update_mock, |
||
1442 | _sched_add_mock, |
||
1443 | mock_use_uni_tags, |
||
1444 | httpx_mock |
||
1445 | ): |
||
1446 | """Test update a circuit circuit.""" |
||
1447 | 1 | self.napp.controller.loop = asyncio.get_running_loop() |
|
1448 | 1 | mock_use_uni_tags.return_value = True |
|
1449 | 1 | evc_deploy.return_value = True |
|
1450 | 1 | interface_by_id_mock.return_value = get_uni_mocked().interface |
|
1451 | 1 | unis = [ |
|
1452 | get_uni_mocked(switch_dpid="00:00:00:00:00:00:00:01"), |
||
1453 | get_uni_mocked(switch_dpid="00:00:00:00:00:00:00:02"), |
||
1454 | ] |
||
1455 | 1 | uni_from_dict_mock.side_effect = 2 * unis |
|
1456 | |||
1457 | 1 | response = MagicMock() |
|
1458 | 1 | response.status_code = 201 |
|
1459 | 1 | httpx_mock.return_value = response |
|
1460 | |||
1461 | 1 | payloads = [ |
|
1462 | { |
||
1463 | "name": "my evc1", |
||
1464 | "uni_a": { |
||
1465 | "interface_id": "00:00:00:00:00:00:00:01:1", |
||
1466 | "tag": {"tag_type": 'vlan', "value": 80}, |
||
1467 | }, |
||
1468 | "uni_z": { |
||
1469 | "interface_id": "00:00:00:00:00:00:00:02:2", |
||
1470 | "tag": {"tag_type": 'vlan', "value": 1}, |
||
1471 | }, |
||
1472 | "dynamic_backup_path": True, |
||
1473 | }, |
||
1474 | { |
||
1475 | "primary_path": [ |
||
1476 | { |
||
1477 | "endpoint_a": {"id": "00:00:00:00:00:00:00:01:1"}, |
||
1478 | "endpoint_b": {"id": "00:00:00:00:00:00:00:02:2"}, |
||
1479 | } |
||
1480 | ] |
||
1481 | }, |
||
1482 | { |
||
1483 | "sb_priority": 3 |
||
1484 | }, |
||
1485 | { |
||
1486 | # It works only with 'enabled' and not with 'enable' |
||
1487 | "enabled": True |
||
1488 | }, |
||
1489 | { |
||
1490 | "sb_priority": 100 |
||
1491 | } |
||
1492 | ] |
||
1493 | |||
1494 | 1 | evc_as_dict_mock.return_value = payloads[0] |
|
1495 | 1 | response = await self.api_client.post( |
|
1496 | f"{self.base_endpoint}/v2/evc/", |
||
1497 | json=payloads[0], |
||
1498 | ) |
||
1499 | 1 | assert 201 == response.status_code |
|
1500 | |||
1501 | 1 | evc_deploy.reset_mock() |
|
1502 | 1 | evc_as_dict_mock.return_value = payloads[1] |
|
1503 | 1 | current_data = response.json() |
|
1504 | 1 | circuit_id = current_data["circuit_id"] |
|
1505 | 1 | response = await self.api_client.patch( |
|
1506 | f"{self.base_endpoint}/v2/evc/{circuit_id}", |
||
1507 | json=payloads[1], |
||
1508 | ) |
||
1509 | # evc_deploy.assert_called_once() |
||
1510 | 1 | assert 200 == response.status_code |
|
1511 | |||
1512 | 1 | evc_deploy.reset_mock() |
|
1513 | 1 | evc_as_dict_mock.return_value = payloads[2] |
|
1514 | 1 | response = await self.api_client.patch( |
|
1515 | f"{self.base_endpoint}/v2/evc/{circuit_id}", |
||
1516 | json=payloads[2], |
||
1517 | ) |
||
1518 | 1 | evc_deploy.assert_not_called() |
|
1519 | 1 | assert 200 == response.status_code |
|
1520 | |||
1521 | 1 | evc_deploy.reset_mock() |
|
1522 | 1 | evc_as_dict_mock.return_value = payloads[3] |
|
1523 | 1 | response = await self.api_client.patch( |
|
1524 | f"{self.base_endpoint}/v2/evc/{circuit_id}", |
||
1525 | json=payloads[3], |
||
1526 | ) |
||
1527 | 1 | evc_deploy.assert_called_once() |
|
1528 | 1 | assert 200 == response.status_code |
|
1529 | |||
1530 | 1 | evc_deploy.reset_mock() |
|
1531 | 1 | response = await self.api_client.patch( |
|
1532 | f"{self.base_endpoint}/v2/evc/{circuit_id}", |
||
1533 | content=b'{"priority":5,}', |
||
1534 | headers={"Content-Type": "application/json"} |
||
1535 | ) |
||
1536 | evc_deploy.assert_not_called() |
||
1537 | assert 400 == response.status_code |
||
1538 | |||
1539 | response = await self.api_client.patch( |
||
1540 | f"{self.base_endpoint}/v2/evc/1234", |
||
1541 | json=payloads[1], |
||
1542 | ) |
||
1543 | current_data = response.json() |
||
1544 | expected_data = "circuit_id 1234 not found" |
||
1545 | assert current_data["description"] == expected_data |
||
1546 | assert 404 == response.status_code |
||
1547 | |||
1548 | self.napp.circuits[circuit_id]._active = False |
||
1549 | evc_deploy.reset_mock() |
||
1550 | response = await self.api_client.patch( |
||
1551 | f"{self.base_endpoint}/v2/evc/{circuit_id}", |
||
1552 | json=payloads[4] |
||
1553 | ) |
||
1554 | 1 | assert 200 == response.status_code |
|
1555 | 1 | evc_deploy.assert_called_once() |
|
1556 | |||
1557 | 1 | View Code Duplication | @patch("napps.kytos.mef_eline.main.Main._check_no_tag_duplication") |
1558 | 1 | @patch("napps.kytos.mef_eline.models.evc.EVC._tag_lists_equal") |
|
1559 | 1 | @patch("napps.kytos.mef_eline.main.Main._use_uni_tags") |
|
1560 | 1 | @patch("napps.kytos.mef_eline.models.evc.EVC.deploy") |
|
1561 | 1 | @patch("napps.kytos.mef_eline.scheduler.Scheduler.add") |
|
1562 | 1 | @patch("napps.kytos.mef_eline.main.Main._uni_from_dict") |
|
1563 | 1 | @patch("napps.kytos.mef_eline.controllers.ELineController.upsert_evc") |
|
1564 | 1 | @patch("napps.kytos.mef_eline.models.evc.EVC._validate") |
|
1565 | 1 | @patch("napps.kytos.mef_eline.main.EVC.as_dict") |
|
1566 | 1 | async def test_update_circuit_invalid_json( |
|
1567 | self, |
||
1568 | evc_as_dict_mock, |
||
1569 | validate_mock, |
||
1570 | mongo_controller_upsert_mock, |
||
1571 | uni_from_dict_mock, |
||
1572 | sched_add_mock, |
||
1573 | evc_deploy_mock, |
||
1574 | mock_use_uni_tags, |
||
1575 | mock_tags_equal, |
||
1576 | mock_check_duplicate |
||
1577 | ): |
||
1578 | """Test update a circuit circuit.""" |
||
1579 | 1 | self.napp.controller.loop = asyncio.get_running_loop() |
|
1580 | 1 | validate_mock.return_value = True |
|
1581 | 1 | mongo_controller_upsert_mock.return_value = True |
|
1582 | 1 | sched_add_mock.return_value = True |
|
1583 | 1 | evc_deploy_mock.return_value = True |
|
1584 | 1 | mock_use_uni_tags.return_value = True |
|
1585 | 1 | mock_tags_equal.return_value = True |
|
1586 | 1 | mock_check_duplicate.return_value = True |
|
1587 | 1 | uni1 = create_autospec(UNI) |
|
1588 | 1 | uni2 = create_autospec(UNI) |
|
1589 | 1 | uni1.interface = create_autospec(Interface) |
|
1590 | 1 | uni2.interface = create_autospec(Interface) |
|
1591 | 1 | uni1.interface.switch = "00:00:00:00:00:00:00:01" |
|
1592 | 1 | uni2.interface.switch = "00:00:00:00:00:00:00:02" |
|
1593 | 1 | uni_from_dict_mock.side_effect = [uni1, uni2, uni1, uni2] |
|
1594 | |||
1595 | 1 | payload1 = { |
|
1596 | "name": "my evc1", |
||
1597 | "uni_a": { |
||
1598 | "interface_id": "00:00:00:00:00:00:00:01:1", |
||
1599 | "tag": {"tag_type": 'vlan', "value": 80}, |
||
1600 | }, |
||
1601 | "uni_z": { |
||
1602 | "interface_id": "00:00:00:00:00:00:00:02:2", |
||
1603 | "tag": {"tag_type": 'vlan', "value": 1}, |
||
1604 | }, |
||
1605 | "dynamic_backup_path": True, |
||
1606 | } |
||
1607 | |||
1608 | 1 | payload2 = { |
|
1609 | "dynamic_backup_path": False, |
||
1610 | } |
||
1611 | |||
1612 | 1 | evc_as_dict_mock.return_value = payload1 |
|
1613 | 1 | response = await self.api_client.post( |
|
1614 | f"{self.base_endpoint}/v2/evc/", |
||
1615 | json=payload1 |
||
1616 | ) |
||
1617 | 1 | assert 201 == response.status_code |
|
1618 | |||
1619 | 1 | evc_as_dict_mock.return_value = payload2 |
|
1620 | 1 | current_data = response.json() |
|
1621 | 1 | circuit_id = current_data["circuit_id"] |
|
1622 | 1 | response = await self.api_client.patch( |
|
1623 | f"{self.base_endpoint}/v2/evc/{circuit_id}", |
||
1624 | json=payload2 |
||
1625 | ) |
||
1626 | current_data = response.json() |
||
1627 | assert 400 == response.status_code |
||
1628 | assert "must have a primary path or" in current_data["description"] |
||
1629 | |||
1630 | 1 | @patch("napps.kytos.mef_eline.main.Main._check_no_tag_duplication") |
|
1631 | 1 | @patch("napps.kytos.mef_eline.models.evc.EVC._tag_lists_equal") |
|
1632 | 1 | @patch("napps.kytos.mef_eline.main.Main._use_uni_tags") |
|
1633 | 1 | @patch("napps.kytos.mef_eline.models.evc.EVC.deploy") |
|
1634 | 1 | @patch("napps.kytos.mef_eline.scheduler.Scheduler.add") |
|
1635 | 1 | @patch("napps.kytos.mef_eline.main.Main._uni_from_dict") |
|
1636 | 1 | @patch("napps.kytos.mef_eline.main.Main._link_from_dict") |
|
1637 | 1 | @patch("napps.kytos.mef_eline.controllers.ELineController.upsert_evc") |
|
1638 | 1 | @patch("napps.kytos.mef_eline.models.evc.EVC._validate") |
|
1639 | 1 | @patch("napps.kytos.mef_eline.main.EVC.as_dict") |
|
1640 | 1 | @patch("napps.kytos.mef_eline.models.path.Path.is_valid") |
|
1641 | 1 | async def test_update_circuit_invalid_path( |
|
1642 | self, |
||
1643 | is_valid_mock, |
||
1644 | evc_as_dict_mock, |
||
1645 | validate_mock, |
||
1646 | mongo_controller_upsert_mock, |
||
1647 | link_from_dict_mock, |
||
1648 | uni_from_dict_mock, |
||
1649 | sched_add_mock, |
||
1650 | evc_deploy_mock, |
||
1651 | mock_use_uni_tags, |
||
1652 | mock_tags_equal, |
||
1653 | mock_check_duplicate |
||
1654 | ): |
||
1655 | """Test update a circuit circuit.""" |
||
1656 | 1 | self.napp.controller.loop = asyncio.get_running_loop() |
|
1657 | 1 | is_valid_mock.side_effect = InvalidPath("error") |
|
1658 | 1 | validate_mock.return_value = True |
|
1659 | 1 | mongo_controller_upsert_mock.return_value = True |
|
1660 | 1 | sched_add_mock.return_value = True |
|
1661 | 1 | evc_deploy_mock.return_value = True |
|
1662 | 1 | mock_use_uni_tags.return_value = True |
|
1663 | 1 | link_from_dict_mock.return_value = 1 |
|
1664 | 1 | mock_tags_equal.return_value = True |
|
1665 | 1 | mock_check_duplicate.return_value = True |
|
1666 | 1 | uni1 = create_autospec(UNI) |
|
1667 | 1 | uni2 = create_autospec(UNI) |
|
1668 | 1 | uni1.interface = create_autospec(Interface) |
|
1669 | 1 | uni2.interface = create_autospec(Interface) |
|
1670 | 1 | uni1.interface.switch = "00:00:00:00:00:00:00:01" |
|
1671 | 1 | uni2.interface.switch = "00:00:00:00:00:00:00:02" |
|
1672 | 1 | uni_from_dict_mock.side_effect = [uni1, uni2, uni1, uni2] |
|
1673 | |||
1674 | 1 | payload1 = { |
|
1675 | "name": "my evc1", |
||
1676 | "uni_a": { |
||
1677 | "interface_id": "00:00:00:00:00:00:00:01:1", |
||
1678 | "tag": {"tag_type": 'vlan', "value": 80}, |
||
1679 | }, |
||
1680 | "uni_z": { |
||
1681 | "interface_id": "00:00:00:00:00:00:00:02:2", |
||
1682 | "tag": {"tag_type": 'vlan', "value": 1}, |
||
1683 | }, |
||
1684 | "dynamic_backup_path": True, |
||
1685 | } |
||
1686 | |||
1687 | 1 | payload2 = { |
|
1688 | "primary_path": [ |
||
1689 | { |
||
1690 | "endpoint_a": {"id": "00:00:00:00:00:00:00:01:1"}, |
||
1691 | "endpoint_b": {"id": "00:00:00:00:00:00:00:02:2"}, |
||
1692 | } |
||
1693 | ] |
||
1694 | } |
||
1695 | |||
1696 | 1 | evc_as_dict_mock.return_value = payload1 |
|
1697 | 1 | response = await self.api_client.post( |
|
1698 | f"{self.base_endpoint}/v2/evc/", |
||
1699 | json=payload1, |
||
1700 | ) |
||
1701 | 1 | assert 201 == response.status_code |
|
1702 | |||
1703 | 1 | evc_as_dict_mock.return_value = payload2 |
|
1704 | 1 | current_data = response.json() |
|
1705 | 1 | circuit_id = current_data["circuit_id"] |
|
1706 | 1 | response = await self.api_client.patch( |
|
1707 | f"{self.base_endpoint}/v2/evc/{circuit_id}", |
||
1708 | json=payload2, |
||
1709 | ) |
||
1710 | current_data = response.json() |
||
1711 | expected_data = "primary_path is not a valid path: error" |
||
1712 | assert 400 == response.status_code |
||
1713 | assert current_data["description"] == expected_data |
||
1714 | |||
1715 | 1 | def test_link_from_dict_non_existent_intf(self): |
|
1716 | """Test _link_from_dict non existent intf.""" |
||
1717 | 1 | self.napp.controller.get_interface_by_id = MagicMock(return_value=None) |
|
1718 | 1 | link_dict = { |
|
1719 | "endpoint_a": {"id": "a"}, |
||
1720 | "endpoint_b": {"id": "b"} |
||
1721 | } |
||
1722 | 1 | with pytest.raises(ValueError): |
|
1723 | 1 | self.napp._link_from_dict(link_dict, "current_path") |
|
1724 | |||
1725 | 1 | def test_link_from_dict_vlan_metadata(self): |
|
1726 | """Test that link_from_dict only accepts vlans for current_path |
||
1727 | and failover_path.""" |
||
1728 | 1 | intf = MagicMock(id="01:1") |
|
1729 | 1 | self.napp.controller.get_interface_by_id = MagicMock(return_value=intf) |
|
1730 | 1 | link_dict = { |
|
1731 | 'id': 'mock_link', |
||
1732 | 'endpoint_a': {'id': '00:00:00:00:00:00:00:01:4'}, |
||
1733 | 'endpoint_b': {'id': '00:00:00:00:00:00:00:05:2'}, |
||
1734 | 'metadata': {'s_vlan': {'tag_type': 'vlan', 'value': 1}} |
||
1735 | } |
||
1736 | 1 | link = self.napp._link_from_dict(link_dict, "current_path") |
|
1737 | 1 | assert link.metadata.get('s_vlan', None) |
|
1738 | |||
1739 | 1 | link = self.napp._link_from_dict(link_dict, "failover_path") |
|
1740 | 1 | assert link.metadata.get('s_vlan', None) |
|
1741 | |||
1742 | 1 | link = self.napp._link_from_dict(link_dict, "primary_path") |
|
1743 | 1 | assert link.metadata.get('s_vlan', None) is None |
|
1744 | |||
1745 | 1 | def test_uni_from_dict_non_existent_intf(self): |
|
1746 | """Test _link_from_dict non existent intf.""" |
||
1747 | 1 | self.napp.controller.get_interface_by_id = MagicMock(return_value=None) |
|
1748 | 1 | uni_dict = { |
|
1749 | "interface_id": "aaa", |
||
1750 | } |
||
1751 | 1 | with pytest.raises(ValueError): |
|
1752 | 1 | self.napp._uni_from_dict(uni_dict) |
|
1753 | |||
1754 | 1 | View Code Duplication | @patch("napps.kytos.mef_eline.main.Main._check_no_tag_duplication") |
1755 | 1 | @patch("napps.kytos.mef_eline.models.evc.EVC._tag_lists_equal") |
|
1756 | 1 | @patch("napps.kytos.mef_eline.main.Main._use_uni_tags") |
|
1757 | 1 | @patch("napps.kytos.mef_eline.models.evc.EVC.deploy") |
|
1758 | 1 | @patch("napps.kytos.mef_eline.scheduler.Scheduler.add") |
|
1759 | 1 | @patch("napps.kytos.mef_eline.main.Main._uni_from_dict") |
|
1760 | 1 | @patch("napps.kytos.mef_eline.models.evc.EVC._validate") |
|
1761 | 1 | @patch("napps.kytos.mef_eline.controllers.ELineController.upsert_evc") |
|
1762 | 1 | async def test_update_evc_no_json_mime( |
|
1763 | self, |
||
1764 | mongo_controller_upsert_mock, |
||
1765 | validate_mock, |
||
1766 | uni_from_dict_mock, |
||
1767 | sched_add_mock, |
||
1768 | evc_deploy_mock, |
||
1769 | mock_use_uni_tags, |
||
1770 | mock_tags_equal, |
||
1771 | mock_check_duplicate |
||
1772 | ): |
||
1773 | """Test update a circuit with wrong mimetype.""" |
||
1774 | 1 | self.napp.controller.loop = asyncio.get_running_loop() |
|
1775 | 1 | validate_mock.return_value = True |
|
1776 | 1 | sched_add_mock.return_value = True |
|
1777 | 1 | evc_deploy_mock.return_value = True |
|
1778 | 1 | mock_use_uni_tags.return_value = True |
|
1779 | 1 | mock_tags_equal.return_value = True |
|
1780 | 1 | mock_check_duplicate.return_value = True |
|
1781 | 1 | uni1 = create_autospec(UNI) |
|
1782 | 1 | uni2 = create_autospec(UNI) |
|
1783 | 1 | uni1.interface = create_autospec(Interface) |
|
1784 | 1 | uni2.interface = create_autospec(Interface) |
|
1785 | 1 | uni1.interface.switch = "00:00:00:00:00:00:00:01" |
|
1786 | 1 | uni2.interface.switch = "00:00:00:00:00:00:00:02" |
|
1787 | 1 | uni_from_dict_mock.side_effect = [uni1, uni2, uni1, uni2] |
|
1788 | 1 | mongo_controller_upsert_mock.return_value = True |
|
1789 | |||
1790 | 1 | payload1 = { |
|
1791 | "name": "my evc1", |
||
1792 | "uni_a": { |
||
1793 | "interface_id": "00:00:00:00:00:00:00:01:1", |
||
1794 | "tag": {"tag_type": 'vlan', "value": 80}, |
||
1795 | }, |
||
1796 | "uni_z": { |
||
1797 | "interface_id": "00:00:00:00:00:00:00:02:2", |
||
1798 | "tag": {"tag_type": 'vlan', "value": 1}, |
||
1799 | }, |
||
1800 | "dynamic_backup_path": True, |
||
1801 | } |
||
1802 | |||
1803 | 1 | payload2 = {"dynamic_backup_path": False} |
|
1804 | |||
1805 | 1 | response = await self.api_client.post( |
|
1806 | f"{self.base_endpoint}/v2/evc/", |
||
1807 | json=payload1, |
||
1808 | ) |
||
1809 | 1 | assert 201 == response.status_code |
|
1810 | |||
1811 | 1 | current_data = response.json() |
|
1812 | 1 | circuit_id = current_data["circuit_id"] |
|
1813 | 1 | response = await self.api_client.patch( |
|
1814 | f"{self.base_endpoint}/v2/evc/{circuit_id}", data=payload2 |
||
1815 | ) |
||
1816 | current_data = response.json() |
||
1817 | assert 415 == response.status_code |
||
1818 | assert "application/json" in current_data["description"] |
||
1819 | |||
1820 | 1 | async def test_delete_no_evc(self): |
|
1821 | """Test delete when EVC does not exist.""" |
||
1822 | 1 | url = f"{self.base_endpoint}/v2/evc/123" |
|
1823 | 1 | response = await self.api_client.delete(url) |
|
1824 | current_data = response.json() |
||
1825 | expected_data = "circuit_id 123 not found" |
||
1826 | assert current_data["description"] == expected_data |
||
1827 | assert 404 == response.status_code |
||
1828 | |||
1829 | 1 | @patch("napps.kytos.mef_eline.main.Main._check_no_tag_duplication") |
|
1830 | 1 | @patch("napps.kytos.mef_eline.models.evc.EVC._tag_lists_equal") |
|
1831 | 1 | @patch("napps.kytos.mef_eline.models.evc.EVC.remove_uni_tags") |
|
1832 | 1 | @patch("napps.kytos.mef_eline.main.Main._use_uni_tags") |
|
1833 | 1 | @patch("napps.kytos.mef_eline.models.evc.EVC.remove_current_flows") |
|
1834 | 1 | @patch("napps.kytos.mef_eline.models.evc.EVC.deploy") |
|
1835 | 1 | @patch("napps.kytos.mef_eline.scheduler.Scheduler.add") |
|
1836 | 1 | @patch("napps.kytos.mef_eline.main.Main._uni_from_dict") |
|
1837 | 1 | @patch("napps.kytos.mef_eline.controllers.ELineController.upsert_evc") |
|
1838 | 1 | @patch("napps.kytos.mef_eline.models.evc.EVC._validate") |
|
1839 | 1 | @patch("napps.kytos.mef_eline.main.EVC.as_dict") |
|
1840 | 1 | async def test_delete_archived_evc( |
|
1841 | self, |
||
1842 | evc_as_dict_mock, |
||
1843 | validate_mock, |
||
1844 | mongo_controller_upsert_mock, |
||
1845 | uni_from_dict_mock, |
||
1846 | sched_add_mock, |
||
1847 | evc_deploy_mock, |
||
1848 | remove_current_flows_mock, |
||
1849 | mock_use_uni, |
||
1850 | mock_remove_tags, |
||
1851 | mock_tags_equal, |
||
1852 | mock_check_duplicate |
||
1853 | ): |
||
1854 | """Try to delete an archived EVC""" |
||
1855 | 1 | self.napp.controller.loop = asyncio.get_running_loop() |
|
1856 | 1 | validate_mock.return_value = True |
|
1857 | 1 | mongo_controller_upsert_mock.return_value = True |
|
1858 | 1 | sched_add_mock.return_value = True |
|
1859 | 1 | evc_deploy_mock.return_value = True |
|
1860 | 1 | mock_use_uni.return_value = True |
|
1861 | 1 | mock_tags_equal.return_value = True |
|
1862 | 1 | mock_check_duplicate.return_value = True |
|
1863 | 1 | uni1 = create_autospec(UNI) |
|
1864 | 1 | uni2 = create_autospec(UNI) |
|
1865 | 1 | uni1.interface = create_autospec(Interface) |
|
1866 | 1 | uni2.interface = create_autospec(Interface) |
|
1867 | 1 | uni1.interface.switch = "00:00:00:00:00:00:00:01" |
|
1868 | 1 | uni2.interface.switch = "00:00:00:00:00:00:00:02" |
|
1869 | 1 | uni_from_dict_mock.side_effect = [uni1, uni2, uni1, uni2] |
|
1870 | |||
1871 | 1 | payload1 = { |
|
1872 | "name": "my evc1", |
||
1873 | "uni_a": { |
||
1874 | "interface_id": "00:00:00:00:00:00:00:01:1", |
||
1875 | "tag": {"tag_type": 'vlan', "value": 80}, |
||
1876 | }, |
||
1877 | "uni_z": { |
||
1878 | "interface_id": "00:00:00:00:00:00:00:02:2", |
||
1879 | "tag": {"tag_type": 'vlan', "value": 1}, |
||
1880 | }, |
||
1881 | "dynamic_backup_path": True, |
||
1882 | } |
||
1883 | |||
1884 | 1 | evc_as_dict_mock.return_value = payload1 |
|
1885 | 1 | response = await self.api_client.post( |
|
1886 | f"{self.base_endpoint}/v2/evc/", |
||
1887 | json=payload1 |
||
1888 | ) |
||
1889 | 1 | assert 201 == response.status_code |
|
1890 | 1 | assert len(self.napp.circuits) == 1 |
|
1891 | 1 | current_data = response.json() |
|
1892 | 1 | circuit_id = current_data["circuit_id"] |
|
1893 | 1 | self.napp.circuits[circuit_id].archive() |
|
1894 | |||
1895 | 1 | response = await self.api_client.delete( |
|
1896 | f"{self.base_endpoint}/v2/evc/{circuit_id}" |
||
1897 | ) |
||
1898 | 1 | assert 200 == response.status_code |
|
1899 | 1 | assert mock_remove_tags.call_count == 0 |
|
1900 | 1 | assert remove_current_flows_mock.call_count == 0 |
|
1901 | 1 | assert len(self.napp.circuits) == 0 |
|
1902 | |||
1903 | 1 | response = await self.api_client.delete( |
|
1904 | f"{self.base_endpoint}/v2/evc/{circuit_id}" |
||
1905 | ) |
||
1906 | current_data = response.json() |
||
1907 | expected_data = f"circuit_id {circuit_id} not found" |
||
1908 | assert current_data["description"] == expected_data |
||
1909 | assert 404 == response.status_code |
||
1910 | assert len(self.napp.circuits) == 0 |
||
1911 | |||
1912 | 1 | @patch("napps.kytos.mef_eline.main.emit_event") |
|
1913 | 1 | async def test_delete_circuit(self, emit_event_mock): |
|
1914 | """Test delete_circuit""" |
||
1915 | 1 | evc = MagicMock() |
|
1916 | 1 | evc.archived = False |
|
1917 | 1 | circuit_id = '1' |
|
1918 | 1 | self.napp.circuits[circuit_id] = evc |
|
1919 | 1 | response = await self.api_client.delete( |
|
1920 | f"{self.base_endpoint}/v2/evc/{circuit_id}" |
||
1921 | ) |
||
1922 | 1 | assert 200 == response.status_code |
|
1923 | 1 | assert evc.deactivate.call_count == 1 |
|
1924 | 1 | assert evc.disable.call_count == 1 |
|
1925 | 1 | evc.remove_current_flows.assert_called_once_with( |
|
1926 | sync=False |
||
1927 | ) |
||
1928 | 1 | evc.remove_failover_flows.assert_called_once_with( |
|
1929 | sync=False |
||
1930 | ) |
||
1931 | 1 | assert evc.archive.call_count == 1 |
|
1932 | 1 | assert evc.remove_uni_tags.call_count == 1 |
|
1933 | 1 | assert evc.sync.call_count == 1 |
|
1934 | 1 | assert not self.napp.circuits |
|
1935 | 1 | assert emit_event_mock.call_count == 1 |
|
1936 | |||
1937 | 1 | def test_handle_link_up(self): |
|
1938 | """Test handle_link_up method.""" |
||
1939 | 1 | evc_mock = create_autospec(EVC) |
|
1940 | 1 | evc_mock.service_level, evc_mock.creation_time = 0, 1 |
|
1941 | 1 | evc_mock.is_enabled = MagicMock(side_effect=[ |
|
1942 | True, False, True, True, True |
||
1943 | ]) |
||
1944 | 1 | evc_mock.lock = MagicMock() |
|
1945 | 1 | evc_mock.archived = False |
|
1946 | 1 | evcs = [evc_mock, evc_mock, evc_mock] |
|
1947 | 1 | event = KytosEvent(name="test", content={"link": "abc"}) |
|
1948 | 1 | self.napp.circuits = dict(zip(["1", "2", "3"], evcs)) |
|
1949 | 1 | self.napp.handle_link_up(event) |
|
1950 | 1 | assert evc_mock.handle_link_up.call_count == 2 |
|
1951 | 1 | evc_mock.handle_link_up.assert_called_with("abc") |
|
1952 | |||
1953 | 1 | def test_handle_link_down( |
|
1954 | self |
||
1955 | ): |
||
1956 | """Test handle_link_down method.""" |
||
1957 | 1 | evc1 = MagicMock(id="1") |
|
1958 | 1 | evc1.is_affected_by_link.return_value = True |
|
1959 | 1 | evc1.is_failover_path_affected_by_link.return_value = True |
|
1960 | |||
1961 | 1 | evc2 = MagicMock(id="2") |
|
1962 | 1 | evc2.is_affected_by_link.return_value = True |
|
1963 | 1 | evc2.failover_path = Path([]) |
|
1964 | |||
1965 | 1 | default_undeploy = [evc1, evc2] |
|
1966 | |||
1967 | 1 | evc3 = MagicMock(id="3") |
|
1968 | 1 | evc3.is_affected_by_link.return_value = True |
|
1969 | 1 | evc3.is_failover_path_affected_by_link.return_value = False |
|
1970 | |||
1971 | 1 | evc4 = MagicMock(id="4") |
|
1972 | 1 | evc4.is_affected_by_link.return_value = True |
|
1973 | 1 | evc4.is_failover_path_affected_by_link.return_value = False |
|
1974 | |||
1975 | 1 | default_swap_to_failover = [evc3, evc4] |
|
1976 | |||
1977 | 1 | evc5 = MagicMock(id="5") |
|
1978 | 1 | evc5.is_affected_by_link.return_value = False |
|
1979 | 1 | evc5.is_failover_path_affected_by_link.return_value = True |
|
1980 | |||
1981 | 1 | evc6 = MagicMock(id="6") |
|
1982 | 1 | evc6.is_affected_by_link.return_value = False |
|
1983 | 1 | evc6.is_failover_path_affected_by_link.return_value = True |
|
1984 | |||
1985 | 1 | default_clear_failover = [evc5, evc6] |
|
1986 | |||
1987 | 1 | self.napp.get_evcs_by_svc_level = MagicMock() |
|
1988 | |||
1989 | 1 | self.napp.get_evcs_by_svc_level.return_value = [ |
|
1990 | evc1, |
||
1991 | evc2, |
||
1992 | evc3, |
||
1993 | evc4, |
||
1994 | evc5, |
||
1995 | evc6, |
||
1996 | ] |
||
1997 | |||
1998 | 1 | self.napp.execute_swap_to_failover = MagicMock() |
|
1999 | |||
2000 | 1 | swap_to_failover_success = [evc3] |
|
2001 | 1 | swap_to_failover_failure = [evc4] |
|
2002 | |||
2003 | 1 | self.napp.execute_swap_to_failover.return_value =\ |
|
2004 | swap_to_failover_success, swap_to_failover_failure |
||
2005 | |||
2006 | 1 | self.napp.execute_clear_failover = MagicMock() |
|
2007 | |||
2008 | 1 | clear_failover_success = [evc5, evc3] |
|
2009 | 1 | clear_failover_failure = [evc6] |
|
2010 | |||
2011 | 1 | self.napp.execute_clear_failover.return_value =\ |
|
2012 | clear_failover_success, clear_failover_failure |
||
2013 | |||
2014 | 1 | self.napp.execute_undeploy = MagicMock() |
|
2015 | |||
2016 | 1 | undeploy_success = [evc1, evc4, evc6] |
|
2017 | 1 | undeploy_failure = [evc2] |
|
2018 | |||
2019 | 1 | self.napp.execute_undeploy.return_value =\ |
|
2020 | undeploy_success, undeploy_failure |
||
2021 | |||
2022 | 1 | link = MagicMock(id="123") |
|
2023 | 1 | event = KytosEvent(name="test", content={"link": link}) |
|
2024 | |||
2025 | 1 | self.napp.handle_link_down(event) |
|
2026 | |||
2027 | 1 | self.napp.execute_swap_to_failover.assert_called_with( |
|
2028 | default_swap_to_failover |
||
2029 | ) |
||
2030 | |||
2031 | 1 | self.napp.execute_clear_failover.assert_called_with( |
|
2032 | [*default_clear_failover, *swap_to_failover_success] |
||
2033 | ) |
||
2034 | |||
2035 | 1 | self.napp.execute_undeploy.assert_called_with([ |
|
2036 | *default_undeploy, |
||
2037 | *swap_to_failover_failure, |
||
2038 | *clear_failover_failure |
||
2039 | ]) |
||
2040 | |||
2041 | 1 | self.napp.mongo_controller.update_evcs.assert_called_with([ |
|
2042 | evc3.as_dict(), |
||
2043 | evc5.as_dict(), |
||
2044 | evc1.as_dict(), |
||
2045 | evc4.as_dict(), |
||
2046 | evc6.as_dict(), |
||
2047 | ]) |
||
2048 | |||
2049 | 1 | View Code Duplication | @patch("napps.kytos.mef_eline.main.emit_event") |
2050 | 1 | @patch("napps.kytos.mef_eline.main.send_flow_mods_http") |
|
2051 | 1 | def test_execute_swap_to_failover( |
|
2052 | self, |
||
2053 | send_flow_mods_mock: MagicMock, |
||
2054 | emit_main_mock: MagicMock, |
||
2055 | ): |
||
2056 | """Test execute_swap_to_failover method.""" |
||
2057 | 1 | evc1 = MagicMock(id="1") |
|
2058 | 1 | good_path = MagicMock(id="GoodPath") |
|
2059 | 1 | bad_path = MagicMock(id="BadPath") |
|
2060 | 1 | evc1.current_path = bad_path |
|
2061 | 1 | evc1.failover_path = good_path |
|
2062 | 1 | evc2 = MagicMock(id="2") |
|
2063 | |||
2064 | 1 | self.napp.prepare_swap_to_failover_flow = { |
|
2065 | evc1: {"1": ["Flow1"]}, |
||
2066 | evc2: None |
||
2067 | }.get |
||
2068 | |||
2069 | 1 | self.napp.prepare_swap_to_failover_event = { |
|
2070 | evc1: "FailoverEvent1", |
||
2071 | }.get |
||
2072 | |||
2073 | 1 | success, failure = self.napp.execute_swap_to_failover([evc1, evc2]) |
|
2074 | |||
2075 | 1 | assert success == [evc1] |
|
2076 | 1 | assert failure == [evc2] |
|
2077 | |||
2078 | 1 | send_flow_mods_mock.assert_called_with( |
|
2079 | {"1": ["Flow1"]}, |
||
2080 | "install" |
||
2081 | ) |
||
2082 | |||
2083 | 1 | assert evc1.current_path == good_path |
|
2084 | 1 | assert evc1.failover_path == bad_path |
|
2085 | |||
2086 | 1 | emit_main_mock.assert_called_with( |
|
2087 | self.napp.controller, |
||
2088 | "failover_link_down", |
||
2089 | content={"1": "FailoverEvent1"} |
||
2090 | ) |
||
2091 | |||
2092 | 1 | View Code Duplication | @patch("napps.kytos.mef_eline.main.emit_event") |
2093 | 1 | @patch("napps.kytos.mef_eline.main.send_flow_mods_http") |
|
2094 | 1 | def test_execute_swap_to_failover_exception( |
|
2095 | self, |
||
2096 | send_flow_mods_mock: MagicMock, |
||
2097 | emit_main_mock: MagicMock, |
||
2098 | ): |
||
2099 | """Test handle_link_down method when an exception occurs.""" |
||
2100 | 1 | evc1 = MagicMock(id="1") |
|
2101 | 1 | good_path = MagicMock(id="GoodPath") |
|
2102 | 1 | bad_path = MagicMock(id="BadPath") |
|
2103 | 1 | evc1.current_path = bad_path |
|
2104 | 1 | evc1.failover_path = good_path |
|
2105 | 1 | evc2 = MagicMock(id="2") |
|
2106 | |||
2107 | 1 | self.napp.prepare_swap_to_failover_flow = { |
|
2108 | evc1: {"1": ["Flow1"]}, |
||
2109 | evc2: None |
||
2110 | }.get |
||
2111 | |||
2112 | 1 | self.napp.prepare_swap_to_failover_event = { |
|
2113 | evc1: "FailoverEvent1", |
||
2114 | }.get |
||
2115 | |||
2116 | 1 | send_flow_mods_mock.side_effect = FlowModException( |
|
2117 | "Flowmod failed to send" |
||
2118 | ) |
||
2119 | |||
2120 | 1 | success, failure = self.napp.execute_swap_to_failover([evc1, evc2]) |
|
2121 | |||
2122 | 1 | assert success == [] |
|
2123 | 1 | assert failure == [evc1, evc2] |
|
2124 | |||
2125 | 1 | send_flow_mods_mock.assert_called_with( |
|
2126 | {"1": ["Flow1"]}, |
||
2127 | "install" |
||
2128 | ) |
||
2129 | |||
2130 | 1 | assert evc1.current_path == bad_path |
|
2131 | 1 | assert evc1.failover_path == good_path |
|
2132 | |||
2133 | 1 | emit_main_mock.assert_not_called() |
|
2134 | |||
2135 | 1 | @patch("napps.kytos.mef_eline.main.emit_event") |
|
2136 | 1 | @patch("napps.kytos.mef_eline.main.send_flow_mods_http") |
|
2137 | 1 | def test_execute_clear_failover( |
|
2138 | self, |
||
2139 | send_flow_mods_mock: MagicMock, |
||
2140 | emit_main_mock: MagicMock, |
||
2141 | ): |
||
2142 | """Test execute_clear_failover method.""" |
||
2143 | 1 | evc1 = MagicMock(id="1") |
|
2144 | 1 | good_path = MagicMock(id="GoodPath") |
|
2145 | 1 | bad_path = MagicMock(id="BadPath") |
|
2146 | 1 | evc1.current_path = good_path |
|
2147 | 1 | evc1.failover_path = bad_path |
|
2148 | 1 | evc2 = MagicMock(id="2") |
|
2149 | |||
2150 | 1 | self.napp.prepare_clear_failover_flow = { |
|
2151 | evc1: {"1": ["Flow1"]}, |
||
2152 | evc2: None |
||
2153 | }.get |
||
2154 | |||
2155 | 1 | self.napp.prepare_clear_failover_event = { |
|
2156 | evc1: "FailoverEvent1", |
||
2157 | }.get |
||
2158 | |||
2159 | 1 | success, failure = self.napp.execute_clear_failover([evc1, evc2]) |
|
2160 | |||
2161 | 1 | assert success == [evc1] |
|
2162 | 1 | assert failure == [evc2] |
|
2163 | |||
2164 | 1 | send_flow_mods_mock.assert_called_with( |
|
2165 | {"1": ["Flow1"]}, |
||
2166 | "delete" |
||
2167 | ) |
||
2168 | |||
2169 | 1 | assert evc1.current_path == good_path |
|
2170 | 1 | assert not evc1.failover_path |
|
2171 | |||
2172 | 1 | bad_path.make_vlans_available.assert_called_once_with( |
|
2173 | self.napp.controller |
||
2174 | ) |
||
2175 | |||
2176 | 1 | emit_main_mock.assert_called_with( |
|
2177 | self.napp.controller, |
||
2178 | "failover_old_path", |
||
2179 | content={"1": "FailoverEvent1"} |
||
2180 | ) |
||
2181 | |||
2182 | 1 | @patch("napps.kytos.mef_eline.main.emit_event") |
|
2183 | 1 | @patch("napps.kytos.mef_eline.main.send_flow_mods_http") |
|
2184 | 1 | def test_execute_clear_failover_exception( |
|
2185 | self, |
||
2186 | send_flow_mods_mock: MagicMock, |
||
2187 | emit_main_mock: MagicMock, |
||
2188 | ): |
||
2189 | """Test execute_clear_failover method when an exception occurs.""" |
||
2190 | 1 | evc1 = MagicMock(id="1") |
|
2191 | 1 | good_path = MagicMock(id="GoodPath") |
|
2192 | 1 | bad_path = MagicMock(id="BadPath") |
|
2193 | 1 | evc1.current_path = good_path |
|
2194 | 1 | evc1.failover_path = bad_path |
|
2195 | 1 | evc2 = MagicMock(id="2") |
|
2196 | |||
2197 | 1 | self.napp.prepare_clear_failover_flow = { |
|
2198 | evc1: {"1": ["Flow1"]}, |
||
2199 | evc2: None |
||
2200 | }.get |
||
2201 | |||
2202 | 1 | self.napp.prepare_clear_failover_event = { |
|
2203 | evc1: "FailoverEvent1", |
||
2204 | }.get |
||
2205 | |||
2206 | 1 | send_flow_mods_mock.side_effect = FlowModException( |
|
2207 | "Flowmod failed to send" |
||
2208 | ) |
||
2209 | |||
2210 | 1 | success, failure = self.napp.execute_clear_failover([evc1, evc2]) |
|
2211 | |||
2212 | 1 | assert success == [] |
|
2213 | 1 | assert failure == [evc1, evc2] |
|
2214 | |||
2215 | 1 | send_flow_mods_mock.assert_called_with( |
|
2216 | {"1": ["Flow1"]}, |
||
2217 | "delete" |
||
2218 | ) |
||
2219 | |||
2220 | 1 | assert evc1.current_path == good_path |
|
2221 | 1 | assert evc1.failover_path == bad_path |
|
2222 | |||
2223 | 1 | emit_main_mock.assert_not_called() |
|
2224 | |||
2225 | 1 | @patch("napps.kytos.mef_eline.main.emit_event") |
|
2226 | 1 | @patch("napps.kytos.mef_eline.main.send_flow_mods_http") |
|
2227 | 1 | def test_execute_undeploy( |
|
2228 | self, |
||
2229 | send_flow_mods_mock: MagicMock, |
||
2230 | emit_main_mock: MagicMock, |
||
2231 | ): |
||
2232 | """Test execute_undeploy method.""" |
||
2233 | 1 | evc1 = MagicMock(id="1") |
|
2234 | 1 | bad_path1 = MagicMock(id="GoodPath") |
|
2235 | 1 | bad_path2 = MagicMock(id="BadPath") |
|
2236 | 1 | evc1.current_path = bad_path1 |
|
2237 | 1 | evc1.failover_path = bad_path2 |
|
2238 | 1 | evc2 = MagicMock(id="2") |
|
2239 | |||
2240 | 1 | self.napp.prepare_undeploy_flow = { |
|
2241 | evc1: {"1": ["Flow1"]}, |
||
2242 | evc2: None |
||
2243 | }.get |
||
2244 | |||
2245 | 1 | success, failure = self.napp.execute_undeploy([evc1, evc2]) |
|
2246 | |||
2247 | 1 | assert success == [evc1] |
|
2248 | 1 | assert failure == [evc2] |
|
2249 | |||
2250 | 1 | send_flow_mods_mock.assert_called_with( |
|
2251 | {"1": ["Flow1"]}, |
||
2252 | "delete" |
||
2253 | ) |
||
2254 | |||
2255 | 1 | assert not evc1.current_path |
|
2256 | 1 | assert not evc1.failover_path |
|
2257 | |||
2258 | 1 | assert evc2.current_path |
|
2259 | 1 | assert evc2.failover_path |
|
2260 | |||
2261 | 1 | bad_path1.make_vlans_available.assert_called_once_with( |
|
2262 | self.napp.controller |
||
2263 | ) |
||
2264 | 1 | bad_path2.make_vlans_available.assert_called_once_with( |
|
2265 | self.napp.controller |
||
2266 | ) |
||
2267 | |||
2268 | 1 | evc1.deactivate.assert_called() |
|
2269 | 1 | evc2.deactivate.assert_not_called() |
|
2270 | |||
2271 | 1 | emit_main_mock.assert_called_with( |
|
2272 | self.napp.controller, |
||
2273 | "need_redeploy", |
||
2274 | content={"evc_id": "1"} |
||
2275 | ) |
||
2276 | |||
2277 | 1 | @patch("napps.kytos.mef_eline.main.emit_event") |
|
2278 | 1 | @patch("napps.kytos.mef_eline.main.send_flow_mods_http") |
|
2279 | 1 | def test_execute_undeploy_exception( |
|
2280 | self, |
||
2281 | send_flow_mods_mock: MagicMock, |
||
2282 | emit_main_mock: MagicMock, |
||
2283 | ): |
||
2284 | """Test execute_undeploy method method when an exception occurs.""" |
||
2285 | 1 | evc1 = MagicMock(id="1") |
|
2286 | 1 | bad_path1 = MagicMock(id="GoodPath") |
|
2287 | 1 | bad_path2 = MagicMock(id="BadPath") |
|
2288 | 1 | evc1.current_path = bad_path1 |
|
2289 | 1 | evc1.failover_path = bad_path2 |
|
2290 | 1 | evc2 = MagicMock(id="2") |
|
2291 | |||
2292 | 1 | self.napp.prepare_undeploy_flow = { |
|
2293 | evc1: {"1": ["Flow1"]}, |
||
2294 | evc2: None |
||
2295 | }.get |
||
2296 | |||
2297 | 1 | send_flow_mods_mock.side_effect = FlowModException( |
|
2298 | "Flowmod failed to send" |
||
2299 | ) |
||
2300 | |||
2301 | 1 | success, failure = self.napp.execute_undeploy([evc1, evc2]) |
|
2302 | |||
2303 | 1 | assert success == [] |
|
2304 | 1 | assert failure == [evc1, evc2] |
|
2305 | |||
2306 | 1 | send_flow_mods_mock.assert_called_with( |
|
2307 | {"1": ["Flow1"]}, |
||
2308 | "delete" |
||
2309 | ) |
||
2310 | |||
2311 | 1 | assert evc1.current_path |
|
2312 | 1 | assert evc1.failover_path |
|
2313 | |||
2314 | 1 | assert evc2.current_path |
|
2315 | 1 | assert evc2.failover_path |
|
2316 | |||
2317 | 1 | bad_path1.make_vlans_available.assert_not_called() |
|
2318 | 1 | bad_path2.make_vlans_available.assert_not_called() |
|
2319 | |||
2320 | 1 | evc1.deactivate.assert_not_called() |
|
2321 | 1 | evc2.deactivate.assert_not_called() |
|
2322 | |||
2323 | 1 | emit_main_mock.assert_not_called() |
|
2324 | |||
2325 | 1 | @patch("napps.kytos.mef_eline.main.emit_event") |
|
2326 | 1 | def test_handle_evc_affected_by_link_down(self, emit_event_mock): |
|
2327 | """Test handle_evc_affected_by_link_down method.""" |
||
2328 | 1 | uni = create_autospec(UNI) |
|
2329 | 1 | evc1 = MagicMock( |
|
2330 | id="1", |
||
2331 | metadata="data_mocked", |
||
2332 | _active="true", |
||
2333 | _enabled="false", |
||
2334 | uni_a=uni, |
||
2335 | uni_z=uni, |
||
2336 | ) |
||
2337 | 1 | evc1.name = "name_mocked" |
|
2338 | 1 | evc1.handle_link_down.return_value = True |
|
2339 | 1 | evc2 = MagicMock( |
|
2340 | id="2", |
||
2341 | metadata="mocked_data", |
||
2342 | _active="false", |
||
2343 | _enabled="true", |
||
2344 | uni_a=uni, |
||
2345 | uni_z=uni, |
||
2346 | ) |
||
2347 | 1 | evc2.name = "mocked_name" |
|
2348 | 1 | evc2.handle_link_down.return_value = False |
|
2349 | 1 | self.napp.circuits = {"1": evc1, "2": evc2} |
|
2350 | |||
2351 | 1 | event = KytosEvent(name="e1", content={ |
|
2352 | "evc_id": "3", |
||
2353 | "link": MagicMock(), |
||
2354 | }) |
||
2355 | 1 | self.napp.handle_evc_affected_by_link_down(event) |
|
2356 | 1 | emit_event_mock.assert_not_called() |
|
2357 | 1 | event.content["evc_id"] = "1" |
|
2358 | 1 | self.napp.handle_evc_affected_by_link_down(event) |
|
2359 | 1 | emit_event_mock.assert_called_with( |
|
2360 | self.napp.controller, "redeployed_link_down", content={ |
||
2361 | "id": "1", |
||
2362 | "evc_id": "1", |
||
2363 | "name": "name_mocked", |
||
2364 | "metadata": "data_mocked", |
||
2365 | "active": "true", |
||
2366 | "enabled": "false", |
||
2367 | "uni_a": uni.as_dict(), |
||
2368 | "uni_z": uni.as_dict(), |
||
2369 | } |
||
2370 | ) |
||
2371 | |||
2372 | 1 | event.content["evc_id"] = "2" |
|
2373 | 1 | self.napp.handle_evc_affected_by_link_down(event) |
|
2374 | 1 | emit_event_mock.assert_called_with( |
|
2375 | self.napp.controller, "error_redeploy_link_down", content={ |
||
2376 | "evc_id": "2", |
||
2377 | "id": "2", |
||
2378 | "name": "mocked_name", |
||
2379 | "metadata": "mocked_data", |
||
2380 | "active": "false", |
||
2381 | "enabled": "true", |
||
2382 | "uni_a": uni.as_dict(), |
||
2383 | "uni_z": uni.as_dict(), |
||
2384 | } |
||
2385 | ) |
||
2386 | |||
2387 | 1 | async def test_add_metadata(self): |
|
2388 | """Test method to add metadata""" |
||
2389 | 1 | self.napp.controller.loop = asyncio.get_running_loop() |
|
2390 | 1 | evc_mock = create_autospec(EVC) |
|
2391 | 1 | evc_mock.metadata = {} |
|
2392 | 1 | evc_mock.id = 1234 |
|
2393 | 1 | self.napp.circuits = {"1234": evc_mock} |
|
2394 | |||
2395 | 1 | payload = {"metadata1": 1, "metadata2": 2} |
|
2396 | 1 | response = await self.api_client.post( |
|
2397 | f"{self.base_endpoint}/v2/evc/1234/metadata", |
||
2398 | json=payload |
||
2399 | ) |
||
2400 | |||
2401 | 1 | assert response.status_code == 201 |
|
2402 | 1 | evc_mock.extend_metadata.assert_called_with(payload) |
|
2403 | |||
2404 | 1 | async def test_add_metadata_malformed_json(self): |
|
2405 | """Test method to add metadata with a malformed json""" |
||
2406 | 1 | self.napp.controller.loop = asyncio.get_running_loop() |
|
2407 | 1 | payload = b'{"metadata1": 1, "metadata2": 2,}' |
|
2408 | 1 | response = await self.api_client.post( |
|
2409 | f"{self.base_endpoint}/v2/evc/1234/metadata", |
||
2410 | content=payload, |
||
2411 | headers={"Content-Type": "application/json"} |
||
2412 | ) |
||
2413 | |||
2414 | assert response.status_code == 400 |
||
2415 | assert "body contains invalid API" in response.json()["description"] |
||
2416 | |||
2417 | 1 | async def test_add_metadata_no_body(self): |
|
2418 | """Test method to add metadata with no body""" |
||
2419 | 1 | self.napp.controller.loop = asyncio.get_running_loop() |
|
2420 | 1 | response = await self.api_client.post( |
|
2421 | f"{self.base_endpoint}/v2/evc/1234/metadata" |
||
2422 | ) |
||
2423 | assert response.status_code == 400 |
||
2424 | assert response.json()["description"] == \ |
||
2425 | "Missing required request body" |
||
2426 | |||
2427 | 1 | async def test_add_metadata_no_evc(self): |
|
2428 | """Test method to add metadata with no evc""" |
||
2429 | 1 | self.napp.controller.loop = asyncio.get_running_loop() |
|
2430 | 1 | payload = {"metadata1": 1, "metadata2": 2} |
|
2431 | 1 | response = await self.api_client.post( |
|
2432 | f"{self.base_endpoint}/v2/evc/1234/metadata", |
||
2433 | json=payload, |
||
2434 | ) |
||
2435 | assert response.status_code == 404 |
||
2436 | assert response.json()["description"] == \ |
||
2437 | "circuit_id 1234 not found." |
||
2438 | |||
2439 | 1 | async def test_add_metadata_wrong_content_type(self): |
|
2440 | """Test method to add metadata with wrong content type""" |
||
2441 | 1 | self.napp.controller.loop = asyncio.get_running_loop() |
|
2442 | 1 | payload = {"metadata1": 1, "metadata2": 2} |
|
2443 | 1 | response = await self.api_client.post( |
|
2444 | f"{self.base_endpoint}/v2/evc/1234/metadata", |
||
2445 | data=payload, |
||
2446 | headers={"Content-Type": "application/xml"} |
||
2447 | ) |
||
2448 | assert response.status_code == 415 |
||
2449 | assert "application/xml" in response.json()["description"] |
||
2450 | |||
2451 | 1 | async def test_get_metadata(self): |
|
2452 | """Test method to get metadata""" |
||
2453 | 1 | evc_mock = create_autospec(EVC) |
|
2454 | 1 | evc_mock.metadata = {'metadata1': 1, 'metadata2': 2} |
|
2455 | 1 | evc_mock.id = 1234 |
|
2456 | 1 | self.napp.circuits = {"1234": evc_mock} |
|
2457 | |||
2458 | 1 | response = await self.api_client.get( |
|
2459 | f"{self.base_endpoint}/v2/evc/1234/metadata", |
||
2460 | ) |
||
2461 | 1 | assert response.status_code == 200 |
|
2462 | 1 | assert response.json() == {"metadata": evc_mock.metadata} |
|
2463 | |||
2464 | 1 | async def test_delete_metadata(self): |
|
2465 | """Test method to delete metadata""" |
||
2466 | 1 | evc_mock = create_autospec(EVC) |
|
2467 | 1 | evc_mock.metadata = {'metadata1': 1, 'metadata2': 2} |
|
2468 | 1 | evc_mock.id = 1234 |
|
2469 | 1 | self.napp.circuits = {"1234": evc_mock} |
|
2470 | |||
2471 | 1 | response = await self.api_client.delete( |
|
2472 | f"{self.base_endpoint}/v2/evc/1234/metadata/metadata1", |
||
2473 | ) |
||
2474 | 1 | assert response.status_code == 200 |
|
2475 | |||
2476 | 1 | async def test_delete_metadata_no_evc(self): |
|
2477 | """Test method to delete metadata with no evc""" |
||
2478 | 1 | response = await self.api_client.delete( |
|
2479 | f"{self.base_endpoint}/v2/evc/1234/metadata/metadata1", |
||
2480 | ) |
||
2481 | assert response.status_code == 404 |
||
2482 | assert response.json()["description"] == \ |
||
2483 | "circuit_id 1234 not found." |
||
2484 | |||
2485 | 1 | @patch('napps.kytos.mef_eline.main.Main._load_evc') |
|
2486 | 1 | def test_load_all_evcs(self, load_evc_mock): |
|
2487 | """Test load_evcs method""" |
||
2488 | 1 | mock_circuits = { |
|
2489 | 'circuits': { |
||
2490 | 1: 'circuit_1', |
||
2491 | 2: 'circuit_2', |
||
2492 | 3: 'circuit_3', |
||
2493 | 4: 'circuit_4' |
||
2494 | } |
||
2495 | } |
||
2496 | 1 | self.napp.mongo_controller.get_circuits.return_value = mock_circuits |
|
2497 | 1 | self.napp.circuits = {2: 'circuit_2', 3: 'circuit_3'} |
|
2498 | 1 | self.napp.load_all_evcs() |
|
2499 | 1 | load_evc_mock.assert_has_calls([call('circuit_1'), call('circuit_4')]) |
|
2500 | 1 | assert self.napp.controller.buffers.app.put.call_count > 1 |
|
2501 | 1 | call_args = self.napp.controller.buffers.app.put.call_args[0] |
|
2502 | 1 | assert call_args[0].name == "kytos/mef_eline.evcs_loaded" |
|
2503 | 1 | assert dict(call_args[0].content) == mock_circuits["circuits"] |
|
2504 | 1 | timeout_d = {"timeout": 1} |
|
2505 | 1 | assert self.napp.controller.buffers.app.put.call_args[1] == timeout_d |
|
2506 | |||
2507 | 1 | @patch('napps.kytos.mef_eline.main.Main._evc_from_dict') |
|
2508 | 1 | def test_load_evc(self, evc_from_dict_mock): |
|
2509 | """Test _load_evc method""" |
||
2510 | # pylint: disable=protected-access |
||
2511 | # case 1: early return with ValueError exception |
||
2512 | 1 | evc_from_dict_mock.side_effect = ValueError("err") |
|
2513 | 1 | evc_dict = MagicMock() |
|
2514 | 1 | assert not self.napp._load_evc(evc_dict) |
|
2515 | |||
2516 | # case 2: early return with KytosTagError exception |
||
2517 | 1 | evc_from_dict_mock.side_effect = KytosTagError("") |
|
2518 | 1 | assert not self.napp._load_evc(evc_dict) |
|
2519 | |||
2520 | # case 3: archived evc |
||
2521 | 1 | evc = MagicMock() |
|
2522 | 1 | evc.archived = True |
|
2523 | 1 | evc_from_dict_mock.side_effect = None |
|
2524 | 1 | evc_from_dict_mock.return_value = evc |
|
2525 | 1 | assert not self.napp._load_evc(evc_dict) |
|
2526 | |||
2527 | # case 4: success creating |
||
2528 | 1 | evc.archived = False |
|
2529 | 1 | evc.id = 1 |
|
2530 | 1 | self.napp.sched = MagicMock() |
|
2531 | |||
2532 | 1 | result = self.napp._load_evc(evc_dict) |
|
2533 | 1 | assert result == evc |
|
2534 | 1 | self.napp.sched.add.assert_called_with(evc) |
|
2535 | 1 | assert self.napp.circuits[1] == evc |
|
2536 | |||
2537 | 1 | def test_handle_flow_mod_error(self): |
|
2538 | """Test handle_flow_mod_error method""" |
||
2539 | 1 | flow = MagicMock() |
|
2540 | 1 | flow.cookie = 0xaa00000000000011 |
|
2541 | 1 | event = MagicMock() |
|
2542 | 1 | event.content = {'flow': flow, 'error_command': 'add'} |
|
2543 | 1 | evc = create_autospec(EVC) |
|
2544 | 1 | evc.archived = False |
|
2545 | 1 | evc.remove_current_flows = MagicMock() |
|
2546 | 1 | evc.lock = MagicMock() |
|
2547 | 1 | self.napp.circuits = {"00000000000011": evc} |
|
2548 | 1 | self.napp.handle_flow_mod_error(event) |
|
2549 | 1 | evc.remove_current_flows.assert_called_once() |
|
2550 | |||
2551 | 1 | def test_handle_flow_mod_error_return(self): |
|
2552 | """Test handle_flow_mod_error method with early return.""" |
||
2553 | 1 | flow = MagicMock() |
|
2554 | 1 | flow.cookie = 0xaa00000000000011 |
|
2555 | 1 | event = MagicMock() |
|
2556 | 1 | event.content = {'flow': flow, 'error_command': 'delete'} |
|
2557 | |||
2558 | 1 | evc = create_autospec(EVC) |
|
2559 | 1 | evc.archived = False |
|
2560 | 1 | evc.is_enabled.return_value = True |
|
2561 | |||
2562 | # Flow command is not 'add' |
||
2563 | 1 | self.napp.circuits = {"00000000000011": evc} |
|
2564 | 1 | self.napp.handle_flow_mod_error(event) |
|
2565 | 1 | assert not evc.remove_current_flows.call_count |
|
2566 | |||
2567 | # EVC is not enabled |
||
2568 | 1 | event.content["error_command"] = "add" |
|
2569 | 1 | evc.is_enabled.return_value = False |
|
2570 | 1 | self.napp.handle_flow_mod_error(event) |
|
2571 | 1 | assert not evc.remove_current_flows.call_count |
|
2572 | |||
2573 | # EVC is archived |
||
2574 | 1 | evc.is_enabled.return_value = True |
|
2575 | 1 | evc.archived = True |
|
2576 | 1 | self.napp.handle_flow_mod_error(event) |
|
2577 | 1 | assert not evc.remove_current_flows.call_count |
|
2578 | |||
2579 | # EVC does not exist in self.circuits |
||
2580 | 1 | evc.archived = False |
|
2581 | 1 | self.napp.circuits = {} |
|
2582 | 1 | self.napp.handle_flow_mod_error(event) |
|
2583 | 1 | assert not evc.remove_current_flows.call_count |
|
2584 | |||
2585 | 1 | @patch("kytos.core.Controller.get_interface_by_id") |
|
2586 | 1 | def test_uni_from_dict(self, _get_interface_by_id_mock): |
|
2587 | """Test _uni_from_dict method.""" |
||
2588 | # pylint: disable=protected-access |
||
2589 | # case1: early return on empty dict |
||
2590 | 1 | assert not self.napp._uni_from_dict(None) |
|
2591 | |||
2592 | # case2: invalid interface raises ValueError |
||
2593 | 1 | _get_interface_by_id_mock.return_value = None |
|
2594 | 1 | uni_dict = { |
|
2595 | "interface_id": "00:01:1", |
||
2596 | "tag": {"tag_type": 'vlan', "value": 81}, |
||
2597 | } |
||
2598 | 1 | with pytest.raises(ValueError): |
|
2599 | 1 | self.napp._uni_from_dict(uni_dict) |
|
2600 | |||
2601 | # case3: success creation |
||
2602 | 1 | uni_mock = get_uni_mocked(switch_id="00:01") |
|
2603 | 1 | _get_interface_by_id_mock.return_value = uni_mock.interface |
|
2604 | 1 | uni = self.napp._uni_from_dict(uni_dict) |
|
2605 | 1 | assert uni == uni_mock |
|
2606 | |||
2607 | # case4: success creation of tag list |
||
2608 | 1 | uni_dict["tag"]["value"] = [[1, 10]] |
|
2609 | 1 | uni = self.napp._uni_from_dict(uni_dict) |
|
2610 | 1 | assert isinstance(uni.user_tag, TAGRange) |
|
2611 | |||
2612 | # case5: success creation without tag |
||
2613 | 1 | uni_mock.user_tag = None |
|
2614 | 1 | del uni_dict["tag"] |
|
2615 | 1 | uni = self.napp._uni_from_dict(uni_dict) |
|
2616 | 1 | assert uni == uni_mock |
|
2617 | |||
2618 | 1 | def test_handle_flow_delete(self): |
|
2619 | """Test handle_flow_delete method""" |
||
2620 | 1 | flow = MagicMock() |
|
2621 | 1 | flow.cookie = 0xaa00000000000011 |
|
2622 | 1 | event = MagicMock() |
|
2623 | 1 | event.content = {'flow': flow} |
|
2624 | 1 | evc = create_autospec(EVC) |
|
2625 | 1 | evc.set_flow_removed_at = MagicMock() |
|
2626 | 1 | self.napp.circuits = {"00000000000011": evc} |
|
2627 | 1 | self.napp.handle_flow_delete(event) |
|
2628 | 1 | evc.set_flow_removed_at.assert_called_once() |
|
2629 | |||
2630 | 1 | View Code Duplication | async def test_add_bulk_metadata(self): |
2631 | """Test add_bulk_metadata method""" |
||
2632 | 1 | self.napp.controller.loop = asyncio.get_running_loop() |
|
2633 | 1 | evc_mock = create_autospec(EVC) |
|
2634 | 1 | evc_mock.id = 1234 |
|
2635 | 1 | self.napp.circuits = {"1234": evc_mock} |
|
2636 | 1 | payload = { |
|
2637 | "circuit_ids": ["1234"], |
||
2638 | "metadata1": 1, |
||
2639 | "metadata2": 2 |
||
2640 | } |
||
2641 | 1 | response = await self.api_client.post( |
|
2642 | f"{self.base_endpoint}/v2/evc/metadata", |
||
2643 | json=payload |
||
2644 | ) |
||
2645 | 1 | assert response.status_code == 201 |
|
2646 | 1 | args = self.napp.mongo_controller.update_evcs_metadata.call_args[0] |
|
2647 | 1 | ids = payload.pop("circuit_ids") |
|
2648 | 1 | assert args[0] == ids |
|
2649 | 1 | assert args[1] == payload |
|
2650 | 1 | assert args[2] == "add" |
|
2651 | 1 | calls = self.napp.mongo_controller.update_evcs_metadata.call_count |
|
2652 | 1 | assert calls == 1 |
|
2653 | 1 | evc_mock.extend_metadata.assert_called_with(payload) |
|
2654 | |||
2655 | 1 | async def test_add_bulk_metadata_empty_list(self): |
|
2656 | """Test add_bulk_metadata method empty list""" |
||
2657 | 1 | self.napp.controller.loop = asyncio.get_running_loop() |
|
2658 | 1 | evc_mock = create_autospec(EVC) |
|
2659 | 1 | evc_mock.id = 1234 |
|
2660 | 1 | self.napp.circuits = {"1234": evc_mock} |
|
2661 | 1 | payload = { |
|
2662 | "circuit_ids": [], |
||
2663 | "metadata1": 1, |
||
2664 | "metadata2": 2 |
||
2665 | } |
||
2666 | 1 | response = await self.api_client.post( |
|
2667 | f"{self.base_endpoint}/v2/evc/metadata", |
||
2668 | json=payload |
||
2669 | ) |
||
2670 | assert response.status_code == 400 |
||
2671 | assert "invalid" in response.json()["description"] |
||
2672 | |||
2673 | 1 | async def test_add_bulk_metadata_no_id(self): |
|
2674 | """Test add_bulk_metadata with unknown evc id""" |
||
2675 | 1 | self.napp.controller.loop = asyncio.get_running_loop() |
|
2676 | 1 | evc_mock = create_autospec(EVC) |
|
2677 | 1 | evc_mock.id = 1234 |
|
2678 | 1 | self.napp.circuits = {"1234": evc_mock} |
|
2679 | 1 | payload = { |
|
2680 | "circuit_ids": ["1234", "4567"] |
||
2681 | } |
||
2682 | 1 | response = await self.api_client.post( |
|
2683 | f"{self.base_endpoint}/v2/evc/metadata", |
||
2684 | json=payload |
||
2685 | ) |
||
2686 | assert response.status_code == 404 |
||
2687 | |||
2688 | 1 | async def test_add_bulk_metadata_no_circuits(self): |
|
2689 | """Test add_bulk_metadata without circuit_ids""" |
||
2690 | 1 | self.napp.controller.loop = asyncio.get_running_loop() |
|
2691 | 1 | evc_mock = create_autospec(EVC) |
|
2692 | 1 | evc_mock.id = 1234 |
|
2693 | 1 | self.napp.circuits = {"1234": evc_mock} |
|
2694 | 1 | payload = { |
|
2695 | "metadata": "data" |
||
2696 | } |
||
2697 | 1 | response = await self.api_client.post( |
|
2698 | f"{self.base_endpoint}/v2/evc/metadata", |
||
2699 | json=payload |
||
2700 | ) |
||
2701 | assert response.status_code == 400 |
||
2702 | |||
2703 | 1 | View Code Duplication | async def test_delete_bulk_metadata(self): |
2704 | """Test delete_metadata method""" |
||
2705 | 1 | self.napp.controller.loop = asyncio.get_running_loop() |
|
2706 | 1 | evc_mock = create_autospec(EVC) |
|
2707 | 1 | evc_mock.id = 1234 |
|
2708 | 1 | self.napp.circuits = {"1234": evc_mock} |
|
2709 | 1 | payload = { |
|
2710 | "circuit_ids": ["1234"] |
||
2711 | } |
||
2712 | 1 | response = await self.api_client.request( |
|
2713 | "DELETE", |
||
2714 | f"{self.base_endpoint}/v2/evc/metadata/metadata1", |
||
2715 | json=payload |
||
2716 | ) |
||
2717 | 1 | assert response.status_code == 200 |
|
2718 | 1 | args = self.napp.mongo_controller.update_evcs_metadata.call_args[0] |
|
2719 | 1 | assert args[0] == payload["circuit_ids"] |
|
2720 | 1 | assert args[1] == {"metadata1": ""} |
|
2721 | 1 | assert args[2] == "del" |
|
2722 | 1 | calls = self.napp.mongo_controller.update_evcs_metadata.call_count |
|
2723 | 1 | assert calls == 1 |
|
2724 | 1 | assert evc_mock.remove_metadata.call_count == 1 |
|
2725 | |||
2726 | 1 | async def test_delete_bulk_metadata_error(self): |
|
2727 | """Test bulk_delete_metadata with ciruit erroring""" |
||
2728 | 1 | self.napp.controller.loop = asyncio.get_running_loop() |
|
2729 | 1 | evc_mock = create_autospec(EVC) |
|
2730 | 1 | evcs = [evc_mock, evc_mock] |
|
2731 | 1 | self.napp.circuits = dict(zip(["1", "2"], evcs)) |
|
2732 | 1 | payload = {"circuit_ids": ["1", "2", "3"]} |
|
2733 | 1 | response = await self.api_client.request( |
|
2734 | "DELETE", |
||
2735 | f"{self.base_endpoint}/v2/evc/metadata/metadata1", |
||
2736 | json=payload |
||
2737 | ) |
||
2738 | assert response.status_code == 404, response.data |
||
2739 | assert response.json()["description"] == ["3"] |
||
2740 | |||
2741 | 1 | async def test_use_uni_tags(self): |
|
2742 | """Test _use_uni_tags""" |
||
2743 | 1 | self.napp.controller.loop = asyncio.get_running_loop() |
|
2744 | 1 | evc_mock = create_autospec(EVC) |
|
2745 | 1 | evc_mock.uni_a = "uni_a_mock" |
|
2746 | 1 | evc_mock.uni_z = "uni_z_mock" |
|
2747 | 1 | self.napp._use_uni_tags(evc_mock) |
|
2748 | 1 | assert evc_mock._use_uni_vlan.call_count == 2 |
|
2749 | 1 | assert evc_mock._use_uni_vlan.call_args[0][0] == evc_mock.uni_z |
|
2750 | |||
2751 | # One UNI tag is not available |
||
2752 | 1 | evc_mock._use_uni_vlan.side_effect = [KytosTagError(""), None] |
|
2753 | 1 | with pytest.raises(KytosTagError): |
|
2754 | 1 | self.napp._use_uni_tags(evc_mock) |
|
2755 | 1 | assert evc_mock._use_uni_vlan.call_count == 3 |
|
2756 | 1 | assert evc_mock.make_uni_vlan_available.call_count == 0 |
|
2757 | |||
2758 | 1 | evc_mock._use_uni_vlan.side_effect = [None, KytosTagError("")] |
|
2759 | 1 | with pytest.raises(KytosTagError): |
|
2760 | 1 | self.napp._use_uni_tags(evc_mock) |
|
2761 | 1 | assert evc_mock._use_uni_vlan.call_count == 5 |
|
2762 | 1 | assert evc_mock.make_uni_vlan_available.call_count == 1 |
|
2763 | |||
2764 | 1 | def test_check_no_tag_duplication(self): |
|
2765 | """Test _check_no_tag_duplication""" |
||
2766 | 1 | evc = MagicMock() |
|
2767 | 1 | evc.check_no_tag_duplicate = MagicMock() |
|
2768 | 1 | evc.archived = False |
|
2769 | 1 | evc.id = "1" |
|
2770 | 1 | self.napp.circuits = {"1": evc} |
|
2771 | 1 | evc_id = "2" |
|
2772 | 1 | uni_a = get_uni_mocked(valid=True) |
|
2773 | 1 | uni_z = get_uni_mocked(valid=True) |
|
2774 | 1 | self.napp._check_no_tag_duplication(evc_id, uni_a, uni_z) |
|
2775 | 1 | assert evc.check_no_tag_duplicate.call_count == 0 |
|
2776 | |||
2777 | 1 | uni_a.user_tag = None |
|
2778 | 1 | uni_z.user_tag = None |
|
2779 | 1 | self.napp._check_no_tag_duplication(evc_id, uni_a, uni_z) |
|
2780 | 1 | assert evc.check_no_tag_duplicate.call_count == 2 |
|
2781 | |||
2782 | 1 | self.napp._check_no_tag_duplication(evc_id, uni_a, None) |
|
2783 | 1 | assert evc.check_no_tag_duplicate.call_count == 3 |
|
2784 | |||
2785 | 1 | self.napp._check_no_tag_duplication(evc_id, None, None) |
|
2786 | 1 | assert evc.check_no_tag_duplicate.call_count == 3 |
|
2787 | |||
2788 | 1 | @patch("napps.kytos.mef_eline.main.time") |
|
2789 | 1 | @patch("napps.kytos.mef_eline.main.Main.handle_interface_link_up") |
|
2790 | 1 | @patch("napps.kytos.mef_eline.main.Main.handle_interface_link_down") |
|
2791 | 1 | def test_handle_on_interface_link_change( |
|
2792 | self, |
||
2793 | mock_down, |
||
2794 | mock_up, |
||
2795 | mock_time |
||
2796 | ): |
||
2797 | """Test handle_on_interface_link_change""" |
||
2798 | 1 | mock_time.sleep.return_value = True |
|
2799 | 1 | mock_intf = Mock() |
|
2800 | 1 | mock_intf.id = "mock_intf" |
|
2801 | |||
2802 | # Created/link_up |
||
2803 | 1 | name = '.*.switch.interface.created' |
|
2804 | 1 | content = {"interface": mock_intf} |
|
2805 | 1 | event = KytosEvent(name=name, content=content) |
|
2806 | 1 | self.napp.handle_on_interface_link_change(event) |
|
2807 | 1 | assert mock_down.call_count == 0 |
|
2808 | 1 | assert mock_up.call_count == 1 |
|
2809 | |||
2810 | # Deleted/link_down |
||
2811 | 1 | name = '.*.switch.interface.deleted' |
|
2812 | 1 | event = KytosEvent(name=name, content=content) |
|
2813 | 1 | self.napp.handle_on_interface_link_change(event) |
|
2814 | 1 | assert mock_down.call_count == 1 |
|
2815 | 1 | assert mock_up.call_count == 1 |
|
2816 | |||
2817 | # Event delay |
||
2818 | 1 | self.napp._intf_events[mock_intf.id]["last_acquired"] = "mock_time" |
|
2819 | 1 | for _ in range(1, 6): |
|
2820 | 1 | self.napp.handle_on_interface_link_change(event) |
|
2821 | 1 | assert mock_down.call_count == 1 |
|
2822 | 1 | assert mock_up.call_count == 1 |
|
2823 | |||
2824 | 1 | self.napp._intf_events[mock_intf.id].pop("last_acquired") |
|
2825 | 1 | self.napp.handle_on_interface_link_change(event) |
|
2826 | 1 | assert mock_down.call_count == 2 |
|
2827 | 1 | assert mock_up.call_count == 1 |
|
2828 | |||
2829 | # Out of order event |
||
2830 | 1 | event = KytosEvent(name=name, content=content) |
|
2831 | 1 | self.napp._intf_events[mock_intf.id]["event"] = Mock(timestamp=now()) |
|
2832 | |||
2833 | 1 | self.napp.handle_on_interface_link_change(event) |
|
2834 | 1 | assert mock_down.call_count == 2 |
|
2835 | assert mock_up.call_count == 1 |
||
2836 |