From b18c70eac7ce06ecde5261109f50d37be7a04f92 Mon Sep 17 00:00:00 2001 From: Rity Menon Date: Wed, 12 Mar 2025 11:37:33 -0700 Subject: [PATCH] dcmanager strategy-step keywords Change-Id: Id940f8d0a3a7cc2e21e85ba3e3215e5caa61abc0 --- .../dcmanager_strategy_step_keywords.py | 47 +++++++++++ .../objects/dcmanager_strategy_step_object.py | 82 +++++++++++++++++++ .../objects/dcmanager_strategy_step_output.py | 66 +++++++++++++++ .../dcmanager_strategy_step_show_output.py | 65 +++++++++++++++ 4 files changed, 260 insertions(+) create mode 100644 keywords/cloud_platform/dcmanager/dcmanager_strategy_step_keywords.py create mode 100644 keywords/cloud_platform/dcmanager/objects/dcmanager_strategy_step_object.py create mode 100644 keywords/cloud_platform/dcmanager/objects/dcmanager_strategy_step_output.py create mode 100644 keywords/cloud_platform/dcmanager/objects/dcmanager_strategy_step_show_output.py diff --git a/keywords/cloud_platform/dcmanager/dcmanager_strategy_step_keywords.py b/keywords/cloud_platform/dcmanager/dcmanager_strategy_step_keywords.py new file mode 100644 index 00000000..6b5dccfa --- /dev/null +++ b/keywords/cloud_platform/dcmanager/dcmanager_strategy_step_keywords.py @@ -0,0 +1,47 @@ +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.dcmanager.objects.dcmanager_strategy_step_output import DcmanagerStrategyStepOutput +from keywords.cloud_platform.dcmanager.objects.dcmanager_strategy_step_show_output import DcmanagerStrategyStepShowOutput + + +class DcmanagerStrategyStepKeywords(BaseKeyword): + """ + This class contains all the keywords related to the 'dcmanager strategy-step' commands. + """ + + def __init__(self, ssh_connection: SSHConnection) -> None: + """ + Initializes DcmanagerStrategyStepKeywords. + + Args: + ssh_connection (SSHConnection): The SSH connection object used for executing commands. + """ + self.ssh_connection = ssh_connection + + def get_dcmanager_strategy_step_list(self) -> DcmanagerStrategyStepOutput: + """ + Gets the dcmanager strategy-step list. + + Returns: + DcmanagerStrategyStepOutput: An object containing the list of strategy steps. + """ + command = source_openrc("dcmanager strategy-step list") + output = self.ssh_connection.send(command) + self.validate_success_return_code(self.ssh_connection) + return DcmanagerStrategyStepOutput(output) + + def get_dcmanager_strategy_step_show(self, subcloud_name: str) -> DcmanagerStrategyStepShowOutput: + """ + Gets the dcmanager strategy-step show. + + Args: + subcloud_name (str): The subcloud name. + + Returns: + DcmanagerStrategyStepShowOutput: An object containing details of the strategy step. + """ + command = source_openrc(f"dcmanager strategy-step show {subcloud_name}") + output = self.ssh_connection.send(command) + self.validate_success_return_code(self.ssh_connection) + return DcmanagerStrategyStepShowOutput(output) diff --git a/keywords/cloud_platform/dcmanager/objects/dcmanager_strategy_step_object.py b/keywords/cloud_platform/dcmanager/objects/dcmanager_strategy_step_object.py new file mode 100644 index 00000000..4cbd02a9 --- /dev/null +++ b/keywords/cloud_platform/dcmanager/objects/dcmanager_strategy_step_object.py @@ -0,0 +1,82 @@ +from typing import Optional + + +class DcmanagerStrategyStepObject: + """ + This class represents a dcmanager strategy-step as an object. + """ + + def __init__(self) -> None: + """Initializes a DcmanagerStrategyStepObject with default values.""" + self.cloud: Optional[str] = None + self.stage: Optional[int] = None + self.state: Optional[str] = None + self.details: Optional[str] = None + self.started_at: Optional[str] = None + self.finished_at: Optional[str] = None + self.created_at: Optional[str] = None + self.updated_at: Optional[str] = None + + def set_cloud(self, cloud: str) -> None: + """Sets the cloud of the strategy-step.""" + self.cloud = cloud + + def get_cloud(self) -> Optional[str]: + """Gets the cloud of the strategy-step.""" + return self.cloud + + def set_stage(self, stage: int) -> None: + """Sets the stage of the strategy-step.""" + self.stage = stage + + def get_stage(self) -> Optional[int]: + """Gets the stage of the strategy-step.""" + return self.stage + + def set_state(self, state: str) -> None: + """Sets the state of the strategy-step.""" + self.state = state + + def get_state(self) -> Optional[str]: + """Gets the state of the strategy-step.""" + return self.state + + def set_details(self, details: str) -> None: + """Sets the details of the strategy-step.""" + self.details = details + + def get_details(self) -> Optional[str]: + """Gets the details of the strategy-step.""" + return self.details + + def set_started_at(self, started_at: str) -> None: + """Sets the started_at of the strategy-step.""" + self.started_at = started_at + + def get_started_at(self) -> Optional[str]: + """Gets the started_at of the strategy-step.""" + return self.started_at + + def set_finished_at(self, finished_at: str) -> None: + """Sets the finished_at of the strategy-step.""" + self.finished_at = finished_at + + def get_finished_at(self) -> Optional[str]: + """Gets the finished_at of the strategy-step.""" + return self.finished_at + + def set_created_at(self, created_at: str) -> None: + """Sets the creation timestamp of the patch-strategy-config.""" + self.created_at = created_at + + def get_created_at(self) -> Optional[str]: + """Gets the creation timestamp of the patch-strategy-config.""" + return self.created_at + + def set_updated_at(self, updated_at: str) -> None: + """Sets the last updated timestamp of the patch-strategy-config.""" + self.updated_at = updated_at + + def get_updated_at(self) -> Optional[str]: + """Gets the last updated timestamp of the patch-strategy-config.""" + return self.updated_at diff --git a/keywords/cloud_platform/dcmanager/objects/dcmanager_strategy_step_output.py b/keywords/cloud_platform/dcmanager/objects/dcmanager_strategy_step_output.py new file mode 100644 index 00000000..747f1a3d --- /dev/null +++ b/keywords/cloud_platform/dcmanager/objects/dcmanager_strategy_step_output.py @@ -0,0 +1,66 @@ +from typing import Dict, List + +from framework.exceptions.keyword_exception import KeywordException +from framework.logging.automation_logger import get_logger +from keywords.cloud_platform.dcmanager.dcmanager_table_parser import DcManagerTableParser +from keywords.cloud_platform.dcmanager.objects.dcmanager_strategy_step_object import DcmanagerStrategyStepObject + + +class DcmanagerStrategyStepOutput: + """ + Parses the output of the 'dcmanager strategy-step list' command into a list of DcmanagerStrategyStepObject instances. + """ + + def __init__(self, dcmanager_strategy: str) -> None: + """ + Initializes DcmanagerStrategyStepOutput. + + Args: + dcmanager_strategy (str): Output of the 'dcmanager strategy-step list' command. + + Raises: + KeywordException: If the output format is invalid. + """ + self.dcmanager_strategy_step: List[DcmanagerStrategyStepObject] = [] + dc_table_parser = DcManagerTableParser(dcmanager_strategy) + output_values = dc_table_parser.get_output_values_list() + + for value in output_values: + if self.is_valid_output(value): + dcmanager_strategy_step = DcmanagerStrategyStepObject() + dcmanager_strategy_step.set_cloud(value["cloud"]) + dcmanager_strategy_step.set_stage(value["stage"]) + dcmanager_strategy_step.set_state(value["state"]) + dcmanager_strategy_step.set_details(value["details"]) + dcmanager_strategy_step.set_started_at(value["started_at"]) + dcmanager_strategy_step.set_finished_at(value["finished_at"]) + self.dcmanager_strategy_step.append(dcmanager_strategy_step) + else: + raise KeywordException(f"The output line {value} was not valid") + + def get_dcmanager_strategy_step_list(self) -> List[DcmanagerStrategyStepObject]: + """ + Retrieves the parsed dcmanager strategy-step list. + + Returns: + List[DcmanagerStrategyStepObject]: A list of parsed DcmanagerStrategyStepObject instances. + """ + return self.dcmanager_strategy_step + + @staticmethod + def is_valid_output(value: Dict[str, str]) -> bool: + """ + Checks if the output dictionary contains all required fields. + + Args: + value (Dict[str, str]): The dictionary of output values. + + Returns: + bool: True if the output contains all required fields, False otherwise. + """ + required_fields = ["cloud", "stage", "state", "details", "started_at", "finished_at"] + 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/dcmanager/objects/dcmanager_strategy_step_show_output.py b/keywords/cloud_platform/dcmanager/objects/dcmanager_strategy_step_show_output.py new file mode 100644 index 00000000..16e90d8d --- /dev/null +++ b/keywords/cloud_platform/dcmanager/objects/dcmanager_strategy_step_show_output.py @@ -0,0 +1,65 @@ +from typing import Dict + +from framework.exceptions.keyword_exception import KeywordException +from framework.logging.automation_logger import get_logger +from keywords.cloud_platform.dcmanager.dcmanager_vertical_table_parser import DcManagerVerticalTableParser +from keywords.cloud_platform.dcmanager.objects.dcmanager_strategy_step_object import DcmanagerStrategyStepObject + + +class DcmanagerStrategyStepShowOutput: + """ + Parses the output of the 'dcmanager strategy-step show' command into a DcmanagerStrategyStepObject instance. + """ + + def __init__(self, dcmanager_strategy: str) -> None: + """ + Initializes DcmanagerStrategyStepObject. + + Args: + dcmanager_strategy (str): Output of the 'dcmanager strategy-step show' command. + + Raises: + KeywordException: If the output format is invalid. + """ + dc_vertical_table_parser = DcManagerVerticalTableParser(dcmanager_strategy) + output_values = dc_vertical_table_parser.get_output_values_dict() + + if self.is_valid_output(output_values): + self.dcmanager_strategy_step = DcmanagerStrategyStepObject() + self.dcmanager_strategy_step.set_cloud(output_values["cloud"]) + self.dcmanager_strategy_step.set_stage(output_values["stage"]) + self.dcmanager_strategy_step.set_state(output_values["state"]) + self.dcmanager_strategy_step.set_details(output_values["details"]) + self.dcmanager_strategy_step.set_started_at(output_values["started_at"]) + self.dcmanager_strategy_step.set_finished_at(output_values["finished_at"]) + self.dcmanager_strategy_step.set_created_at(output_values["created_at"]) + self.dcmanager_strategy_step.set_updated_at(output_values["updated_at"]) + else: + raise KeywordException(f"The output line {output_values} was not valid") + + def get_dcmanager_strategy_step_show(self) -> DcmanagerStrategyStepObject: + """ + Retrieves the parsed dcmanager strategy-step show object. + + Returns: + DcmanagerStrategyStepObject: The parsed dcmanager strategy-step show object. + """ + return self.dcmanager_strategy_step + + @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 = ["cloud", "stage", "state", "details", "started_at", "finished_at", "created_at", "updated_at"] + 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