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
This commit is contained in:
croy
@@ -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
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
@@ -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()
|
||||
|
||||
Reference in New Issue
Block a user