|
@@ 299-354 (lines=56) @@
|
| 296 |
|
msg = f'Failed to re-deploy {evc} after link down.' |
| 297 |
|
log_mocked.debug.assert_called_once_with(msg) |
| 298 |
|
|
| 299 |
|
@patch('napps.kytos.mef_eline.models.log') |
| 300 |
|
@patch('napps.kytos.mef_eline.models.EVCDeploy.deploy') |
| 301 |
|
@patch('napps.kytos.mef_eline.models.EVCDeploy._send_flow_mods') |
| 302 |
|
@patch('napps.kytos.mef_eline.models.DynamicPathManager.get_best_path') |
| 303 |
|
@patch(DEPLOY_TO_PRIMARY_PATH) |
| 304 |
|
@patch('napps.kytos.mef_eline.models.Path.status', EntityStatus.DOWN) |
| 305 |
|
def test_handle_link_down_case_4(self, deploy_to_mocked, |
| 306 |
|
get_best_path_mocked, |
| 307 |
|
_send_flow_mods_mocked, |
| 308 |
|
deploy_mocked, |
| 309 |
|
log_mocked): |
| 310 |
|
"""Test if circuit with dynamic path is return success.""" |
| 311 |
|
deploy_mocked.return_value = True |
| 312 |
|
deploy_to_mocked.return_value = False |
| 313 |
|
primary_path = [ |
| 314 |
|
get_link_mocked(endpoint_a_port=9, endpoint_b_port=10, |
| 315 |
|
metadata={"s_vlan": 5}, |
| 316 |
|
status=EntityStatus.DOWN), |
| 317 |
|
get_link_mocked(endpoint_a_port=11, endpoint_b_port=12, |
| 318 |
|
metadata={"s_vlan": 6}, |
| 319 |
|
status=EntityStatus.UP), |
| 320 |
|
] |
| 321 |
|
backup_path = [ |
| 322 |
|
get_link_mocked(endpoint_a_port=9, endpoint_b_port=10, |
| 323 |
|
metadata={"s_vlan": 5}, |
| 324 |
|
status=EntityStatus.DOWN), |
| 325 |
|
get_link_mocked(endpoint_a_port=13, endpoint_b_port=14, |
| 326 |
|
metadata={"s_vlan": 6}, |
| 327 |
|
status=EntityStatus.UP), |
| 328 |
|
] |
| 329 |
|
attributes = { |
| 330 |
|
"controller": get_controller_mock(), |
| 331 |
|
"name": "circuit_name", |
| 332 |
|
"uni_a": get_uni_mocked(is_valid=True), |
| 333 |
|
"uni_z": get_uni_mocked(is_valid=True), |
| 334 |
|
"primary_path": primary_path, |
| 335 |
|
"backup_path": backup_path, |
| 336 |
|
"enabled": True, |
| 337 |
|
"dynamic_backup_path": True |
| 338 |
|
} |
| 339 |
|
|
| 340 |
|
evc = EVC(**attributes) |
| 341 |
|
evc.current_path = evc.backup_path |
| 342 |
|
|
| 343 |
|
# storehouse mock |
| 344 |
|
evc._storehouse.box = Mock() # pylint: disable=protected-access |
| 345 |
|
evc._storehouse.box.data = {} # pylint: disable=protected-access |
| 346 |
|
|
| 347 |
|
current_handle_link_down = evc.handle_link_down() |
| 348 |
|
|
| 349 |
|
self.assertEqual(get_best_path_mocked.call_count, 1) |
| 350 |
|
self.assertEqual(deploy_to_mocked.call_count, 1) |
| 351 |
|
|
| 352 |
|
self.assertTrue(current_handle_link_down) |
| 353 |
|
msg = f"{evc} deployed after link down." |
| 354 |
|
log_mocked.debug.assert_called_with(msg) |
| 355 |
|
|
| 356 |
|
@patch('napps.kytos.mef_eline.models.EVCDeploy.deploy') |
| 357 |
|
@patch('napps.kytos.mef_eline.models.LinkProtection.deploy_to') |
|
@@ 250-297 (lines=48) @@
|
| 247 |
|
msg = f"{evc} deployed after link down." |
| 248 |
|
log_mocked.debug.assert_called_once_with(msg) |
| 249 |
|
|
| 250 |
|
@patch('napps.kytos.mef_eline.models.log') |
| 251 |
|
@patch('napps.kytos.mef_eline.models.EVCDeploy.deploy') |
| 252 |
|
@patch(DEPLOY_TO_PRIMARY_PATH) |
| 253 |
|
@patch('napps.kytos.mef_eline.models.DynamicPathManager.get_paths') |
| 254 |
|
@patch('napps.kytos.mef_eline.models.Path.status', EntityStatus.DOWN) |
| 255 |
|
def test_handle_link_down_case_3(self, get_paths_mocked, |
| 256 |
|
deploy_to_mocked, deploy_mocked, |
| 257 |
|
log_mocked): |
| 258 |
|
"""Test if circuit without dynamic path is return failed.""" |
| 259 |
|
deploy_mocked.return_value = False |
| 260 |
|
deploy_to_mocked.return_value = False |
| 261 |
|
primary_path = [ |
| 262 |
|
get_link_mocked(endpoint_a_port=9, endpoint_b_port=10, |
| 263 |
|
metadata={"s_vlan": 5}, |
| 264 |
|
status=EntityStatus.DOWN), |
| 265 |
|
get_link_mocked(endpoint_a_port=11, endpoint_b_port=12, |
| 266 |
|
metadata={"s_vlan": 6}, |
| 267 |
|
status=EntityStatus.UP), |
| 268 |
|
] |
| 269 |
|
backup_path = [ |
| 270 |
|
get_link_mocked(endpoint_a_port=9, endpoint_b_port=10, |
| 271 |
|
metadata={"s_vlan": 5}, |
| 272 |
|
status=EntityStatus.DOWN), |
| 273 |
|
get_link_mocked(endpoint_a_port=13, endpoint_b_port=14, |
| 274 |
|
metadata={"s_vlan": 6}, |
| 275 |
|
status=EntityStatus.UP), |
| 276 |
|
] |
| 277 |
|
attributes = { |
| 278 |
|
"controller": get_controller_mock(), |
| 279 |
|
"name": "circuit_name", |
| 280 |
|
"uni_a": get_uni_mocked(is_valid=True), |
| 281 |
|
"uni_z": get_uni_mocked(is_valid=True), |
| 282 |
|
"primary_path": primary_path, |
| 283 |
|
"backup_path": backup_path, |
| 284 |
|
"enabled": True |
| 285 |
|
} |
| 286 |
|
|
| 287 |
|
evc = EVC(**attributes) |
| 288 |
|
evc.current_path = evc.backup_path |
| 289 |
|
current_handle_link_down = evc.handle_link_down() |
| 290 |
|
|
| 291 |
|
self.assertEqual(get_paths_mocked.call_count, 0) |
| 292 |
|
self.assertEqual(deploy_mocked.call_count, 0) |
| 293 |
|
self.assertEqual(deploy_to_mocked.call_count, 1) |
| 294 |
|
|
| 295 |
|
self.assertFalse(current_handle_link_down) |
| 296 |
|
msg = f'Failed to re-deploy {evc} after link down.' |
| 297 |
|
log_mocked.debug.assert_called_once_with(msg) |
| 298 |
|
|
| 299 |
|
@patch('napps.kytos.mef_eline.models.log') |
| 300 |
|
@patch('napps.kytos.mef_eline.models.EVCDeploy.deploy') |
ajoaoff /
mef_eline
