From d17b22d2c29892f5353cf692f9140b9ae952ce4e Mon Sep 17 00:00:00 2001 From: croy Date: Thu, 9 Jan 2025 16:18:07 -0500 Subject: [PATCH] Replacing track_status by validate_app_status In system_application_list_keyword, replacing track_status with validate_app_status to make use of the new validation library. Change-Id: I13718cdaad127c219f9b12f20bedd812613a313e --- .../object/system_application_apply_input.py | 53 ------------------- .../system_application_apply_keywords.py | 38 +++++-------- .../system_application_list_keywords.py | 47 +--------------- .../system_application_remove_keywords.py | 14 ++--- .../system_application_upload_keywords.py | 12 +---- .../cloud_platform/sanity/test_sanity.py | 10 +--- 6 files changed, 20 insertions(+), 154 deletions(-) delete mode 100644 keywords/cloud_platform/system/application/object/system_application_apply_input.py diff --git a/keywords/cloud_platform/system/application/object/system_application_apply_input.py b/keywords/cloud_platform/system/application/object/system_application_apply_input.py deleted file mode 100644 index b428fd3e..00000000 --- a/keywords/cloud_platform/system/application/object/system_application_apply_input.py +++ /dev/null @@ -1,53 +0,0 @@ -class SystemApplicationApplyInput: - """ - Class to support the parameters for 'system application-apply' command. - An example of using this command is: - - 'system application-upload -n hello-kitty' - - This class is able to generate this command just by previously setting the parameters. - """ - - def __init__(self): - """ - Constructor - """ - self.app_name = 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. - """ - return self.app_name - - 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 diff --git a/keywords/cloud_platform/system/application/system_application_apply_keywords.py b/keywords/cloud_platform/system/application/system_application_apply_keywords.py index 1bbdd45d..01952da5 100644 --- a/keywords/cloud_platform/system/application/system_application_apply_keywords.py +++ b/keywords/cloud_platform/system/application/system_application_apply_keywords.py @@ -1,8 +1,6 @@ from framework.logging.automation_logger import get_logger from keywords.base_keyword import BaseKeyword from keywords.cloud_platform.command_wrappers import source_openrc -from keywords.cloud_platform.system.application.object.system_application_apply_input import SystemApplicationApplyInput -from keywords.cloud_platform.system.application.object.system_application_list_status_tracking_input import SystemApplicationListStatusTrackingInput 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.system_application_list_keywords import SystemApplicationListKeywords @@ -22,26 +20,23 @@ class SystemApplicationApplyKeywords(BaseKeyword): """ self.ssh_connection = ssh_connection - def system_application_apply(self, system_application_apply_input: SystemApplicationApplyInput) -> SystemApplicationOutput: + def system_application_apply(self, app_name: str) -> SystemApplicationOutput: """ Executes the applying of an application file by executing the command 'system application-apply'. This method returns upon the completion of the 'system application-apply' command, that is, when the 'status' is 'applied'. Executes the installation of an application by executing the command 'system application-apply'. Args: - system_application_apply_input (SystemApplicationApplyInput): the object representing the parameters for - executing the 'system application-apply' command. + app_name (str): the name of the application to apply Returns: SystemApplicationOutput: an object representing status values related to the current installation process of the application, as a result of the execution of the 'system application-apply' command. """ - # Gets the app name. - app_name = system_application_apply_input.get_app_name() # Gets the command 'system application-apply' with its parameters configured. - cmd = self.get_command(system_application_apply_input) + cmd = self.get_command(app_name) # Executes the command 'system application-apply'. output = self.ssh_connection.send(source_openrc(cmd)) @@ -49,19 +44,11 @@ class SystemApplicationApplyKeywords(BaseKeyword): system_application_output = SystemApplicationOutput(output) # Tracks the execution of the command 'system application-apply' until its completion or a timeout. - - # Setups the status tracker. - system_application_list_status_tracking_input = SystemApplicationListStatusTrackingInput(app_name, SystemApplicationStatusEnum.APPLIED) - system_application_list_status_tracking_input.set_timeout_in_seconds(system_application_apply_input.get_timeout_in_seconds()) - system_application_list_status_tracking_input.set_check_interval_in_seconds(system_application_apply_input.get_check_interval_in_seconds()) - - # Tracks the status of the application. system_application_list_keywords = SystemApplicationListKeywords(self.ssh_connection) - system_application_list_output = system_application_list_keywords.track_status(system_application_list_status_tracking_input) + system_application_list_keywords.validate_app_status(app_name, "applied") # If the execution arrived here the status of the application is 'applied'. - application = system_application_list_output.get_application(app_name) - system_application_output.get_system_application_object().set_status(application.get_status()) + system_application_output.get_system_application_object().set_status("applied") return system_application_output @@ -85,26 +72,25 @@ class SystemApplicationApplyKeywords(BaseKeyword): get_logger().log_exception(f"An error occurred while verifying whether the application named {app_name} is already applied.") raise ex - def get_command(self, system_application_apply_input: SystemApplicationApplyInput) -> str: + def get_command(self, app_name: str) -> str: """ - Generates a string representing the 'system application-apply' command with parameters based on the values in - the 'system_application_apply_input' argument. + Generates a string representing the 'system application-apply' command + Args: - system_application_apply_input (SystemApplicationApplyInput): an instance of SystemApplicationApplyInput - configured with the parameters needed to execute the 'system application-apply' command properly. + app_name (str): Name of the application to Apply Returns: str: a string representing the 'system application-apply' command, configured according to the parameters in the 'system_application_apply_input' argument. """ - app_name_as_param = system_application_apply_input.get_app_name() - if String.is_empty(app_name_as_param): + + if String.is_empty(app_name): error_message = "The app_name property must be specified." get_logger().log_exception(error_message) raise ValueError(error_message) # Assembles the command 'system application-apply'. - cmd = f'system application-apply {app_name_as_param}' + cmd = f'system application-apply {app_name}' return cmd diff --git a/keywords/cloud_platform/system/application/system_application_list_keywords.py b/keywords/cloud_platform/system/application/system_application_list_keywords.py index 51f77d4e..c3fe7049 100644 --- a/keywords/cloud_platform/system/application/system_application_list_keywords.py +++ b/keywords/cloud_platform/system/application/system_application_list_keywords.py @@ -37,51 +37,6 @@ class SystemApplicationListKeywords(BaseKeyword): return system_application_list_output - def track_status(self, system_application_list_status_tracking_input: SystemApplicationListStatusTrackingInput) -> SystemApplicationListOutput: - """ - - Args: - system_application_list_status_tracking_input: - - Returns: - SystemApplicationListOutput: an instance of the SystemApplicationOutput object representing the - applications on the host, as a result of the execution of the 'system application-list' command. - If no exception is raised during the execution of this method, the 'status' property of this instance will - have the same value as defined in the 'system_application_list_status_tracking_input' argument. - - """ - system_application_list_output = None - - # Gets the input values. - expected_status = system_application_list_status_tracking_input.get_expected_status() - app_name = system_application_list_status_tracking_input.get_app_name() - check_interval_in_seconds = system_application_list_status_tracking_input.get_check_interval_in_seconds() - timeout_in_seconds = system_application_list_status_tracking_input.get_timeout_in_seconds() - - # Tracks the target status. - timeout = True - end_time = time.time() + timeout_in_seconds - while time.time() < end_time: - system_application_list_output = self.get_system_application_list() - system_application_list_object = system_application_list_output.get_application(app_name) - if system_application_list_object is not None: - if system_application_list_object.get_status() == expected_status.value: - timeout = False - get_logger().log_info(f"Application {app_name} reached the '{system_application_list_object.get_status()}' status.") - break - else: - get_logger().log_info( - f"Waiting for {check_interval_in_seconds} more seconds to the application {app_name} to reach the {expected_status.value} status. Maximum total time to wait: {timeout_in_seconds} s. Remaining time: {end_time - time.time():.3f} s. Current status: '{system_application_list_object.get_status()}'." - ) - time.sleep(check_interval_in_seconds) - - # Verifies whether a timeout has occurred. - if timeout: - message = f"Application {app_name} couldn't reach the expected status {expected_status.value}. Reason: timeout ({timeout_in_seconds} s). Note: the expected process was not interrupted." - raise TimeoutError(message) - - return system_application_list_output - def validate_app_status(self, application_name: str, status: str): """ This function will validate that the application specified reaches the desired status. @@ -99,5 +54,5 @@ class SystemApplicationListKeywords(BaseKeyword): return application_status message = f"Application {application_name}'s status is {status}" - validate_equals_with_retry(get_app_status, status, message) + validate_equals_with_retry(get_app_status, status, message, timeout=120) diff --git a/keywords/cloud_platform/system/application/system_application_remove_keywords.py b/keywords/cloud_platform/system/application/system_application_remove_keywords.py index edf6c3b5..72f08177 100644 --- a/keywords/cloud_platform/system/application/system_application_remove_keywords.py +++ b/keywords/cloud_platform/system/application/system_application_remove_keywords.py @@ -41,20 +41,12 @@ class SystemApplicationRemoveKeywords(BaseKeyword): self.validate_success_return_code(self.ssh_connection) system_application_output = SystemApplicationOutput(output) - # Tracks the execution of the command 'system application-apply' until its completion or a timeout. - - # Setups the status tracker. Note: there's no 'removed' status. When an app is removed its status comes back to 'uploaded'. - system_application_list_status_tracking_input = SystemApplicationListStatusTrackingInput(app_name, SystemApplicationStatusEnum.UPLOADED) - system_application_list_status_tracking_input.set_timeout_in_seconds(system_application_remove_input.get_timeout_in_seconds()) - system_application_list_status_tracking_input.set_check_interval_in_seconds(system_application_remove_input.get_check_interval_in_seconds()) - - # Tracks the status of the application. + # Tracks the execution of the command 'system application-remove' until its completion or a timeout. system_application_list_keywords = SystemApplicationListKeywords(self.ssh_connection) - system_application_list_output = system_application_list_keywords.track_status(system_application_list_status_tracking_input) + system_application_list_keywords.validate_app_status(app_name, 'uploaded') # If the execution arrived here the status of the application is 'uploaded', that is, the application was removed. - application = system_application_list_output.get_application(app_name) - system_application_output.get_system_application_object().set_status(application.get_status()) + system_application_output.get_system_application_object().set_status('uploaded') return system_application_output diff --git a/keywords/cloud_platform/system/application/system_application_upload_keywords.py b/keywords/cloud_platform/system/application/system_application_upload_keywords.py index 40e5fed4..324b6bc7 100644 --- a/keywords/cloud_platform/system/application/system_application_upload_keywords.py +++ b/keywords/cloud_platform/system/application/system_application_upload_keywords.py @@ -64,19 +64,11 @@ class SystemApplicationUploadKeywords(BaseKeyword): system_application_output = SystemApplicationOutput(output) # Tracks the execution of the command 'system application-upload' until its completion or a timeout. - - # Setups the status tracker. - system_application_list_status_tracking_input = SystemApplicationListStatusTrackingInput(app_name, SystemApplicationStatusEnum.UPLOADED) - system_application_list_status_tracking_input.set_timeout_in_seconds(system_application_upload_input.get_timeout_in_seconds()) - system_application_list_status_tracking_input.set_check_interval_in_seconds(system_application_upload_input.get_check_interval_in_seconds()) - - # Tracks the status of the application. system_application_list_keywords = SystemApplicationListKeywords(self.ssh_connection) - system_application_list_output = system_application_list_keywords.track_status(system_application_list_status_tracking_input) + system_application_list_keywords.validate_app_status(app_name, 'uploaded') # If the execution arrived here the status of the application is 'uploaded'. - application = system_application_list_output.get_application(app_name) - system_application_output.get_system_application_object().set_status(application.get_status()) + system_application_output.get_system_application_object().set_status('uploaded') return system_application_output diff --git a/testcases/cloud_platform/sanity/test_sanity.py b/testcases/cloud_platform/sanity/test_sanity.py index b88c44d2..3c002781 100644 --- a/testcases/cloud_platform/sanity/test_sanity.py +++ b/testcases/cloud_platform/sanity/test_sanity.py @@ -18,7 +18,6 @@ from keywords.cloud_platform.fault_management.alarms.alarm_list_keywords import from keywords.cloud_platform.fault_management.fm_client_cli.fm_client_cli_keywords import FaultManagementClientCLIKeywords from keywords.cloud_platform.fault_management.fm_client_cli.object.fm_client_cli_object import FaultManagementClientCLIObject from keywords.cloud_platform.ssh.lab_connection_keywords import LabConnectionKeywords -from keywords.cloud_platform.system.application.object.system_application_apply_input import SystemApplicationApplyInput 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_status_enum import SystemApplicationStatusEnum @@ -556,13 +555,8 @@ def test_dc_install_custom_app(): # Step 3: Apply the custom app on the active controller - # Setups the apply input object. - system_application_apply_input = SystemApplicationApplyInput() - system_application_apply_input.set_app_name(app_name) - system_application_apply_input.set_timeout_in_seconds(120) - # Executes the application installation - system_application_apply_output = SystemApplicationApplyKeywords(ssh_connection).system_application_apply(system_application_apply_input) + system_application_apply_output = SystemApplicationApplyKeywords(ssh_connection).system_application_apply(app_name) # Asserts that the applying process concluded successfully system_application_object = system_application_apply_output.get_system_application_object() @@ -640,7 +634,7 @@ def test_dc_install_custom_app(): # Step 7: Apply the custom app on the current subcloud. # Executes the application installation. There is no need to change anything in the apply input object 'system_application_apply_input'. - system_application_apply_output = SystemApplicationApplyKeywords(ssh_subcloud_connection).system_application_apply(system_application_apply_input) + system_application_apply_output = SystemApplicationApplyKeywords(ssh_subcloud_connection).system_application_apply(app_name) # Asserts that the applying process concluded successfully on the current subcloud. system_application_object = system_application_apply_output.get_system_application_object()