starlingx/test
Code Issues Proposed changes

Refactor USMConfig and update default.json5 schema

Browse Source 
- Replace  boolean flags with usm_operation_type = "upgrade"
  or "patch".
- Add getters and setters for all configuration fields.
- Introduce validate_config() to explicitly finalize and check
  config validity.
- Update default.json5 structure with clean field layout and
  updated examples:
  - Add copy_from_remote, remote_server, and upload timing fields
  - Clarify usage of ISO, SIG, patch_path, and patch_dir
  - Replace to_release_ids string with list of strings
  - Ensure comments match current upgrade and patching logic

Change-Id: Ie16ac3caf4aad8a24a98617eb278e6a6f1f7e2ae
Signed-off-by: Andrew Vaillancourt <andrew.vaillancourt@windriver.com>
This commit is contained in:
Andrew Vaillancourt
2025-05-01 16:15:49 -04:00
parent cb60a4ecba
commit 038cf6cfb8
4 changed files with 316 additions and 136 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
config/configuration_file_locations_manager.py
View File

@@ -237,7 +237,7 @@ class ConfigurationFileLocationsManager:
action="store",
type="str",
dest="usm_config_file",
help="The Usm config file",
help="The USM config file",
)
parser.add_option(

14
config/configuration_manager.py
View File

@@ -9,7 +9,7 @@ from config.logger.objects.logger_config import LoggerConfig
from config.ptp.objects.ptp_config import PTPConfig
from config.rest_api.objects.rest_api_config import RestAPIConfig
from config.security.objects.security_config import SecurityConfig
from config.usm.objects.usm_config import UsmConfig
from config.usm.objects.usm_config import USMConfig
from config.web.objects.web_config import WebConfig
from framework.resources.resource_finder import get_stx_resource_path
@@ -31,7 +31,7 @@ class ConfigurationManagerClass:
self.database_config: DatabaseConfig = None
self.rest_api_config: RestAPIConfig = None
self.security_config: SecurityConfig = None
self.usm_config: UsmConfig = None
self.usm_config: USMConfig = None
self.app_config: AppConfig = None
self.configuration_locations_manager = None
@@ -120,7 +120,7 @@ class ConfigurationManagerClass:
self.database_config = DatabaseConfig(database_config_file)
self.rest_api_config = RestAPIConfig(rest_api_config_file)
self.security_config = SecurityConfig(security_config_file)
self.usm_config = UsmConfig(usm_config_file)
self.usm_config = USMConfig(usm_config_file)
self.app_config = AppConfig(app_config_file)
self.loaded = True
except FileNotFoundError as e:
@@ -227,12 +227,12 @@ class ConfigurationManagerClass:
"""
return self.security_config
def get_usm_config(self) -> UsmConfig:
def get_usm_config(self) -> USMConfig:
"""
Getter for usm config
Returns:
UsmConfig: the usm config
USMConfig: the usm config
"""
return self.usm_config
@@ -246,11 +246,11 @@ class ConfigurationManagerClass:
"""
return self.app_config
def get_config_pytest_args(self) -> [str]:
def get_config_pytest_args(self) -> list[str]:
"""
Returns the configuration file locations as pytest args.
Returns ([str]):
Returns (list[str]): A list of strings representing pytest arguments for configuration file locations.
"""
pytest_config_args = []

86
config/usm/files/default.json5
View File

@@ -1,44 +1,64 @@
{
// Indicates the current operation is for Major rel upgrade
"usm_is_upgrade": true,
// Type of operation: "upgrade" or "patch"
"usm_operation_type": "upgrade",
// Indicates the current operation is patching the system.
"usm_is_patch": "",
// Whether a reboot is required after the operation
"requires_reboot": false,
// Absolute path to ISO for Major rel upgrade
"iso_path": "/path/to/iso/file",
// If true, copy files from a remote build server to the controller
"copy_from_remote": false,
// Absolute path to the Signature path for Major rel upgrade.
"upgrade_license": "/path/to/lic/file/stx.sig",
// ISO and SIG source paths
// Required if usm_operation_type = "upgrade"
// If copy_from_remote=true, these must be full remote paths
// If copy_from_remote=false, these are assumed to already exist on the controller
"iso_path": "/home/sysadmin/starlingx-10.0.0.iso",
"sig_path": "/home/sysadmin/starlingx-10.0.0.sig",
// Absolute path to the .patch file for Patching.
"patch_path": "/home/sysadmin",
// Patch file paths
// Required if usm_operation_type = "patch"
// Exactly one of the following must be set:
// - patch_path: to upload a single patch file
// - patch_dir: to upload a directory containing multiple .patch files
// Leave both empty for upgrade operations
//
// Examples:
// "patch_path": "/home/sysadmin/starlingx-10.0.0.patch"
// "patch_dir": "/home/sysadmin/patches/"
"patch_path": "",
"patch_dir": "",
// Path of directory where one or more .patch files are located.
"patch_dir": "/home/sysadmin/patch_dir",
// Destination directory on the controller where files will be copied
"dest_dir": "/home/sysadmin/upload_test/",
// Directory on the lab where the files need to be stored.
"dest_dir": "/home/sysadmin/upload_test/",
// Expected release IDs to validate a successful software upload or patching.
//
// This should always be a list of one or more version strings.
// In most upgrade workflows, this will contain a single release ID.
// For example:
// "to_release_ids": ["starlingx-10.0.0"]
//
// In some patch workflows, especially when applying multiple patches,
// this list can contain multiple target release IDs expected after upload.
// For example:
// "to_release_ids": ["starlingx-10.0.0", "starlingx-10.0.1"]
"to_release_ids": ["starlingx-10.0.0"],
// Address of the server where the iso/patch files are located.
"build_server": "host-name-for-build-server",
// Remote build server credentials (only used if copy_from_remote = true)
"remote_server": "",
"remote_server_username": "",
"remote_server_password": "",
// Username to authenticate to build_server
"build_server_username": "build_server_username",
// Optional CLI arguments passed to the 'software upload' or 'software upgrade' command
// Example: "--force"
"upgrade_arguments": "",
// Password to authenticate to build_server
"build_server_password": "build_server_password",
// Extra attributes for future workflows (e.g., patch staging, etc.)
"extra_attributes": {
// Example: "staged": "true"
},
// Arguments for upgrade/patching testing
"upgrade_arguments": "",
// Release ids (one or more) to which the system is upgraded.
// Used for either multiple patches or major release + patch(es)
"to_release_ids": "stx-major.minor.version",
// Some extra attributes for patching USM, config
"extra_attributes": {
"": "",
},
}
// Polling configuration for checking upload success
"upload_poll_interval_sec": 30,
"upload_timeout_sec": 1800
}

