starlingx/test
Code Issues Proposed changes

Merge "ValidationResponse and fix to App Applied Sanity tests"

Browse Source
This commit is contained in:
Zuul
2025-06-25 19:40:21 +00:00
committed by Gerrit Code Review
parent e8dc36380c d51862dba3
commit 905faeca26
4 changed files with 116 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files

54
framework/validation/validation.py
View File

@@ -3,6 +3,7 @@ from time import sleep
from typing import Any, Callable
from framework.logging.automation_logger import get_logger
from framework.validation.validation_response import ValidationResponse
def validate_equals(observed_value: Any, expected_value: Any, validation_description: str) -> None:
@@ -35,7 +36,7 @@ def validate_equals_with_retry(
validation_description: str,
timeout: int = 30,
polling_sleep_time: int = 5,
) -> None:
) -> object:
"""
Validates that function_to_execute will return the expected value in the specified amount of time.
@@ -46,8 +47,8 @@ def validate_equals_with_retry(
timeout (int): The maximum time (in seconds) to wait for the match.
polling_sleep_time (int): The interval of time to wait between calls to function_to_execute.
Raises:
TimeoutError: raised when validate does not equal in the required time
Returns:
object: Returns the value_to_return of the ValidationResponse associated with the function_to_execute.
"""
get_logger().log_info(f"Attempting Validation - {validation_description}")
@@ -58,14 +59,20 @@ def validate_equals_with_retry(
# Compute the actual value that we are trying to validate.
result = function_to_execute()
if isinstance(result, ValidationResponse):
value_to_validate = result.get_value_to_validate()
value_to_return = result.get_value_to_return()
else:
value_to_validate = result
value_to_return = result
if result == expected_value:
if value_to_validate == expected_value:
get_logger().log_info(f"Validation Successful - {validation_description}")
return
return value_to_return
else:
get_logger().log_info("Validation Failed")
get_logger().log_info(f"Expected: {expected_value}")
get_logger().log_info(f"Observed: {result}")
get_logger().log_info(f"Observed: {value_to_validate}")
if time.time() < end_time:
get_logger().log_info(f"Retrying in {polling_sleep_time}s")
@@ -129,7 +136,7 @@ def validate_str_contains_with_retry(
validation_description: str,
timeout: int = 30,
polling_sleep_time: int = 5,
) -> None:
) -> object:
"""
This function will validate if the observed value contains the expected value.
@@ -140,8 +147,8 @@ def validate_str_contains_with_retry(
timeout (int): The maximum time (in seconds) to wait for the match.
polling_sleep_time (int): The interval of time to wait between calls to function_to_execute.
Returns: None
Returns:
object: Returns the value_to_return of the ValidationResponse associated with the function_to_execute.
Raises:
Exception: when validate fails
@@ -155,14 +162,20 @@ def validate_str_contains_with_retry(
# Compute the actual value that we are trying to validate.
result = function_to_execute()
if isinstance(result, ValidationResponse):
value_to_validate = result.get_value_to_validate()
value_to_return = result.get_value_to_return()
else:
value_to_validate = result
value_to_return = result
if expected_value in result:
if expected_value in value_to_validate:
get_logger().log_info(f"Validation Successful - {validation_description}")
return
return value_to_return
else:
get_logger().log_info("Validation Failed")
get_logger().log_info(f"Expected: {expected_value}")
get_logger().log_info(f"Observed: {result}")
get_logger().log_info(f"Observed: {value_to_validate}")
if time.time() < end_time:
get_logger().log_info(f"Retrying in {polling_sleep_time}s")
@@ -201,7 +214,7 @@ def validate_list_contains_with_retry(
validation_description: str,
timeout: int = 30,
polling_sleep_time: int = 5,
) -> str:
) -> object:
"""
This function will validate if the observed value contains the expected value.
@@ -213,7 +226,8 @@ def validate_list_contains_with_retry(
polling_sleep_time (int): The interval of time to wait between calls to function_to_execute.
Returns: str
Returns:
object: Returns the value_to_return of the ValidationResponse associated with the function_to_execute.
Raises:
Exception: when validate fails
@@ -227,14 +241,20 @@ def validate_list_contains_with_retry(
# Compute the actual value that we are trying to validate.
result = function_to_execute()
if isinstance(result, ValidationResponse):
value_to_validate = result.get_value_to_validate()
value_to_return = result.get_value_to_return()
else:
value_to_validate = result
value_to_return = result
if result in expected_values:
if value_to_validate in expected_values:
get_logger().log_info(f"Validation Successful - {validation_description}")
return result
return value_to_return
else:
get_logger().log_info("Validation Failed")
get_logger().log_info(f"Expected: {expected_values}")
get_logger().log_info(f"Observed: {result}")
get_logger().log_info(f"Observed: {value_to_validate}")
if time.time() < end_time:
get_logger().log_info(f"Retrying in {polling_sleep_time}s")

29
framework/validation/validation_response.py Normal file
View File

@@ -0,0 +1,29 @@
class ValidationResponse:
"""
Class that models the response from a ValidationWithRetry.
The purpose of this object is to store a return value as well as a value to validate.
"""
def __init__(self, value_to_validate: object, value_to_return: object):
"""
Constructor for ValidationResponse
Args:
value_to_validate (object): What we want to validate.
value_to_return (object): The object that we want to return if the validation is successful.
"""
self.value_to_validate: object = value_to_validate
self.value_to_return: object = value_to_return
def get_value_to_validate(self) -> object:
"""
Getter for the value to validate
"""
return self.value_to_validate
def get_value_to_return(self) -> object:
"""
Getter for the value to return
"""
return self.value_to_return

56
keywords/cloud_platform/system/application/system_application_show_keywords.py
View File

@@ -1,8 +1,10 @@
from framework.logging.automation_logger import get_logger
from framework.ssh.ssh_connection import SSHConnection
from framework.validation.validation import validate_equals_with_retry
from framework.validation.validation import validate_equals_with_retry, validate_str_contains_with_retry
from framework.validation.validation_response import ValidationResponse
from keywords.base_keyword import BaseKeyword
from keywords.cloud_platform.command_wrappers import source_openrc
from keywords.cloud_platform.system.application.object.system_application_show_object import SystemApplicationShowObject
from keywords.cloud_platform.system.application.object.system_application_show_output import SystemApplicationShowOutput
@@ -45,28 +47,56 @@ class SystemApplicationShowKeywords(BaseKeyword):
get_logger().log_exception(f"Failed to run 'system application-show {app_name}'")
raise ex
def validate_app_status(self, app_name: str, expected_status: str, expected_active: str, expected_progress: str) -> None:
def validate_app_status(self, app_name: str, expected_status: str) -> SystemApplicationShowObject:
"""
Validates that the application reaches the expected status, active state, and progress.
Validates that the application reaches the expected status.
Args:
app_name (str): Name of the application
expected_status (str): Expected status (e.g., 'applied')
expected_active (str): Expected active state (e.g., 'True')
expected_progress (str): Expected progress (e.g., 'completed')
Returns:
SystemApplicationShowObject:
"""
def get_status():
return self.get_system_application_show(app_name).get_system_application_object().get_status()
system_application_object = self.get_system_application_show(app_name).get_system_application_object()
return ValidationResponse(system_application_object.get_status(), system_application_object)
return validate_equals_with_retry(get_status, expected_status, f"Application '{app_name}' status is '{expected_status}'", timeout=300)
def validate_app_active(self, app_name: str, expected_active: str) -> SystemApplicationShowObject:
"""
Validates that the application reaches the expected active.
Args:
app_name (str): Name of the application
expected_active (str): Expected active state (e.g., 'True')
Returns:
SystemApplicationShowObject:
"""
def get_active():
return self.get_system_application_show(app_name).get_system_application_object().get_active()
system_application_object = self.get_system_application_show(app_name).get_system_application_object()
return ValidationResponse(system_application_object.get_active(), system_application_object)
return validate_equals_with_retry(get_active, expected_active, f"Application '{app_name}' active value is '{expected_active}'", timeout=300)
def validate_app_progress_contains(self, app_name: str, expected_progress: str) -> SystemApplicationShowObject:
"""
Validates that the application reaches the expected progress.
Args:
app_name (str): Name of the application
expected_progress (str): Expected progress (e.g., 'completed')
Returns:
SystemApplicationShowObject:
"""
def get_progress():
return self.get_system_application_show(app_name).get_system_application_object().get_progress()
system_application_object = self.get_system_application_show(app_name).get_system_application_object()
return ValidationResponse(system_application_object.get_progress(), system_application_object)
validate_equals_with_retry(get_status, expected_status, f"Waiting for application '{app_name}' to reach status '{expected_status}'", timeout=300)
validate_equals_with_retry(get_active, expected_active, f"Waiting for application '{app_name}' to become active '{expected_active}'", timeout=300)
validate_equals_with_retry(get_progress, expected_progress, f"Waiting for application '{app_name}' to complete progress '{expected_progress}'", timeout=300)
return validate_str_contains_with_retry(get_progress, expected_progress, f"Application '{app_name}' progress is '{expected_progress}'", timeout=300)

8
testcases/cloud_platform/sanity/test_sanity.py
View File

@@ -130,7 +130,9 @@ def test_cert_manager_applied():
"""
ssh_connection = LabConnectionKeywords().get_active_controller_ssh()
SystemApplicationShowKeywords(ssh_connection).validate_app_status(app_name="cert-manager", expected_status="applied", expected_active="True", expected_progress="completed")
system_application_show_output = SystemApplicationShowKeywords(ssh_connection).validate_app_progress_contains("cert-manager", "completed")
validate_equals(system_application_show_output.get_status(), "applied", "cert-manager's status is applied")
validate_equals(system_application_show_output.get_active(), "True", "cert-manager's active is True")
@mark.p0
@@ -145,7 +147,9 @@ def test_nginx_ingress_controller_applied():
"""
ssh_connection = LabConnectionKeywords().get_active_controller_ssh()
SystemApplicationShowKeywords(ssh_connection).validate_app_status(app_name="nginx-ingress-controller", expected_status="applied", expected_active="True", expected_progress="completed")
system_application_show_output = SystemApplicationShowKeywords(ssh_connection).validate_app_progress_contains("nginx-ingress-controller", "completed")
validate_equals(system_application_show_output.get_status(), "applied", "nginx-ingress-controller's status is applied")
validate_equals(system_application_show_output.get_active(), "True", "nginx-ingress-controller's active is True")
@mark.p0