starlingx/test
Code Issues Proposed changes

Adding Keywords

Browse Source 
Added keywords for
    - sw-manager sw-deploy-strategy
    - software upload

Change-Id: Ic87ce7fa13762e140ce816371541a5a43d5c21fd
Signed-off-by: Abhishek jaiswal <abhishek.jaiswal@windriver.com>
This commit is contained in:
Abhishek jaiswal
2025-06-06 06:56:19 -04:00
parent 21bcdb0618
commit 8821b92369
7 changed files with 371 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
keywords/cloud_platform/dcmanager/dcmanager_prestage_strategy_keywords.py
View File

@@ -9,7 +9,7 @@ class DcmanagerPrestageStrategyKeywords(BaseKeyword):
This class contains all the keywords related to the 'dcmanager prestage-strategy' commands.
"""
def __init__(self, ssh_connection: SSHConnection) -> None:
def __init__(self, ssh_connection: SSHConnection) -> str:
"""
Initializes DcmanagerPrestageStrategyKeywords.
@@ -18,6 +18,20 @@ class DcmanagerPrestageStrategyKeywords(BaseKeyword):
"""
self.ssh_connection = ssh_connection
def get_dcmanager_prestage_strategy_abort(self) -> None:
"""Gets the prestage-strategy abort."""
command = source_openrc("dcmanager prestage-strategy abort")
output = self.ssh_connection.send(command)
self.validate_success_return_code(self.ssh_connection)
return output
def get_dcmanager_prestage_strategy_apply(self) -> None:
"""Gets the prestage-strategy apply."""
command = source_openrc("dcmanager prestage-strategy apply")
output = self.ssh_connection.send(command)
self.validate_success_return_code(self.ssh_connection)
return output
def get_dcmanager_prestage_strategy_create(self) -> DcmanagerPrestageStrategyShowOutput:
"""
Gets the prestage-strategy create.

127
keywords/cloud_platform/swmanager/objects/swmanager_sw_deploy_strategy_object.py Normal file
View File

