starlingx/test
Code Issues Proposed changes

dcmanager patch-strategy-config keywords

Browse Source 
Change-Id: If1383a7f937f6619a3b83a0f88ecd72475ae1a1e
This commit is contained in:
Rity Menon
2025-03-06 11:34:31 -08:00
parent 0f11de29d0
commit a01f7f61da
4 changed files with 269 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

48
keywords/cloud_platform/dcmanager/dcmanager_patch_strategy_config_keywords.py Normal file
View File

@@ -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)

82
keywords/cloud_platform/dcmanager/objects/dcmanager_patch_strategy_config_object.py Normal file
View File

@@ -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

70
keywords/cloud_platform/dcmanager/objects/dcmanager_patch_strategy_config_output.py Normal file
View File

@@ -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

69
keywords/cloud_platform/dcmanager/objects/dcmanager_patch_strategy_config_show_output.py Normal file
View File

@@ -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