fix admintech api and unittests

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,15 +142,15 @@ 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
         Returns:
             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")

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