@@ -0,0 +1,127 @@
from typing import Optional
class SwManagerSwDeployStrategyObject:
"""
This class represents a sw-manager sw-deploy-strategy as an object.
"""
def __init__(self) -> None:
"""Initializes a SwmanagerFwUpdateStrategyObject with default values."""
self.strategy_uuid: Optional[str] = None
self.release_id: 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
def set_strategy_uuid(self, strategy_uuid: str) -> None:
"""Sets the strategy_uuid of the sw-deploy-strategy."""
self.strategy_uuid = strategy_uuid
def get_strategy_uuid(self) -> Optional[str]:
"""Gets the strategy_uuid of the sw-deploy-strategy."""
return self.strategy_uuid
def set_release_id(self, release_id: str) -> None:
"""Sets the release_id of the sw-deploy-strategy."""
self.release_id = release_id
def get_release_id(self) -> Optional[str]:
"""Gets the release_id of the sw-deploy-strategy."""
return self.release_id
def set_controller_apply_type(self, controller_apply_type: str) -> None:
"""Sets the controller_apply_type of the sw-deploy-strategy."""
self.controller_apply_type = controller_apply_type
def get_controller_apply_type(self) -> Optional[str]:
"""Gets the controller_apply_type of the sw-deploy-strategy."""
return self.controller_apply_type
def set_storage_apply_type(self, storage_apply_type: str) -> None:
"""Sets the storage_apply_type of the sw-deploy-strategy."""
self.storage_apply_type = storage_apply_type
def get_storage_apply_type(self) -> Optional[str]:
"""Gets the storage_apply_type of the sw-deploy-strategy."""
return self.storage_apply_type
def set_worker_apply_type(self, worker_apply_type: str) -> None:
"""Sets the worker_apply_type of the sw-deploy-strategy."""
self.worker_apply_type = worker_apply_type
def get_worker_apply_type(self) -> Optional[str]:
"""Gets the worker_apply_type of the sw-deploy-strategy."""
return self.worker_apply_type
def set_default_instance_action(self, default_instance_action: str) -> None:
"""Sets the default_instance_action of the sw-deploy-strategy."""
self.default_instance_action = default_instance_action
def get_default_instance_action(self) -> Optional[str]:
"""Gets the default_instance_action of the sw-deploy-strategy."""
return self.default_instance_action
def set_alarm_restrictions(self, alarm_restrictions: str) -> None:
"""Sets the alarm_restrictions of the sw-deploy-strategy."""
self.alarm_restrictions = alarm_restrictions
def get_alarm_restrictions(self) -> Optional[str]:
"""Gets the alarm_restrictions of the sw-deploy-strategy."""
return self.alarm_restrictions
def set_current_phase(self, current_phase: str) -> None:
"""Sets the current_phase of the sw-deploy-strategy."""
self.current_phase = current_phase
def get_current_phase(self) -> Optional[str]:
"""Gets the current_phase of the sw-deploy-strategy."""
return self.current_phase
def set_current_stage(self, current_stage: str) -> None:
"""Sets the current_stage of the sw-deploy-strategy."""
self.current_stage = current_stage
def get_current_stage(self) -> Optional[str]:
"""Gets the current_stage of the sw-deploy-strategy."""
return self.current_stage
def set_current_step(self, current_step: str) -> None:
"""Sets the current_step of the sw-deploy-strategy."""
self.current_step = current_step
def get_current_step(self) -> Optional[str]:
"""Gets the current_step of the sw-deploy-strategy."""
return self.current_step
def set_current_phase_completion(self, current_phase_completion: str) -> None:
"""Sets the current_phase_completion of the sw-deploy-strategy."""
self.current_phase_completion = current_phase_completion
def get_current_phase_completion(self) -> Optional[str]:
"""Gets the current_phase_completion of the sw-deploy-strategy."""
return self.current_phase_completion
def set_state(self, state: str) -> None:
"""Sets the state of the sw-deploy-strategy."""
self.state = state
def get_state(self) -> Optional[str]:
"""Gets the state of the sw-deploy-strategy."""
return self.state
def set_inprogress(self, inprogress: str) -> None:
"""Sets the inprogress of the sw-deploy-strategy."""
self.inprogress = inprogress
def get_inprogress(self) -> Optional[str]:
"""Gets the inprogress of the sw-deploy-strategy."""
return self.inprogress

