Added a treatment to upload

release and patch to DC systems,
adding os-region-name option to the software upload
command.

Change-Id: I1ec65afb106798f978eb17053010c0b02b4e0798
Signed-off-by: Elson Oliveira <eclaudio@windriver.com>
This commit is contained in:
croy
2025-11-14 15:54:05 -05:00
committed by Elson Claudio de Oliveira
parent dfdb0dd780
commit 46f5aa0ce9
2 changed files with 25 additions and 136 deletions

View File

@@ -8,6 +8,7 @@ from keywords.cloud_platform.command_wrappers import source_openrc
from keywords.cloud_platform.upgrade.software_show_keywords import SoftwareShowKeywords
from config.configuration_manager import ConfigurationManager
class USMKeywords(BaseKeyword):
"""
Keywords for USM software operations.
@@ -19,13 +20,14 @@ class USMKeywords(BaseKeyword):
self.ssh_connection = ssh_connection
self.usm_config = ConfigurationManager.get_usm_config()
def upload_patch_file(self, patch_file_path: str, sudo: bool = False) -> SoftwareUploadOutput:
def upload_patch_file(self, patch_file_path: str, sudo: bool = False, os_region_name: str = "" ) -> SoftwareUploadOutput:
"""
Upload a single patch file using 'software upload'.
Args:
patch_file_path (str): Absolute path to a .patch file.
sudo (bool): Option to pass the command with sudo.
os_region_name: Use Os region name option for upload if it is specified
Raises:
KeywordException: On failure to upload.
@@ -34,7 +36,10 @@ class USMKeywords(BaseKeyword):
SoftwareUploadOutput: Parsed output containing details of the uploaded patch.
"""
get_logger().log_info(f"Uploading patch file: {patch_file_path}")
base_cmd = f"software upload {patch_file_path}"
upload_option = ""
if os_region_name:
upload_option = f"--os-region-name {os_region_name}"
base_cmd = f"software {upload_option} upload {patch_file_path}"
cmd = source_openrc(base_cmd)
timeout = self.usm_config.get_upload_patch_timeout_sec()
if sudo:
@@ -45,19 +50,23 @@ class USMKeywords(BaseKeyword):
get_logger().log_info("Upload completed:\n" + "\n".join(output))
return SoftwareUploadOutput(output)
def upload_patch_dir(self, patch_dir_path: str, sudo: bool = False) -> None:
def upload_patch_dir(self, patch_dir_path: str, sudo: bool = False, os_region_name: str = "") -> None:
"""
Upload all patches in a directory using 'software upload-dir'.
Args:
patch_dir_path (str): Absolute path to a directory of .patch files.
sudo (bool): Option to pass the command with sudo.
os_region_name: OS region name option for upload if it is specified
Raises:
KeywordException: On failure to upload.
"""
get_logger().log_info(f"Uploading patch directory: {patch_dir_path}")
base_cmd = f"software upload-dir {patch_dir_path}"
upload_option = ""
if os_region_name:
upload_option = f"--os-region-name {os_region_name}"
base_cmd = f"software {upload_option} upload-dir {patch_dir_path}"
cmd = source_openrc(base_cmd)
timeout = self.usm_config.get_upload_patch_timeout_sec()
if sudo:
@@ -88,7 +97,7 @@ class USMKeywords(BaseKeyword):
self.validate_success_return_code(self.ssh_connection)
return output
def upload_release(self, iso_path: str, sig_path: str, sudo: bool = False) -> None:
def upload_release(self, iso_path: str, sig_path: str, sudo: bool = False, os_region_name: str = "") -> None:
"""
Upload a full software release using 'software upload'.
@@ -96,12 +105,16 @@ class USMKeywords(BaseKeyword):
iso_path (str): Absolute path to the .iso file.
sig_path (str): Absolute path to the corresponding .sig file.
sudo (bool): Option to pass the command with sudo.
os_region_name: Use Os region name option for upload if it is specified
Raises:
KeywordException: On failure to upload.
"""
get_logger().log_info(f"Uploading software release: ISO={iso_path}, SIG={sig_path}")
base_cmd = f"software upload {iso_path} {sig_path}"
upload_option = ""
if os_region_name:
upload_option = f"--os-region-name {os_region_name}"
base_cmd = f"software {upload_option} upload {iso_path} {sig_path}"
cmd = source_openrc(base_cmd)
timeout = self.usm_config.get_upload_release_timeout_sec()
if sudo:
@@ -111,7 +124,7 @@ class USMKeywords(BaseKeyword):
self.validate_success_return_code(self.ssh_connection)
get_logger().log_info("Release upload completed:\n" + "\n".join(output))
def upload_and_verify_patch_file(self, patch_file_path: str, expected_release_id: str, timeout: int, poll_interval: int, sudo: bool = False) -> None:
def upload_and_verify_patch_file(self, patch_file_path: str, expected_release_id: str, timeout: int, poll_interval: int, sudo: bool = False, os_region_name: str="") -> None:
"""Upload a patch and verify that it becomes available.
This method is used for USM patching operations. It uploads a `.patch` file
@@ -124,11 +137,12 @@ class USMKeywords(BaseKeyword):
timeout (int): Maximum number of seconds to wait for the release to appear.
poll_interval (int): Interval (in seconds) between poll attempts.
sudo (bool): Option to pass the command with sudo.
os_region_name: Use Os region name option for upload if it is specified
Raises:
KeywordException: If upload fails or release does not become available in time.
"""
self.upload_patch_file(patch_file_path, sudo)
self.upload_patch_file(patch_file_path, sudo, os_region_name=os_region_name)
validate_equals_with_retry(
function_to_execute=lambda: SoftwareShowKeywords(self.ssh_connection).get_release_state(expected_release_id),
@@ -138,7 +152,7 @@ class USMKeywords(BaseKeyword):
polling_sleep_time=poll_interval,
)
def upload_and_verify_release(self, iso_path: str, sig_path: str, expected_release_id: str, timeout: int, poll_interval: int, sudo: bool = False) -> None:
def upload_and_verify_release(self, iso_path: str, sig_path: str, expected_release_id: str, timeout: int, poll_interval: int, sudo: bool = False, os_region_name: str = "") -> None:
"""Upload a software release and verify that it becomes available.
This method is used for USM upgrade operations. It uploads a `.iso` and `.sig`
@@ -152,11 +166,12 @@ class USMKeywords(BaseKeyword):
timeout (int): Maximum number of seconds to wait for the release to appear.
poll_interval (int): Interval (in seconds) between poll attempts.
sudo (bool): Option to pass the command with sudo.
os_region_name (str): Region name used when upload to the DC Systems
Raises:
KeywordException: If upload fails or release does not become available in time.
"""
self.upload_release(iso_path, sig_path, sudo)
self.upload_release(iso_path, sig_path, sudo, os_region_name)
validate_equals_with_retry(
function_to_execute=lambda: SoftwareShowKeywords(self.ssh_connection).get_release_state(expected_release_id),

View File

@@ -1,126 +0,0 @@
"""
Basic USM Upload Tests (Foundation Layer)
=========================================
This module provides **foundational test coverage** for the USM (Upgrade and Software Management) system,
specifically focusing on uploading software releases and patches. These tests verify that an `.iso` + `.sig`
pair (for major upgrades) or `.patch` file (for patches) can be uploaded successfully to a StarlingX controller
and reach the "available" state.
Scope:
------
These tests are **intentionally minimal** and serve as a starting point. They do not implement full
end-to-end upgrade or patch deployment flows. Their goal is to validate:
- Configuration parsing via `UsmConfig`.
- Optional remote-to-local file copy using `rsync` via `FileKeywords`.
- Uploading files via the `software upload` or `software upload-dir` CLI.
- Polling for availability using `software show`.
Key Concepts:
-------------
- The config file is parsed using `ConfigurationManager.get_usm_config()` and provides:
- ISO/SIG/PATCH paths (`get_iso_path()`, `get_sig_path()`, `get_patch_path()`).
- Destination and expected release ID (`get_to_release_ids()`).
- Upload timeouts and polling intervals.
- If `copy_from_remote` is `True`, contributors should use
`FileKeywords(ssh_connection).copy_from_remote(remote_path, local_path, ...)`
to retrieve the necessary `.iso`, `.sig`, or `.patch` files from a remote server
before calling the upload keyword. This supports workflows where files are staged
on build hosts or CI/CD artifacts servers.
How to Extend:
--------------
This module is meant to be built upon. Contributors are encouraged to:
- Validate `software list` and `software show` parsing logic.
- Chain upload -> deploy precheck -> deploy start -> deploy complete steps for full upgrade flows.
- Cover patch rollback, deploy delete, and state recovery.
Location of Supporting Logic:
-----------------------------
- Upload Keywords: `keywords/cloud_platform/upgrade/usm_keywords.py`.
- Contains `upload_patch_file()`, `upload_release()`, and verification methods that call `software show`.
- Release State Polling: `keywords/cloud_platform/upgrade/software_show_keywords.py`.
- Wraps `software show` and extracts release state (e.g., `"available"`).
- Config Management: `config/usm/objects/usm_config.py`.
- Parses structured JSON5 configuration and validates upgrade parameters.
- Remote Copy Support: `keywords/files/file_keywords.py`.
- Implements `copy_from_remote()` for fetching `.iso`, `.sig`, or `.patch` files using `rsync`.
These tests form a solid base for contributors to validate the upload mechanism
before tackling the broader USM lifecycle.
"""
from config.configuration_manager import ConfigurationManager
from framework.logging.automation_logger import get_logger
from keywords.cloud_platform.ssh.lab_connection_keywords import LabConnectionKeywords
from keywords.cloud_platform.upgrade.usm_keywords import USMKeywords
def test_usm_upload_release_from_local():
"""
Upload a USM ISO already present on the controller and verify the upload was successful.
Assumes that:
- ISO and SIG files are already present at the expected paths.
- No remote copy is required (copy_from_remote is False).
- The release ID is known and matches the ISO being uploaded.
"""
get_logger().log_info("Starting local upload test for USM release ISO.")
ssh_connection = LabConnectionKeywords().get_active_controller_ssh()
usm_keywords = USMKeywords(ssh_connection)
usm_config = ConfigurationManager.get_usm_config()
iso_path = usm_config.get_iso_path()
sig_path = usm_config.get_sig_path()
release_id = usm_config.get_to_release_ids()[0]
timeout = usm_config.get_upload_release_timeout_sec()
poll_interval = usm_config.get_upload_poll_interval_sec()
get_logger().log_test_case_step(f"Uploading software release: ISO={iso_path}, SIG={sig_path}")
usm_keywords.upload_and_verify_release(
iso_path=iso_path,
sig_path=sig_path,
expected_release_id=release_id,
timeout=timeout,
poll_interval=poll_interval,
sudo=False
)
get_logger().log_info(f"Upload verification complete for release: {release_id}")
def test_usm_upload_patch_from_local():
"""
Upload a USM patch file already present on the controller and verify it becomes available.
Assumes that:
- The patch file is already located at the configured path.
- No remote copy is required (copy_from_remote is False).
- The expected release ID after applying the patch is known.
"""
get_logger().log_info("Starting local upload test for USM patch file.")
ssh_connection = LabConnectionKeywords().get_active_controller_ssh()
usm_keywords = USMKeywords(ssh_connection)
usm_config = ConfigurationManager.get_usm_config()
patch_file_path = usm_config.get_patch_path()
expected_release_id = usm_config.get_to_release_ids()[0]
timeout = usm_config.get_upload_patch_timeout_sec()
poll_interval = usm_config.get_upload_poll_interval_sec()
get_logger().log_test_case_step(f"Uploading patch file: {patch_file_path}")
usm_keywords.upload_and_verify_patch_file(
patch_file_path=patch_file_path,
expected_release_id=expected_release_id,
timeout=timeout,
poll_interval=poll_interval,
sudo=False
)
get_logger().log_info(f"Upload verification complete for patch release: {expected_release_id}")