starlingx/test
Code Issues Proposed changes

Merge "dcmanager patch-strategy keywords"

Browse Source
This commit is contained in:
Zuul
2025-03-17 17:59:33 +00:00
committed by Gerrit Code Review
parent 07fb7d10ba f7c5bb531b
commit cadeb11f01
3 changed files with 202 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

55
keywords/cloud_platform/dcmanager/dcmanager_patch_strategy_keywords.py Normal file
View File

@@ -0,0 +1,55 @@
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_show_output import DcmanagerPatchStrategyShowOutput
class DcmanagerPatchStrategyKeywords(BaseKeyword):
"""
This class contains all the keywords related to the 'dcmanager patch-strategy' commands.
"""
def __init__(self, ssh_connection: SSHConnection) -> None:
"""
Initializes DcmanagerPatchStrategyKeywords.
Args:
ssh_connection (SSHConnection): The SSH connection object used for executing commands.
"""
self.ssh_connection = ssh_connection
def get_dcmanager_patch_strategy_show(self) -> DcmanagerPatchStrategyShowOutput:
"""
Gets the patch-strategy show.
Returns:
DcmanagerPatchStrategyShowOutput: An object containing details of the patch strategy .
"""
command = source_openrc("dcmanager patch-strategy show")
output = self.ssh_connection.send(command)
self.validate_success_return_code(self.ssh_connection)
return DcmanagerPatchStrategyShowOutput(output)
def get_dcmanager_patch_strategy_create(self) -> DcmanagerPatchStrategyShowOutput:
"""
Gets the patch-strategy create.
Returns:
DcmanagerPatchStrategyShowOutput: An object containing details of the patch strategy .
"""
command = source_openrc("dcmanager patch-strategy create")
output = self.ssh_connection.send(command)
self.validate_success_return_code(self.ssh_connection)
return DcmanagerPatchStrategyShowOutput(output)
def get_dcmanager_patch_strategy_delete(self) -> DcmanagerPatchStrategyShowOutput:
"""
Gets the patch-strategy delete.
Returns:
DcmanagerPatchStrategyShowOutput: An object containing details of the patch strategy .
"""
command = source_openrc("dcmanager patch-strategy delete")
output = self.ssh_connection.send(command)
self.validate_success_return_code(self.ssh_connection)
return DcmanagerPatchStrategyShowOutput(output)

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

@@ -0,0 +1,82 @@
from typing import Optional
class DcmanagerPatchStrategyObject:
"""
This class represents a dcmanager patch-strategy as an object.
"""
def __init__(self) -> None:
"""Initializes a DcmanagerPatchStrategyObject with default values."""
self.strategy_type: Optional[str] = None
self.subcloud_apply_type: Optional[str] = None
self.max_parallel_subclouds: Optional[int] = None
self.stop_on_failure: Optional[bool] = None
self.upload_only: Optional[bool] = None
self.state: Optional[str] = None
self.created_at: Optional[str] = None
self.updated_at: Optional[str] = None
def set_strategy_type(self, strategy_type: str) -> None:
"""Sets the strategy_type of the patch-strategy."""
self.strategy_type = strategy_type
def get_strategy_type(self) -> Optional[str]:
"""Gets the strategy_type of the patch-strategy."""
return self.strategy_type
def set_subcloud_apply_type(self, subcloud_apply_type: str) -> None:
"""Sets the subcloud_apply_type of the patch-strategy."""
self.subcloud_apply_type = subcloud_apply_type
def get_subcloud_apply_type(self) -> Optional[str]:
"""Gets the subcloud_apply_type of the patch-strategy."""
return self.subcloud_apply_type
def set_max_parallel_subclouds(self, max_parallel_subclouds: int) -> None:
"""Sets the max_parallel_subclouds of the patch-strategy."""
self.max_parallel_subclouds = max_parallel_subclouds
def get_max_parallel_subclouds(self) -> Optional[int]:
"""Gets the max_parallel_subclouds of the patch-strategy."""
return self.max_parallel_subclouds
def set_stop_on_failure(self, stop_on_failure: bool) -> None:
"""Sets the stop_on_failure of the patch-strategy."""
self.stop_on_failure = stop_on_failure
def get_stop_on_failure(self) -> bool:
"""Gets the stop_on_failure of the patch-strategy."""
return self.stop_on_failure
def set_upload_only(self, upload_only: bool) -> None:
"""Sets the upload_only of the patch-strategy."""
self.upload_only = upload_only
def get_upload_only(self) -> Optional[bool]:
"""Gets the upload_only of the patch-strategy."""
return self.upload_only
def set_state(self, state: str) -> None:
"""Sets the state of the patch-strategy."""
self.state = state
def get_state(self) -> Optional[str]:
"""Gets the state of the patch-strategy."""
return self.state
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

65
keywords/cloud_platform/dcmanager/objects/dcmanager_patch_strategy_show_output.py Normal file
View File

@@ -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_patch_strategy_object import DcmanagerPatchStrategyObject
class DcmanagerPatchStrategyShowOutput:
"""
Parses the output of the 'dcmanager patch-strategy-config show' command into a DcmanagerPatchStrategyObject instance.
"""
def __init__(self, dcmanager_patch: str) -> None:
"""
Initializes DcmanagerPatchStrategyObject.
Args:
dcmanager_patch (str): Output of the 'dcmanager patch-strategy 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 = DcmanagerPatchStrategyObject()
self.dcmanager_patch_strategy.set_strategy_type(output_values["strategy type"])
self.dcmanager_patch_strategy.set_subcloud_apply_type(output_values["subcloud apply type"])
self.dcmanager_patch_strategy.set_max_parallel_subclouds(output_values["max parallel subclouds"])
self.dcmanager_patch_strategy.set_stop_on_failure(output_values["stop on failure"])
self.dcmanager_patch_strategy.set_upload_only(output_values["upload only"])
self.dcmanager_patch_strategy.set_state(output_values["state"])
self.dcmanager_patch_strategy.set_created_at(output_values["created_at"])
self.dcmanager_patch_strategy.set_updated_at(output_values["updated_at"])
else:
raise KeywordException(f"The output line {output_values} was not valid")
def get_dcmanager_patch_strategy_show(self) -> DcmanagerPatchStrategyObject:
"""
Retrieves the parsed dcmanager patch-strategy show object.
Returns:
DcmanagerPatchStrategyObject: The parsed dcmanager patch-strategy show object.
"""
return self.dcmanager_patch_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 type", "subcloud apply type", "max parallel subclouds", "stop on failure", "upload only", "state", "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