Merge "Adding validate_str_contains_with_retires"

framework/validation/validation.py

@@ -123,6 +123,55 @@ def validate_str_contains(observed_value: str, expected_value: str, validation_d
         raise Exception("Validation Failed")
 
 
+def validate_str_contains_with_retry(
+    function_to_execute: Callable[[], Any],
+    expected_value: str,
+    validation_description: str,
+    timeout: int = 30,
+    polling_sleep_time: int = 5,
+) -> None:
+    """
+    This function will validate if the observed value contains the expected value.
+
+    Args:
+        function_to_execute (Callable[[], Any]): The function to be executed repeatedly, taking no arguments and returning any value.
+        expected_value(str): the value we are expecting to see in the observed value str.
+        validation_description (str): Description of this validation for logging purposes.
+        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
+
+    Raises:
+        Exception: when validate fails
+
+    """
+    get_logger().log_info(f"Attempting Validation - {validation_description}")
+    end_time = time.time() + timeout
+
+    # Attempt the validation
+    while True:
+
+        # Compute the actual value that we are trying to validate.
+        result = function_to_execute()
+
+        if expected_value in result:
+            get_logger().log_info(f"Validation Successful - {validation_description}")
+            return
+        else:
+            get_logger().log_info("Validation Failed")
+            get_logger().log_info(f"Expected: {expected_value}")
+            get_logger().log_info(f"Observed: {result}")
+
+            if time.time() < end_time:
+                get_logger().log_info(f"Retrying in {polling_sleep_time}s")
+                sleep(polling_sleep_time)
+                # Move on to the next iteration
+            else:
+                raise TimeoutError(f"Timeout performing validation - {validation_description}")
+
+
 def validate_list_contains(observed_value: Any, expected_values: Any, validation_description: str) -> None:
     """
     This function validates if the observed value matches ANY of the expected values with associated logging.

unit_tests/framework/validation/validaton_test.py

@@ -1,8 +1,14 @@
+import pytest
+
 from config.configuration_file_locations_manager import (
     ConfigurationFileLocationsManager,
 )
 from config.configuration_manager import ConfigurationManager
-from framework.validation.validation import validate_not_equals, validate_str_contains
+from framework.validation.validation import (
+    validate_not_equals,
+    validate_str_contains,
+    validate_str_contains_with_retry,
+)
 
 
 def test_validate_str_contains():
@@ -20,11 +26,44 @@ def test_validate_str_contains_fails():
     """
     configuration_locations_manager = ConfigurationFileLocationsManager()
     ConfigurationManager.load_configs(configuration_locations_manager)
-    try:
+    with pytest.raises(Exception):
         validate_str_contains("observed value contains: <word not found>", "success", "Test that the word success appears")
-        assert False, "Validation passed when it should not have"  # if test succeeds, we should never get to this line
+
-    except Exception as e:
+
-        assert e.__str__() == "Validation Failed", "Validation failed as expected."
+def test_validate_str_contains_with_retries():
+    """
+    Test to validate str contains with retries
+
+    """
+    configuration_locations_manager = ConfigurationFileLocationsManager()
+    ConfigurationManager.load_configs(configuration_locations_manager)
+    counter = 0
+
+    def function_for_validate_with_retries():
+        nonlocal counter
+        if counter == 0:
+            counter += 1
+            return "this was not what you are looking for"
+        else:
+            counter = 0
+            return "successful"
+
+    validate_str_contains_with_retry(function_for_validate_with_retries, "success", "validate success string", polling_sleep_time=0.01)
+
+
+def test_validate_str_contains_with_retries_fail():
+    """
+    Test to validate str contains with retries when it fails
+
+    """
+    configuration_locations_manager = ConfigurationFileLocationsManager()
+    ConfigurationManager.load_configs(configuration_locations_manager)
+
+    def function_for_validate_with_retries():
+        return "this was not what you are looking for"
+
+    with pytest.raises(TimeoutError):
+        validate_str_contains_with_retry(function_for_validate_with_retries, "success", "validate success string", polling_sleep_time=0.01, timeout=0.05)
 
 
 def test_validate_not_equals_str():