starlingx/test
Code Issues Proposed changes

sw-manager fw-update-strategy keyword

Browse Source 
Change-Id: I493f95af62ee067874449cf5d8c35c23e74a6817
This commit is contained in:
Rity Menon
2025-03-20 10:21:47 -07:00
parent 07fb7d10ba
commit a4a97c3870
4 changed files with 285 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

136
keywords/cloud_platform/swmanager/objects/swmanager_fw_update_strategy_object.py Normal file
View File

@@ -0,0 +1,136 @@
from typing import Optional
class SwmanagerFwUpdateStrategyObject:
"""
This class represents a sw-manager fw-update-strategy as an object.
"""
def __init__(self) -> None:
"""Initializes a SwmanagerFwUpdateStrategyObject with default values."""
self.strategy_uuid: Optional[str] = None
self.controller_apply_type: Optional[str] = None
self.storage_apply_type: Optional[str] = None
self.worker_apply_type: Optional[str] = None
self.default_instance_action: Optional[str] = None
self.alarm_restrictions: Optional[str] = None
self.current_phase: Optional[str] = None
self.current_stage: Optional[str] = None
self.current_step: Optional[str] = None
self.current_phase_completion: Optional[str] = None
self.state: Optional[str] = None
self.inprogress: Optional[str] = None
self.build_result: Optional[str] = None
self.build_reason: Optional[str] = None
def set_strategy_uuid(self, strategy_uuid: str) -> None:
"""Sets the strategy_uuid of the fw-update-strategy."""
self.strategy_uuid = strategy_uuid
def get_strategy_uuid(self) -> Optional[str]:
"""Gets the strategy_uuid of the fw-update-strategy."""
return self.strategy_uuid
def set_controller_apply_type(self, controller_apply_type: str) -> None:
"""Sets the controller_apply_type of the fw-update-strategy."""
self.controller_apply_type = controller_apply_type
def get_controller_apply_type(self) -> Optional[str]:
"""Gets the controller_apply_type of the fw-update-strategy."""
return self.controller_apply_type
def set_storage_apply_type(self, storage_apply_type: str) -> None:
"""Sets the storage_apply_type of the fw-update-strategy."""
self.storage_apply_type = storage_apply_type
def get_storage_apply_type(self) -> Optional[str]:
"""Gets the storage_apply_type of the fw-update-strategy."""
return self.storage_apply_type
def set_worker_apply_type(self, worker_apply_type: str) -> None:
"""Sets the worker_apply_type of the fw-update-strategy."""
self.worker_apply_type = worker_apply_type
def get_worker_apply_type(self) -> Optional[str]:
"""Gets the worker_apply_type of the fw-update-strategy."""
return self.worker_apply_type
def set_default_instance_action(self, default_instance_action: str) -> None:
"""Sets the default_instance_action of the fw-update-strategy."""
self.default_instance_action = default_instance_action
def get_default_instance_action(self) -> Optional[str]:
"""Gets the default_instance_action of the fw-update-strategy."""
return self.default_instance_action
def set_alarm_restrictions(self, alarm_restrictions: str) -> None:
"""Sets the alarm_restrictions of the fw-update-strategy."""
self.alarm_restrictions = alarm_restrictions
def get_alarm_restrictions(self) -> Optional[str]:
"""Gets the alarm_restrictions of the fw-update-strategy."""
return self.alarm_restrictions
def set_current_phase(self, current_phase: str) -> None:
"""Sets the current_phase of the fw-update-strategy."""
self.current_phase = current_phase
def get_current_phase(self) -> Optional[str]:
"""Gets the current_phase of the fw-update-strategy."""
return self.current_phase
def set_current_stage(self, current_stage: str) -> None:
"""Sets the current_stage of the fw-update-strategy."""
self.current_stage = current_stage
def get_current_stage(self) -> Optional[str]:
"""Gets the current_stage of the fw-update-strategy."""
return self.current_stage
def set_current_step(self, current_step: str) -> None:
"""Sets the current_step of the fw-update-strategy."""
self.current_step = current_step
def get_current_step(self) -> Optional[str]:
"""Gets the current_step of the fw-update-strategy."""
return self.current_step
def set_current_phase_completion(self, current_phase_completion: str) -> None:
"""Sets the current_phase_completion of the fw-update-strategy."""
self.current_phase_completion = current_phase_completion
def get_current_phase_completion(self) -> Optional[str]:
"""Gets the current_phase_completion of the fw-update-strategy."""
return self.current_phase_completion
def set_state(self, state: str) -> None:
"""Sets the state of the fw-update-strategy."""
self.state = state
def get_state(self) -> Optional[str]:
"""Gets the state of the fw-update-strategy."""
return self.state
def set_inprogress(self, inprogress: str) -> None:
"""Sets the inprogress of the fw-update-strategy."""
self.inprogress = inprogress
def get_inprogress(self) -> Optional[str]:
"""Gets the inprogress of the fw-update-strategy."""
return self.inprogress
def set_build_result(self, build_result: str) -> None:
"""Sets the build_result of the fw-update-strategy."""
self.build_result = build_result
def get_build_result(self) -> Optional[str]:
"""Gets the build_result of the fw-update-strategy."""
return self.build_result
def set_build_reason(self, build_reason: str) -> None:
"""Sets the build_reason of the fw-update-strategy."""
self.build_reason = build_reason
def get_build_reason(self) -> Optional[str]:
"""Gets the build_reason of the fw-update-strategy."""
return self.build_reason

74
keywords/cloud_platform/swmanager/objects/swmanager_fw_update_strategy_show_output.py Normal file
View File

@@ -0,0 +1,74 @@
from typing import Dict
from framework.exceptions.keyword_exception import KeywordException
from framework.logging.automation_logger import get_logger
from keywords.cloud_platform.swmanager.objects.swmanager_fw_update_strategy_object import SwmanagerFwUpdateStrategyObject
from keywords.cloud_platform.swmanager.swmanager_vertical_table_parser import SwManagerVerticalTableParser
class SwmanagerFwUpdateStrategyShowOutput:
"""
Parses the output of the 'sw-manager fw-update-strategy show' command into a SwmanagerFwUpdateStrategyObject instance.
"""
def __init__(self, swmanager_fw_update: str) -> None:
"""
Initializes SwmanagerFwUpdateStrategyObject.
Args:
swmanager_fw_update (str): Output of the 'sw-manager fw-update-strategy show' command.
Raises:
KeywordException: If the output format is invalid.
"""
swmanager_vertical_table_parser = SwManagerVerticalTableParser(swmanager_fw_update)
output_values = swmanager_vertical_table_parser.get_output_values_dict()
if self.is_valid_output(output_values):
self.swmanager_fw_update_strategy = SwmanagerFwUpdateStrategyObject()
strat_fw_update = output_values["Strategy Firmware Update Strategy"]
self.swmanager_fw_update_strategy.set_strategy_uuid(strat_fw_update["strategy-uuid"])
self.swmanager_fw_update_strategy.set_controller_apply_type(strat_fw_update["controller-apply-type"])
self.swmanager_fw_update_strategy.set_storage_apply_type(strat_fw_update["storage-apply-type"])
self.swmanager_fw_update_strategy.set_worker_apply_type(strat_fw_update["worker-apply-type"])
self.swmanager_fw_update_strategy.set_default_instance_action(strat_fw_update["default-instance-action"])
self.swmanager_fw_update_strategy.set_alarm_restrictions(strat_fw_update["alarm-restrictions"])
self.swmanager_fw_update_strategy.set_current_phase(strat_fw_update["current-phase"])
self.swmanager_fw_update_strategy.set_current_stage(strat_fw_update["current-stage"])
self.swmanager_fw_update_strategy.set_current_step(strat_fw_update["current-step"])
self.swmanager_fw_update_strategy.set_current_phase_completion(strat_fw_update["current-phase-completion"])
self.swmanager_fw_update_strategy.set_state(strat_fw_update["state"])
self.swmanager_fw_update_strategy.set_build_result(strat_fw_update["build-result"])
self.swmanager_fw_update_strategy.set_build_reason(strat_fw_update["build-reason"])
else:
raise KeywordException(f"The output line {output_values} was not valid")
def get_swmanager_fw_update_strategy_show(self) -> SwmanagerFwUpdateStrategyObject:
"""
Retrieves the parsed sw-manager fw-update-strategy show object.
Returns:
SwmanagerFwUpdateStrategyObject: The parsed sw-manager fw-update-strategy show object.
"""
return self.swmanager_fw_update_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-uuid", "controller-apply-type", "storage-apply-type", "worker-apply-type", "default-instance-action", "alarm-restrictions", "current-phase", "current-stage", "current-step", "current-phase-completion", "state", "build-result", "build-reason"]
if "Strategy Firmware Update Strategy" not in value:
get_logger().log_error("Strategy Firmware Update Strategy is not in the output value")
return False
for field in required_fields:
if field not in value["Strategy Firmware Update Strategy"]:
get_logger().log_error(f"{field} is not in the output value")
return False
return True

31
keywords/cloud_platform/swmanager/swmanager_fw_update_strategy_keywords.py Normal file
View File

@@ -0,0 +1,31 @@
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.swmanager.objects.swmanager_fw_update_strategy_show_output import SwmanagerFwUpdateStrategyShowOutput
class SwmanagerFwUpdateStrategyKeywords(BaseKeyword):
"""
This class contains all the keywords related to the 'sw-manager fw-update-strategy' commands.
"""
def __init__(self, ssh_connection: SSHConnection) -> None:
"""
Initializes SwmanagerFwUpdateStrategyKeywords.
Args:
ssh_connection (SSHConnection): The SSH connection object used for executing commands.
"""
self.ssh_connection = ssh_connection
def get_swmanager_fw_update_strategy_show(self) -> SwmanagerFwUpdateStrategyShowOutput:
"""
Gets the sw-manager fw-update-strategy show.
Returns:
SwmanagerFwUpdateStrategyShowOutput: An object containing details of the fw_update_strategy.
"""
command = source_openrc("sw-manager fw-update-strategy show")
output = self.ssh_connection.send(command)
self.validate_success_return_code(self.ssh_connection)
return SwmanagerFwUpdateStrategyShowOutput(output)

44
keywords/cloud_platform/swmanager/swmanager_vertical_table_parser.py Normal file
View File

@@ -0,0 +1,44 @@
import yaml
class SwManagerVerticalTableParser:
"""
Class for SwManager vertical table parsing.
In a vertical table, attributes and their corresponding values are arranged in
structured text, with each attribute listed alongside its value.
As an example of a typical argument for the constructor of this class,
consider the output of the command 'sw-manager fw-update-strategy show':
```
Strategy Firmware Update Strategy:
strategy-uuid: 52bf402f-54ab-4fee-8155-4e0b23bd3463
controller-apply-type: ignore
storage-apply-type: ignore
worker-apply-type: serial
default-instance-action: stop-start
alarm-restrictions: strict
current-phase: build
current-phase-completion: 100%
state: build-failed
build-result: failed
build-reason: alarms ['100.104'] from platform are present
```
"""
def __init__(self, swmanager_output):
self.swmanager_output = swmanager_output
def get_output_values_dict(self) -> dict:
"""Retrieves the output values from `self.swmanager_output` as a dictionary.
This method parses the YAML content stored in `self.swmanager_output` and returns
it as a dictionary. Each key-value pair in the resulting dictionary corresponds
to entries from the parsed YAML data.
Returns:
dict: A dictionary representation of the parsed YAML content.
"""
output_values_dict = yaml.safe_load("".join(self.swmanager_output))
return output_values_dict