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