350
config/usm/objects/usm_config.py
View File

@@ -1,172 +1,332 @@
import json5
class UsmConfig:
class USMConfig:
"""
Class to hold configuration for USM upgrade/patch tests.
Encapsulates configuration for USM upgrade and patch operations.
Call `validate_config()` after setting fields to ensure configuration consistency.
"""
def __init__(self, config):
def __init__(self, config_path: str):
"""
Load and parse the USM config file and validate contents.
Args:
config_path (str): Path to the JSON5 USM config file.
Raises:
FileNotFoundError: If the config file cannot be found.
ValueError: If any required field is invalid.
"""
try:
json_data = open(config)
with open(config_path) as f:
usm_dict = json5.load(f)
except FileNotFoundError:
print(f"Could not find the USM config file: {config}")
raise
raise FileNotFoundError(f"Could not find USM config file: {config_path}")
usm_dict = json5.load(json_data)
self.usm_is_upgrade = usm_dict.get("usm_is_upgrade", None)
self.usm_is_patch = usm_dict.get("usm_is_patch", None)
self.iso_path = usm_dict.get("iso_path", None)
self.upgrade_license = usm_dict.get("upgrade_license", None)
self.patch_path = usm_dict.get("patch_path", None)
self.patch_dir = usm_dict.get("patch_dir", None)
self.dest_dir = usm_dict.get("dest_dir", None)
self.build_server = usm_dict.get("build_server", None)
self.build_server_username = usm_dict.get("build_server_username", None)
self.build_server_password = usm_dict.get("build_server_password", None)
self.upgrade_arguments = usm_dict.get("upgrade_arguments", None)
self.to_release_ids = usm_dict.get("to_release_ids", None)
self.extra_attributes = usm_dict.get("extra_attributes", {})
self._usm_operation_type = usm_dict.get("usm_operation_type")
self._requires_reboot = usm_dict.get("requires_reboot")
self._copy_from_remote = usm_dict.get("copy_from_remote")
self._iso_path = usm_dict.get("iso_path")
self._sig_path = usm_dict.get("sig_path")
self._patch_path = usm_dict.get("patch_path")
self._patch_dir = usm_dict.get("patch_dir")
self._dest_dir = usm_dict.get("dest_dir")
self._to_release_ids = usm_dict.get("to_release_ids")
self._remote_server = usm_dict.get("remote_server")
self._remote_server_username = usm_dict.get("remote_server_username")
self._remote_server_password = usm_dict.get("remote_server_password")
self._upgrade_arguments = usm_dict.get("upgrade_arguments")
self._extra_attributes = usm_dict.get("extra_attributes")
self._upload_poll_interval_sec = usm_dict.get("upload_poll_interval_sec")
self._upload_timeout_sec = usm_dict.get("upload_timeout_sec")
def get_usm_is_upgrade(self) -> str:
self.validate_config()
def validate_config(self) -> None:
"""
Getter for the usm_is_upgrade
Validate config values for logical consistency.
Indicates the current operation is for Major rel upgrade
This includes:
- Checking operation type is either 'upgrade' or 'patch'.
- Ensuring expected release IDs are present.
- Validating required fields for remote copy.
- Confirming ISO/SIG or patch fields based on operation type.
Raises:
ValueError: If any config field is missing or inconsistent.
"""
if self._usm_operation_type not in ("upgrade", "patch"):
raise ValueError("Invalid usm_operation_type: must be 'upgrade' or 'patch'")
if not isinstance(self._to_release_ids, list) or not self._to_release_ids:
raise ValueError("to_release_ids must be a non-empty list")
if self._copy_from_remote:
if not (self._remote_server and self._remote_server_username and self._remote_server_password):
raise ValueError("Remote server credentials required when copy_from_remote is true")
if self._usm_operation_type == "upgrade":
if not self._iso_path or not self._sig_path:
raise ValueError("Upgrade requires source_iso_path and source_sig_path")
if self._usm_operation_type == "patch":
if not self._patch_path and not self._patch_dir:
raise ValueError("Patch requires either patch_path or patch_dir")
def get_usm_operation_type(self) -> str:
"""Get the USM operation type.
Returns:
str: the usm_is_upgrade
str: Either "upgrade" or "patch".
"""
return self.usm_is_upgrade
return self._usm_operation_type
def get_usm_is_patch(self) -> str:
def set_usm_operation_type(self, value: str) -> None:
"""Set the USM operation type.
Args:
value (str): Either "upgrade" or "patch".
"""
Getter for the usm_is_patch
self._usm_operation_type = value
Indicates the current operation is patching the system.
def get_requires_reboot(self) -> bool:
"""Get whether a reboot is required after operation.
Returns:
str: the usm_is_patch
bool: True if a reboot is required.
"""
return self.usm_is_patch
return self._requires_reboot
def set_requires_reboot(self, value: bool) -> None:
"""Set whether a reboot is required after operation.
Args:
value (bool): True if reboot is required.
"""
self._requires_reboot = value
def get_copy_from_remote(self) -> bool:
"""Check if files should be copied from a remote server.
Returns:
bool: True if ISO/SIG or patch files should be pulled from a remote build server.
"""
return self._copy_from_remote
def set_copy_from_remote(self, value: bool) -> None:
"""Specify whether to copy files from a remote build server.
Args:
value (bool): True to copy files from remote, False if they already exist on the controller.
"""
self._copy_from_remote = value
def get_iso_path(self) -> str:
"""
Getter for the iso_path
Absolute path to ISO for Major rel upgrade
"""Get the path to the ISO file.
Returns:
str: the iso_path
str: Absolute path to the ISO file for upgrade.
"""
return self.iso_path
return self._iso_path
def get_upgrade_license(self) -> str:
def set_iso_path(self, value: str) -> None:
"""Set the path to the ISO file.
Args:
value (str): Absolute path to the ISO file.
"""
Getter for the upgrade_license
self._iso_path = value
Absolute path to the Signature path for Major rel upgrade.
def get_sig_path(self) -> str:
"""Get the path to the signature file.
Returns:
str: the upgrade_license
str: Absolute path to the SIG file.
"""
return self.upgrade_license
return self._sig_path
def set_sig_path(self, value: str) -> None:
"""Set the path to the signature file.
Args:
value (str): Absolute path to the SIG file.
"""
self._sig_path = value
def get_patch_path(self) -> str:
"""
Getter for the patch_path
Absolute path to the .patch file for Patching.
"""Get the path to a single patch file.
Returns:
str: the patch_path
str: Absolute path to a single .patch file.
"""
return self.patch_path
return self._patch_path
def set_patch_path(self, value: str) -> None:
"""Set the path to a single patch file.
Args:
value (str): Absolute path to a single .patch file.
"""
self._patch_path = value
def get_patch_dir(self) -> str:
"""
Getter for the patch_dir
Path of directory where one or more .patch files are located.
"""Get the path to a patch directory.
Returns:
str: the patch_dir
str: Directory containing multiple .patch files.
"""
return self.patch_dir
return self._patch_dir
def set_patch_dir(self, value: str) -> None:
"""Set the path to a patch directory.
Args:
value (str): Directory containing multiple .patch files.
"""
self._patch_dir = value
def get_dest_dir(self) -> str:
"""
Getter for the dest_dir
Directory on the lab where the files need to be stored.
"""Get the destination directory on the controller.
Returns:
str: the dest_dir
str: Directory where ISO/SIG or patch files will be copied.
"""
return self.dest_dir
return self._dest_dir
def get_build_server(self) -> str:
def set_dest_dir(self, value: str) -> None:
"""Set the destination directory on the controller.
Args:
value (str): Path on controller where files will be copied.
"""
Getter for the build_server
self._dest_dir = value
Address of the server where the iso/patch files are located.
def get_to_release_ids(self) -> list[str]:
"""Get the expected release IDs.
Returns:
str: the build_server
list[str]: List of release versions used to validate success.
"""
return self.build_server
return self._to_release_ids
def get_build_server_username(self) -> str:
def set_to_release_ids(self, value: list[str]) -> None:
"""Set the expected release IDs.
Args:
value (list[str]): One or more release version strings.
"""
Getter for the build_server_username
self._to_release_ids = value
Username to authenticate to build_server
def get_remote_server(self) -> str:
"""Get the remote server address.
Returns:
str: the build_server_username
str: Hostname or IP of the remote server.
"""
return self.build_server_username
return self._remote_server
def get_build_server_password(self) -> str:
def set_remote_server(self, value: str) -> None:
"""Set the remote server address.
Args:
value (str): Hostname or IP of the remote server.
"""
Getter for the build_server_password
self._remote_server = value
Password to authenticate to build_server
def get_remote_server_username(self) -> str:
"""Get the remote server username.
Returns:
str: the build_server_password
str: Username for authenticating with the remote server.
"""
return self.build_server_password
return self._remote_server_username
def set_remote_server_username(self, value: str) -> None:
"""Set the remote server username.
Args:
value (str): Username for authenticating with the remote server.
"""
self._remote_server_username = value
def get_remote_server_password(self) -> str:
"""Get the remote server password.
Returns:
str: Password for authenticating with the remote server.
"""
return self._remote_server_password
def set_remote_server_password(self, value: str) -> None:
"""Set the remote server password.
Args:
value (str): Password for authenticating with the remote server.
"""
self._remote_server_password = value
def get_upgrade_arguments(self) -> str:
"""
Getter for the upgrade_arguments
Arguments for upgrade/patching testing
"""Get optional CLI arguments for upload or upgrade.
Returns:
str: the upgrade_arguments
str: Extra CLI flags like "--force".
"""
return self.upgrade_arguments
return self._upgrade_arguments
def get_to_release_ids(self) -> str:
def set_upgrade_arguments(self, value: str) -> None:
"""Set optional CLI arguments for upload or upgrade.
Args:
value (str): Extra CLI flags like "--force".
"""
Getter for the to_release_ids
Release ids (one or more) to which the system is upgraded. Used for either multiple patches or major release + patch(es)
Returns:
str: the to_release_ids
"""
return self.to_release_ids
self._upgrade_arguments = value
def get_extra_attributes(self) -> dict:
"""
Getter for the extra_attributes
Some extra attributes for patching USM, config
"""Get extra user-defined attributes.
Returns:
dict: the extra_attributes
dict: Arbitrary key-value pairs used in future workflows.
"""
return self.extra_attributes
return self._extra_attributes
def set_extra_attributes(self, value: dict) -> None:
"""Set extra user-defined attributes.
Args:
value (dict): Arbitrary key-value pairs for workflows like patch staging.
"""
self._extra_attributes = value
def get_upload_poll_interval_sec(self) -> int:
"""Get polling interval for upload progress.
Returns:
int: Number of seconds between upload status checks.
"""
return self._upload_poll_interval_sec
def set_upload_poll_interval_sec(self, value: int) -> None:
"""Set polling interval for upload progress.
Args:
value (int): Number of seconds between upload status checks.
"""
self._upload_poll_interval_sec = value
def get_upload_timeout_sec(self) -> int:
"""Get timeout duration for upload completion.
Returns:
int: Maximum seconds to wait for upload to complete.
"""
return self._upload_timeout_sec
def set_upload_timeout_sec(self, value: int) -> None:
"""Set timeout duration for upload completion.
Args:
value (int): Maximum seconds to wait for upload to complete.
"""
self._upload_timeout_sec = value