ajoaoff /
mef_eline
| Total Complexity | 21 |
| Total Lines | 676 |
| Duplicated Lines | 5.18 % |
| 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:
| 1 | """Method to thest EVCDeploy class.""" |
||
| 2 | import sys |
||
| 3 | from unittest import TestCase |
||
| 4 | from unittest.mock import MagicMock, Mock, patch |
||
| 5 | |||
| 6 | from kytos.core.interface import Interface |
||
| 7 | from kytos.core.switch import Switch |
||
| 8 | |||
| 9 | # pylint: disable=wrong-import-position |
||
| 10 | sys.path.insert(0, '/var/lib/kytos/napps/..') |
||
| 11 | # pylint: enable=wrong-import-position |
||
| 12 | |||
| 13 | from napps.kytos.mef_eline.models import EVC, Path # NOQA |
||
| 14 | from napps.kytos.mef_eline.settings import MANAGER_URL # NOQA |
||
| 15 | from napps.kytos.mef_eline.exceptions import FlowModException # NOQA |
||
| 16 | from tests.helpers import get_link_mocked,\ |
||
| 17 | get_uni_mocked, get_controller_mock # NOQA |
||
| 18 | |||
| 19 | |||
| 20 | class TestEVC(TestCase): # pylint: disable=too-many-public-methods |
||
| 21 | """Tests to verify EVC class.""" |
||
| 22 | |||
| 23 | def test_primary_links_zipped(self): |
||
| 24 | """Test primary links zipped method.""" |
||
| 25 | |||
| 26 | @staticmethod |
||
| 27 | @patch('napps.kytos.mef_eline.models.log') |
||
| 28 | def test_should_deploy_case1(log_mock): |
||
| 29 | """Test should deploy method without primary links.""" |
||
| 30 | log_mock.debug.return_value = True |
||
| 31 | attributes = { |
||
| 32 | "controller": get_controller_mock(), |
||
| 33 | "name": "custom_name", |
||
| 34 | "uni_a": get_uni_mocked(is_valid=True), |
||
| 35 | "uni_z": get_uni_mocked(is_valid=True) |
||
| 36 | } |
||
| 37 | |||
| 38 | evc = EVC(**attributes) |
||
| 39 | evc.should_deploy() |
||
| 40 | log_mock.debug.assert_called_with('Path is empty.') |
||
| 41 | |||
| 42 | @patch('napps.kytos.mef_eline.models.log') |
||
| 43 | def test_should_deploy_case2(self, log_mock): |
||
| 44 | """Test should deploy method with disable circuit.""" |
||
| 45 | log_mock.debug.return_value = True |
||
| 46 | attributes = { |
||
| 47 | "controller": get_controller_mock(), |
||
| 48 | "name": "custom_name", |
||
| 49 | "uni_a": get_uni_mocked(is_valid=True), |
||
| 50 | "uni_z": get_uni_mocked(is_valid=True), |
||
| 51 | "primary_links": [get_link_mocked(), get_link_mocked()] |
||
| 52 | } |
||
| 53 | evc = EVC(**attributes) |
||
| 54 | |||
| 55 | self.assertFalse(evc.should_deploy(attributes['primary_links'])) |
||
| 56 | log_mock.debug.assert_called_with(f'{evc} is disabled.') |
||
| 57 | |||
| 58 | @patch('napps.kytos.mef_eline.models.log') |
||
| 59 | def test_should_deploy_case3(self, log_mock): |
||
| 60 | """Test should deploy method with enabled and not active circuit.""" |
||
| 61 | log_mock.debug.return_value = True |
||
| 62 | attributes = { |
||
| 63 | "controller": get_controller_mock(), |
||
| 64 | "name": "custom_name", |
||
| 65 | "uni_a": get_uni_mocked(is_valid=True), |
||
| 66 | "uni_z": get_uni_mocked(is_valid=True), |
||
| 67 | "primary_links": [get_link_mocked(), get_link_mocked()], |
||
| 68 | "enabled": True |
||
| 69 | } |
||
| 70 | evc = EVC(**attributes) |
||
| 71 | self.assertTrue(evc.should_deploy(attributes['primary_links'])) |
||
| 72 | log_mock.debug.assert_called_with(f'{evc} will be deployed.') |
||
| 73 | |||
| 74 | @patch('napps.kytos.mef_eline.models.log') |
||
| 75 | def test_should_deploy_case4(self, log_mock): |
||
| 76 | """Test should deploy method with enabled and active circuit.""" |
||
| 77 | log_mock.debug.return_value = True |
||
| 78 | attributes = { |
||
| 79 | "controller": get_controller_mock(), |
||
| 80 | "name": "custom_name", |
||
| 81 | "uni_a": get_uni_mocked(is_valid=True), |
||
| 82 | "uni_z": get_uni_mocked(is_valid=True), |
||
| 83 | "primary_links": [get_link_mocked(), get_link_mocked()], |
||
| 84 | "enabled": True, |
||
| 85 | "active": True |
||
| 86 | } |
||
| 87 | evc = EVC(**attributes) |
||
| 88 | self.assertFalse(evc.should_deploy(attributes['primary_links'])) |
||
| 89 | |||
| 90 | View Code Duplication | @patch('napps.kytos.mef_eline.models.requests') |
|
|
Duplication
introduced
by
|
|||
| 91 | def test_send_flow_mods_case1(self, requests_mock): |
||
| 92 | """Test if you are sending flow_mods.""" |
||
| 93 | flow_mods = {"id": 20} |
||
| 94 | switch = Mock(spec=Switch, id=1) |
||
| 95 | |||
| 96 | response = MagicMock() |
||
| 97 | response.status_code = 201 |
||
| 98 | requests_mock.post.return_value = response |
||
| 99 | |||
| 100 | # pylint: disable=protected-access |
||
| 101 | EVC._send_flow_mods(switch, flow_mods) |
||
| 102 | |||
| 103 | expected_endpoint = f"{MANAGER_URL}/flows/{switch.id}" |
||
| 104 | expected_data = {"flows": flow_mods} |
||
| 105 | self.assertEqual(requests_mock.post.call_count, 1) |
||
| 106 | requests_mock.post.assert_called_once_with(expected_endpoint, |
||
| 107 | json=expected_data) |
||
| 108 | |||
| 109 | View Code Duplication | @patch('napps.kytos.mef_eline.models.requests') |
|
| 110 | def test_send_flow_mods_case2(self, requests_mock): |
||
| 111 | """Test if you are sending flow_mods.""" |
||
| 112 | flow_mods = {"id": 20} |
||
| 113 | switch = Mock(spec=Switch, id=1) |
||
| 114 | response = MagicMock() |
||
| 115 | response.status_code = 201 |
||
| 116 | requests_mock.post.return_value = response |
||
| 117 | |||
| 118 | # pylint: disable=protected-access |
||
| 119 | EVC._send_flow_mods(switch, flow_mods, command='delete') |
||
| 120 | |||
| 121 | expected_endpoint = f"{MANAGER_URL}/delete/{switch.id}" |
||
| 122 | expected_data = {"flows": flow_mods} |
||
| 123 | self.assertEqual(requests_mock.post.call_count, 1) |
||
| 124 | requests_mock.post.assert_called_once_with(expected_endpoint, |
||
| 125 | json=expected_data) |
||
| 126 | |||
| 127 | def test_prepare_flow_mod(self): |
||
| 128 | """Test prepare flow_mod method.""" |
||
| 129 | interface_a = Interface('eth0', 1, Mock(spec=Switch)) |
||
| 130 | interface_z = Interface('eth1', 3, Mock(spec=Switch)) |
||
| 131 | attributes = { |
||
| 132 | "controller": get_controller_mock(), |
||
| 133 | "name": "custom_name", |
||
| 134 | "uni_a": get_uni_mocked(is_valid=True), |
||
| 135 | "uni_z": get_uni_mocked(is_valid=True), |
||
| 136 | "primary_links": [get_link_mocked(), get_link_mocked()], |
||
| 137 | "enabled": True, |
||
| 138 | "active": True |
||
| 139 | } |
||
| 140 | evc = EVC(**attributes) |
||
| 141 | |||
| 142 | # pylint: disable=protected-access |
||
| 143 | flow_mod = evc._prepare_flow_mod(interface_a, interface_z) |
||
| 144 | expected_flow_mod = { |
||
| 145 | 'match': {'in_port': interface_a.port_number}, |
||
| 146 | 'cookie': evc.get_cookie(), |
||
| 147 | 'actions': [ |
||
| 148 | {'action_type': 'output', |
||
| 149 | 'port': interface_z.port_number} |
||
| 150 | ] |
||
| 151 | } |
||
| 152 | self.assertEqual(expected_flow_mod, flow_mod) |
||
| 153 | |||
| 154 | def test_prepare_pop_flow(self): |
||
| 155 | """Test prepare pop flow method.""" |
||
| 156 | attributes = { |
||
| 157 | "controller": get_controller_mock(), |
||
| 158 | "name": "custom_name", |
||
| 159 | "uni_a": get_uni_mocked(interface_port=1, is_valid=True), |
||
| 160 | "uni_z": get_uni_mocked(interface_port=2, is_valid=True), |
||
| 161 | } |
||
| 162 | evc = EVC(**attributes) |
||
| 163 | interface_a = evc.uni_a.interface |
||
| 164 | interface_z = evc.uni_z.interface |
||
| 165 | in_vlan = 10 |
||
| 166 | |||
| 167 | # pylint: disable=protected-access |
||
| 168 | flow_mod = evc._prepare_pop_flow(interface_a, interface_z, |
||
| 169 | None, in_vlan) |
||
| 170 | |||
| 171 | expected_flow_mod = { |
||
| 172 | 'match': {'in_port': interface_a.port_number, 'dl_vlan': in_vlan}, |
||
| 173 | 'cookie': evc.get_cookie(), |
||
| 174 | 'actions': [ |
||
| 175 | {'action_type': 'pop_vlan'}, |
||
| 176 | {'action_type': 'output', |
||
| 177 | 'port': interface_z.port_number |
||
| 178 | } |
||
| 179 | ] |
||
| 180 | } |
||
| 181 | self.assertEqual(expected_flow_mod, flow_mod) |
||
| 182 | |||
| 183 | def test_prepare_push_flow(self): |
||
| 184 | """Test prepare push flow method.""" |
||
| 185 | attributes = { |
||
| 186 | "controller": get_controller_mock(), |
||
| 187 | "name": "custom_name", |
||
| 188 | "uni_a": get_uni_mocked(interface_port=1, is_valid=True), |
||
| 189 | "uni_z": get_uni_mocked(interface_port=2, is_valid=True), |
||
| 190 | } |
||
| 191 | evc = EVC(**attributes) |
||
| 192 | interface_a = evc.uni_a.interface |
||
| 193 | interface_z = evc.uni_z.interface |
||
| 194 | out_vlan_a = 20 |
||
| 195 | |||
| 196 | for in_vlan_a in (10, None): |
||
| 197 | with self.subTest(in_vlan_a=in_vlan_a): |
||
| 198 | # pylint: disable=protected-access |
||
| 199 | flow_mod = evc._prepare_push_flow(interface_a, interface_z, |
||
| 200 | in_vlan_a, out_vlan_a) |
||
| 201 | |||
| 202 | expected_flow_mod = { |
||
| 203 | 'match': {'in_port': interface_a.port_number}, |
||
| 204 | 'cookie': evc.get_cookie(), |
||
| 205 | 'actions': [ |
||
| 206 | {'action_type': 'push_vlan', 'tag_type': 's'}, |
||
| 207 | {'action_type': 'set_vlan', 'vlan_id': out_vlan_a}, |
||
| 208 | { |
||
| 209 | 'action_type': 'output', |
||
| 210 | 'port': interface_z.port_number |
||
| 211 | } |
||
| 212 | ] |
||
| 213 | } |
||
| 214 | if in_vlan_a: |
||
| 215 | expected_flow_mod['match']['dl_vlan'] = in_vlan_a |
||
| 216 | expected_flow_mod['actions'].insert(0, { |
||
| 217 | 'action_type': 'pop_vlan' |
||
| 218 | }) |
||
| 219 | self.assertEqual(expected_flow_mod, flow_mod) |
||
| 220 | |||
| 221 | @staticmethod |
||
| 222 | @patch('napps.kytos.mef_eline.models.EVC._send_flow_mods') |
||
| 223 | def test_install_uni_flows(send_flow_mods_mock): |
||
| 224 | """Test install uni flows method. |
||
| 225 | |||
| 226 | This test will verify the flows send to the send_flow_mods method. |
||
| 227 | """ |
||
| 228 | uni_a = get_uni_mocked(interface_port=2, tag_value=82, |
||
| 229 | switch_id="switch_uni_a", is_valid=True) |
||
| 230 | uni_z = get_uni_mocked(interface_port=3, tag_value=83, |
||
| 231 | switch_id="switch_uni_z", is_valid=True) |
||
| 232 | |||
| 233 | attributes = { |
||
| 234 | "controller": get_controller_mock(), |
||
| 235 | "name": "custom_name", |
||
| 236 | "uni_a": uni_a, |
||
| 237 | "uni_z": uni_z, |
||
| 238 | "primary_links": [ |
||
| 239 | get_link_mocked(endpoint_a_port=9, endpoint_b_port=10, |
||
| 240 | metadata={"s_vlan": 5}), |
||
| 241 | get_link_mocked(endpoint_a_port=11, endpoint_b_port=12, |
||
| 242 | metadata={"s_vlan": 6}) |
||
| 243 | ] |
||
| 244 | } |
||
| 245 | evc = EVC(**attributes) |
||
| 246 | |||
| 247 | # pylint: disable=protected-access |
||
| 248 | evc._install_uni_flows(attributes['primary_links']) |
||
| 249 | |||
| 250 | expected_flow_mod_a = [ |
||
| 251 | {'match': {'in_port': uni_a.interface.port_number, |
||
| 252 | 'dl_vlan': uni_a.user_tag.value}, |
||
| 253 | 'cookie': evc.get_cookie(), |
||
| 254 | 'actions': [ |
||
| 255 | {'action_type': 'pop_vlan'}, |
||
| 256 | {'action_type': 'push_vlan', 'tag_type': 's'}, |
||
| 257 | {'action_type': 'set_vlan', |
||
| 258 | 'vlan_id': evc.primary_links[0].get_metadata('s_vlan').value}, |
||
| 259 | {'action_type': 'output', |
||
| 260 | 'port': evc.primary_links[0].endpoint_a.port_number} |
||
| 261 | ]}, |
||
| 262 | {'match': { |
||
| 263 | 'in_port': evc.primary_links[0].endpoint_a.port_number, |
||
| 264 | 'dl_vlan': evc.primary_links[0].get_metadata('s_vlan').value |
||
| 265 | }, |
||
| 266 | 'cookie': evc.get_cookie(), |
||
| 267 | 'actions': [ |
||
| 268 | {'action_type': 'pop_vlan'}, |
||
| 269 | {'action_type': 'push_vlan', 'tag_type': 'c'}, |
||
| 270 | {'action_type': 'set_vlan', 'vlan_id': uni_a.user_tag.value}, |
||
| 271 | {'action_type': 'output', 'port': uni_a.interface.port_number} |
||
| 272 | ] |
||
| 273 | } |
||
| 274 | ] |
||
| 275 | |||
| 276 | send_flow_mods_mock.assert_any_call(uni_a.interface.switch, |
||
| 277 | expected_flow_mod_a) |
||
| 278 | |||
| 279 | expected_flow_mod_z = [ |
||
| 280 | {'match': {'in_port': uni_z.interface.port_number, |
||
| 281 | 'dl_vlan': uni_z.user_tag.value}, |
||
| 282 | 'cookie': evc.get_cookie(), |
||
| 283 | 'actions': [ |
||
| 284 | {'action_type': 'pop_vlan'}, |
||
| 285 | {'action_type': 'push_vlan', 'tag_type': 's'}, |
||
| 286 | {'action_type': 'set_vlan', |
||
| 287 | 'vlan_id': evc.primary_links[-1].get_metadata('s_vlan').value |
||
| 288 | }, |
||
| 289 | {'action_type': 'output', |
||
| 290 | 'port': evc.primary_links[-1].endpoint_b.port_number} |
||
| 291 | ] |
||
| 292 | }, |
||
| 293 | {'match': { |
||
| 294 | 'in_port': evc.primary_links[-1].endpoint_b.port_number, |
||
| 295 | 'dl_vlan': evc.primary_links[-1].get_metadata('s_vlan').value |
||
| 296 | }, |
||
| 297 | 'cookie': evc.get_cookie(), |
||
| 298 | 'actions': [ |
||
| 299 | {'action_type': 'pop_vlan'}, |
||
| 300 | {'action_type': 'push_vlan', 'tag_type': 'c'}, |
||
| 301 | {'action_type': 'set_vlan', 'vlan_id': uni_z.user_tag.value}, |
||
| 302 | {'action_type': 'output', 'port': uni_z.interface.port_number} |
||
| 303 | ] |
||
| 304 | } |
||
| 305 | ] |
||
| 306 | |||
| 307 | send_flow_mods_mock.assert_any_call(uni_z.interface.switch, |
||
| 308 | expected_flow_mod_z) |
||
| 309 | |||
| 310 | @staticmethod |
||
| 311 | @patch('napps.kytos.mef_eline.models.EVC._send_flow_mods') |
||
| 312 | def test_install_nni_flows(send_flow_mods_mock): |
||
| 313 | """Test install nni flows method. |
||
| 314 | |||
| 315 | This test will verify the flows send to the send_flow_mods method. |
||
| 316 | """ |
||
| 317 | uni_a = get_uni_mocked(interface_port=2, tag_value=82, |
||
| 318 | switch_id="switch_uni_a", is_valid=True) |
||
| 319 | uni_z = get_uni_mocked(interface_port=3, tag_value=83, |
||
| 320 | switch_id="switch_uni_z", is_valid=True) |
||
| 321 | |||
| 322 | attributes = { |
||
| 323 | "controller": get_controller_mock(), |
||
| 324 | "name": "custom_name", |
||
| 325 | "uni_a": uni_a, |
||
| 326 | "uni_z": uni_z, |
||
| 327 | "primary_links": [ |
||
| 328 | get_link_mocked(endpoint_a_port=9, endpoint_b_port=10, |
||
| 329 | metadata={"s_vlan": 5}), |
||
| 330 | get_link_mocked(endpoint_a_port=11, endpoint_b_port=12, |
||
| 331 | metadata={"s_vlan": 6}) |
||
| 332 | ] |
||
| 333 | } |
||
| 334 | evc = EVC(**attributes) |
||
| 335 | |||
| 336 | # pylint: disable=protected-access |
||
| 337 | evc._install_nni_flows(attributes['primary_links']) |
||
| 338 | |||
| 339 | in_vlan = evc.primary_links[0].get_metadata('s_vlan').value |
||
| 340 | out_vlan = evc.primary_links[-1].get_metadata('s_vlan').value |
||
| 341 | |||
| 342 | in_port = evc.primary_links[0].endpoint_b.port_number |
||
| 343 | out_port = evc.primary_links[-1].endpoint_a.port_number |
||
| 344 | |||
| 345 | expected_flow_mods = [ |
||
| 346 | { |
||
| 347 | 'match': {'in_port': in_port, 'dl_vlan': in_vlan}, |
||
| 348 | 'cookie': evc.get_cookie(), |
||
| 349 | 'actions': [ |
||
| 350 | {'action_type': 'set_vlan', 'vlan_id': out_vlan}, |
||
| 351 | {'action_type': 'output', 'port': out_port} |
||
| 352 | ] |
||
| 353 | }, |
||
| 354 | { |
||
| 355 | 'match': {'in_port': out_port, 'dl_vlan': out_vlan}, |
||
| 356 | 'cookie': evc.get_cookie(), |
||
| 357 | 'actions': [ |
||
| 358 | {'action_type': 'set_vlan', 'vlan_id': in_vlan}, |
||
| 359 | {'action_type': 'output', 'port': in_port} |
||
| 360 | ] |
||
| 361 | } |
||
| 362 | ] |
||
| 363 | |||
| 364 | switch = evc.primary_links[0].endpoint_b.switch |
||
| 365 | send_flow_mods_mock.assert_called_once_with(switch, expected_flow_mods) |
||
| 366 | |||
| 367 | @patch('requests.post') |
||
| 368 | @patch('napps.kytos.mef_eline.storehouse.StoreHouse.save_evc') |
||
| 369 | @patch('napps.kytos.mef_eline.models.log') |
||
| 370 | @patch('napps.kytos.mef_eline.models.Path.choose_vlans') |
||
| 371 | @patch('napps.kytos.mef_eline.models.EVC._install_nni_flows') |
||
| 372 | @patch('napps.kytos.mef_eline.models.EVC._install_uni_flows') |
||
| 373 | @patch('napps.kytos.mef_eline.models.EVC.activate') |
||
| 374 | @patch('napps.kytos.mef_eline.models.EVC.should_deploy') |
||
| 375 | def test_deploy_successfully(self, *args): |
||
| 376 | """Test if all methods to deploy are called.""" |
||
| 377 | # pylint: disable=too-many-locals |
||
| 378 | (should_deploy_mock, activate_mock, |
||
| 379 | install_uni_flows_mock, install_nni_flows, chose_vlans_mock, |
||
| 380 | log_mock, _, requests_mock) = args |
||
| 381 | |||
| 382 | response = MagicMock() |
||
| 383 | response.status_code = 201 |
||
| 384 | requests_mock.return_value = response |
||
| 385 | |||
| 386 | should_deploy_mock.return_value = True |
||
| 387 | uni_a = get_uni_mocked(interface_port=2, tag_value=82, |
||
| 388 | switch_id="switch_uni_a", is_valid=True) |
||
| 389 | uni_z = get_uni_mocked(interface_port=3, tag_value=83, |
||
| 390 | switch_id="switch_uni_z", is_valid=True) |
||
| 391 | |||
| 392 | primary_links = [ |
||
| 393 | get_link_mocked(endpoint_a_port=9, endpoint_b_port=10, |
||
| 394 | metadata={"s_vlan": 5}), |
||
| 395 | get_link_mocked(endpoint_a_port=11, endpoint_b_port=12, |
||
| 396 | metadata={"s_vlan": 6}) |
||
| 397 | ] |
||
| 398 | |||
| 399 | attributes = { |
||
| 400 | "controller": get_controller_mock(), |
||
| 401 | "name": "custom_name", |
||
| 402 | "uni_a": uni_a, |
||
| 403 | "uni_z": uni_z, |
||
| 404 | "primary_links": primary_links, |
||
| 405 | "queue_id": 5 |
||
| 406 | } |
||
| 407 | # Setup path to deploy |
||
| 408 | path = Path() |
||
| 409 | path.append(primary_links[0]) |
||
| 410 | path.append(primary_links[1]) |
||
| 411 | |||
| 412 | evc = EVC(**attributes) |
||
| 413 | |||
| 414 | deployed = evc.deploy_to_path(path) |
||
| 415 | |||
| 416 | self.assertEqual(should_deploy_mock.call_count, 1) |
||
| 417 | self.assertEqual(activate_mock.call_count, 1) |
||
| 418 | self.assertEqual(install_uni_flows_mock.call_count, 1) |
||
| 419 | self.assertEqual(install_nni_flows.call_count, 1) |
||
| 420 | self.assertEqual(chose_vlans_mock.call_count, 1) |
||
| 421 | log_mock.info.assert_called_with(f"{evc} was deployed.") |
||
| 422 | self.assertTrue(deployed) |
||
| 423 | |||
| 424 | @patch('requests.post') |
||
| 425 | @patch('napps.kytos.mef_eline.models.log') |
||
| 426 | @patch('napps.kytos.mef_eline.models.EVC.discover_new_paths', |
||
| 427 | return_value=[]) |
||
| 428 | @patch('napps.kytos.mef_eline.models.Path.choose_vlans') |
||
| 429 | @patch('napps.kytos.mef_eline.models.EVC._install_nni_flows') |
||
| 430 | @patch('napps.kytos.mef_eline.models.EVC._install_uni_flows') |
||
| 431 | @patch('napps.kytos.mef_eline.models.EVC.activate') |
||
| 432 | @patch('napps.kytos.mef_eline.models.EVC.should_deploy', |
||
| 433 | return_value=False) |
||
| 434 | @patch('napps.kytos.mef_eline.models.EVC.sync') |
||
| 435 | def test_deploy_fail(self, *args): |
||
| 436 | """Test if all methods is ignored when the should_deploy is false.""" |
||
| 437 | # pylint: disable=too-many-locals |
||
| 438 | (sync_mock, should_deploy_mock, activate_mock, install_uni_flows_mock, |
||
| 439 | install_nni_flows, choose_vlans_mock, |
||
| 440 | discover_new_paths, log_mock, requests_mock) = args |
||
| 441 | |||
| 442 | response = MagicMock() |
||
| 443 | response.status_code = 201 |
||
| 444 | requests_mock.return_value = response |
||
| 445 | |||
| 446 | uni_a = get_uni_mocked(interface_port=2, tag_value=82, |
||
| 447 | switch_id="switch_uni_a", |
||
| 448 | switch_dpid="switch_dpid_uni_a", |
||
| 449 | is_valid=True) |
||
| 450 | uni_z = get_uni_mocked(interface_port=3, tag_value=83, |
||
| 451 | switch_id="switch_uni_z", |
||
| 452 | switch_dpid="switch_dpid_uni_a", |
||
| 453 | is_valid=True) |
||
| 454 | |||
| 455 | attributes = { |
||
| 456 | "controller": get_controller_mock(), |
||
| 457 | "name": "custom_name", |
||
| 458 | "uni_a": uni_a, |
||
| 459 | "uni_z": uni_z, |
||
| 460 | "primary_links": [ |
||
| 461 | get_link_mocked(endpoint_a_port=9, endpoint_b_port=10, |
||
| 462 | metadata={"s_vlan": 5}), |
||
| 463 | get_link_mocked(endpoint_a_port=11, endpoint_b_port=12, |
||
| 464 | metadata={"s_vlan": 6}) |
||
| 465 | ] |
||
| 466 | } |
||
| 467 | |||
| 468 | evc = EVC(**attributes) |
||
| 469 | deployed = evc.deploy_to_path() |
||
| 470 | |||
| 471 | self.assertEqual(discover_new_paths.call_count, 1) |
||
| 472 | self.assertEqual(should_deploy_mock.call_count, 1) |
||
| 473 | self.assertEqual(activate_mock.call_count, 0) |
||
| 474 | self.assertEqual(install_uni_flows_mock.call_count, 0) |
||
| 475 | self.assertEqual(install_nni_flows.call_count, 0) |
||
| 476 | self.assertEqual(choose_vlans_mock.call_count, 0) |
||
| 477 | self.assertEqual(log_mock.info.call_count, 0) |
||
| 478 | self.assertEqual(sync_mock.call_count, 1) |
||
| 479 | self.assertFalse(deployed) |
||
| 480 | |||
| 481 | @patch('napps.kytos.mef_eline.models.log') |
||
| 482 | @patch('napps.kytos.mef_eline.models.EVC.discover_new_paths', |
||
| 483 | return_value=[]) |
||
| 484 | @patch('napps.kytos.mef_eline.models.Path.choose_vlans') |
||
| 485 | @patch('napps.kytos.mef_eline.models.EVC._install_nni_flows') |
||
| 486 | @patch('napps.kytos.mef_eline.models.EVC.should_deploy') |
||
| 487 | @patch('napps.kytos.mef_eline.models.EVC.remove_current_flows') |
||
| 488 | @patch('napps.kytos.mef_eline.models.EVC.sync') |
||
| 489 | def test_deploy_error(self, *args): |
||
| 490 | """Test if all methods is ignored when the should_deploy is false.""" |
||
| 491 | # pylint: disable=too-many-locals |
||
| 492 | (sync_mock, remove_current_flows, should_deploy_mock, |
||
| 493 | install_nni_flows, choose_vlans_mock, |
||
| 494 | discover_new_paths, log_mock) = args |
||
| 495 | |||
| 496 | install_nni_flows.side_effect = FlowModException |
||
| 497 | should_deploy_mock.return_value = True |
||
| 498 | uni_a = get_uni_mocked(interface_port=2, tag_value=82, |
||
| 499 | switch_id="switch_uni_a", is_valid=True) |
||
| 500 | uni_z = get_uni_mocked(interface_port=3, tag_value=83, |
||
| 501 | switch_id="switch_uni_z", is_valid=True) |
||
| 502 | |||
| 503 | primary_links = [ |
||
| 504 | get_link_mocked(endpoint_a_port=9, endpoint_b_port=10, |
||
| 505 | metadata={"s_vlan": 5}), |
||
| 506 | get_link_mocked(endpoint_a_port=11, endpoint_b_port=12, |
||
| 507 | metadata={"s_vlan": 6}) |
||
| 508 | ] |
||
| 509 | |||
| 510 | attributes = { |
||
| 511 | "controller": get_controller_mock(), |
||
| 512 | "name": "custom_name", |
||
| 513 | "uni_a": uni_a, |
||
| 514 | "uni_z": uni_z, |
||
| 515 | "primary_links": primary_links, |
||
| 516 | "queue_id": 5 |
||
| 517 | } |
||
| 518 | # Setup path to deploy |
||
| 519 | path = Path() |
||
| 520 | path.append(primary_links[0]) |
||
| 521 | path.append(primary_links[1]) |
||
| 522 | |||
| 523 | evc = EVC(**attributes) |
||
| 524 | |||
| 525 | deployed = evc.deploy_to_path(path) |
||
| 526 | |||
| 527 | self.assertEqual(discover_new_paths.call_count, 0) |
||
| 528 | self.assertEqual(should_deploy_mock.call_count, 1) |
||
| 529 | self.assertEqual(install_nni_flows.call_count, 1) |
||
| 530 | self.assertEqual(choose_vlans_mock.call_count, 1) |
||
| 531 | self.assertEqual(log_mock.error.call_count, 1) |
||
| 532 | self.assertEqual(sync_mock.call_count, 0) |
||
| 533 | self.assertEqual(remove_current_flows.call_count, 2) |
||
| 534 | self.assertFalse(deployed) |
||
| 535 | |||
| 536 | @patch('napps.kytos.mef_eline.models.EVC.deploy_to_path') |
||
| 537 | @patch('napps.kytos.mef_eline.models.EVC.discover_new_paths') |
||
| 538 | def test_deploy_to_backup_path1(self, discover_new_paths_mocked, |
||
| 539 | deploy_to_path_mocked): |
||
| 540 | """Test deployment when dynamic_backup_path is False in same switch""" |
||
| 541 | uni_a = get_uni_mocked(interface_port=2, tag_value=82, |
||
| 542 | is_valid=True) |
||
| 543 | uni_z = get_uni_mocked(interface_port=3, tag_value=83, |
||
| 544 | is_valid=True) |
||
| 545 | |||
| 546 | switch = Mock(spec=Switch) |
||
| 547 | uni_a.interface.switch = switch |
||
| 548 | uni_z.interface.switch = switch |
||
| 549 | |||
| 550 | attributes = { |
||
| 551 | "controller": get_controller_mock(), |
||
| 552 | "name": "custom_name", |
||
| 553 | "uni_a": uni_a, |
||
| 554 | "uni_z": uni_z, |
||
| 555 | "enabled": True, |
||
| 556 | "dynamic_backup_path": False |
||
| 557 | } |
||
| 558 | |||
| 559 | evc = EVC(**attributes) |
||
| 560 | discover_new_paths_mocked.return_value = [] |
||
| 561 | deploy_to_path_mocked.return_value = True |
||
| 562 | |||
| 563 | # storehouse initialization mock |
||
| 564 | evc._storehouse.box = Mock() # pylint: disable=protected-access |
||
| 565 | evc._storehouse.box.data = {} # pylint: disable=protected-access |
||
| 566 | |||
| 567 | deployed = evc.deploy_to_backup_path() |
||
| 568 | |||
| 569 | deploy_to_path_mocked.assert_called_once_with() |
||
| 570 | self.assertEqual(deployed, True) |
||
| 571 | |||
| 572 | @patch('requests.post') |
||
| 573 | @patch('napps.kytos.mef_eline.storehouse.StoreHouse.save_evc') |
||
| 574 | @patch('napps.kytos.mef_eline.models.log') |
||
| 575 | @patch('napps.kytos.mef_eline.models.Path.choose_vlans') |
||
| 576 | @patch('napps.kytos.mef_eline.models.EVC._install_nni_flows') |
||
| 577 | @patch('napps.kytos.mef_eline.models.EVC._install_uni_flows') |
||
| 578 | @patch('napps.kytos.mef_eline.models.EVC.activate') |
||
| 579 | @patch('napps.kytos.mef_eline.models.EVC.should_deploy') |
||
| 580 | @patch('napps.kytos.mef_eline.models.EVC.discover_new_paths') |
||
| 581 | def test_deploy_without_path_case1(self, *args): |
||
| 582 | """Test if not path is found a dynamic path is used.""" |
||
| 583 | # pylint: disable=too-many-locals |
||
| 584 | (discover_new_paths_mocked, should_deploy_mock, activate_mock, |
||
| 585 | install_uni_flows_mock, install_nni_flows, chose_vlans_mock, |
||
| 586 | log_mock, _, requests_mock) = args |
||
| 587 | |||
| 588 | response = MagicMock() |
||
| 589 | response.status_code = 201 |
||
| 590 | requests_mock.return_value = response |
||
| 591 | |||
| 592 | should_deploy_mock.return_value = False |
||
| 593 | uni_a = get_uni_mocked(interface_port=2, tag_value=82, |
||
| 594 | switch_id="switch_uni_a", is_valid=True) |
||
| 595 | uni_z = get_uni_mocked(interface_port=3, tag_value=83, |
||
| 596 | switch_id="switch_uni_z", is_valid=True) |
||
| 597 | |||
| 598 | attributes = { |
||
| 599 | "controller": get_controller_mock(), |
||
| 600 | "name": "custom_name", |
||
| 601 | "uni_a": uni_a, |
||
| 602 | "uni_z": uni_z, |
||
| 603 | "enabled": True, |
||
| 604 | "dynamic_backup_path": False |
||
| 605 | } |
||
| 606 | |||
| 607 | dynamic_backup_path = Path([ |
||
| 608 | get_link_mocked(endpoint_a_port=9, endpoint_b_port=10, |
||
| 609 | metadata={"s_vlan": 5}), |
||
| 610 | get_link_mocked(endpoint_a_port=11, endpoint_b_port=12, |
||
| 611 | metadata={"s_vlan": 6}) |
||
| 612 | ]) |
||
| 613 | |||
| 614 | evc = EVC(**attributes) |
||
| 615 | discover_new_paths_mocked.return_value = [dynamic_backup_path] |
||
| 616 | |||
| 617 | deployed = evc.deploy_to_path() |
||
| 618 | |||
| 619 | self.assertEqual(should_deploy_mock.call_count, 1) |
||
| 620 | self.assertEqual(discover_new_paths_mocked.call_count, 1) |
||
| 621 | self.assertEqual(activate_mock.call_count, 1) |
||
| 622 | self.assertEqual(install_uni_flows_mock.call_count, 1) |
||
| 623 | self.assertEqual(install_nni_flows.call_count, 1) |
||
| 624 | self.assertEqual(chose_vlans_mock.call_count, 1) |
||
| 625 | log_mock.info.assert_called_with(f"{evc} was deployed.") |
||
| 626 | self.assertTrue(deployed) |
||
| 627 | |||
| 628 | @patch('napps.kytos.mef_eline.models.EVC._send_flow_mods') |
||
| 629 | def test_remove_current_flows(self, send_flow_mods_mocked): |
||
| 630 | """Test remove current flows.""" |
||
| 631 | uni_a = get_uni_mocked(interface_port=2, tag_value=82, |
||
| 632 | switch_id="switch_uni_a", is_valid=True) |
||
| 633 | uni_z = get_uni_mocked(interface_port=3, tag_value=83, |
||
| 634 | switch_id="switch_uni_z", is_valid=True) |
||
| 635 | |||
| 636 | switch_a = Switch('00:00:00:00:00:01') |
||
| 637 | switch_b = Switch('00:00:00:00:00:02') |
||
| 638 | switch_c = Switch('00:00:00:00:00:03') |
||
| 639 | |||
| 640 | attributes = { |
||
| 641 | "controller": get_controller_mock(), |
||
| 642 | "name": "custom_name", |
||
| 643 | "uni_a": uni_a, |
||
| 644 | "uni_z": uni_z, |
||
| 645 | "active": True, |
||
| 646 | "enabled": True, |
||
| 647 | "primary_links": [ |
||
| 648 | get_link_mocked(switch_a=switch_a, |
||
| 649 | switch_b=switch_b, |
||
| 650 | endpoint_a_port=9, endpoint_b_port=10, |
||
| 651 | metadata={"s_vlan": 5}), |
||
| 652 | get_link_mocked(switch_a=switch_b, |
||
| 653 | switch_b=switch_c, |
||
| 654 | endpoint_a_port=11, endpoint_b_port=12, |
||
| 655 | metadata={"s_vlan": 6}) |
||
| 656 | ] |
||
| 657 | } |
||
| 658 | |||
| 659 | evc = EVC(**attributes) |
||
| 660 | |||
| 661 | # storehouse initialization mock |
||
| 662 | evc._storehouse.box = Mock() # pylint: disable=protected-access |
||
| 663 | evc._storehouse.box.data = {} # pylint: disable=protected-access |
||
| 664 | |||
| 665 | evc.current_path = evc.primary_links |
||
| 666 | evc.remove_current_flows() |
||
| 667 | |||
| 668 | self.assertEqual(send_flow_mods_mocked.call_count, 5) |
||
| 669 | self.assertFalse(evc.is_active()) |
||
| 670 | flows = [{'cookie': evc.get_cookie(), |
||
| 671 | 'cookie_mask': 18446744073709551615}] |
||
| 672 | switch_1 = evc.primary_links[0].endpoint_a.switch |
||
| 673 | switch_2 = evc.primary_links[0].endpoint_b.switch |
||
| 674 | send_flow_mods_mocked.assert_any_call(switch_1, flows, 'delete') |
||
| 675 | send_flow_mods_mocked.assert_any_call(switch_2, flows, 'delete') |
||
| 676 |
Loading history...