74
keywords/cloud_platform/swmanager/objects/swmanager_sw_deploy_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_sw_deploy_strategy_object import SwManagerSwDeployStrategyObject
from keywords.cloud_platform.swmanager.swmanager_vertical_table_parser import SwManagerVerticalTableParser
class SwManagerSwDeployStrategyShowOutput:
"""
Parses the output of the 'sw-manager sw-deploy-strategy show' command into a SwManagerSwDeployStrategyObject instance.
"""
def __init__(self, swmanager_sw_deploy: str) -> None:
"""
Initializes SwManagerSwDeployStrategyObject.
Args:
swmanager_sw_deploy (str): Output of the 'sw-manager sw-deploy-strategy show' command.
Raises:
KeywordException: If the output format is invalid.
"""
swmanager_vertical_table_parser = SwManagerVerticalTableParser(swmanager_sw_deploy)
output_values = swmanager_vertical_table_parser.get_output_values_dict()
if self.is_valid_output(output_values):
self.swmanager_sw_deploy_strategy = SwManagerSwDeployStrategyObject()
sw_deploy_strat = output_values["Strategy Software Deploy Strategy"]
self.swmanager_sw_deploy_strategy.set_strategy_uuid(sw_deploy_strat["strategy-uuid"])
self.swmanager_sw_deploy_strategy.set_release_id(sw_deploy_strat["release-id"])
self.swmanager_sw_deploy_strategy.set_controller_apply_type(sw_deploy_strat["controller-apply-type"])
self.swmanager_sw_deploy_strategy.set_storage_apply_type(sw_deploy_strat["storage-apply-type"])
self.swmanager_sw_deploy_strategy.set_worker_apply_type(sw_deploy_strat["worker-apply-type"])
self.swmanager_sw_deploy_strategy.set_default_instance_action(sw_deploy_strat["default-instance-action"])
self.swmanager_sw_deploy_strategy.set_alarm_restrictions(sw_deploy_strat["alarm-restrictions"])
self.swmanager_sw_deploy_strategy.set_current_phase(sw_deploy_strat["current-phase"])
self.swmanager_sw_deploy_strategy.set_current_stage(sw_deploy_strat["current-stage"])
self.swmanager_sw_deploy_strategy.set_current_step(sw_deploy_strat["current-step"])
self.swmanager_sw_deploy_strategy.set_current_phase_completion(sw_deploy_strat["current-phase-completion"])
self.swmanager_sw_deploy_strategy.set_state(sw_deploy_strat["state"])
self.swmanager_sw_deploy_strategy.set_inprogress(sw_deploy_strat["inprogress"])
else:
raise KeywordException(f"The output line {output_values} was not valid")
def get_swmanager_sw_deploy_strategy_show(self) -> SwManagerSwDeployStrategyObject:
"""
Retrieves the parsed sw-manager sw-deploy-strategy show object.
Returns:
SwManagerSwDeployStrategyObject: The parsed sw-manager sw-deploy-strategy show object.
"""
return self.swmanager_sw_deploy_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", "release-id", "controller-apply-type", "storage-apply-type", "worker-apply-type", "default-instance-action", "alarm-restrictions", "current-phase", "current-stage", "current-step", "current-phase-completion", "state", "inprogress"]
if "Strategy Software Deploy Strategy" not in value:
get_logger().log_error("Strategy Software Deploy Strategy is not in the output value")
return False
for field in required_fields:
if field not in value["Strategy Software Deploy Strategy"]:
get_logger().log_error(f"{field} is not in the output value")
return False
return True

72
keywords/cloud_platform/swmanager/swmanager_sw_deploy_strategy_keywords.py Normal file
View File

@@ -0,0 +1,72 @@
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_sw_deploy_strategy_show_output import SwManagerSwDeployStrategyShowOutput
class SwmanagerSwDeployStrategyKeywords(BaseKeyword):
"""
This class contains all the keywords related to the 'sw-manager sw-deploy-strategy' commands.
"""
def __init__(self, ssh_connection: SSHConnection) -> None:
"""Initializes SwmanagerSwDeployStrategyKeywords.
Args:
ssh_connection (SSHConnection): The SSH connection object used for executing commands.
"""
self.ssh_connection = ssh_connection
def get_sw_deploy_strategy_abort(self):
"""Gets the sw-deploy-strategy abort."""
command = source_openrc("sw-manager sw-deploy-strategy abort")
self.ssh_connection.send(command)
self.validate_success_return_code(self.ssh_connection)
def get_sw_deploy_strategy_apply(self):
"""Gets the sw-deploy-strategy apply."""
command = source_openrc("sw-manager sw-deploy-strategy apply")
self.ssh_connection.send(command)
self.validate_success_return_code(self.ssh_connection)
def get_sw_deploy_strategy_create(self, release: str, delete: bool) -> SwManagerSwDeployStrategyShowOutput:
"""Gets the sw-deploy-strategy create.
Args:
release (str): The release to be used for the strategy.
delete (bool): If True, pass --delete as an argument to the command.
Returns:
SwManagerSwDeployStrategyShowOutput: The output of the prestage strategy.
"""
if delete:
delete_arg = "--delete"
else:
delete_arg = ""
command = source_openrc(f"sw-manager sw-deploy-strategy create {delete_arg} {release}")
output = self.ssh_connection.send(command)
self.validate_success_return_code(self.ssh_connection)
return SwManagerSwDeployStrategyShowOutput(output)
def get_sw_deploy_strategy_show(self) -> SwManagerSwDeployStrategyShowOutput:
"""Gets the sw-deploy-strategy show.
Returns:
SwManagerSwDeployStrategyShowOutput: An object containing details of the prestage strategy .
"""
command = source_openrc("sw-manager sw-deploy-strategy show")
output = self.ssh_connection.send(command)
self.validate_success_return_code(self.ssh_connection)
return SwManagerSwDeployStrategyShowOutput(output)
def get_sw_deploy_strategy_delete(self) -> None:
"""Gets the sw-deploy-strategy delete.
Returns:
None: None.
"""
command = source_openrc("sw-manager sw-deploy-strategy delete")
output = self.ssh_connection.send(command)
self.validate_success_return_code(self.ssh_connection)
return output

