Merge branch 'main' into add/parcel/aaa

.flake8

@@ -1,5 +1,7 @@
 [flake8]
-per-file-ignores = catalystwan/models/policy/__init__.py: F401
+per-file-ignores = 
+    catalystwan/models/policy/__init__.py: F401
+    catalystwan/models/configuration/feature_profile/sdwan/policy_object/__init__.py: F401
 max-line-length = 120
 inline-quotes = double
 # https://black.readthedocs.io/en/stable/faq.html#why-are-flake8-s-e203-and-w503-violated

README.md

@@ -121,11 +121,11 @@ critical_alarms = session.api.alarms.get(from_time=n).filter(severity=Severity.C
 session.api.users.get()
 
 # Create user
-new_user = User(userName="new_user", password="new_user", group=["netadmin"], description="new user")
+new_user = User(username="new_user", password="new_user", group=["netadmin"], description="new user")
 session.api.users.create(new_user)
 
 # Update user data
-new_user_update = UserUpdateRequest(userName="new_user", group=["netadmin", "netops"], locale="en_US", description="updated-new_user-description", resGroupName="global")
+new_user_update = UserUpdateRequest(username="new_user", group=["netadmin", "netops"], locale="en_US", description="updated-new_user-description")
 session.api.users.update(new_user_update)
 
 # Update user password
@@ -324,14 +324,14 @@ urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
 ## Catching Exceptions
 ```python
 try:
-	session.api.users.delete_user("XYZ")
-except vManageBadRequestError as error:
-	# Process an error.
-	logger.error(error.info.details)
-
-# message = 'Delete users request failed' 
-# details = 'No user with name XYZ was found' 
-# code = 'USER0006'
+    session.api.users.delete("bogus-user-name")
+except ManagerHTTPError as error:
+    # Process an error.
+    print(error.response.status_code)
+    print(error.info.code)
+    print(error.info.message)
+    print(error.info.details)
+
 ```
 
 ## [Supported API endpoints](https://github.com/CiscoDevNet/catalystwan/blob/main/ENDPOINTS.md)

catalystwan/abstractions.py

@@ -0,0 +1,52 @@
+from typing import Optional, Protocol, Type, TypeVar
+
+from packaging.version import Version  # type: ignore
+
+from catalystwan.typed_list import DataSequence
+from catalystwan.utils.session_type import SessionType
+
+T = TypeVar("T")
+
+
+class APIEndpointClientResponse(Protocol):
+    """
+    Interface to response object. Fits "requests.Response"
+    but set of methods is minimal to allow easy migration to another client if needed
+    """
+
+    @property
+    def text(self) -> str:
+        ...
+
+    @property
+    def content(self) -> bytes:
+        ...
+
+    def dataobj(self, cls: Type[T], sourcekey: Optional[str]) -> T:
+        ...
+
+    def dataseq(self, cls: Type[T], sourcekey: Optional[str]) -> DataSequence[T]:
+        ...
+
+    def json(self) -> dict:
+        ...
+
+
+class APIEndpointClient(Protocol):
+    """
+    Interface to client object.
+    We only need a 'request' function and few vmanage session properties obtained from server.
+    Matched to fit "requests.Session" but migration to other client is possible.
+    At his point not very clean as injection of custom kwargs is possible (and sometimes used)
+    """
+
+    def request(self, method: str, url: str, **kwargs) -> APIEndpointClientResponse:
+        ...
+
+    @property
+    def api_version(self) -> Optional[Version]:
+        ...
+
+    @property
+    def session_type(self) -> Optional[SessionType]:
+        ...

catalystwan/api/admin_tech_api.py

@@ -12,7 +12,7 @@
 from requests.exceptions import HTTPError
 
 from catalystwan.dataclasses import AdminTech, DeviceAdminTech
-from catalystwan.exceptions import ManagerError
+from catalystwan.exceptions import CatalystwanException
 from catalystwan.utils.creation_tools import create_dataclass
 
 if TYPE_CHECKING:
@@ -21,15 +21,15 @@
 logger = logging.getLogger(__name__)
 
 
-class GenerateAdminTechLogError(ManagerError):
+class GenerateAdminTechLogError(CatalystwanException):
     pass
 
 
-class DownloadAdminTechLogError(ManagerError):
+class DownloadAdminTechLogError(CatalystwanException):
     pass
 
 
-class RequestTokenIdNotFound(ManagerError):
+class RequestTokenIdNotFound(CatalystwanException):
     pass
 
 

catalystwan/api/administration.py

@@ -31,7 +31,7 @@
     UserRole,
     UserUpdateRequest,
 )
-from catalystwan.exceptions import CatalystwanDeprecationWarning, InvalidOperationError
+from catalystwan.exceptions import CatalystwanDeprecationWarning, CatalystwanException
 from catalystwan.typed_list import DataSequence
 from catalystwan.utils.creation_tools import asdict, create_dataclass
 
@@ -393,7 +393,7 @@ def update(self, payload: Union[Organization, Certificate, Password, Vbond]) ->
         elif isinstance(payload, Vbond):
             response = self.__update_vbond(json_payload)
         else:
-            raise InvalidOperationError(f"Not supported payload type: {type(payload).__name__}")
+            raise CatalystwanException(f"Not supported payload type: {type(payload).__name__}")
 
         return True if response.status_code == HTTPStatus.OK else False
 

catalystwan/api/basic_api.py

@@ -9,7 +9,7 @@
 
 from catalystwan.dataclasses import BfdSessionData, Connection, Device, WanInterface
 from catalystwan.endpoints.real_time_monitoring.reboot_history import RebootEntry
-from catalystwan.exceptions import InvalidOperationError
+from catalystwan.exceptions import CatalystwanException
 from catalystwan.typed_list import DataSequence
 from catalystwan.utils.creation_tools import create_dataclass
 from catalystwan.utils.operation_status import OperationStatus
@@ -128,7 +128,7 @@ def wait_for_state():
         if response.get("id"):
             action_id = response["id"]
         else:
-            raise InvalidOperationError("Failed to push edges list certificates")
+            raise CatalystwanException("Failed to push edges list certificates")
 
         return True if wait_for_state() else False
 

catalystwan/api/config_device_inventory_api.py

@@ -29,19 +29,19 @@ def unlock(self, device_uuid: str, device_type: str, device_details: list) -> Ta
         devices = []
         for device_detail in device_details:
             unlock_device_detail = UnlockDeviceDetail(
-                deviceId=device_detail["deviceId"], deviceIP=device_detail["deviceIP"]
+                device_id=device_detail["deviceId"], device_ip=device_detail["deviceIP"]
             )
             devices.append(unlock_device_detail)
 
-        payload = DeviceUnlockPayload(deviceType=device_type, devices=devices)
+        payload = DeviceUnlockPayload(device_type=device_type, devices=devices)
 
         task_id = self.endpoint.unlock(device_uuid=device_uuid, payload=payload).parentTaskId
         return Task(self.session, task_id=task_id)
 
     def generate_bootstrap_cfg(
         self,
         device_uuid: UUID,
-        configtype: ConfigType = ConfigType.CLOUDINIT,
+        configtype: ConfigType = "cloudinit",
         incl_def_root_cert: bool = False,
         version: str = "v1",
     ) -> BoostrapConfigurationDetails: