starlingx/test
Code Issues Proposed changes

dcmanager strategy-step keywords

Browse Source 
Change-Id: Id940f8d0a3a7cc2e21e85ba3e3215e5caa61abc0
This commit is contained in:
Rity Menon
2025-03-12 11:37:33 -07:00
parent 549888131e
commit b18c70eac7
4 changed files with 260 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

47
keywords/cloud_platform/dcmanager/dcmanager_strategy_step_keywords.py Normal file
View File

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

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

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

66
keywords/cloud_platform/dcmanager/objects/dcmanager_strategy_step_output.py Normal file
View File

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

65
keywords/cloud_platform/dcmanager/objects/dcmanager_strategy_step_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_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