starlingx/test
Code Issues Proposed changes

Add platform-integ-app rollback testcase

Browse Source 
Change-Id: I90b138bb358dc76fea5ce2b326e50ed61988623c
Signed-off-by: Rity Menon <rity.menon@windriver.com>
This commit is contained in:
Rity Menon
2025-11-10 16:03:31 -08:00
parent fed32bdded
commit c2d861c7c6
7 changed files with 317 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
config/app/files/default.json5
View File

@@ -1,5 +1,7 @@
{
base_application_path: "/usr/local/share/applications/helm/",
base_application_localhost: "fake_path",
istio_app_name: "istio",
metric_server_app_name: "metrics-server",
oidc_app_name: "oidc-auth-apps",
@@ -8,4 +10,5 @@
node_feature_discovery_app_name: "node-feature-discovery",
node_interface_metrics_exporter_app_name: "node-interface-metrics-exporter",
platform_integ_apps_app_name: "platform-integ-apps"
}

13
config/app/objects/app_config.py
View File

@@ -3,7 +3,7 @@ import json5
class AppConfig:
"""
Class to hold App config
Class to hold App config.
"""
def __init__(self, config: str):
@@ -15,6 +15,7 @@ class AppConfig:
app_dict = json5.load(json_data)
self.base_application_path = app_dict["base_application_path"]
self.base_application_localhost = app_dict["base_application_localhost"]
self.istio_app_name = app_dict["istio_app_name"]
self.metric_server_app_name = app_dict["metric_server_app_name"]
self.oidc_app_name = app_dict["oidc_app_name"]
@@ -113,3 +114,13 @@ class AppConfig:
"""
return self.platform_integ_apps_app_name
def get_base_application_localhost(self) -> str:
"""
Getter for base application path
Returns:
str: the base application localhost
"""
return self.base_application_localhost

87
keywords/cloud_platform/system/application/object/system_application_update_input.py Normal file
View File

@@ -0,0 +1,87 @@
from keywords.python.string import String
class SystemApplicationUpdateInput:
"""
Class to support the parameters for 'system application-update' command.
An example of using this command is:
'system application-update hello-kitty-min-k8s-version'
This class is able to generate this command just by previously setting the parameters.
"""
def __init__(self):
"""
Constructor
"""
self.app_name = None
self.app_version = None
self.tar_file_path = None
self.timeout_in_seconds = 60
self.check_interval_in_seconds = 3
def set_app_name(self, app_name: str):
"""
Setter for the 'app_name' parameter.
"""
self.app_name = app_name
def get_app_name(self) -> str:
"""
Getter for this 'app_name' parameter.
"""
if String.is_empty(self.app_name) and not String.is_empty(self.get_tar_file_path()):
tar_file_path = self.get_tar_file_path()
filename = tar_file_path.split("/")[-1]
return filename.split("-")[0]
return self.app_name
def set_app_version(self, app_version: str):
"""
Setter for the 'app_version' parameter.
"""
self.app_version = app_version
def get_app_version(self) -> str:
"""
Getter for this 'app_version' parameter.
"""
return self.app_version
def set_tar_file_path(self, tar_file_path: str):
"""
Setter for the 'tar_file_path' parameter.
"""
self.tar_file_path = tar_file_path
def get_tar_file_path(self) -> str:
"""
Getter for this 'tar_file_path' parameter.
"""
return self.tar_file_path
def set_timeout_in_seconds(self, timeout_in_seconds: int):
"""
Setter for the 'timeout_in_seconds' parameter.
"""
self.timeout_in_seconds = timeout_in_seconds
def get_timeout_in_seconds(self) -> int:
"""
Getter for this 'timeout_in_seconds' parameter.
"""
return self.timeout_in_seconds
def set_check_interval_in_seconds(self, check_interval_in_seconds: int):
"""
Setter for the 'check_interval_in_seconds' parameter.
"""
self.check_interval_in_seconds = check_interval_in_seconds
def get_check_interval_in_seconds(self) -> int:
"""
Getter for this 'check_interval_in_seconds' parameter.
"""
return self.check_interval_in_seconds

92
keywords/cloud_platform/system/application/system_application_update_keywords.py Normal file
View File

@@ -0,0 +1,92 @@
from framework.logging.automation_logger import get_logger
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.system.application.object.system_application_output import SystemApplicationOutput
from keywords.cloud_platform.system.application.object.system_application_status_enum import SystemApplicationStatusEnum
from keywords.cloud_platform.system.application.object.system_application_update_input import SystemApplicationUpdateInput
from keywords.cloud_platform.system.application.system_application_list_keywords import SystemApplicationListKeywords
from keywords.python.string import String
class SystemApplicationUpdateKeywords(BaseKeyword):
"""
Class for System application update keywords
"""
def __init__(self, ssh_connection: SSHConnection):
"""
Constructor
Args:
ssh_connection (SSHConnection): SSH connection object
"""
self.ssh_connection = ssh_connection
def system_application_update(self, system_application_update_input: SystemApplicationUpdateInput) -> SystemApplicationOutput:
"""
Executes the update of an application by executing the command 'system application-update'.
This method returns upon the completion of the 'system application-update' command, that is, when the 'status' is 'applied'.
Args:
system_application_update_input (SystemApplicationUpdateInput): the object representing the parameters for
executing the 'system application-update' command.
Returns:
SystemApplicationOutput: an object representing status values related to the current updating process of
the application, as a result of the execution of the 'system application-update' command.
"""
# Gets the command 'system application-update' with its parameters configured.
cmd = self.get_command(system_application_update_input)
# Determine app name for status tracking
app_name = system_application_update_input.get_app_name()
# Executes the command 'system application-update'.
output = self.ssh_connection.send(source_openrc(cmd))
self.validate_success_return_code(self.ssh_connection)
system_application_output = SystemApplicationOutput(output)
# Tracks the execution of the command 'system application-update' until its completion or a timeout.
system_application_list_keywords = SystemApplicationListKeywords(self.ssh_connection)
system_application_list_keywords.validate_app_status(app_name, SystemApplicationStatusEnum.APPLIED.value)
# If the execution arrived here the status of the application is 'applied'.
system_application_output.get_system_application_object().set_status(SystemApplicationStatusEnum.APPLIED.value)
return system_application_output
def get_command(self, system_application_update_input: SystemApplicationUpdateInput) -> str:
"""
Generates a string representing the 'system application-update' command with parameters.
Based on the values in the 'system_application_update_input' argument.
Args:
system_application_update_input (SystemApplicationUpdateInput): an instance of SystemApplicationUpdateInput
configured with the parameters needed to execute the 'system application-update' command properly.
Returns:
str: a string representing the 'system application-update' command, configured according to the parameters
in the 'system_application_update_input' argument.
"""
# Either 'tar_file_path' or 'app_name' property is required.
tar_file_path = system_application_update_input.get_tar_file_path()
app_name = system_application_update_input.get_app_name()
if String.is_empty(tar_file_path) and String.is_empty(app_name):
error_message = "Either tar_file_path or app_name property must be specified."
get_logger().log_exception(error_message)
raise ValueError(error_message)
# Assembles the command - prioritize tar_file_path if provided
if not String.is_empty(tar_file_path):
cmd = f"system application-update {tar_file_path}"
else:
cmd = f"system application-update {app_name}"
return cmd

121
testcases/cloud_platform/regression/storage/test_platform_integ_apps.py
View File

@@ -1,17 +1,24 @@
from pytest import mark
from pytest import FixtureRequest, mark
from config.configuration_manager import ConfigurationManager
from framework.logging.automation_logger import get_logger
from framework.validation.validation import validate_equals
from framework.resources.resource_finder import get_stx_resource_path
from framework.ssh.secure_transfer_file.secure_transfer_file import SecureTransferFile
from framework.ssh.secure_transfer_file.secure_transfer_file_enum import TransferDirection
from framework.ssh.secure_transfer_file.secure_transfer_file_input_object import SecureTransferFileInputObject
from framework.validation.validation import validate_equals, validate_not_equals
from keywords.ceph.ceph_status_keywords import CephStatusKeywords
from keywords.cloud_platform.ssh.lab_connection_keywords import LabConnectionKeywords
from keywords.cloud_platform.system.application.object.system_application_delete_input import SystemApplicationDeleteInput
from keywords.cloud_platform.system.application.object.system_application_remove_input import SystemApplicationRemoveInput
from keywords.cloud_platform.system.application.object.system_application_update_input import SystemApplicationUpdateInput
from keywords.cloud_platform.system.application.object.system_application_upload_input import SystemApplicationUploadInput
from keywords.cloud_platform.system.application.system_application_apply_keywords import SystemApplicationApplyKeywords
from keywords.cloud_platform.system.application.system_application_delete_keywords import SystemApplicationDeleteKeywords
from keywords.cloud_platform.system.application.system_application_list_keywords import SystemApplicationListKeywords
from keywords.cloud_platform.system.application.system_application_remove_keywords import SystemApplicationRemoveKeywords
from keywords.cloud_platform.system.application.system_application_show_keywords import SystemApplicationShowKeywords
from keywords.cloud_platform.system.application.system_application_update_keywords import SystemApplicationUpdateKeywords
from keywords.cloud_platform.system.application.system_application_upload_keywords import SystemApplicationUploadKeywords
@@ -92,3 +99,113 @@ def test_delete_platform_integ_app(request):
system_application_delete_input.set_app_name(platform_integ_apps_name)
app_delete_response = SystemApplicationDeleteKeywords(active_ssh_connection).get_system_application_delete(system_application_delete_input)
validate_equals(app_delete_response.rstrip(), "Application platform-integ-apps deleted.", "Application deletion.")
@mark.p2
def test_rollback_platform_integ_app(request: FixtureRequest):
"""
Rollback platform-integ-apps application to a previous version.
Test Steps:
- Record current version of platform-integ-apps
- Transfer tarball from local machine to /home/sysadmin
- Mount /usr with read-write permissions
- Copy tarball to /usr/local/share/applications/helm/
- Execute system application-update with tarball filename
- Verify the platform-integ-apps version has changed (rollback)
Args:
request (FixtureRequest): pytest request fixture for test setup and teardown
"""
active_ssh_connection = LabConnectionKeywords().get_active_controller_ssh()
platform_integ_apps_name = setup(request, active_ssh_connection)
# Record current version before rollback
get_logger().log_test_case_step("Record current platform-integ-apps version")
current_app_info = SystemApplicationShowKeywords(active_ssh_connection).get_system_application_show(platform_integ_apps_name)
current_version = current_app_info.get_system_application_object().get_version()
def teardown():
get_logger().log_teardown_step("Test- Teardown: Check if restore needed")
# Check current version on system
current_app_info = SystemApplicationShowKeywords(active_ssh_connection).get_system_application_show(platform_integ_apps_name)
system_version = current_app_info.get_system_application_object().get_version()
if system_version != current_version:
get_logger().log_teardown_step("Restoring original platform-integ-apps version")
# Remove the rolled back version
system_application_remove_input = SystemApplicationRemoveInput()
system_application_remove_input.set_app_name(platform_integ_apps_name)
SystemApplicationRemoveKeywords(active_ssh_connection).system_application_remove(system_application_remove_input)
# Delete the rolled back version
system_application_delete_input = SystemApplicationDeleteInput()
system_application_delete_input.set_app_name(platform_integ_apps_name)
SystemApplicationDeleteKeywords(active_ssh_connection).get_system_application_delete(system_application_delete_input)
# Copy original tarball from /home/sysadmin to base_application_path
get_logger().log_teardown_step("Copy original tarball to base application path")
active_ssh_connection.send_as_sudo(f"mv /home/sysadmin/platform-integ-app*.tgz {app_config.get_base_application_path()}")
# Move rollback tarball from base_application_path to /home/sysadmin
get_logger().log_teardown_step("Move rollback tarball to /home/sysadmin")
active_ssh_connection.send_as_sudo(f"mv {app_config.get_base_application_path()}{tarball_filename} /home/sysadmin/")
# Upload original version
system_application_upload_input = SystemApplicationUploadInput()
system_application_upload_input.set_app_name(platform_integ_apps_name)
system_application_upload_input.set_tar_file_path(f"{app_config.get_base_application_path()}platform-integ-app*.tgz")
SystemApplicationUploadKeywords(active_ssh_connection).system_application_upload(system_application_upload_input)
# Apply original version
SystemApplicationApplyKeywords(active_ssh_connection).system_application_apply(app_name=platform_integ_apps_name)
# Delete tarball file from /home/sysadmin
get_logger().log_teardown_step("Delete tarball file from /home/sysadmin")
active_ssh_connection.send_as_sudo(f"rm -f /home/sysadmin/{tarball_filename}")
else:
get_logger().log_teardown_step("No restore needed - version unchanged")
request.addfinalizer(teardown)
# Transfer tarball from local machine to /home/sysadmin
get_logger().log_test_case_step("Transfer rollback tarball from local machine to /home/sysadmin")
app_config = ConfigurationManager.get_app_config()
local_path = get_stx_resource_path(app_config.get_base_application_localhost())
tarball_filename = app_config.get_base_application_localhost().split("/")[-1]
temp_remote_path = f"/home/sysadmin/{tarball_filename}"
sftp_client = active_ssh_connection.get_sftp_client()
transfer_input = SecureTransferFileInputObject()
transfer_input.set_sftp_client(sftp_client)
transfer_input.set_origin_path(local_path)
transfer_input.set_destination_path(temp_remote_path)
transfer_input.set_transfer_direction(TransferDirection.FROM_LOCAL_TO_REMOTE)
transfer_input.set_force(True)
SecureTransferFile(transfer_input).transfer_file()
# Mount /usr to be able to write the tarball
get_logger().log_test_case_step("Mount /usr with read-write permissions")
active_ssh_connection.send_as_sudo("mount -o rw,remount /usr")
# Copy platform_integ_app*.tgz from base_application_path to /home/sysadmin
get_logger().log_test_case_step("Move platform_integ_app*.tgz to /home/sysadmin")
active_ssh_connection.send_as_sudo(f"mv {app_config.get_base_application_path()}platform-integ-app*.tgz /home/sysadmin/")
# Copy tarball from /home/sysadmin to base_application_path
get_logger().log_test_case_step("Move tarball to base application path")
active_ssh_connection.send_as_sudo(f"mv {temp_remote_path} {app_config.get_base_application_path()}")
# Rollback platform-integ-apps with tarball
get_logger().log_test_case_step("Rollback platform-integ-apps with tarball")
system_application_update_input = SystemApplicationUpdateInput()
system_application_update_input.set_app_name(platform_integ_apps_name)
system_application_update_input.set_tar_file_path(f"{app_config.get_base_application_path()}{tarball_filename}")
SystemApplicationUpdateKeywords(active_ssh_connection).system_application_update(system_application_update_input)
# Verify the application version has changed (rollback)
get_logger().log_test_case_step("Verify platform-integ-apps version has changed after rollback")
rollback_app_info = SystemApplicationShowKeywords(active_ssh_connection).get_system_application_show(platform_integ_apps_name)
rollback_version = rollback_app_info.get_system_application_object().get_version()
validate_not_equals(current_version, rollback_version, "Application version should have changed after rollback")

2
unit_tests/config/app/app_config_test.py
View File

@@ -22,6 +22,7 @@ def test_default_app_config():
assert default_config.get_node_feature_discovery_app_name() == "node-feature-discovery", "node feature discovery default app name was incorrect"
assert default_config.get_node_interface_metrics_exporter_app_name() == "node-interface-metrics-exporter", "node interface metrics exporter default app name was incorrect"
assert default_config.get_platform_integ_apps_app_name() == "platform-integ-apps", "platform integ apps default app name was incorrect"
assert default_config.get_base_application_localhost() == "fake_path", "default base path localhost was incorrect"
def test_custom_app_config():
@@ -45,3 +46,4 @@ def test_custom_app_config():
assert custom_config.get_node_feature_discovery_app_name() == "node-feature-discovery_custom", "node feature discovery custom name was incorrect"
assert custom_config.get_node_interface_metrics_exporter_app_name() == "node-interface-metrics-exporter_custom", "node interface metrics exporter custom name was incorrect"
assert custom_config.get_platform_integ_apps_app_name() == "platform-integ-apps_custom", "platform integ apps custom name was incorrect"
assert custom_config.get_base_application_localhost() == "fake_path", "custom base path localhost was incorrect"

3
unit_tests/config/app/custom_app_config.json5
View File

@@ -7,5 +7,6 @@
power_manager_app_name: "kubernetes-power-manager_custom",
node_feature_discovery_app_name: "node-feature-discovery_custom",
node_interface_metrics_exporter_app_name: "node-interface-metrics-exporter_custom",
platform_integ_apps_app_name: "platform-integ-apps_custom"
platform_integ_apps_app_name: "platform-integ-apps_custom",
base_application_localhost: "fake_path"
}