diff --git a/catalystwan/api/partition_manager_api.py b/catalystwan/api/partition_manager_api.py index 63b855630..229b313c2 100644 --- a/catalystwan/api/partition_manager_api.py +++ b/catalystwan/api/partition_manager_api.py @@ -5,12 +5,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 @@ -48,7 +48,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 @@ -82,7 +84,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 398730139..c1a7559b6 100644 --- a/catalystwan/api/software_action_api.py +++ b/catalystwan/api/software_action_api.py @@ -5,14 +5,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 @@ -54,7 +55,7 @@ def __init__(self, session: ManagerSession) -> None: def activate( self, - devices: DataSequence[Device], + devices: DataSequence[DeviceDetailsResponse], version_to_activate: Optional[str] = "", image: Optional[str] = "", ) -> Task: @@ -78,7 +79,7 @@ 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!") payload_devices = self.device_versions.get_device_available(version, devices) for device in payload_devices: @@ -98,12 +99,16 @@ 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 @@ -126,27 +131,72 @@ def install( 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 @@ -154,7 +204,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( diff --git a/catalystwan/api/versions_utils.py b/catalystwan/api/versions_utils.py index 2496bd581..1c724acf5 100644 --- a/catalystwan/api/versions_utils.py +++ b/catalystwan/api/versions_utils.py @@ -6,9 +6,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 @@ -112,6 +112,41 @@ def get_image_version(self, software_image: str) -> Union[str, None]: 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 @@ -157,8 +192,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 @@ -166,15 +210,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: @@ -192,7 +237,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 @@ -200,7 +245,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 @@ -208,7 +253,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 @@ -216,7 +261,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: @@ -225,7 +270,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 @@ -233,28 +278,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 @@ -262,13 +311,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 @@ -276,5 +327,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 7a27be5b4..aec391db3 100644 --- a/catalystwan/endpoints/configuration/software_actions.py +++ b/catalystwan/endpoints/configuration/software_actions.py @@ -156,6 +156,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 ce50fa533..70232d2df 100644 --- a/catalystwan/endpoints/configuration_device_actions.py +++ b/catalystwan/endpoints/configuration_device_actions.py @@ -15,6 +15,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"] @@ -76,19 +78,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): diff --git a/catalystwan/tests/test_partition_manager_api.py b/catalystwan/tests/test_partition_manager_api.py index f23eeef07..b07d85558 100644 --- a/catalystwan/tests/test_partition_manager_api.py +++ b/catalystwan/tests/test_partition_manager_api.py @@ -3,20 +3,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 8b1155e7d..1352c17b9 100644 --- a/catalystwan/tests/test_software_action_api.py +++ b/catalystwan/tests/test_software_action_api.py @@ -3,18 +3,18 @@ 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_device_inventory import DeviceDetailsResponse 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", @@ -57,7 +57,7 @@ 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", diff --git a/catalystwan/tests/test_version_utils.py b/catalystwan/tests/test_version_utils.py index adfe06b0f..94fc93bdd 100644 --- a/catalystwan/tests/test_version_utils.py +++ b/catalystwan/tests/test_version_utils.py @@ -2,18 +2,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", diff --git a/catalystwan/utils/upgrades_helper.py b/catalystwan/utils/upgrades_helper.py index 5d07afdc8..75b749100 100644 --- a/catalystwan/utils/upgrades_helper.py +++ b/catalystwan/utils/upgrades_helper.py @@ -5,8 +5,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 @@ -15,10 +15,13 @@ class Family(Enum): VEDGE = "vedge" VMANAGE = "vmanage" + VEDGE_X86 = "vedge-x86" + C8000V = "c8000v" class VersionType(Enum): VMANAGE = "vmanage" + REMOTE = "remote" class DeviceType(Enum): @@ -45,19 +48,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(