From cb4fc9c00f506cff2958c6fd90327c502d0cfc04 Mon Sep 17 00:00:00 2001 From: sbasan Date: Thu, 11 Apr 2024 13:34:59 +0200 Subject: [PATCH] fix admintech api and unittests --- catalystwan/api/admin_tech_api.py | 14 ++++++-------- catalystwan/tests/test_admin_tech_api.py | 20 ++++++++++---------- 2 files changed, 16 insertions(+), 18 deletions(-) diff --git a/catalystwan/api/admin_tech_api.py b/catalystwan/api/admin_tech_api.py index 088251f2..9e0eb8a7 100644 --- a/catalystwan/api/admin_tech_api.py +++ b/catalystwan/api/admin_tech_api.py @@ -10,11 +10,9 @@ from pathlib import Path from typing import TYPE_CHECKING, List, Optional -from requests import Response # type: ignore -from requests.exceptions import HTTPError # type: ignore - from catalystwan.dataclasses import AdminTech, DeviceAdminTech -from catalystwan.exceptions import CatalystwanException +from catalystwan.exceptions import CatalystwanException, ManagerHTTPError +from catalystwan.response import ManagerResponse from catalystwan.utils.creation_tools import create_dataclass if TYPE_CHECKING: @@ -117,7 +115,7 @@ def generate( json=body, timeout=request_timeout, ) - except HTTPError as http_error: + except ManagerHTTPError as http_error: response = http_error.response # type: ignore if response.status_code == 200: return response.json()["fileName"] @@ -131,7 +129,7 @@ def generate( polling_timer -= polling_interval raise GenerateAdminTechLogError(f"It is not possible to generate admintech log for {device_id}") - def _get_token_id(self, filename: str, timeout: int = 3600, interval: int = 30) -> str: + def _get_token_id(self, filename: str, timeout: int, interval: int) -> str: # Wait for the file to be ready and obtain the token_id end_time = time.time() + timeout while time.time() < end_time: @@ -144,7 +142,7 @@ def _get_token_id(self, filename: str, timeout: int = 3600, interval: int = 30) f"requestTokenId of admin tech generation request not found for file name: {filename}" ) - def delete(self, filename: str) -> Response: + def delete(self, filename: str, timeout: int = 3600, interval: int = 30) -> ManagerResponse: """Deletes admin tech logs for a device. Args: filename: name of admin_tech file @@ -152,7 +150,7 @@ def delete(self, filename: str) -> Response: response: http response for delete operation """ - token_id = self._get_token_id(filename) + token_id = self._get_token_id(filename, timeout, interval) response = self.session.delete(f"/dataservice/device/tools/admintech/{token_id}") if response.status_code == 200: logger.info(f"Deleted AdminTech file {filename} on remote") diff --git a/catalystwan/tests/test_admin_tech_api.py b/catalystwan/tests/test_admin_tech_api.py index 7716a04a..2a7788a2 100644 --- a/catalystwan/tests/test_admin_tech_api.py +++ b/catalystwan/tests/test_admin_tech_api.py @@ -63,7 +63,7 @@ def setUp(self): self.download_file = io.BytesIO(self.download_file_content.encode()) @patch("catalystwan.session.ManagerSession") - @patch("requests.Response") + @patch("catalystwan.response.ManagerResponse") def test_get(self, mock_session, mock_response): # Arrange mock_session.post.return_value = mock_response @@ -78,7 +78,7 @@ def test_get(self, mock_session, mock_response): self.assertIsInstance(admintechs[0], DeviceAdminTech) @patch("catalystwan.session.ManagerSession") - @patch("requests.Response") + @patch("catalystwan.response.ManagerResponse") def test_get_all(self, mock_session, mock_response): # Arrange mock_session.get.return_value = mock_response @@ -90,7 +90,7 @@ def test_get_all(self, mock_session, mock_response): self.assertEqual(len(admintechs), len(self.admin_tech_infos["data"])) @patch("catalystwan.session.ManagerSession") - @patch("requests.Response") + @patch("catalystwan.response.ManagerResponse") def test_generate(self, mock_session, mock_response): # Arrange mock_session.post.return_value = mock_response @@ -106,7 +106,7 @@ def test_generate(self, mock_session, mock_response): self.assertEqual(filename, self.admin_tech_generate_response["fileName"]) @patch("catalystwan.session.ManagerSession") - @patch("requests.Response") + @patch("catalystwan.response.ManagerResponse") def test_generate_in_progress_error_retry(self, mock_session, mock_response): # Arrange mock_session.post.return_value = mock_response @@ -124,7 +124,7 @@ def test_generate_in_progress_error_retry(self, mock_session, mock_response): self.assertEqual(mock_session.post.call_count, count) @patch("catalystwan.session.ManagerSession") - @patch("requests.Response") + @patch("catalystwan.response.ManagerResponse") def test_generate_error(self, mock_session, mock_response): # Arrange mock_session.post.return_value = mock_response @@ -142,7 +142,7 @@ def test_generate_error(self, mock_session, mock_response): mock_session.post.assert_called_once() @patch("catalystwan.session.ManagerSession") - @patch("requests.Response") + @patch("catalystwan.response.ManagerResponse") def test_delete(self, mock_session, mock_response): # Arrange filename = self.admin_tech_generate_response["fileName"] @@ -155,17 +155,17 @@ def test_delete(self, mock_session, mock_response): mock_session.delete.assert_called_once_with(f"/dataservice/device/tools/admintech/{token_id}") @patch("catalystwan.session.ManagerSession") - @patch("requests.Response") + @patch("catalystwan.response.ManagerResponse") def test_delete_token_not_found(self, mock_session, mock_response): # Arrange mock_session.get.return_value = mock_response mock_response.json.return_value = self.admin_tech_infos # Act/Assert with self.assertRaises(RequestTokenIdNotFound): - AdminTechAPI(mock_session).delete("fake-filename.tar.gz") + AdminTechAPI(mock_session).delete("fake-filename.tar.gz", timeout=0.01, interval=0.01) @patch("catalystwan.session.ManagerSession") - @patch("requests.Response") + @patch("catalystwan.response.ManagerResponse") def test_download(self, mock_session, mock_response): # Arrange filename = self.admin_tech_generate_response["fileName"] @@ -180,7 +180,7 @@ def test_download(self, mock_session, mock_response): self.assertEqual(download_path, Path(tmpdir) / filename) @patch("catalystwan.session.ManagerSession") - @patch("requests.Response") + @patch("catalystwan.response.ManagerResponse") def test_download_error(self, mock_session, mock_response): # Arrange mock_session.get.return_value = mock_response