From 8821b923695f503ae698a79a20ac33ccc42662ac Mon Sep 17 00:00:00 2001 From: Abhishek jaiswal Date: Fri, 6 Jun 2025 06:56:19 -0400 Subject: [PATCH] Adding Keywords Added keywords for - sw-manager sw-deploy-strategy - software upload Change-Id: Ic87ce7fa13762e140ce816371541a5a43d5c21fd Signed-off-by: Abhishek jaiswal --- .../dcmanager_prestage_strategy_keywords.py | 16 ++- .../swmanager_sw_deploy_strategy_object.py | 127 ++++++++++++++++++ ...wmanager_sw_deploy_strategy_show_output.py | 74 ++++++++++ .../swmanager_sw_deploy_strategy_keywords.py | 72 ++++++++++ .../upgrade/objects/software_upload_object.py | 23 ++++ .../upgrade/objects/software_upload_output.py | 54 ++++++++ .../cloud_platform/upgrade/usm_keywords.py | 7 +- 7 files changed, 371 insertions(+), 2 deletions(-) create mode 100644 keywords/cloud_platform/swmanager/objects/swmanager_sw_deploy_strategy_object.py create mode 100644 keywords/cloud_platform/swmanager/objects/swmanager_sw_deploy_strategy_show_output.py create mode 100644 keywords/cloud_platform/swmanager/swmanager_sw_deploy_strategy_keywords.py create mode 100644 keywords/cloud_platform/upgrade/objects/software_upload_object.py create mode 100644 keywords/cloud_platform/upgrade/objects/software_upload_output.py diff --git a/keywords/cloud_platform/dcmanager/dcmanager_prestage_strategy_keywords.py b/keywords/cloud_platform/dcmanager/dcmanager_prestage_strategy_keywords.py index 47ab25b7..bd36e833 100644 --- a/keywords/cloud_platform/dcmanager/dcmanager_prestage_strategy_keywords.py +++ b/keywords/cloud_platform/dcmanager/dcmanager_prestage_strategy_keywords.py @@ -9,7 +9,7 @@ class DcmanagerPrestageStrategyKeywords(BaseKeyword): This class contains all the keywords related to the 'dcmanager prestage-strategy' commands. """ - def __init__(self, ssh_connection: SSHConnection) -> None: + def __init__(self, ssh_connection: SSHConnection) -> str: """ Initializes DcmanagerPrestageStrategyKeywords. @@ -18,6 +18,20 @@ class DcmanagerPrestageStrategyKeywords(BaseKeyword): """ self.ssh_connection = ssh_connection + def get_dcmanager_prestage_strategy_abort(self) -> None: + """Gets the prestage-strategy abort.""" + command = source_openrc("dcmanager prestage-strategy abort") + output = self.ssh_connection.send(command) + self.validate_success_return_code(self.ssh_connection) + return output + + def get_dcmanager_prestage_strategy_apply(self) -> None: + """Gets the prestage-strategy apply.""" + command = source_openrc("dcmanager prestage-strategy apply") + output = self.ssh_connection.send(command) + self.validate_success_return_code(self.ssh_connection) + return output + def get_dcmanager_prestage_strategy_create(self) -> DcmanagerPrestageStrategyShowOutput: """ Gets the prestage-strategy create. diff --git a/keywords/cloud_platform/swmanager/objects/swmanager_sw_deploy_strategy_object.py b/keywords/cloud_platform/swmanager/objects/swmanager_sw_deploy_strategy_object.py new file mode 100644 index 00000000..5824d9f6 --- /dev/null +++ b/keywords/cloud_platform/swmanager/objects/swmanager_sw_deploy_strategy_object.py @@ -0,0 +1,127 @@ +from typing import Optional + + +class SwManagerSwDeployStrategyObject: + """ + This class represents a sw-manager sw-deploy-strategy as an object. + """ + + def __init__(self) -> None: + """Initializes a SwmanagerFwUpdateStrategyObject with default values.""" + self.strategy_uuid: Optional[str] = None + self.release_id: Optional[str] = None + self.controller_apply_type: Optional[str] = None + self.storage_apply_type: Optional[str] = None + self.worker_apply_type: Optional[str] = None + self.default_instance_action: Optional[str] = None + self.alarm_restrictions: Optional[str] = None + self.current_phase: Optional[str] = None + self.current_stage: Optional[str] = None + self.current_step: Optional[str] = None + self.current_phase_completion: Optional[str] = None + self.state: Optional[str] = None + self.inprogress: Optional[str] = None + + def set_strategy_uuid(self, strategy_uuid: str) -> None: + """Sets the strategy_uuid of the sw-deploy-strategy.""" + self.strategy_uuid = strategy_uuid + + def get_strategy_uuid(self) -> Optional[str]: + """Gets the strategy_uuid of the sw-deploy-strategy.""" + return self.strategy_uuid + + def set_release_id(self, release_id: str) -> None: + """Sets the release_id of the sw-deploy-strategy.""" + self.release_id = release_id + + def get_release_id(self) -> Optional[str]: + """Gets the release_id of the sw-deploy-strategy.""" + return self.release_id + + def set_controller_apply_type(self, controller_apply_type: str) -> None: + """Sets the controller_apply_type of the sw-deploy-strategy.""" + self.controller_apply_type = controller_apply_type + + def get_controller_apply_type(self) -> Optional[str]: + """Gets the controller_apply_type of the sw-deploy-strategy.""" + return self.controller_apply_type + + def set_storage_apply_type(self, storage_apply_type: str) -> None: + """Sets the storage_apply_type of the sw-deploy-strategy.""" + self.storage_apply_type = storage_apply_type + + def get_storage_apply_type(self) -> Optional[str]: + """Gets the storage_apply_type of the sw-deploy-strategy.""" + return self.storage_apply_type + + def set_worker_apply_type(self, worker_apply_type: str) -> None: + """Sets the worker_apply_type of the sw-deploy-strategy.""" + self.worker_apply_type = worker_apply_type + + def get_worker_apply_type(self) -> Optional[str]: + """Gets the worker_apply_type of the sw-deploy-strategy.""" + return self.worker_apply_type + + def set_default_instance_action(self, default_instance_action: str) -> None: + """Sets the default_instance_action of the sw-deploy-strategy.""" + self.default_instance_action = default_instance_action + + def get_default_instance_action(self) -> Optional[str]: + """Gets the default_instance_action of the sw-deploy-strategy.""" + return self.default_instance_action + + def set_alarm_restrictions(self, alarm_restrictions: str) -> None: + """Sets the alarm_restrictions of the sw-deploy-strategy.""" + self.alarm_restrictions = alarm_restrictions + + def get_alarm_restrictions(self) -> Optional[str]: + """Gets the alarm_restrictions of the sw-deploy-strategy.""" + return self.alarm_restrictions + + def set_current_phase(self, current_phase: str) -> None: + """Sets the current_phase of the sw-deploy-strategy.""" + self.current_phase = current_phase + + def get_current_phase(self) -> Optional[str]: + """Gets the current_phase of the sw-deploy-strategy.""" + return self.current_phase + + def set_current_stage(self, current_stage: str) -> None: + """Sets the current_stage of the sw-deploy-strategy.""" + self.current_stage = current_stage + + def get_current_stage(self) -> Optional[str]: + """Gets the current_stage of the sw-deploy-strategy.""" + return self.current_stage + + def set_current_step(self, current_step: str) -> None: + """Sets the current_step of the sw-deploy-strategy.""" + self.current_step = current_step + + def get_current_step(self) -> Optional[str]: + """Gets the current_step of the sw-deploy-strategy.""" + return self.current_step + + def set_current_phase_completion(self, current_phase_completion: str) -> None: + """Sets the current_phase_completion of the sw-deploy-strategy.""" + self.current_phase_completion = current_phase_completion + + def get_current_phase_completion(self) -> Optional[str]: + """Gets the current_phase_completion of the sw-deploy-strategy.""" + return self.current_phase_completion + + def set_state(self, state: str) -> None: + """Sets the state of the sw-deploy-strategy.""" + self.state = state + + def get_state(self) -> Optional[str]: + """Gets the state of the sw-deploy-strategy.""" + return self.state + + def set_inprogress(self, inprogress: str) -> None: + """Sets the inprogress of the sw-deploy-strategy.""" + self.inprogress = inprogress + + def get_inprogress(self) -> Optional[str]: + """Gets the inprogress of the sw-deploy-strategy.""" + return self.inprogress diff --git a/keywords/cloud_platform/swmanager/objects/swmanager_sw_deploy_strategy_show_output.py b/keywords/cloud_platform/swmanager/objects/swmanager_sw_deploy_strategy_show_output.py new file mode 100644 index 00000000..a7ae10cb --- /dev/null +++ b/keywords/cloud_platform/swmanager/objects/swmanager_sw_deploy_strategy_show_output.py @@ -0,0 +1,74 @@ +from typing import Dict + +from framework.exceptions.keyword_exception import KeywordException +from framework.logging.automation_logger import get_logger +from keywords.cloud_platform.swmanager.objects.swmanager_sw_deploy_strategy_object import SwManagerSwDeployStrategyObject +from keywords.cloud_platform.swmanager.swmanager_vertical_table_parser import SwManagerVerticalTableParser + + +class SwManagerSwDeployStrategyShowOutput: + """ + Parses the output of the 'sw-manager sw-deploy-strategy show' command into a SwManagerSwDeployStrategyObject instance. + """ + + def __init__(self, swmanager_sw_deploy: str) -> None: + """ + Initializes SwManagerSwDeployStrategyObject. + + Args: + swmanager_sw_deploy (str): Output of the 'sw-manager sw-deploy-strategy show' command. + + Raises: + KeywordException: If the output format is invalid. + """ + swmanager_vertical_table_parser = SwManagerVerticalTableParser(swmanager_sw_deploy) + output_values = swmanager_vertical_table_parser.get_output_values_dict() + + if self.is_valid_output(output_values): + self.swmanager_sw_deploy_strategy = SwManagerSwDeployStrategyObject() + sw_deploy_strat = output_values["Strategy Software Deploy Strategy"] + self.swmanager_sw_deploy_strategy.set_strategy_uuid(sw_deploy_strat["strategy-uuid"]) + self.swmanager_sw_deploy_strategy.set_release_id(sw_deploy_strat["release-id"]) + self.swmanager_sw_deploy_strategy.set_controller_apply_type(sw_deploy_strat["controller-apply-type"]) + self.swmanager_sw_deploy_strategy.set_storage_apply_type(sw_deploy_strat["storage-apply-type"]) + self.swmanager_sw_deploy_strategy.set_worker_apply_type(sw_deploy_strat["worker-apply-type"]) + self.swmanager_sw_deploy_strategy.set_default_instance_action(sw_deploy_strat["default-instance-action"]) + self.swmanager_sw_deploy_strategy.set_alarm_restrictions(sw_deploy_strat["alarm-restrictions"]) + self.swmanager_sw_deploy_strategy.set_current_phase(sw_deploy_strat["current-phase"]) + self.swmanager_sw_deploy_strategy.set_current_stage(sw_deploy_strat["current-stage"]) + self.swmanager_sw_deploy_strategy.set_current_step(sw_deploy_strat["current-step"]) + self.swmanager_sw_deploy_strategy.set_current_phase_completion(sw_deploy_strat["current-phase-completion"]) + self.swmanager_sw_deploy_strategy.set_state(sw_deploy_strat["state"]) + self.swmanager_sw_deploy_strategy.set_inprogress(sw_deploy_strat["inprogress"]) + else: + raise KeywordException(f"The output line {output_values} was not valid") + + def get_swmanager_sw_deploy_strategy_show(self) -> SwManagerSwDeployStrategyObject: + """ + Retrieves the parsed sw-manager sw-deploy-strategy show object. + + Returns: + SwManagerSwDeployStrategyObject: The parsed sw-manager sw-deploy-strategy show object. + """ + return self.swmanager_sw_deploy_strategy + + @staticmethod + def is_valid_output(value: Dict[str, str]) -> bool: + """ + Checks if the output contains all the required fields. + + Args: + value (Dict[str, str]): The dictionary of output values. + + Returns: + bool: True if all required fields are present, False otherwise. + """ + required_fields = ["strategy-uuid", "release-id", "controller-apply-type", "storage-apply-type", "worker-apply-type", "default-instance-action", "alarm-restrictions", "current-phase", "current-stage", "current-step", "current-phase-completion", "state", "inprogress"] + if "Strategy Software Deploy Strategy" not in value: + get_logger().log_error("Strategy Software Deploy Strategy is not in the output value") + return False + for field in required_fields: + if field not in value["Strategy Software Deploy Strategy"]: + get_logger().log_error(f"{field} is not in the output value") + return False + return True diff --git a/keywords/cloud_platform/swmanager/swmanager_sw_deploy_strategy_keywords.py b/keywords/cloud_platform/swmanager/swmanager_sw_deploy_strategy_keywords.py new file mode 100644 index 00000000..a02e86ed --- /dev/null +++ b/keywords/cloud_platform/swmanager/swmanager_sw_deploy_strategy_keywords.py @@ -0,0 +1,72 @@ +from framework.ssh.ssh_connection import SSHConnection +from keywords.base_keyword import BaseKeyword +from keywords.cloud_platform.command_wrappers import source_openrc +from keywords.cloud_platform.swmanager.objects.swmanager_sw_deploy_strategy_show_output import SwManagerSwDeployStrategyShowOutput + + +class SwmanagerSwDeployStrategyKeywords(BaseKeyword): + """ + This class contains all the keywords related to the 'sw-manager sw-deploy-strategy' commands. + """ + + def __init__(self, ssh_connection: SSHConnection) -> None: + """Initializes SwmanagerSwDeployStrategyKeywords. + + Args: + ssh_connection (SSHConnection): The SSH connection object used for executing commands. + """ + self.ssh_connection = ssh_connection + + def get_sw_deploy_strategy_abort(self): + """Gets the sw-deploy-strategy abort.""" + command = source_openrc("sw-manager sw-deploy-strategy abort") + self.ssh_connection.send(command) + self.validate_success_return_code(self.ssh_connection) + + def get_sw_deploy_strategy_apply(self): + """Gets the sw-deploy-strategy apply.""" + command = source_openrc("sw-manager sw-deploy-strategy apply") + self.ssh_connection.send(command) + self.validate_success_return_code(self.ssh_connection) + + def get_sw_deploy_strategy_create(self, release: str, delete: bool) -> SwManagerSwDeployStrategyShowOutput: + """Gets the sw-deploy-strategy create. + + Args: + release (str): The release to be used for the strategy. + delete (bool): If True, pass --delete as an argument to the command. + + Returns: + SwManagerSwDeployStrategyShowOutput: The output of the prestage strategy. + """ + if delete: + delete_arg = "--delete" + else: + delete_arg = "" + + command = source_openrc(f"sw-manager sw-deploy-strategy create {delete_arg} {release}") + output = self.ssh_connection.send(command) + self.validate_success_return_code(self.ssh_connection) + return SwManagerSwDeployStrategyShowOutput(output) + + def get_sw_deploy_strategy_show(self) -> SwManagerSwDeployStrategyShowOutput: + """Gets the sw-deploy-strategy show. + + Returns: + SwManagerSwDeployStrategyShowOutput: An object containing details of the prestage strategy . + """ + command = source_openrc("sw-manager sw-deploy-strategy show") + output = self.ssh_connection.send(command) + self.validate_success_return_code(self.ssh_connection) + return SwManagerSwDeployStrategyShowOutput(output) + + def get_sw_deploy_strategy_delete(self) -> None: + """Gets the sw-deploy-strategy delete. + + Returns: + None: None. + """ + command = source_openrc("sw-manager sw-deploy-strategy delete") + output = self.ssh_connection.send(command) + self.validate_success_return_code(self.ssh_connection) + return output diff --git a/keywords/cloud_platform/upgrade/objects/software_upload_object.py b/keywords/cloud_platform/upgrade/objects/software_upload_object.py new file mode 100644 index 00000000..28cf0b88 --- /dev/null +++ b/keywords/cloud_platform/upgrade/objects/software_upload_object.py @@ -0,0 +1,23 @@ +class SoftwareUploadObject: + """Class to hold attributes of a software upload as returned by software upload command""" + + def __init__(self, uploaded_file: str, release: str) -> None: + """Initializes a SoftwareUploadObject with the uploaded file and release.""" + self.uploaded_file = uploaded_file + self.release = release + + def get_uploaded_file(self) -> str: + """Getter for uploaded_file. + + Returns: + str: the uploaded file name + """ + return self.uploaded_file + + def get_release(self) -> str: + """Getter for release. + + Returns: + str: the release name + """ + return self.release diff --git a/keywords/cloud_platform/upgrade/objects/software_upload_output.py b/keywords/cloud_platform/upgrade/objects/software_upload_output.py new file mode 100644 index 00000000..5333563d --- /dev/null +++ b/keywords/cloud_platform/upgrade/objects/software_upload_output.py @@ -0,0 +1,54 @@ +"""Software Upload Output.""" + +from typing import Dict + +from framework.logging.automation_logger import get_logger +from keywords.cloud_platform.system.system_table_parser import SystemTableParser +from keywords.cloud_platform.upgrade.objects.software_upload_object import SoftwareUploadObject + + +class SoftwareUploadOutput: + """This class parses o/p 'software show' command into an object of type SoftwareUploadObject.""" + + def __init__(self, software_upload_output: list[str]): + """Instance of the class. + + Args: + software_upload_output (list[str]): output of 'software show' command as a list of strings. + """ + self.software_show: SoftwareUploadObject = [] + system_table_parser = SystemTableParser(software_upload_output) + self.output_values = system_table_parser.get_output_values_list() + + for value in self.output_values: + software_show_object = SoftwareUploadObject( + value["Uploaded File"], + value["Release"], + ) + self.software_show.append(software_show_object) + + def get_software_uploaded(self) -> SoftwareUploadObject: + """Get the uploaded data. + + Returns: + SoftwareUploadObject: the list of software upload objects + + """ + return self.software_show + + @staticmethod + def is_valid_output(value: Dict[str, str]) -> bool: + """Checks if the output contains all the required fields. + + Args: + value (Dict[str, str]): The dictionary of output values. + + Returns: + bool: True if all required fields are present, False otherwise. + """ + required_fields = ["Release", "Uploaded File"] + for field in required_fields: + if field not in value: + get_logger().log_error(f"{field} is not in the output value") + return False + return True diff --git a/keywords/cloud_platform/upgrade/usm_keywords.py b/keywords/cloud_platform/upgrade/usm_keywords.py index c4c94355..1d59bf28 100644 --- a/keywords/cloud_platform/upgrade/usm_keywords.py +++ b/keywords/cloud_platform/upgrade/usm_keywords.py @@ -3,6 +3,7 @@ from framework.logging.automation_logger import get_logger from framework.ssh.ssh_connection import SSHConnection from framework.validation.validation import validate_equals_with_retry from keywords.base_keyword import BaseKeyword +from keywords.cloud_platform.upgrade.objects.software_upload_output import SoftwareUploadOutput from keywords.cloud_platform.upgrade.software_show_keywords import SoftwareShowKeywords @@ -16,7 +17,7 @@ class USMKeywords(BaseKeyword): def __init__(self, ssh_connection: SSHConnection): self.ssh_connection = ssh_connection - def upload_patch_file(self, patch_file_path: str) -> None: + def upload_patch_file(self, patch_file_path: str) -> SoftwareUploadOutput: """ Upload a single patch file using 'software upload'. @@ -25,11 +26,15 @@ class USMKeywords(BaseKeyword): Raises: KeywordException: On failure to upload. + + Returns: + SoftwareUploadOutput: Parsed output containing details of the uploaded patch. """ get_logger().log_info(f"Uploading patch file: {patch_file_path}") output = self.ssh_connection.send_as_sudo(f"software upload {patch_file_path}") self.validate_success_return_code(self.ssh_connection) get_logger().log_info("Upload completed:\n" + "\n".join(output)) + return SoftwareUploadOutput(output) def upload_patch_dir(self, patch_dir_path: str) -> None: """