23
keywords/cloud_platform/upgrade/objects/software_upload_object.py Normal file
View File

@@ -0,0 +1,23 @@
class SoftwareUploadObject:
"""Class to hold attributes of a software upload as returned by software upload command"""
def __init__(self, uploaded_file: str, release: str) -> None:
"""Initializes a SoftwareUploadObject with the uploaded file and release."""
self.uploaded_file = uploaded_file
self.release = release
def get_uploaded_file(self) -> str:
"""Getter for uploaded_file.
Returns:
str: the uploaded file name
"""
return self.uploaded_file
def get_release(self) -> str:
"""Getter for release.
Returns:
str: the release name
"""
return self.release

54
keywords/cloud_platform/upgrade/objects/software_upload_output.py Normal file
View File

@@ -0,0 +1,54 @@
"""Software Upload Output."""
from typing import Dict
from framework.logging.automation_logger import get_logger
from keywords.cloud_platform.system.system_table_parser import SystemTableParser
from keywords.cloud_platform.upgrade.objects.software_upload_object import SoftwareUploadObject
class SoftwareUploadOutput:
"""This class parses o/p 'software show' command into an object of type SoftwareUploadObject."""
def __init__(self, software_upload_output: list[str]):
"""Instance of the class.
Args:
software_upload_output (list[str]): output of 'software show' command as a list of strings.
"""
self.software_show: SoftwareUploadObject = []
system_table_parser = SystemTableParser(software_upload_output)
self.output_values = system_table_parser.get_output_values_list()
for value in self.output_values:
software_show_object = SoftwareUploadObject(
value["Uploaded File"],
value["Release"],
)
self.software_show.append(software_show_object)
def get_software_uploaded(self) -> SoftwareUploadObject:
"""Get the uploaded data.
Returns:
SoftwareUploadObject: the list of software upload objects
"""
return self.software_show
@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 = ["Release", "Uploaded File"]
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

7
keywords/cloud_platform/upgrade/usm_keywords.py
View File

@@ -3,6 +3,7 @@ from framework.logging.automation_logger import get_logger
from framework.ssh.ssh_connection import SSHConnection
from framework.validation.validation import validate_equals_with_retry
from keywords.base_keyword import BaseKeyword
from keywords.cloud_platform.upgrade.objects.software_upload_output import SoftwareUploadOutput
from keywords.cloud_platform.upgrade.software_show_keywords import SoftwareShowKeywords
@@ -16,7 +17,7 @@ class USMKeywords(BaseKeyword):
def __init__(self, ssh_connection: SSHConnection):
self.ssh_connection = ssh_connection
def upload_patch_file(self, patch_file_path: str) -> None:
def upload_patch_file(self, patch_file_path: str) -> SoftwareUploadOutput:
"""
Upload a single patch file using 'software upload'.
@@ -25,11 +26,15 @@ class USMKeywords(BaseKeyword):
Raises:
KeywordException: On failure to upload.
Returns:
SoftwareUploadOutput: Parsed output containing details of the uploaded patch.
"""
get_logger().log_info(f"Uploading patch file: {patch_file_path}")
output = self.ssh_connection.send_as_sudo(f"software upload {patch_file_path}")
self.validate_success_return_code(self.ssh_connection)
get_logger().log_info("Upload completed:\n" + "\n".join(output))
return SoftwareUploadOutput(output)
def upload_patch_dir(self, patch_dir_path: str) -> None:
"""