Add new way of authentication

.github/workflows/linting.yml

@@ -42,6 +42,6 @@ jobs:
       - name: Checking if imports are sorted correctly
         run: poetry run isort --check --diff -l 120 --profile black vmngclient
       - name: Check static-typing
-        run: poetry run mypy --show-error-codes --show-error-context --pretty vmngclient
+        run: poetry run mypy --show-error-codes --show-error-context --pretty --install-types --non-interactive vmngclient
       - name: Check code style
         run: poetry run flake8 --max-line-length 120 vmngclient

.pre-commit-config.yaml

@@ -32,15 +32,18 @@ repos:
 
   # Code analysis (only checks staged files)
 
-  - repo: local
+  - repo: https://github.com/pre-commit/mirrors-mypy
+    rev: v0.991
     hooks:
       - id: mypy
         name: mypy
         entry: "poetry run mypy"
-        language: system
-        types: [python]
-        require_serial: true
-        args: ["--show-error-codes", "--show-error-context", "--pretty"]
+        # language: system
+        # types: [python]
+        # require_serial: true
+        args: ["--show-error-codes", "--show-error-context", "--pretty", "--show-absolute-path"]
+        additional_dependencies: ['types-requests', 'types-attrs']
+        verbose: true
 
 
   - repo: https://github.com/PyCQA/flake8

pyproject.toml

@@ -14,13 +14,14 @@ attrs = "^21.4.0"
 aiohttp = "^3.8.1"
 ciscoconfparse = "^1.6.40"
 tenacity = "^8.1.0"
+parameterized = "^0.8.1"
 
 [tool.poetry.dev-dependencies]
 pytest = "^7.1.2"
 pytest-mock = "^3.7.0"
 isort = "^5.10.1"
 pre-commit = "^2.19.0"
-mypy = "^0.971"
+mypy = "0.991"
 flake8 = "^5.0.4"
 Sphinx = "^5.2.3"
 

vmngclient/api/admin_tech_api.py

@@ -4,13 +4,14 @@
 import json
 import shutil
 import time
-from http.client import HTTPResponse
 from pathlib import Path
 from typing import List, Optional, cast
 from urllib.error import HTTPError
 
+from requests import Response
+
 from vmngclient.dataclasses import AdminTech
-from vmngclient.session import Session
+from vmngclient.session import vManageSession
 from vmngclient.utils.creation_tools import create_dataclass
 
 
@@ -29,7 +30,7 @@ class AdminTechAPI:
         session: logged in API client session
     """
 
-    def __init__(self, session: Session) -> None:
+    def __init__(self, session: vManageSession) -> None:
         self.session = session
 
     def __str__(self) -> str:
@@ -44,7 +45,7 @@ def get(self, device_id: str) -> AdminTech:
             AdminTech object for given device
         """
         body = {'deviceIP': device_id}
-        response = self.session.post_data('/dataservice/device/tools/admintechlist', data=body)
+        response = self.session.post(url='/dataservice/device/tools/admintechlist', data=body).json()
         return create_dataclass(AdminTech, response[0])
 
     def get_all(self) -> List[AdminTech]:
