diff --git a/config/configuration_file_locations_manager.py b/config/configuration_file_locations_manager.py index 9372fdbe..9260e66a 100644 --- a/config/configuration_file_locations_manager.py +++ b/config/configuration_file_locations_manager.py @@ -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( diff --git a/config/configuration_manager.py b/config/configuration_manager.py index 8dcc236e..aa8ebf26 100644 --- a/config/configuration_manager.py +++ b/config/configuration_manager.py @@ -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 = [] diff --git a/config/usm/files/default.json5 b/config/usm/files/default.json5 index 5d4c40b4..a62d344b 100644 --- a/config/usm/files/default.json5 +++ b/config/usm/files/default.json5 @@ -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": { - "": "", - }, - -} \ No newline at end of file + // Polling configuration for checking upload success + "upload_poll_interval_sec": 30, + "upload_timeout_sec": 1800 +} diff --git a/config/usm/objects/usm_config.py b/config/usm/objects/usm_config.py index 78aba5f9..5a5be396 100644 --- a/config/usm/objects/usm_config.py +++ b/config/usm/objects/usm_config.py @@ -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