Upgrades with Remote Server

catalystwan/api/partition_manager_api.py

@@ -7,12 +7,12 @@
 
 from catalystwan.api.task_status_api import Task
 from catalystwan.api.versions_utils import DeviceVersions, RepositoryAPI
-from catalystwan.dataclasses import Device
 from catalystwan.endpoints.configuration_device_actions import (
     PartitionActionPayload,
     RemovePartitionActionPayload,
     RemovePartitionDevice,
 )
+from catalystwan.endpoints.configuration_device_inventory import DeviceDetailsResponse
 from catalystwan.exceptions import EmptyVersionPayloadError
 from catalystwan.typed_list import DataSequence
 from catalystwan.utils.upgrades_helper import get_install_specification, validate_personality_homogeneity
@@ -50,7 +50,9 @@ def __init__(self, session: ManagerSession) -> None:
         self.repository = RepositoryAPI(self.session)
         self.device_version = DeviceVersions(self.session)
 
-    def set_default_partition(self, devices: DataSequence[Device], partition: Optional[str] = None) -> Task:
+    def set_default_partition(
+        self, devices: DataSequence[DeviceDetailsResponse], partition: Optional[str] = None
+    ) -> Task:
         """
         Set default software versions for devices
 
@@ -84,7 +86,7 @@ def set_default_partition(self, devices: DataSequence[Device], partition: Option
         return Task(self.session, partition_action.id)
 
     def remove_partition(
-        self, devices: DataSequence[Device], partition: Optional[str] = None, force: bool = False
+        self, devices: DataSequence[DeviceDetailsResponse], partition: Optional[str] = None, force: bool = False
     ) -> Task:
         """
         Remove chosen software version from device

catalystwan/api/software_action_api.py

@@ -7,14 +7,15 @@
 
 from catalystwan.api.task_status_api import Task
 from catalystwan.api.versions_utils import DeviceVersions, RepositoryAPI
-from catalystwan.dataclasses import Device
 from catalystwan.endpoints.configuration_device_actions import (
     InstallActionPayload,
+    InstallData,
     InstallDevice,
     InstallInput,
     PartitionActionPayload,
 )
-from catalystwan.exceptions import EmptyVersionPayloadError, VersionDeclarationError  # type: ignore
+from catalystwan.endpoints.configuration_device_inventory import DeviceDetailsResponse
+from catalystwan.exceptions import EmptyVersionPayloadError, ImageNotInRepositoryError  # type: ignore
 from catalystwan.typed_list import DataSequence
 from catalystwan.utils.personality import Personality
 from catalystwan.utils.upgrades_helper import get_install_specification, validate_personality_homogeneity
@@ -37,13 +38,12 @@ class SoftwareActionAPI:
     session = create_manager_session(...)
 
     # Prepare devices list
-    devices = session.api.devices.get()
-    vsmarts = devices.filter(personality=Personality.VSMART)
+    controllers = session.endpoints.configuration_device_inventory.get_device_details('controllers')
+    vsmarts = controllers.filter(personality=Personality.VSMART)
     software_image = "viptela-20.7.2-x86_64.tar.gz"
 
     # Upgrade
-    upgrade_id = SoftwareActionAPI(session).install(devices = vmanages,
-    software_image=software_image)
+    upgrade_id = SoftwareActionAPI(session).install(devices = vsmarts, software_image=software_image)
 
     # Check upgrade status
     TaskAPI(session, software_action_id).wait_for_completed()
@@ -56,20 +56,23 @@ def __init__(self, session: ManagerSession) -> None:
 
     def activate(
         self,
-        devices: DataSequence[Device],
+        devices: DataSequence[DeviceDetailsResponse],
         version_to_activate: Optional[str] = "",
         image: Optional[str] = "",
     ) -> Task:
         """
-        Set chosen version as current version
+        Set chosen version as current version. Requires that selected devices have already version_to_activate
+        or image present in their available files.
 
         Args:
-            devices (List[Device]): For those devices software will be activated
+            devices (List[DeviceDetailsResponse]): For those devices software will be activated
             version_to_activate (Optional[str]): version to be set as current version
-            image (Optional[str]): path to software image or its name from available files
+            image (Optional[str]): software image name in available files
 
-            Notice: Have to pass one of those arguments (version_to_activate,
-            image)
+            Notice: Have to pass one of those arguments (version_to_activate, image)
+
+        Raises:
+            EmptyVersionPayloadError: If selected version_to_activate or image not detected in available files
 
         Returns:
             str: Activate software action id
@@ -80,7 +83,12 @@ def activate(
         elif version_to_activate and not image:
             version = cast(str, version_to_activate)
         else:
-            raise VersionDeclarationError("You can not provide software_image and image version at the same time!")
+            raise ValueError("You can not provide software_image and image version at the same time!")
+
+        if not version:
+            raise ImageNotInRepositoryError(
+                "Based on provided arguments, software version to activate on device(s) cannot be detected."
+            )
 
         payload_devices = self.device_versions.get_device_available(version, devices)
         for device in payload_devices:
@@ -100,63 +108,121 @@ def activate(
 
     def install(
         self,
-        devices: DataSequence[Device],
+        devices: DataSequence[DeviceDetailsResponse],
         reboot: bool = False,
         sync: bool = True,
-        image: str = "",
-        image_version: str = "",
+        v_edge_vpn: int = 0,
+        v_smart_vpn: int = 0,
+        image: Optional[str] = None,
+        image_version: Optional[str] = None,
         downgrade_check: bool = True,
+        remote_server_name: Optional[str] = None,
+        remote_image_filename: Optional[str] = None,
     ) -> Task:
         """
         Method to install new software
 
         Args:
