diff --git a/keywords/cloud_platform/dcmanager/dcmanager_patch_strategy_config_keywords.py b/keywords/cloud_platform/dcmanager/dcmanager_patch_strategy_config_keywords.py new file mode 100644 index 00000000..113d53f4 --- /dev/null +++ b/keywords/cloud_platform/dcmanager/dcmanager_patch_strategy_config_keywords.py @@ -0,0 +1,48 @@ +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_patch_strategy_config_output import ( + DcmanagerPatchStrategyConfigOutput, +) +from keywords.cloud_platform.dcmanager.objects.dcmanager_patch_strategy_config_show_output import ( + DcmanagerPatchStrategyConfigShowOutput, +) + + +class DcmanagerPatchStrategyConfigKeywords(BaseKeyword): + """ + This class contains all the keywords related to the 'dcmanager patch-strategy-config' commands. + """ + + def __init__(self, ssh_connection: SSHConnection) -> None: + """ + Initializes DcmanagerPatchStrategyConfigKeywords. + + Args: + ssh_connection (SSHConnection): The SSH connection object used for executing commands. + """ + self.ssh_connection = ssh_connection + + def get_dcmanager_patch_strategy_config_list(self) -> DcmanagerPatchStrategyConfigOutput: + """ + Gets the dcmanager patch-strategy-config list. + + Returns: + DcmanagerPatchStrategyConfigOutput: An object containing the list of patch strategy configs. + """ + command = source_openrc("dcmanager patch-strategy-config list") + output = self.ssh_connection.send(command) + self.validate_success_return_code(self.ssh_connection) + return DcmanagerPatchStrategyConfigOutput(output) + + def get_dcmanager_patch_strategy_config_show(self) -> DcmanagerPatchStrategyConfigShowOutput: + """ + Gets the patch-strategy-config show. + + Returns: + DcmanagerPatchStrategyConfigShowOutput: An object containing details of the patch strategy config. + """ + command = source_openrc("dcmanager patch-strategy-config show") + output = self.ssh_connection.send(command) + self.validate_success_return_code(self.ssh_connection) + return DcmanagerPatchStrategyConfigShowOutput(output) diff --git a/keywords/cloud_platform/dcmanager/objects/dcmanager_patch_strategy_config_object.py b/keywords/cloud_platform/dcmanager/objects/dcmanager_patch_strategy_config_object.py new file mode 100644 index 00000000..1f2c7da2 --- /dev/null +++ b/keywords/cloud_platform/dcmanager/objects/dcmanager_patch_strategy_config_object.py @@ -0,0 +1,82 @@ +from typing import Optional + + +class DcmanagerPatchStrategyConfigObject: + """ + This class represents a dcmanager patch-strategy-config as an object. + """ + + def __init__(self) -> None: + """Initializes a DcmanagerPatchStrategyConfigObject with default values.""" + self.cloud: Optional[str] = None + self.storage_apply_type: Optional[str] = None + self.worker_apply_type: Optional[str] = None + self.max_parallel_workers: int = -1 + self.alarm_restriction_type: Optional[str] = None + self.default_instance_action: 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 patch-strategy-config.""" + self.cloud = cloud + + def get_cloud(self) -> Optional[str]: + """Gets the cloud of the patch-strategy-config.""" + return self.cloud + + def set_storage_apply_type(self, storage_apply_type: str) -> None: + """Sets the storage_apply_type of the patch-strategy-config.""" + self.storage_apply_type = storage_apply_type + + def get_storage_apply_type(self) -> Optional[str]: + """Gets the storage_apply_type of the patch-strategy-config.""" + return self.storage_apply_type + + def set_worker_apply_type(self, worker_apply_type: str) -> None: + """Sets the worker_apply_type of the patch-strategy-config.""" + self.worker_apply_type = worker_apply_type + + def get_worker_apply_type(self) -> Optional[str]: + """Gets the worker_apply_type of the patch-strategy-config.""" + return self.worker_apply_type + + def set_max_parallel_workers(self, max_parallel_workers: int) -> None: + """Sets the max_parallel_workers of the patch-strategy-config.""" + self.max_parallel_workers = max_parallel_workers + + def get_max_parallel_workers(self) -> int: + """Gets the max_parallel_workers of the patch-strategy-config.""" + return self.max_parallel_workers + + def set_alarm_restriction_type(self, alarm_restriction_type: str) -> None: + """Sets the alarm_restriction_type of the patch-strategy-config.""" + self.alarm_restriction_type = alarm_restriction_type + + def get_alarm_restriction_type(self) -> Optional[str]: + """Gets the alarm_restriction_type of the patch-strategy-config.""" + return self.alarm_restriction_type + + def set_default_instance_action(self, default_instance_action: str) -> None: + """Sets the default_instance_action of the patch-strategy-config.""" + self.default_instance_action = default_instance_action + + def get_default_instance_action(self) -> Optional[str]: + """Gets the default_instance_action of the patch-strategy-config.""" + return self.default_instance_action + + 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_patch_strategy_config_output.py b/keywords/cloud_platform/dcmanager/objects/dcmanager_patch_strategy_config_output.py new file mode 100644 index 00000000..e5ddfc9c --- /dev/null +++ b/keywords/cloud_platform/dcmanager/objects/dcmanager_patch_strategy_config_output.py @@ -0,0 +1,70 @@ +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_patch_strategy_config_object import ( + DcmanagerPatchStrategyConfigObject, +) + + +class DcmanagerPatchStrategyConfigOutput: + """ + Parses the output of the 'dcmanager patch-strategy-config list' command into a list of DcmanagerPatchStrategyConfigObject instances. + """ + + def __init__(self, dcmanager_patch: str) -> None: + """ + Initializes DcmanagerPatchStrategyConfigOutput. + + Args: + dcmanager_patch (str): Output of the 'dcmanager patch-strategy-config list' command. + + Raises: + KeywordException: If the output format is invalid. + """ + self.dcmanager_patch_strategy_config: List[DcmanagerPatchStrategyConfigObject] = [] + dc_table_parser = DcManagerTableParser(dcmanager_patch) + output_values = dc_table_parser.get_output_values_list() + + for value in output_values: + if self.is_valid_output(value): + dcmanager_patch_strategy_config = DcmanagerPatchStrategyConfigObject() + dcmanager_patch_strategy_config.set_cloud(value["cloud"]) + dcmanager_patch_strategy_config.set_storage_apply_type(value["storage apply type"]) + dcmanager_patch_strategy_config.set_worker_apply_type(value["worker apply type"]) + dcmanager_patch_strategy_config.set_max_parallel_workers(value["max parallel workers"]) + dcmanager_patch_strategy_config.set_alarm_restriction_type(value["alarm restriction type"]) + dcmanager_patch_strategy_config.set_default_instance_action(value["default instance action"]) + self.dcmanager_patch_strategy_config.append(dcmanager_patch_strategy_config) + else: + raise KeywordException(f"The output line {value} was not valid") + + def get_dcmanager_patch_strategy_config_list(self) -> List[DcmanagerPatchStrategyConfigObject]: + """ + Retrieves the parsed dcmanager patch-strategy-config list. + + Returns: + List[DcmanagerPatchStrategyConfigObject]: A list of parsed DcmanagerPatchStrategyConfigObject instances. + """ + return self.dcmanager_patch_strategy_config + + @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", "storage apply type", "worker apply type", "max parallel workers", "alarm restriction type", "default instance action"] + 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_patch_strategy_config_show_output.py b/keywords/cloud_platform/dcmanager/objects/dcmanager_patch_strategy_config_show_output.py new file mode 100644 index 00000000..d8cfab85 --- /dev/null +++ b/keywords/cloud_platform/dcmanager/objects/dcmanager_patch_strategy_config_show_output.py @@ -0,0 +1,69 @@ +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_patch_strategy_config_object import ( + DcmanagerPatchStrategyConfigObject, +) + + +class DcmanagerPatchStrategyConfigShowOutput: + """ + Parses the output of the 'dcmanager patch-strategy-config show' command into a DcmanagerPatchStrategyConfigObject instance. + """ + + def __init__(self, dcmanager_patch: str) -> None: + """ + Initializes DcmanagerPatchStrategyConfigObject. + + Args: + dcmanager_patch (str): Output of the 'dcmanager patch-strategy-config show' command. + + Raises: + KeywordException: If the output format is invalid. + """ + dc_vertical_table_parser = DcManagerVerticalTableParser(dcmanager_patch) + output_values = dc_vertical_table_parser.get_output_values_dict() + + if self.is_valid_output(output_values): + self.dcmanager_patch_strategy_config = DcmanagerPatchStrategyConfigObject() + self.dcmanager_patch_strategy_config.set_cloud(output_values["cloud"]) + self.dcmanager_patch_strategy_config.set_storage_apply_type(output_values["storage apply type"]) + self.dcmanager_patch_strategy_config.set_worker_apply_type(output_values["worker apply type"]) + self.dcmanager_patch_strategy_config.set_max_parallel_workers(output_values["max parallel workers"]) + self.dcmanager_patch_strategy_config.set_alarm_restriction_type(output_values["alarm restriction type"]) + self.dcmanager_patch_strategy_config.set_default_instance_action(output_values["default instance action"]) + self.dcmanager_patch_strategy_config.set_created_at(output_values["created_at"]) + self.dcmanager_patch_strategy_config.set_updated_at(output_values["updated_at"]) + else: + raise KeywordException(f"The output line {output_values} was not valid") + + def get_dcmanager_patch_strategy_config_show(self) -> DcmanagerPatchStrategyConfigObject: + """ + Retrieves the parsed dcmanager patch-strategy-config show object. + + Returns: + DcmanagerPatchStrategyConfigObject: The parsed dcmanager patch-strategy-config show object. + """ + return self.dcmanager_patch_strategy_config + + @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", "storage apply type", "worker apply type", "max parallel workers", "alarm restriction type", "default instance action", "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