@@ -71,12 +72,10 @@ def generate(
         """
         create_admin_tech_error_msgs = 'Admin tech creation already in progress'
         body = {'deviceIP': device_id, 'exclude-cores': True, 'exclude-tech': False, 'exclude-logs': True}
-        _session_timeout = self.session.timeout
         polling_timer = polling_timeout
         while polling_timer > 0:
             try:
-                self.session.timeout = request_timeout
-                response = self.session.post_json('/dataservice/device/tools/admintech', body)
+                response = self.session.post(url='/dataservice/device/tools/admintech', data=body)
                 return cast(dict, response)['fileName']
             except HTTPError as error:
                 error_details = error.read().decode()
@@ -86,8 +85,6 @@ def generate(
                     raise GenerateAdminTechLogError(f'It is not possible to generate admintech log for {device_id}')
                 time.sleep(polling_interval)
                 polling_timer -= polling_interval
-            finally:
-                self.session.timeout = _session_timeout
         raise GenerateAdminTechLogError(f'It is not possible to generate admintech log for {device_id}')
 
     def _get_token_id(self, device_id) -> str:
@@ -98,7 +95,7 @@ def _get_token_id(self, device_id) -> str:
                 return admin_tech.token_id
         raise RequestTokenIdNotFound(f'Request Id of admin tech generation request not found for device: {device_id}')
 
-    def delete(self, device_id: str) -> HTTPResponse:
+    def delete(self, device_id: str) -> Response:
         """Deletes admin tech logs for a device.
 
         Args:
@@ -123,7 +120,7 @@ def download(self, admin_tech_name: str, download_dir: Optional[Path] = None) ->
         if not download_dir:
             download_dir = Path.cwd()
         download_path = download_dir / admin_tech_name
-        with self.session.get(f'/dataservice/device/tools/admintech/download/{admin_tech_name}') as payload:
+        with self.session.get_data(f'/dataservice/device/tools/admintech/download/{admin_tech_name}') as payload:
             with open(download_path, 'wb') as file:
                 shutil.copyfileobj(payload, file)
         return download_path

vmngclient/api/administration.py

@@ -2,7 +2,7 @@
 from typing import List, Union, cast
 
 from vmngclient.dataclasses import CloudConnectorData, CloudServicesSettings, ServiceConfigurationData, User
-from vmngclient.session import Session
+from vmngclient.session import vManageSession
 from vmngclient.utils.creation_tools import asdict, create_dataclass, get_logger_name
 
 logger = logging.getLogger(get_logger_name(__name__))
@@ -17,7 +17,7 @@ class UserDoesNotExists(Exception):
 
 
 class UsersAPI:
-    def __init__(self, session: Session) -> None:
+    def __init__(self, session: vManageSession) -> None:
         self.session = session
 
     def get_all_users(self) -> List[User]:
@@ -35,7 +35,7 @@ def create_user(self, user: User) -> None:
         url_path = "/dataservice/admin/user"
         data = asdict(user)  # type: ignore
 
-        response = self.session.post_json(url_path, data)
+        response = self.session.post(url=url_path, data=data)
         logger.info(response)
 
     def delete_user(self, username: str) -> bool:
@@ -44,7 +44,7 @@ def delete_user(self, username: str) -> bool:
         url_path = f"/dataservice/admin/user/{username}"
         logger.debug(f"Deleting user {username}.")
         response = self.session.delete(url_path)
-        return True if response.status == 200 else False
+        return True if response.status_code == 200 else False
 
 
 class ClusterManagementAPI:
@@ -54,9 +54,8 @@ class ClusterManagementAPI:
         session: logged in API admin session
     """
 
-    def __init__(self, session: Session) -> None:
+    def __init__(self, session: vManageSession) -> None:
         self.session = session
-        self.session.login()  # consider using login in Session init
 
     def modify_cluster_setup(self, service_configuration: ServiceConfigurationData) -> bool:
         """Updates vManage cluster configuration.
@@ -67,7 +66,7 @@ def modify_cluster_setup(self, service_configuration: ServiceConfigurationData)
         url_path = "/dataservice/clusterManagement/setup"
         data = asdict(service_configuration)  # type: ignore
         response = self.session.put(url_path, data)
-        return True if response.status == 200 else False
+        return True if response.status_code == 200 else False
 
     def get_cluster_management_health_status(self) -> Union[dict, list]:
         """Gets Cluster Management Service Reachability health status.
@@ -80,14 +79,13 @@ def get_cluster_management_health_status(self) -> Union[dict, list]:
 
 
 class AdministrationSettingsAPI:
-    def __init__(self, session: Session) -> None:
+    def __init__(self, session: vManageSession) -> None:
         """Covers Administration Settings API calls.
 
         Args:
             session: logged in API admin session
         """
         self.session = session
-        self.session.login()  # consider using login in Session init
 
     def get_sdavc_cloud_connector_config(self) -> CloudConnectorData:
         """Gets SD-AVC Cloud Connector Config.
@@ -104,14 +102,14 @@ def enable_sdavc_cloud_connector(self, cloud_connector: CloudConnectorData) -> b
         url_path = "/dataservice/sdavc/cloudconnector"
         data = asdict(cloud_connector)  # type: ignore
         response = self.session.post(url_path, data)
-        return True if response.status == 200 else False
+        return True if response.status_code == 200 else False
 
     def disable_sdavc_cloud_connector(self) -> bool:
         """Disables SD-AVC Cloud Connector on vManage."""
         url_path = "/dataservice/sdavc/cloudconnector"
         data = {"cloudEnabled": False}
         response = self.session.put(url_path, data)
-        return True if response.status == 200 else False
+        return True if response.status_code == 200 else False
 
     def get_cloud_services(self) -> CloudServicesSettings:
         url_path = "/dataservice/settings/configuration/cloudservices"
@@ -121,7 +119,7 @@ def get_cloud_services(self) -> CloudServicesSettings:
     def set_cloud_services(self, config: CloudServicesSettings) -> bool:
         url_path = "/dataservice/settings/configuration/cloudservices"
         response = self.session.post(url_path, asdict(config))  # type: ignore
-        return True if response.status == 200 else False
+        return True if response.status_code == 200 else False
 
     def get_cloud_on_ramp_for_saas_mode(self):
         """Get information about Cloud on Ramp for Saas mode"""

vmngclient/api/alarms_api.py

@@ -1,11 +1,11 @@
 import logging
 from enum import Enum
-from typing import Any, Callable, Dict, List, Set
+from typing import Any, Callable, Dict, List, Optional, Set
 
-from tenacity import retry, retry_if_result, stop_after_attempt, wait_fixed
+from tenacity import retry, retry_if_result, stop_after_attempt, wait_fixed  # type: ignore
 
 from vmngclient.dataclasses import AlarmData
-from vmngclient.session import Session
+from vmngclient.session import vManageSession
 from vmngclient.utils.creation_tools import create_dataclass, flatten_dict, get_logger_name
 
 logger = logging.getLogger(get_logger_name(__name__))
@@ -26,10 +26,12 @@ class AlarmsAPI:
 
     URL = '/dataservice/alarms'
 
-    def __init__(self, session: Session):
+    def __init__(self, session: vManageSession):
         self.session = session
 
-    def get_alarms(self, hours: int = None, level: str = None, viewed: bool = None) -> List[AlarmData]:
+    def get_alarms(
+        self, hours: Optional[int] = None, level: Optional[str] = None, viewed: Optional[bool] = None
+    ) -> List[AlarmData]:
         query: Dict[str, Any] = {"query": {"condition": "AND", "rules": []}}
 
         if hours:
@@ -50,22 +52,22 @@ def get_alarms(self, hours: int = None, level: str = None, viewed: bool = None)
                 {"value": [value], "field": "acknowledged", "type": "bool", "operator": "equal"}
             )
 
