diff --git a/catalystwan/api/partition_manager_api.py b/catalystwan/api/partition_manager_api.py index 3531ac64..6299d8b5 100644 --- a/catalystwan/api/partition_manager_api.py +++ b/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 diff --git a/catalystwan/api/software_action_api.py b/catalystwan/api/software_action_api.py index 7d07905d..b7d1502c 100644 --- a/catalystwan/api/software_action_api.py +++ b/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,55 +108,111 @@ 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 @@ -156,7 +220,9 @@ def install( 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() diff --git a/catalystwan/api/versions_utils.py b/catalystwan/api/versions_utils.py index 9bda7725..0c5377a3 100644 --- a/catalystwan/api/versions_utils.py +++ b/catalystwan/api/versions_utils.py @@ -8,9 +8,9 @@ from pydantic import BaseModel, ConfigDict, Field -from catalystwan.dataclasses import Device from catalystwan.endpoints.configuration.software_actions import SoftwareImageDetails from catalystwan.endpoints.configuration_device_actions import PartitionDevice +from catalystwan.endpoints.configuration_device_inventory import DeviceDetailsResponse from catalystwan.exceptions import ImageNotInRepositoryError from catalystwan.typed_list import DataSequence from catalystwan.utils.upgrades_helper import SoftwarePackageUploadPayload @@ -96,7 +96,9 @@ def get_devices_versions_repository(self) -> Dict[str, DeviceSoftwareRepository] def get_image_version(self, software_image: str) -> Union[str, None]: """ - Get proper software image version, based on name in available files + Get proper software image version, based on name in available files. + + If software_image detected in available files, but doesn't include version_name, software_image won't be used. Args: software_image (str): path to software image @@ -109,11 +111,50 @@ def get_image_version(self, software_image: str) -> Union[str, None]: software_images = self.get_all_software_images() for image in software_images: if image.available_files and image_name in image.available_files: - image_version = image.version_name - return image_version + if image.version_name and not image.version_name == "--": + return image.version_name + logger.warning( + f"Detected image {image_name} in available files has version_name: {image.version_name} as value." + "Image will not be used and image version won't be returned." + ) logger.error(f"Software image {image_name} is not in available images") return None + def get_remote_image( + self, remote_image_filename: str, remote_server_name: str + ) -> Union[SoftwareImageDetails, None]: + """ + Get remote software image details, based on name in available files and remote server name. + + Args: + remote_image_filename (str): path to software image on remote server + remote_server_name (str): remote server name + + Returns: + Union[SoftwareImageDetails, None]: remote image image details + """ + + image_name = PurePath(remote_image_filename).name + software_images = self.get_all_software_images() + for image_details in software_images: + if ( + image_details.available_files + and image_details.version_type + and image_name in image_details.available_files + and remote_server_name in image_details.version_type + ): + if not (image_details.remote_server_id and image_details.version_id): + raise ValueError( + f"Requested image: '{image_name}' does not include include required fields for this operation:" + f"image_details.remote_server_id - (current value: {image_details.remote_server_id})" + f"image_details.version_id - (current value: {image_details.version_id})" + ) + return image_details + logger.error( + f"Software image {image_name} is not in available in images from remote server {remote_server_name}" + ) + return None + def upload_image(self, image_path: str) -> None: """ Upload software image ('tar.gz' or 'SPA.bin') to vManage software repository @@ -159,8 +200,17 @@ class DeviceVersions: def __init__(self, session: ManagerSession): self.repository = RepositoryAPI(session) + def _validate_devices_required_fields(self, devices: DataSequence[DeviceDetailsResponse]): + for device in devices: + if not device.uuid or not device.device_ip: + raise ValueError( + f"Provided device '{device.host_name}' doesn't include required fields for this operation:" + f"device.uuid (current value: {device.uuid})" + f"device.device_ip (current value: {device.device_ip})" + ) + def _get_device_list_in( - self, version_to_set_up: str, devices: DataSequence[Device], version_type: str + self, version_to_set_up: str, devices: DataSequence[DeviceDetailsResponse], version_type: str ) -> DataSequence[PartitionDevice]: """ Create devices payload list included requested version, if requested version @@ -168,15 +218,16 @@ def _get_device_list_in( Args: version_to_set_up (str): requested version - devices List[Device]: list of Device dataclass instances + devices List[DeviceDetailsResponse]: list of Device dataclass instances version_type: type of version (installed, available, etc.) Returns: list : list of devices """ + self._validate_devices_required_fields(devices) devices_payload = DataSequence( PartitionDevice, - [PartitionDevice(device_id=device.uuid, device_ip=device.id) for device in devices], + [PartitionDevice(device_id=device.uuid, device_ip=device.device_ip) for device in devices], # type: ignore ) all_dev_versions = self.repository.get_devices_versions_repository() for device in devices_payload: @@ -194,7 +245,7 @@ def _get_device_list_in( return devices_payload def get_device_list_in_installed( - self, version_to_set_up: str, devices: DataSequence[Device] + self, version_to_set_up: str, devices: DataSequence[DeviceDetailsResponse] ) -> DataSequence[PartitionDevice]: """ Create devices payload list included requested version, if requested version @@ -202,7 +253,7 @@ def get_device_list_in_installed( Args: version_to_set_up (str): requested version - devices (List[Device]): devices on which action going to be performed + devices (List[DeviceDetailsResponse]): devices on which action going to be performed Returns: list : list of devices @@ -210,7 +261,7 @@ def get_device_list_in_installed( return self._get_device_list_in(version_to_set_up, devices, "installed_versions") def get_device_available( - self, version_to_set_up: str, devices: DataSequence[Device] + self, version_to_set_up: str, devices: DataSequence[DeviceDetailsResponse] ) -> DataSequence[PartitionDevice]: """ Create devices payload list included requested, if requested version @@ -218,7 +269,7 @@ def get_device_available( Args: version_to_set_up (str): requested version - devices (List[Device]): devices on which action going to be performed + devices (List[DeviceDetailsResponse]): devices on which action going to be performed Returns: @@ -227,7 +278,7 @@ def get_device_available( return self._get_device_list_in(version_to_set_up, devices, "available_versions") def _get_devices_chosen_version( - self, devices: DataSequence[Device], version_type: str + self, devices: DataSequence[DeviceDetailsResponse], version_type: str ) -> DataSequence[PartitionDevice]: """ Create devices payload list included software version key @@ -235,28 +286,32 @@ def _get_devices_chosen_version( Args: version_to_set_up (str): requested version - devices (List[Device]): devices on which action going to be performed + devices (List[DeviceDetailsResponse]): devices on which action going to be performed Returns: list : list of devices """ + self._validate_devices_required_fields(devices) + devices_payload = DataSequence( PartitionDevice, - [PartitionDevice(device_id=device.uuid, device_ip=device.id) for device in devices], + [PartitionDevice(device_id=device.uuid, device_ip=device.device_ip) for device in devices], # type: ignore ) all_dev_versions = self.repository.get_devices_versions_repository() for device in devices_payload: device.version = getattr(all_dev_versions[device.device_id], version_type) return devices_payload - def get_devices_current_version(self, devices: DataSequence[Device]) -> DataSequence[PartitionDevice]: + def get_devices_current_version( + self, devices: DataSequence[DeviceDetailsResponse] + ) -> DataSequence[PartitionDevice]: """ Create devices payload list included current software version key for every device in devices list Args: version_to_set_up (str): requested version - devices (List[Device]): devices on which action going to be performed + devices (List[DeviceDetailsResponse]): devices on which action going to be performed Returns: list : list of devices @@ -264,13 +319,15 @@ def get_devices_current_version(self, devices: DataSequence[Device]) -> DataSequ return self._get_devices_chosen_version(devices, "current_version") - def get_devices_available_versions(self, devices: DataSequence[Device]) -> DataSequence[PartitionDevice]: + def get_devices_available_versions( + self, devices: DataSequence[DeviceDetailsResponse] + ) -> DataSequence[PartitionDevice]: """ Create devices payload list included available software versions key for every device in devices list Args: - devices (List[Device]): devices on which action going to be performed + devices (List[DeviceDetailsResponse]): devices on which action going to be performed Returns: list : list of devices @@ -278,5 +335,9 @@ def get_devices_available_versions(self, devices: DataSequence[Device]) -> DataS return self._get_devices_chosen_version(devices, "available_versions") - def get_device_list(self, devices: DataSequence[Device]) -> List[PartitionDevice]: - return [PartitionDevice(device_id=device.uuid, device_ip=device.id) for device in devices] # type: ignore + def get_device_list(self, devices: DataSequence[DeviceDetailsResponse]) -> List[PartitionDevice]: + self._validate_devices_required_fields(devices) + + return [ + PartitionDevice(device_id=device.uuid, device_ip=device.device_ip) for device in devices # type: ignore + ] diff --git a/catalystwan/endpoints/configuration/software_actions.py b/catalystwan/endpoints/configuration/software_actions.py index 09992ecf..eab9f68f 100644 --- a/catalystwan/endpoints/configuration/software_actions.py +++ b/catalystwan/endpoints/configuration/software_actions.py @@ -158,6 +158,7 @@ class SoftwareImageDetails(BaseModel): vnf_properties_json: Optional[str] = Field( default=None, serialization_alias="vnfPropertiesJson", validation_alias="vnfPropertiesJson" ) + remote_server_id: str = Field(default=None, serialization_alias="remoteServerId", validation_alias="remoteServerId") class ConfigurationSoftwareActions(APIEndpoints): diff --git a/catalystwan/endpoints/configuration_device_actions.py b/catalystwan/endpoints/configuration_device_actions.py index b1deffc5..70017111 100644 --- a/catalystwan/endpoints/configuration_device_actions.py +++ b/catalystwan/endpoints/configuration_device_actions.py @@ -17,6 +17,8 @@ def convert_to_list(element: Union[str, List[str]]) -> List[str]: DeviceType = Literal["vedge", "controller", "vmanage"] +VersionType = Literal["vmanage", "remote"] + PartitionActionType = Literal["removepartition", "defaultpartition", "changepartition"] @@ -78,19 +80,20 @@ class InstallData(BaseModel): family: str version: str version_id: str = Field(serialization_alias="versionId", validation_alias="versionId") + remote_server_id: str = Field(serialization_alias="remoteServerId", validation_alias="remoteServerId") class InstallInput(BaseModel): model_config = ConfigDict(populate_by_name=True) data: Optional[List[InstallData]] = Field(default=None) - family: Optional[str] + family: Optional[str] = Field(default=None) reboot: bool sync: bool v_edge_vpn: int = Field(serialization_alias="vEdgeVPN", validation_alias="vEdgeVPN") v_smart_vpn: int = Field(serialization_alias="vSmartVPN", validation_alias="vSmartVPN") - version: str = Field(default=None) - version_type: str = Field(serialization_alias="versionType", validation_alias="versionType") + version: Optional[str] = Field(default=None) + version_type: VersionType = Field(serialization_alias="versionType", validation_alias="versionType") class InstallDevice(BaseModel): @@ -99,7 +102,7 @@ class InstallDevice(BaseModel): device_id: str = Field(serialization_alias="deviceId", validation_alias="deviceId") device_ip: str = Field(serialization_alias="deviceIP", validation_alias="deviceIP") is_nutella_migration: Optional[bool] = Field( - default=None, serialization_alias="isNutellaMigration", validation_alias="isNutellaMigration" + default=False, serialization_alias="isNutellaMigration", validation_alias="isNutellaMigration" ) diff --git a/catalystwan/exceptions.py b/catalystwan/exceptions.py index e4641c34..48ab79d7 100644 --- a/catalystwan/exceptions.py +++ b/catalystwan/exceptions.py @@ -67,6 +67,8 @@ class ImageNotInRepositoryError(CatalystwanException): class EmptyVersionPayloadError(CatalystwanException): """Used when a version is not found in device available or current versions.""" + pass + class TemplateNotFoundError(CatalystwanException): """Used when a template item is not found.""" diff --git a/catalystwan/tests/test_partition_manager_api.py b/catalystwan/tests/test_partition_manager_api.py index 586c0c96..5203537e 100644 --- a/catalystwan/tests/test_partition_manager_api.py +++ b/catalystwan/tests/test_partition_manager_api.py @@ -5,20 +5,20 @@ from catalystwan.api.partition_manager_api import PartitionManagerAPI from catalystwan.api.versions_utils import DeviceSoftwareRepository, DeviceVersions, RepositoryAPI -from catalystwan.dataclasses import Device from catalystwan.endpoints.configuration_device_actions import ActionId, RemovePartitionDevice +from catalystwan.endpoints.configuration_device_inventory import DeviceDetailsResponse from catalystwan.typed_list import DataSequence class TestPartitionManagerAPI(unittest.TestCase): def setUp(self): self.device = DataSequence( - Device, + DeviceDetailsResponse, [ - Device( + DeviceDetailsResponse( personality="vedge", uuid="mock_uuid", - id="mock_ip", + device_ip="mock_ip", hostname="mock_host", reachability="reachable", local_system_ip="mock_ip", diff --git a/catalystwan/tests/test_software_action_api.py b/catalystwan/tests/test_software_action_api.py index 55cad85e..1563861e 100644 --- a/catalystwan/tests/test_software_action_api.py +++ b/catalystwan/tests/test_software_action_api.py @@ -5,18 +5,20 @@ from catalystwan.api.software_action_api import SoftwareActionAPI from catalystwan.api.versions_utils import DeviceSoftwareRepository, DeviceVersions, RepositoryAPI -from catalystwan.dataclasses import Device -from catalystwan.endpoints.configuration_device_actions import ActionId, InstallDevice +from catalystwan.endpoints.configuration.software_actions import SoftwareImageDetails +from catalystwan.endpoints.configuration_device_actions import ActionId, InstallDevice, PartitionDevice +from catalystwan.endpoints.configuration_device_inventory import DeviceDetailsResponse +from catalystwan.exceptions import ImageNotInRepositoryError from catalystwan.typed_list import DataSequence from catalystwan.utils.upgrades_helper import Family, InstallSpecHelper class TestSoftwareAcionAPI(unittest.TestCase): def setUp(self): - self.device = Device( + self.device = DeviceDetailsResponse( personality="vedge", uuid="mock_uuid", - id="mock_ip", + device_ip="mock_ip", hostname="mock_host", reachability="reachable", local_system_ip="mock_ip", @@ -43,12 +45,9 @@ def setUp(self): self.mock_device_versions = DeviceVersions(self.mock_repository_object) self.mock_software_action_obj = SoftwareActionAPI(mock_session) - @patch("catalystwan.session.ManagerSession") @patch.object(SoftwareActionAPI, "_downgrade_check") @patch.object(RepositoryAPI, "get_image_version") - def test_upgrade_software_if_downgrade_check_is_none( - self, mock_get_image_version, mock_downgrade_check, mock_session - ): + def test_upgrade_software_if_downgrade_check_is_none(self, mock_get_image_version, mock_downgrade_check): # Prepare mock data mock_downgrade_check.return_value = False expected_id = ActionId(id="mock_action_id") @@ -59,13 +58,41 @@ def test_upgrade_software_if_downgrade_check_is_none( # Assert answer = self.mock_software_action_obj.install( - devices=DataSequence(Device, [self.device]), + devices=DataSequence(DeviceDetailsResponse, [self.device]), reboot=True, sync=True, image="path", ) self.assertEqual(answer.task_id, "mock_action_id") + @patch.object(DeviceVersions, "get_device_available") + @patch.object(RepositoryAPI, "get_all_software_images") + @patch.object(RepositoryAPI, "get_devices_versions_repository") + def test_activate_software( + self, mock_get_devices_versions_repository, mock_get_all_software_images, mock_get_device_available + ): + # Prepare mock data + expected_id = ActionId(id="mock_action_id") + mock_get_devices_versions_repository.return_value = self.DeviceSoftwareRepository_obj + mock_get_device_available.return_value = DataSequence( + PartitionDevice, [PartitionDevice(device_id="mock_uuid", device_ip="mock_ip", version="ver2")] + ) + mock_get_all_software_images.return_value = DataSequence( + SoftwareImageDetails, + [SoftwareImageDetails(**{"availableFiles": "vmanage-20.9.1-x86_64.tar.gz", "versionName": "ver2"})], + ) + + self.mock_software_action_obj.session.endpoints.configuration_device_actions.process_mark_change_partition = ( + MagicMock(return_value=expected_id) + ) + + # Assert + answer = self.mock_software_action_obj.activate( + devices=DataSequence(DeviceDetailsResponse, [self.device]), + image="vmanage-20.9.1-x86_64.tar.gz", + ) + self.assertEqual(answer.task_id, "mock_action_id") + @patch.object(RepositoryAPI, "get_devices_versions_repository") def test_downgrade_check_no_incorrect_devices(self, mock_get_devices_versions_repository): # Preapre mock data @@ -93,3 +120,35 @@ def test_downgrade_check_incorrect_devices_exists(self, mock_get_devices_version upgrade_version, Family.VMANAGE.value, ) + + def test_install_software_from_remote_image_not_available_with_downgrade_check(self): + with self.assertRaises(ValueError): + self.mock_software_action_obj.install( + devices=DataSequence(DeviceDetailsResponse, [self.device]), + remote_server_name="dummy", + remote_image_filename="dummy", + ) + + @patch.object(RepositoryAPI, "get_all_software_images") + def test_install_software_from_remote_image_with_wrong_version(self, mock_get_all_software_images): + mock_get_all_software_images.return_value = DataSequence( + SoftwareImageDetails, + [ + SoftwareImageDetails( + **{ + "availableFiles": "vmanage-20.9.1-x86_64.tar.gz", + "versionType": "remote-server-test", + "remoteServerId": "123456789-abcdabcd", + "versionId": "ver1", + } + ) + ], + ) + + with self.assertRaises(ImageNotInRepositoryError): + self.mock_software_action_obj.install( + devices=DataSequence(DeviceDetailsResponse, [self.device]), + remote_server_name="remote-server-test", + remote_image_filename="not-ver1", + downgrade_check=False, + ) diff --git a/catalystwan/tests/test_version_utils.py b/catalystwan/tests/test_version_utils.py index 7534114d..842be872 100644 --- a/catalystwan/tests/test_version_utils.py +++ b/catalystwan/tests/test_version_utils.py @@ -4,18 +4,18 @@ from unittest.mock import MagicMock, Mock, patch from catalystwan.api.versions_utils import DeviceSoftwareRepository, DeviceVersions, RepositoryAPI -from catalystwan.dataclasses import Device from catalystwan.endpoints.configuration.software_actions import SoftwareImageDetails from catalystwan.endpoints.configuration_device_actions import InstalledDeviceData, PartitionDevice +from catalystwan.endpoints.configuration_device_inventory import DeviceDetailsResponse from catalystwan.typed_list import DataSequence class TestRepositoryAPI(unittest.TestCase): def setUp(self): - self.device = Device( + self.device = DeviceDetailsResponse( personality="vedge", uuid="mock_uuid", - id="mock_ip", + device_ip="mock_ip", hostname="mock_host", reachability="reachable", local_system_ip="mock_ip", @@ -57,6 +57,104 @@ def test_get_image_version_if_image_unavailable(self, mock_session): self.assertEqual(answer, image_version, "not same version") + def test_get_remote_image(self): + versions_response = DataSequence( + SoftwareImageDetails, + [ + SoftwareImageDetails( + **{ + "availableFiles": "vmanage-20.9.1-x86_64.tar.gz", + "versionType": "remote-server-test", + "remoteServerId": "123456789-abcdabcd", + "versionId": "abcd-1234", + } + ) + ], + ) + self.mock_repository_object.get_all_software_images = MagicMock(return_value=versions_response) + answer = self.mock_repository_object.get_remote_image("vmanage-20.9.1-x86_64.tar.gz", "remote-server-test") + + self.assertEqual(answer.version_id, "abcd-1234", "not same version") + + def test_get_remote_image_non_existing(self): + versions_response = DataSequence( + SoftwareImageDetails, + [ + SoftwareImageDetails( + **{ + "availableFiles": "vmanage-20.9.1-x86_64.tar.gz", + "versionType": "remote-server-test", + "remoteServerId": "123456789-abcdabcd", + "versionId": "abcd-1234", + } + ) + ], + ) + self.mock_repository_object.get_all_software_images = MagicMock(return_value=versions_response) + answer = self.mock_repository_object.get_remote_image("vmanage-20.10.1-x86_64.tar.gz", "remote-server-test") + + self.assertEqual(answer, None, "not same version") + + def test_get_remote_image_no_version_id(self): + versions_response = DataSequence( + SoftwareImageDetails, + [ + SoftwareImageDetails( + **{ + "availableFiles": "vmanage-20.9.1-x86_64.tar.gz", + "versionType": "remote-server-test", + "remoteServerId": "123456789-abcdabcd", + "versionId": None, + } + ) + ], + ) + self.mock_repository_object.get_all_software_images = MagicMock(return_value=versions_response) + + with self.assertRaises(ValueError): + self.mock_repository_object.get_remote_image("vmanage-20.9.1-x86_64.tar.gz", "remote-server-test") + + def test_get_version_when_same_available_file_present_for_remote_version(self): + versions_response = DataSequence( + SoftwareImageDetails, + [ + SoftwareImageDetails( + **{ + "availableFiles": "vmanage-20.9.1-x86_64.tar.gz", + "versionType": "remote-server-test", + "remoteServerId": "123456789-abcdabcd", + "versionId": "abcd-1234", + } + ), + SoftwareImageDetails(**{"availableFiles": "vmanage-20.9.1-x86_64.tar.gz", "versionName": "20.9.1"}), + ], + ) + self.mock_repository_object.get_all_software_images = MagicMock(return_value=versions_response) + image_version = "20.9.1" + answer = self.mock_repository_object.get_image_version("vmanage-20.9.1-x86_64.tar.gz") + + self.assertEqual(answer, image_version, "not same version") + + def test_get_version_when_remote_available_file_is_matching(self): + versions_response = DataSequence( + SoftwareImageDetails, + [ + SoftwareImageDetails( + **{ + "availableFiles": "vmanage-20.9.1-x86_64.tar.gz", + "versionType": "remote-server-test", + "remoteServerId": "123456789-abcdabcd", + "versionId": "abcd-1234", + } + ), + SoftwareImageDetails(**{"availableFiles": "vmanage-20.10.1-x86_64.tar.gz", "versionName": "20.9.1"}), + ], + ) + self.mock_repository_object.get_all_software_images = MagicMock(return_value=versions_response) + answer = self.mock_repository_object.get_image_version("vmanage-20.9.1-x86_64.tar.gz") + + self.assertEqual(answer, None) + @patch("catalystwan.session.Session") def test_get_devices_versions_repository(self, mock_session): endpoint_mock_response = DataSequence( diff --git a/catalystwan/utils/upgrades_helper.py b/catalystwan/utils/upgrades_helper.py index b1c4c8ee..d8f9a9bd 100644 --- a/catalystwan/utils/upgrades_helper.py +++ b/catalystwan/utils/upgrades_helper.py @@ -7,8 +7,8 @@ from clint.textui.progress import Bar as ProgressBar # type: ignore from requests_toolbelt.multipart.encoder import MultipartEncoder, MultipartEncoderMonitor # type: ignore -from catalystwan.dataclasses import Device from catalystwan.endpoints import CustomPayloadType, PreparedPayload +from catalystwan.endpoints.configuration_device_inventory import DeviceDetailsResponse from catalystwan.exceptions import MultiplePersonalityError from catalystwan.typed_list import DataSequence from catalystwan.utils.personality import Personality @@ -17,10 +17,13 @@ class Family(Enum): VEDGE = "vedge" VMANAGE = "vmanage" + VEDGE_X86 = "vedge-x86" + C8000V = "c8000v" class VersionType(Enum): VMANAGE = "vmanage" + REMOTE = "remote" class DeviceType(Enum): @@ -47,19 +50,23 @@ class InstallSpecHelper(Enum): VSMART = InstallSpecification(Family.VEDGE, VersionType.VMANAGE, DeviceType.CONTROLLER) # type: ignore VBOND = InstallSpecification(Family.VEDGE, VersionType.VMANAGE, DeviceType.CONTROLLER) # type: ignore VEDGE = InstallSpecification(Family.VEDGE, VersionType.VMANAGE, DeviceType.VEDGE) # type: ignore + REMOTE_VMANAGE = InstallSpecification(Family.VMANAGE, VersionType.REMOTE, DeviceType.VMANAGE) # type: ignore + REMOTE_VSMART = InstallSpecification(Family.VEDGE_X86, VersionType.REMOTE, DeviceType.CONTROLLER) # type: ignore + REMOTE_VBOND = InstallSpecification(Family.VEDGE_X86, VersionType.REMOTE, DeviceType.CONTROLLER) # type: ignore + REMOTE_VEDGE = InstallSpecification(Family.C8000V, VersionType.REMOTE, DeviceType.VEDGE) # type: ignore -def get_install_specification(device: Device): +def get_install_specification(device: DeviceDetailsResponse, remote: bool = False): specification_container = { - Personality.VMANAGE: InstallSpecHelper.VMANAGE.value, - Personality.VBOND: InstallSpecHelper.VBOND.value, - Personality.VSMART: InstallSpecHelper.VSMART.value, - Personality.EDGE: InstallSpecHelper.VEDGE.value, + Personality.VMANAGE: InstallSpecHelper.REMOTE_VMANAGE.value if remote else InstallSpecHelper.VMANAGE.value, + Personality.VBOND: InstallSpecHelper.REMOTE_VBOND.value if remote else InstallSpecHelper.VBOND.value, + Personality.VSMART: InstallSpecHelper.REMOTE_VSMART.value if remote else InstallSpecHelper.VSMART.value, + Personality.EDGE: InstallSpecHelper.REMOTE_VEDGE.value if remote else InstallSpecHelper.VEDGE.value, } - return specification_container[device.personality] + return specification_container[device.personality] # type: ignore -def validate_personality_homogeneity(devices: DataSequence[Device]): +def validate_personality_homogeneity(devices: DataSequence[DeviceDetailsResponse]): personalities = set([device.personality for device in devices]) if not len(personalities) == 1: raise MultiplePersonalityError(