-            devices (List[Device]): For those devices software will be activated
-            reboot (bool): reboot device after action end
+            devices (List[DeviceDetailsResponse]): For those devices software will be activated
+            reboot (bool, optional): reboot device after action end
             sync (bool, optional): Synchronize settings. Defaults to True.
+            v_edge_vpn (int, optional): vEdge VPN
+            v_smart_vpn (int, optional): vSmart VPN
             image (str): path to software image or its name from available files
             image_version (str): version of software image
-            downgrade_check (bool): perform a downgrade check when applicable
+            downgrade_check (bool, optional): perform a downgrade check when applicable
+            remote_server_name (str): name of configured Remote Server
+            remote_image_filename (str): filename to choose from selected Remote Server
 
-            Notice: Have to pass one of those arguments (image_version,
-            image)
+            Notice: Have to pass one of those:
+                - image_version
+                - image
+                - remote_server_name and remote_image_filename
 
         Raises:
-            ValueError: Raise error if downgrade in certain cases
+            ValueError: Raise error if downgrade in certain cases or wrong arguments combination provided
+            ImageNotInRepositoryError: If selected image, image_version or remote_image_filename not found
 
         Returns:
             Task: Task object representing started install process
         """
         validate_personality_homogeneity(devices)
-        if image and not image_version:
+
+        if (
+            sum(
+                [
+                    image is not None,
+                    image_version is not None,
+                    all([remote_server_name is not None, remote_image_filename is not None]),
+                ]
+            )
+            != 1
+        ):
+            raise ValueError(
+                "Please provide one option to detect software to install. "
+                "Pick either 'image', 'image_version', or both 'remote_server_name' and 'remote_image_filename'."
+            )
+
+        # FIXME downgrade_check will be supported when software images from Remote Server will have versions fields
+        if remote_server_name and remote_image_filename and downgrade_check:
+            raise ValueError("Downgrade check is not supported for install action for images from Remote Server.")
+
+        version, remote_image_details = None, None
+        if image:
             version = cast(str, self.repository.get_image_version(image))
-        elif image_version and not image:
+        if image_version:
             version = cast(str, image_version)
-        else:
-            raise VersionDeclarationError("You can not provide image and image version at the same time")
+        if remote_server_name and remote_image_filename:
+            remote_image_details = self.repository.get_remote_image(remote_image_filename, remote_server_name)
 
-        install_specification = get_install_specification(devices.first())
+        if not any([version, remote_image_details]):
+            raise ImageNotInRepositoryError(
+                "Based on provided arguments, software version to install on device(s) cannot be detected."
+            )
+
+        install_specification = get_install_specification(devices.first(), remote=bool(remote_image_details))
         install_devices = [
             InstallDevice(**device.model_dump(by_alias=True))
             for device in self.device_versions.get_device_list(devices)
         ]
-        input = InstallInput(
-            v_edge_vpn=0,
-            v_smart_vpn=0,
-            family=install_specification.family.value,
-            version=version,
-            version_type=install_specification.version_type.value,
-            reboot=reboot,
-            sync=sync,
-        )
+
+        if version:
+            input = InstallInput(
+                v_edge_vpn=v_edge_vpn,
+                v_smart_vpn=v_smart_vpn,
+                family=install_specification.family.value,
+                version=version,
+                version_type=install_specification.version_type.value,
+                reboot=reboot,
+                sync=sync,
+            )
+        else:
+            input = InstallInput(
+                v_edge_vpn=v_edge_vpn,
+                v_smart_vpn=v_smart_vpn,
+                data=[
+                    InstallData(
+                        family=install_specification.family.value,
+                        version=remote_image_details.version_id,  # type: ignore
+                        remote_server_id=remote_image_details.remote_server_id,  # type: ignore
+                        version_id=remote_image_details.version_id,  # type: ignore
+                    )
+                ],
+                version_type=install_specification.version_type.value,
+                reboot=reboot,
+                sync=sync,
+            )
+
         device_type = install_specification.device_type.value
         install_payload = InstallActionPayload(
             action="install", input=input, devices=install_devices, device_type=device_type
         )
 
         if downgrade_check and devices.first().personality in (Personality.VMANAGE, Personality.EDGE):
             self._downgrade_check(
-                install_payload.devices, install_payload.input.version, install_specification.family.value
+                install_payload.devices,
+                install_payload.input.version,  # type: ignore
+                install_specification.family.value,  # type: ignore
             )
 
         install_action = self.session.endpoints.configuration_device_actions.process_install_operation(
@@ -172,10 +238,10 @@ def _downgrade_check(self, payload_devices: List[InstallDevice], version_to_upgr
 
         Args:
             version_to_upgrade (str): version to upgrade
-            devices_category (DeviceCategory): devices category
+            payload_devices List[InstallDevice]: list of Devices to check downgrade possibility
 
-        Returns:
-            List[str]: list of devices with no permission to downgrade
+        Raises:
+            ValueError: If for any of the devices upgrade action is denied
         """
         incorrect_devices = []
         devices_versions_repo = self.repository.get_devices_versions_repository()