-        alarms = self.session.post_data(AlarmsAPI.URL, data=query)
+        alarms = self.session.post(url=AlarmsAPI.URL, data=query).json()
 
         logger.info("Actualas alarms collected successfuly.")
 
         return [create_dataclass(AlarmData, flatten_dict(alarm)) for alarm in alarms]
 
-    def get_critical_alarms(self, hours: int = None, viewed: bool = None) -> List[AlarmData]:
+    def get_critical_alarms(self, hours: Optional[int] = None, viewed: Optional[bool] = None) -> List[AlarmData]:
         return self.get_alarms(hours, AlarmLevel.critical.value, viewed)
 
-    def get_major_alarms(self, hours: int = None, viewed: bool = None) -> List[AlarmData]:
+    def get_major_alarms(self, hours: Optional[int] = None, viewed: Optional[bool] = None) -> List[AlarmData]:
         return self.get_alarms(hours, AlarmLevel.major.value, viewed)
 
-    def get_minor_alarms(self, hours: int = None, viewed: bool = None) -> List[AlarmData]:
+    def get_minor_alarms(self, hours: Optional[int] = None, viewed: Optional[bool] = None) -> List[AlarmData]:
         return self.get_alarms(hours, AlarmLevel.minor.value, viewed)
 
-    def get_not_viewed_alarms(self, hours: int = None) -> List[AlarmData]:
+    def get_not_viewed_alarms(self, hours: Optional[int] = None) -> List[AlarmData]:
         return self.get_alarms(hours=hours, viewed=False)
 
     def mark_all_as_viewed(self) -> bool:

