|
@@ 1434-1505 (lines=72) @@
|
| 1431 |
|
assert 200 == response.status_code |
| 1432 |
|
evc_deploy.assert_called_once() |
| 1433 |
|
|
| 1434 |
|
@patch("napps.kytos.mef_eline.main.Main._check_no_tag_duplication") |
| 1435 |
|
@patch("napps.kytos.mef_eline.models.evc.EVC._tag_lists_equal") |
| 1436 |
|
@patch("napps.kytos.mef_eline.main.Main._use_uni_tags") |
| 1437 |
|
@patch("napps.kytos.mef_eline.models.evc.EVC.deploy") |
| 1438 |
|
@patch("napps.kytos.mef_eline.scheduler.Scheduler.add") |
| 1439 |
|
@patch("napps.kytos.mef_eline.main.Main._uni_from_dict") |
| 1440 |
|
@patch("napps.kytos.mef_eline.controllers.ELineController.upsert_evc") |
| 1441 |
|
@patch("napps.kytos.mef_eline.models.evc.EVC._validate") |
| 1442 |
|
@patch("napps.kytos.mef_eline.main.EVC.as_dict") |
| 1443 |
|
async def test_update_circuit_invalid_json( |
| 1444 |
|
self, |
| 1445 |
|
evc_as_dict_mock, |
| 1446 |
|
validate_mock, |
| 1447 |
|
mongo_controller_upsert_mock, |
| 1448 |
|
uni_from_dict_mock, |
| 1449 |
|
sched_add_mock, |
| 1450 |
|
evc_deploy_mock, |
| 1451 |
|
mock_use_uni_tags, |
| 1452 |
|
mock_tags_equal, |
| 1453 |
|
mock_check_duplicate |
| 1454 |
|
): |
| 1455 |
|
"""Test update a circuit circuit.""" |
| 1456 |
|
self.napp.controller.loop = asyncio.get_running_loop() |
| 1457 |
|
validate_mock.return_value = True |
| 1458 |
|
mongo_controller_upsert_mock.return_value = True |
| 1459 |
|
sched_add_mock.return_value = True |
| 1460 |
|
evc_deploy_mock.return_value = True |
| 1461 |
|
mock_use_uni_tags.return_value = True |
| 1462 |
|
mock_tags_equal.return_value = True |
| 1463 |
|
mock_check_duplicate.return_value = True |
| 1464 |
|
uni1 = create_autospec(UNI) |
| 1465 |
|
uni2 = create_autospec(UNI) |
| 1466 |
|
uni1.interface = create_autospec(Interface) |
| 1467 |
|
uni2.interface = create_autospec(Interface) |
| 1468 |
|
uni1.interface.switch = "00:00:00:00:00:00:00:01" |
| 1469 |
|
uni2.interface.switch = "00:00:00:00:00:00:00:02" |
| 1470 |
|
uni_from_dict_mock.side_effect = [uni1, uni2, uni1, uni2] |
| 1471 |
|
|
| 1472 |
|
payload1 = { |
| 1473 |
|
"name": "my evc1", |
| 1474 |
|
"uni_a": { |
| 1475 |
|
"interface_id": "00:00:00:00:00:00:00:01:1", |
| 1476 |
|
"tag": {"tag_type": 'vlan', "value": 80}, |
| 1477 |
|
}, |
| 1478 |
|
"uni_z": { |
| 1479 |
|
"interface_id": "00:00:00:00:00:00:00:02:2", |
| 1480 |
|
"tag": {"tag_type": 'vlan', "value": 1}, |
| 1481 |
|
}, |
| 1482 |
|
"dynamic_backup_path": True, |
| 1483 |
|
} |
| 1484 |
|
|
| 1485 |
|
payload2 = { |
| 1486 |
|
"dynamic_backup_path": False, |
| 1487 |
|
} |
| 1488 |
|
|
| 1489 |
|
evc_as_dict_mock.return_value = payload1 |
| 1490 |
|
response = await self.api_client.post( |
| 1491 |
|
f"{self.base_endpoint}/v2/evc/", |
| 1492 |
|
json=payload1 |
| 1493 |
|
) |
| 1494 |
|
assert 201 == response.status_code |
| 1495 |
|
|
| 1496 |
|
evc_as_dict_mock.return_value = payload2 |
| 1497 |
|
current_data = response.json() |
| 1498 |
|
circuit_id = current_data["circuit_id"] |
| 1499 |
|
response = await self.api_client.patch( |
| 1500 |
|
f"{self.base_endpoint}/v2/evc/{circuit_id}", |
| 1501 |
|
json=payload2 |
| 1502 |
|
) |
| 1503 |
|
current_data = response.json() |
| 1504 |
|
assert 400 == response.status_code |
| 1505 |
|
assert "must have a primary path or" in current_data["description"] |
| 1506 |
|
|
| 1507 |
|
@patch("napps.kytos.mef_eline.main.Main._check_no_tag_duplication") |
| 1508 |
|
@patch("napps.kytos.mef_eline.models.evc.EVC._tag_lists_equal") |
|
@@ 1703-1767 (lines=65) @@
|
| 1700 |
|
with pytest.raises(ValueError): |
| 1701 |
|
self.napp._uni_from_dict(uni_dict) |
| 1702 |
|
|
| 1703 |
|
@patch("napps.kytos.mef_eline.main.Main._check_no_tag_duplication") |
| 1704 |
|
@patch("napps.kytos.mef_eline.models.evc.EVC._tag_lists_equal") |
| 1705 |
|
@patch("napps.kytos.mef_eline.main.Main._use_uni_tags") |
| 1706 |
|
@patch("napps.kytos.mef_eline.models.evc.EVC.deploy") |
| 1707 |
|
@patch("napps.kytos.mef_eline.scheduler.Scheduler.add") |
| 1708 |
|
@patch("napps.kytos.mef_eline.main.Main._uni_from_dict") |
| 1709 |
|
@patch("napps.kytos.mef_eline.models.evc.EVC._validate") |
| 1710 |
|
@patch("napps.kytos.mef_eline.controllers.ELineController.upsert_evc") |
| 1711 |
|
async def test_update_evc_no_json_mime( |
| 1712 |
|
self, |
| 1713 |
|
mongo_controller_upsert_mock, |
| 1714 |
|
validate_mock, |
| 1715 |
|
uni_from_dict_mock, |
| 1716 |
|
sched_add_mock, |
| 1717 |
|
evc_deploy_mock, |
| 1718 |
|
mock_use_uni_tags, |
| 1719 |
|
mock_tags_equal, |
| 1720 |
|
mock_check_duplicate |
| 1721 |
|
): |
| 1722 |
|
"""Test update a circuit with wrong mimetype.""" |
| 1723 |
|
self.napp.controller.loop = asyncio.get_running_loop() |
| 1724 |
|
validate_mock.return_value = True |
| 1725 |
|
sched_add_mock.return_value = True |
| 1726 |
|
evc_deploy_mock.return_value = True |
| 1727 |
|
mock_use_uni_tags.return_value = True |
| 1728 |
|
mock_tags_equal.return_value = True |
| 1729 |
|
mock_check_duplicate.return_value = True |
| 1730 |
|
uni1 = create_autospec(UNI) |
| 1731 |
|
uni2 = create_autospec(UNI) |
| 1732 |
|
uni1.interface = create_autospec(Interface) |
| 1733 |
|
uni2.interface = create_autospec(Interface) |
| 1734 |
|
uni1.interface.switch = "00:00:00:00:00:00:00:01" |
| 1735 |
|
uni2.interface.switch = "00:00:00:00:00:00:00:02" |
| 1736 |
|
uni_from_dict_mock.side_effect = [uni1, uni2, uni1, uni2] |
| 1737 |
|
mongo_controller_upsert_mock.return_value = True |
| 1738 |
|
|
| 1739 |
|
payload1 = { |
| 1740 |
|
"name": "my evc1", |
| 1741 |
|
"uni_a": { |
| 1742 |
|
"interface_id": "00:00:00:00:00:00:00:01:1", |
| 1743 |
|
"tag": {"tag_type": 'vlan', "value": 80}, |
| 1744 |
|
}, |
| 1745 |
|
"uni_z": { |
| 1746 |
|
"interface_id": "00:00:00:00:00:00:00:02:2", |
| 1747 |
|
"tag": {"tag_type": 'vlan', "value": 1}, |
| 1748 |
|
}, |
| 1749 |
|
"dynamic_backup_path": True, |
| 1750 |
|
} |
| 1751 |
|
|
| 1752 |
|
payload2 = {"dynamic_backup_path": False} |
| 1753 |
|
|
| 1754 |
|
response = await self.api_client.post( |
| 1755 |
|
f"{self.base_endpoint}/v2/evc/", |
| 1756 |
|
json=payload1, |
| 1757 |
|
) |
| 1758 |
|
assert 201 == response.status_code |
| 1759 |
|
|
| 1760 |
|
current_data = response.json() |
| 1761 |
|
circuit_id = current_data["circuit_id"] |
| 1762 |
|
response = await self.api_client.patch( |
| 1763 |
|
f"{self.base_endpoint}/v2/evc/{circuit_id}", data=payload2 |
| 1764 |
|
) |
| 1765 |
|
current_data = response.json() |
| 1766 |
|
assert 415 == response.status_code |
| 1767 |
|
assert "application/json" in current_data["description"] |
| 1768 |
|
|
| 1769 |
|
async def test_delete_no_evc(self): |
| 1770 |
|
"""Test delete when EVC does not exist.""" |
kytos-ng /
mef_eline
