From a4a97c3870312f9733ad980ceea9b246d66f2be1 Mon Sep 17 00:00:00 2001 From: Rity Menon Date: Thu, 20 Mar 2025 10:21:47 -0700 Subject: [PATCH] sw-manager fw-update-strategy keyword Change-Id: I493f95af62ee067874449cf5d8c35c23e74a6817 --- .../swmanager_fw_update_strategy_object.py | 136 ++++++++++++++++++ ...wmanager_fw_update_strategy_show_output.py | 74 ++++++++++ .../swmanager_fw_update_strategy_keywords.py | 31 ++++ .../swmanager_vertical_table_parser.py | 44 ++++++ 4 files changed, 285 insertions(+) create mode 100644 keywords/cloud_platform/swmanager/objects/swmanager_fw_update_strategy_object.py create mode 100644 keywords/cloud_platform/swmanager/objects/swmanager_fw_update_strategy_show_output.py create mode 100644 keywords/cloud_platform/swmanager/swmanager_fw_update_strategy_keywords.py create mode 100644 keywords/cloud_platform/swmanager/swmanager_vertical_table_parser.py diff --git a/keywords/cloud_platform/swmanager/objects/swmanager_fw_update_strategy_object.py b/keywords/cloud_platform/swmanager/objects/swmanager_fw_update_strategy_object.py new file mode 100644 index 00000000..0ab7d359 --- /dev/null +++ b/keywords/cloud_platform/swmanager/objects/swmanager_fw_update_strategy_object.py @@ -0,0 +1,136 @@ +from typing import Optional + + +class SwmanagerFwUpdateStrategyObject: + """ + This class represents a sw-manager fw-update-strategy as an object. + """ + + def __init__(self) -> None: + """Initializes a SwmanagerFwUpdateStrategyObject with default values.""" + self.strategy_uuid: 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 + self.build_result: Optional[str] = None + self.build_reason: Optional[str] = None + + def set_strategy_uuid(self, strategy_uuid: str) -> None: + """Sets the strategy_uuid of the fw-update-strategy.""" + self.strategy_uuid = strategy_uuid + + def get_strategy_uuid(self) -> Optional[str]: + """Gets the strategy_uuid of the fw-update-strategy.""" + return self.strategy_uuid + + def set_controller_apply_type(self, controller_apply_type: str) -> None: + """Sets the controller_apply_type of the fw-update-strategy.""" + self.controller_apply_type = controller_apply_type + + def get_controller_apply_type(self) -> Optional[str]: + """Gets the controller_apply_type of the fw-update-strategy.""" + return self.controller_apply_type + + def set_storage_apply_type(self, storage_apply_type: str) -> None: + """Sets the storage_apply_type of the fw-update-strategy.""" + self.storage_apply_type = storage_apply_type + + def get_storage_apply_type(self) -> Optional[str]: + """Gets the storage_apply_type of the fw-update-strategy.""" + return self.storage_apply_type + + def set_worker_apply_type(self, worker_apply_type: str) -> None: + """Sets the worker_apply_type of the fw-update-strategy.""" + self.worker_apply_type = worker_apply_type + + def get_worker_apply_type(self) -> Optional[str]: + """Gets the worker_apply_type of the fw-update-strategy.""" + return self.worker_apply_type + + def set_default_instance_action(self, default_instance_action: str) -> None: + """Sets the default_instance_action of the fw-update-strategy.""" + self.default_instance_action = default_instance_action + + def get_default_instance_action(self) -> Optional[str]: + """Gets the default_instance_action of the fw-update-strategy.""" + return self.default_instance_action + + def set_alarm_restrictions(self, alarm_restrictions: str) -> None: + """Sets the alarm_restrictions of the fw-update-strategy.""" + self.alarm_restrictions = alarm_restrictions + + def get_alarm_restrictions(self) -> Optional[str]: + """Gets the alarm_restrictions of the fw-update-strategy.""" + return self.alarm_restrictions + + def set_current_phase(self, current_phase: str) -> None: + """Sets the current_phase of the fw-update-strategy.""" + self.current_phase = current_phase + + def get_current_phase(self) -> Optional[str]: + """Gets the current_phase of the fw-update-strategy.""" + return self.current_phase + + def set_current_stage(self, current_stage: str) -> None: + """Sets the current_stage of the fw-update-strategy.""" + self.current_stage = current_stage + + def get_current_stage(self) -> Optional[str]: + """Gets the current_stage of the fw-update-strategy.""" + return self.current_stage + + def set_current_step(self, current_step: str) -> None: + """Sets the current_step of the fw-update-strategy.""" + self.current_step = current_step + + def get_current_step(self) -> Optional[str]: + """Gets the current_step of the fw-update-strategy.""" + return self.current_step + + def set_current_phase_completion(self, current_phase_completion: str) -> None: + """Sets the current_phase_completion of the fw-update-strategy.""" + self.current_phase_completion = current_phase_completion + + def get_current_phase_completion(self) -> Optional[str]: + """Gets the current_phase_completion of the fw-update-strategy.""" + return self.current_phase_completion + + def set_state(self, state: str) -> None: + """Sets the state of the fw-update-strategy.""" + self.state = state + + def get_state(self) -> Optional[str]: + """Gets the state of the fw-update-strategy.""" + return self.state + + def set_inprogress(self, inprogress: str) -> None: + """Sets the inprogress of the fw-update-strategy.""" + self.inprogress = inprogress + + def get_inprogress(self) -> Optional[str]: + """Gets the inprogress of the fw-update-strategy.""" + return self.inprogress + + def set_build_result(self, build_result: str) -> None: + """Sets the build_result of the fw-update-strategy.""" + self.build_result = build_result + + def get_build_result(self) -> Optional[str]: + """Gets the build_result of the fw-update-strategy.""" + return self.build_result + + def set_build_reason(self, build_reason: str) -> None: + """Sets the build_reason of the fw-update-strategy.""" + self.build_reason = build_reason + + def get_build_reason(self) -> Optional[str]: + """Gets the build_reason of the fw-update-strategy.""" + return self.build_reason diff --git a/keywords/cloud_platform/swmanager/objects/swmanager_fw_update_strategy_show_output.py b/keywords/cloud_platform/swmanager/objects/swmanager_fw_update_strategy_show_output.py new file mode 100644 index 00000000..55e57a18 --- /dev/null +++ b/keywords/cloud_platform/swmanager/objects/swmanager_fw_update_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_fw_update_strategy_object import SwmanagerFwUpdateStrategyObject +from keywords.cloud_platform.swmanager.swmanager_vertical_table_parser import SwManagerVerticalTableParser + + +class SwmanagerFwUpdateStrategyShowOutput: + """ + Parses the output of the 'sw-manager fw-update-strategy show' command into a SwmanagerFwUpdateStrategyObject instance. + """ + + def __init__(self, swmanager_fw_update: str) -> None: + """ + Initializes SwmanagerFwUpdateStrategyObject. + + Args: + swmanager_fw_update (str): Output of the 'sw-manager fw-update-strategy show' command. + + Raises: + KeywordException: If the output format is invalid. + """ + swmanager_vertical_table_parser = SwManagerVerticalTableParser(swmanager_fw_update) + output_values = swmanager_vertical_table_parser.get_output_values_dict() + + if self.is_valid_output(output_values): + self.swmanager_fw_update_strategy = SwmanagerFwUpdateStrategyObject() + strat_fw_update = output_values["Strategy Firmware Update Strategy"] + self.swmanager_fw_update_strategy.set_strategy_uuid(strat_fw_update["strategy-uuid"]) + self.swmanager_fw_update_strategy.set_controller_apply_type(strat_fw_update["controller-apply-type"]) + self.swmanager_fw_update_strategy.set_storage_apply_type(strat_fw_update["storage-apply-type"]) + self.swmanager_fw_update_strategy.set_worker_apply_type(strat_fw_update["worker-apply-type"]) + self.swmanager_fw_update_strategy.set_default_instance_action(strat_fw_update["default-instance-action"]) + self.swmanager_fw_update_strategy.set_alarm_restrictions(strat_fw_update["alarm-restrictions"]) + self.swmanager_fw_update_strategy.set_current_phase(strat_fw_update["current-phase"]) + self.swmanager_fw_update_strategy.set_current_stage(strat_fw_update["current-stage"]) + self.swmanager_fw_update_strategy.set_current_step(strat_fw_update["current-step"]) + self.swmanager_fw_update_strategy.set_current_phase_completion(strat_fw_update["current-phase-completion"]) + self.swmanager_fw_update_strategy.set_state(strat_fw_update["state"]) + self.swmanager_fw_update_strategy.set_build_result(strat_fw_update["build-result"]) + self.swmanager_fw_update_strategy.set_build_reason(strat_fw_update["build-reason"]) + else: + raise KeywordException(f"The output line {output_values} was not valid") + + def get_swmanager_fw_update_strategy_show(self) -> SwmanagerFwUpdateStrategyObject: + """ + Retrieves the parsed sw-manager fw-update-strategy show object. + + Returns: + SwmanagerFwUpdateStrategyObject: The parsed sw-manager fw-update-strategy show object. + """ + return self.swmanager_fw_update_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", "controller-apply-type", "storage-apply-type", "worker-apply-type", "default-instance-action", "alarm-restrictions", "current-phase", "current-stage", "current-step", "current-phase-completion", "state", "build-result", "build-reason"] + if "Strategy Firmware Update Strategy" not in value: + get_logger().log_error("Strategy Firmware Update Strategy is not in the output value") + return False + for field in required_fields: + if field not in value["Strategy Firmware Update 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_fw_update_strategy_keywords.py b/keywords/cloud_platform/swmanager/swmanager_fw_update_strategy_keywords.py new file mode 100644 index 00000000..906c3d4c --- /dev/null +++ b/keywords/cloud_platform/swmanager/swmanager_fw_update_strategy_keywords.py @@ -0,0 +1,31 @@ +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_fw_update_strategy_show_output import SwmanagerFwUpdateStrategyShowOutput + + +class SwmanagerFwUpdateStrategyKeywords(BaseKeyword): + """ + This class contains all the keywords related to the 'sw-manager fw-update-strategy' commands. + """ + + def __init__(self, ssh_connection: SSHConnection) -> None: + """ + Initializes SwmanagerFwUpdateStrategyKeywords. + + Args: + ssh_connection (SSHConnection): The SSH connection object used for executing commands. + """ + self.ssh_connection = ssh_connection + + def get_swmanager_fw_update_strategy_show(self) -> SwmanagerFwUpdateStrategyShowOutput: + """ + Gets the sw-manager fw-update-strategy show. + + Returns: + SwmanagerFwUpdateStrategyShowOutput: An object containing details of the fw_update_strategy. + """ + command = source_openrc("sw-manager fw-update-strategy show") + output = self.ssh_connection.send(command) + self.validate_success_return_code(self.ssh_connection) + return SwmanagerFwUpdateStrategyShowOutput(output) diff --git a/keywords/cloud_platform/swmanager/swmanager_vertical_table_parser.py b/keywords/cloud_platform/swmanager/swmanager_vertical_table_parser.py new file mode 100644 index 00000000..8c214e12 --- /dev/null +++ b/keywords/cloud_platform/swmanager/swmanager_vertical_table_parser.py @@ -0,0 +1,44 @@ +import yaml + + +class SwManagerVerticalTableParser: + """ + Class for SwManager vertical table parsing. + + In a vertical table, attributes and their corresponding values are arranged in + structured text, with each attribute listed alongside its value. + + As an example of a typical argument for the constructor of this class, + consider the output of the command 'sw-manager fw-update-strategy show': + + ``` + Strategy Firmware Update Strategy: + strategy-uuid: 52bf402f-54ab-4fee-8155-4e0b23bd3463 + controller-apply-type: ignore + storage-apply-type: ignore + worker-apply-type: serial + default-instance-action: stop-start + alarm-restrictions: strict + current-phase: build + current-phase-completion: 100% + state: build-failed + build-result: failed + build-reason: alarms ['100.104'] from platform are present + ``` + """ + + def __init__(self, swmanager_output): + self.swmanager_output = swmanager_output + + def get_output_values_dict(self) -> dict: + """Retrieves the output values from `self.swmanager_output` as a dictionary. + + This method parses the YAML content stored in `self.swmanager_output` and returns + it as a dictionary. Each key-value pair in the resulting dictionary corresponds + to entries from the parsed YAML data. + + Returns: + dict: A dictionary representation of the parsed YAML content. + """ + output_values_dict = yaml.safe_load("".join(self.swmanager_output)) + return output_values_dict