vmngclient/api/basic_api.py

@@ -2,12 +2,12 @@
 import json
 from contextlib import contextmanager
 from enum import Enum
-from typing import Iterator, List, Union, cast
+from typing import Iterator, List, Union
 
-from tenacity import retry, retry_if_result, stop_after_attempt, wait_fixed
+from tenacity import retry, retry_if_result, stop_after_attempt, wait_fixed  # type: ignore
 
 from vmngclient.dataclasses import BfdSessionData, Connection, Device, Reboot, WanInterface
-from vmngclient.session import Session
+from vmngclient.session import vManageSession
 from vmngclient.utils.creation_tools import create_dataclass
 from vmngclient.utils.operation_status import OperationStatus
 from vmngclient.utils.personality import Personality
@@ -32,7 +32,7 @@ class DevicesAPI:
         devices: List with system status for all devices
     """
 
-    def __init__(self, session: Session) -> None:
+    def __init__(self, session: vManageSession) -> None:
         self.session = session
 
     def __str__(self) -> str:
@@ -168,7 +168,7 @@ def wait_for_state():
             status_api = f'/dataservice/device/action/status/{action_id}'
             return self.session.get_data(f'{status_api}')
 
-        response = cast(dict, self.session.post_json('/dataservice/certificate/vedge/list?action=push'))
+        response = self.session.post('/dataservice/certificate/vedge/list?action=push').json()
         if response.get('id'):
             action_id = response['id']
         else:
@@ -198,7 +198,7 @@ class DeviceStateAPI:
         session: logged in API client session
     """
 
-    def __init__(self, session: Session) -> None:
+    def __init__(self, session: vManageSession) -> None:
         self.session = session
 
     def __str__(self) -> str:
@@ -275,29 +275,29 @@ def get_device_wan_interfaces(self, device_id: str):
 
     def get_colors(self, device_id: str) -> List[str]:
         url = '/dataservice/device/bfd/state/device/tlocInterfaceMap'
-        colors_raw = DevicesAPI(self.session).session.get(url + f'?deviceId={device_id}')
+        colors_raw = DevicesAPI(self.session).session.get_data(url + f'?deviceId={device_id}')
         json_colors = json.loads(str(colors_raw.read(), 'utf-8'))
         colors = list(json_colors["intfList"].keys())
 
         return colors
 
     @contextmanager
-    def enable_data_stream(self) -> Iterator:
+    def enable_data_stream(self) -> Iterator:  # TODO check
         try:
             url_path = "/dataservice/settings/configuration/vmanagedatastream"
             data_stream_status = self.session.get_data(url_path)[0]
-            query = {
-                "enable": True,
+            query = {  # TODO Dict[str, obj]
+                "enable": "True",
                 "ipType": "systemIp",
                 "serverHostName": "systemIp",
-                "vpn": 0,
+                "vpn": "0",
             }
             url_path = "/dataservice/settings/configuration/vmanagedatastream"
-            self.session.post_data(url_path, query)
+            self.session.post(url=url_path, params=query)
             yield None
         finally:
             url_path = "/dataservice/settings/configuration/vmanagedatastream"
-            self.session.post_data(url_path, data_stream_status)
+            self.session.post(url=url_path, data=data_stream_status)
 
     def get_bfd_sessions(self, device_id: str) -> List[BfdSessionData]:
         items = self.session.get_data(f'/dataservice/device/bfd/sessions?deviceId={device_id}')