Title: Add version-aware patch state verification before backup

Description:
Added patch state verification before backup operation:
- Implemented sw-patch query for builds before 24.09
- Implemented software list for builds after 24.09
- Captures patch/release states before backup is taken

Testing:
- Verified patch state retrieval for 22.12 and 25.09 versions on SX

Change-Id: I59d9ef3b170247b6725cb628911850a7cc93a11f
This commit is contained in:
rmenon
2025-06-06 11:39:29 -04:00
parent 21bcdb0618
commit 1d95be44bd
4 changed files with 88 additions and 2 deletions

View File

@@ -0,0 +1,25 @@
class SoftwareStateOutput:
"""
Class to hold attributes of a software state
"""
def __init__(self, patch_states: dict) -> None:
"""
Constructor
Args:
patch_states (dict): Dictionary containing patch states.
"""
self.patch_states = patch_states
def get_state(self, patch_id: str) -> str:
"""
Get the state for a given patch_id.
Args:
patch_id (str): The patch ID to look up.
Returns:
str: The state of the patch, or None if not found.
"""
return self.patch_states.get(patch_id)

View File

@@ -0,0 +1,52 @@
from wrcp.keywords.cloud_platform.software.patch.software_patch_keywords import SwPatchQueryKeywords
from framework.logging.automation_logger import get_logger
from framework.ssh.ssh_connection import SSHConnection
from keywords.base_keyword import BaseKeyword
from starlingx.keywords.cloud_platform.ansible_playbook.object.software_state_output import SoftwareStateOutput
from starlingx.keywords.cloud_platform.upgrade.software_list_keywords import SoftwareListKeywords
from starlingx.keywords.cloud_platform.version_info.cloud_platform_software_version import CloudPlatformSoftwareVersion
from starlingx.keywords.cloud_platform.version_info.cloud_platform_version_manager import CloudPlatformVersionManager
class SoftwareStateKeywords(BaseKeyword):
"""Query and summarise the patch / software state of a StarlingX controller."""
def __init__(self, ssh_connection: SSHConnection) -> None:
"""Initialise the helper.
Args:
ssh_connection (SSHConnection): SSH connection to the controller.
"""
self.ssh_connection = ssh_connection
self.patch_keywords = SwPatchQueryKeywords(ssh_connection)
self.software_list_keywords = SoftwareListKeywords(ssh_connection)
def get_patch_state(self) -> SoftwareStateOutput:
"""Return the current patch / software state."""
sw_version = CloudPlatformVersionManager.get_sw_version()
get_logger().log_info(f"sw_version: {sw_version}, name: {sw_version.get_name()}, id: {sw_version.get_id()}")
if sw_version.is_after_or_equal_to(CloudPlatformSoftwareVersion.STARLINGX_10_0):
patch_states = self._get_software_list_state()
else:
patch_states = self._get_sw_patch_query_state()
return SoftwareStateOutput(patch_states)
def _get_sw_patch_query_state(self) -> dict:
"""Fetch patch state by running ``sw-patch query``."""
patch_query = self.patch_keywords.get_sw_patch_query()
patches = patch_query.get_patches()
if not patches:
get_logger().log_info("No patches found")
return {}
states = {patch.patch_id: patch.state for patch in patches}
return states
def _get_software_list_state(self) -> dict:
"""Fetch patch state by running ``software list``."""
software_list_output = self.software_list_keywords.get_software_list()
states = {entry["Release"]: entry["State"] for entry in software_list_output.output_values}
return states

View File

@@ -6,7 +6,8 @@ class CloudPlatformSoftwareVersion:
This class enumerates the different versions of the Cloud Platform as well as their chronological release order. This class enumerates the different versions of the Cloud Platform as well as their chronological release order.
""" """
STARLINGX_9_0 = ProductVersion("24.03", 0) STARLINGX_8_0 = ProductVersion("22.12", 0)
STARLINGX_10_0 = ProductVersion("24.09", 1) STARLINGX_9_0 = ProductVersion("24.03", 1)
STARLINGX_10_0 = ProductVersion("24.09", 2)
# Every new version must contain STARLINGX in the variable name. # Every new version must contain STARLINGX in the variable name.

View File

@@ -4,6 +4,7 @@ from framework.logging.automation_logger import get_logger
from framework.validation.validation import validate_equals from framework.validation.validation import validate_equals
from keywords.cloud_platform.ansible_playbook.ansible_playbook_keywords import AnsiblePlaybookKeywords from keywords.cloud_platform.ansible_playbook.ansible_playbook_keywords import AnsiblePlaybookKeywords
from keywords.cloud_platform.ansible_playbook.backup_files_upload_keywords import BackUpFilesUploadKeywords from keywords.cloud_platform.ansible_playbook.backup_files_upload_keywords import BackUpFilesUploadKeywords
from keywords.cloud_platform.ansible_playbook.software_state_keywords import SoftwareStateKeywords
from keywords.cloud_platform.ssh.lab_connection_keywords import LabConnectionKeywords from keywords.cloud_platform.ssh.lab_connection_keywords import LabConnectionKeywords
from keywords.files.file_keywords import FileKeywords from keywords.files.file_keywords import FileKeywords
@@ -23,6 +24,13 @@ def test_backup():
backup_dir = "/opt/backups" backup_dir = "/opt/backups"
local_backup_folder_path = "/tmp/bnr" local_backup_folder_path = "/tmp/bnr"
ssh_connection = LabConnectionKeywords().get_active_controller_ssh() ssh_connection = LabConnectionKeywords().get_active_controller_ssh()
# Patch/Software release state check before backup
patch_keywords = SoftwareStateKeywords(ssh_connection)
patch_state_output = patch_keywords.get_patch_state()
patch_states = patch_state_output.patch_states
get_logger().log_info(f"Patch/Software release states before backup: {patch_states}")
get_logger().log_info("Delete old backup files if present in back up directory") get_logger().log_info("Delete old backup files if present in back up directory")
backup_files = FileKeywords(ssh_connection).get_files_in_dir(backup_dir) backup_files = FileKeywords(ssh_connection).get_files_in_dir(backup_dir)
for backup_file in backup_files: for backup_file in backup_files: