Fix high memory usage in PyTests logging system

Polling functions like `wait_until...` often runs multiple
iterations. When ODL returns large response body
(e.g., PCEP topology dumps), logging every message leads to high
memmory usage. Pytest and Allure retain the complete history of all
logs in virtual memmory.

This patch implements a "Deferred Logging" mechanism:
1. Logs are temporarily captured in a lightweight memory buffer
   during test execution.
2. If an attempt fails (and is not the last one), logs cpatured
   in the buffer are discarded immediately.
3. The buffer is "replayed" to the main logger only if the attempt
   succeeds or if it is the final failure.

Also introduce variable MAX_HTTP_RESPONSE_BODY_LOG_SIZE
for limitting maximum logged HTTP response body size.

JIRA: BGPCEP-1066
Change-Id: I37146943a79622e796fa92b03fe4f4f03a614eac
Signed-off-by: Martin Balaz <[email protected]>

Author: Martin Balaz

tests/libraries/pcep.py

@@ -26,6 +26,7 @@
 ODL_IP = variables.ODL_IP
 RESTCONF_PORT = variables.RESTCONF_PORT
 TOOLS_IP = variables.TOOLS_IP
+MAX_HTTP_RESPONSE_BODY_LOG_SIZE = variables.MAX_HTTP_RESPONSE_BODY_LOG_SIZE
 VARIABLES = pcep_variables.get_variables(TOOLS_IP)
 
 log = logging.getLogger(__name__)
@@ -55,7 +56,6 @@ def get_pcep_topology() -> requests.Response:
         requests.Response:: Pcep topology
     """
     rest_session = AuthStandalone.Init_Session(ODL_IP, "admin", "admin")
-    # Not Logging content, as it may be huge.
     resp = AuthStandalone.Get_Using_Session(
         rest_session,
         (
@@ -68,7 +68,10 @@ def get_pcep_topology() -> requests.Response:
         f"Response code for get pcep topology does not meach expected 200, "
         f"but is {resp.status_code}"
     )
-    log.debug(resp.text)
+    resp_text = utils.truncate_long_text(resp.text, MAX_HTTP_RESPONSE_BODY_LOG_SIZE)
+    log.debug(f"Response: {resp_text}")
+    log.info(f"Response code: {resp.status_code}")
+    log.debug(f"Response headers: {resp.headers}")
     return resp
 
 

tests/libraries/templated_requests.py

@@ -18,6 +18,7 @@
 
 ODL_IP = variables.ODL_IP
 RESTCONF_PORT = variables.RESTCONF_PORT
+MAX_HTTP_RESPONSE_BODY_LOG_SIZE = variables.MAX_HTTP_RESPONSE_BODY_LOG_SIZE
 BASE_URL = f"http://{ODL_IP}:{RESTCONF_PORT}"
 
 ALLOWED_STATUS_CODES = {200, 201, 204}
@@ -68,7 +69,8 @@ def get_from_uri(
         log.error(f"Response headers: {response.headers}")
         raise AssertionError("Unexpected failure in GET response.") from e
     else:
-        log.debug(f"Response: {response.text}")
+        resposne_text = utils.truncate_long_text(response.text, MAX_HTTP_RESPONSE_BODY_LOG_SIZE)
+        log.debug(f"Response: {resposne_text}")
         log.info(f"Response code: {response.status_code}")
         log.debug(f"Response headers: {response.headers}")
 
@@ -121,7 +123,8 @@ def put_to_uri_request(
         log.error(f"Response headers: {response.headers}")
         raise AssertionError("Unexpected failure in PUT response.") from e
     else:
-        log.debug(f"Response: {response.text}")
+        resposne_text = utils.truncate_long_text(response.text, MAX_HTTP_RESPONSE_BODY_LOG_SIZE)
+        log.debug(f"Response: {resposne_text}")
         log.info(f"Response code: {response.status_code}")
         log.debug(f"Response headers: {response.headers}")
 
@@ -174,7 +177,8 @@ def post_to_uri(
         log.error(f"Response headers: {response.headers}")
         raise AssertionError("Unexpected failure in POST response.") from e
     else:
-        log.debug(f"Response: {response.text}")
+        resposne_text = utils.truncate_long_text(response.text, MAX_HTTP_RESPONSE_BODY_LOG_SIZE)
+        log.debug(f"Response: {resposne_text}")
         log.info(f"Response code: {response.status_code}")
         log.debug(f"Response headers: {response.headers}")
 
@@ -220,7 +224,8 @@ def delete_from_uri_request(
         log.error(f"Response headers: {response.headers}")
         raise AssertionError("Unexpected failure in DELETE response.") from e
     else:
-        log.debug(f"Response: {response.text}")
+        resposne_text = utils.truncate_long_text(response.text, MAX_HTTP_RESPONSE_BODY_LOG_SIZE)
+        log.debug(f"Response: {resposne_text}")
         log.info(f"Response code: {response.status_code}")
         log.debug(f"Response headers: {response.headers}")
 

tests/libraries/utils.py

@@ -6,18 +6,96 @@
 # and is available at http://www.eclipse.org/legal/epl-v10.html
 #
 
+from contextlib import contextmanager
 import difflib
 import logging
 import time
 from collections.abc import Callable
-from typing import Any, List, Tuple
+from typing import Any, Generator, List, Tuple
 
 from libraries import infra
 from libraries import norm_json
 
 log = logging.getLogger(__name__)
 
 
+class DeferredLogHandler(logging.Handler):
+    """Stores log records in memory instead of writing them immediately."""
+    def __init__(self):
+        """Initialize the deferred handler with an empty record list."""
+        super().__init__()
+        self.records = []
+
+    def emit(self, record: logging.LogRecord):
+        """Store an emitted log record in local buffer.
+
+        Args:
+            record (logging.LogRecord): Log record to be stored.
+
+        Returns:
+            None
+        """
+        self.records.append(record)
+
+    def flush_to_target(self, target_logger: logging.Logger):
+        """Replay buffered log records to the target logger.
+
+        Args:
+            target_logger (logging.Logger): Logger to which the buffered records
+                should be sent.
+
+        Returns:
+            None
+        """
+        for record in self.records:
+            target_logger.handle(record)
+
+
+@contextmanager
+def deferred_logging() -> Generator['DeferredLogHandler', None, None]:
+    """Context manager for temporary log buffering.
+
+    This replaces the root logger's handlers with a temporary buffer. Logs
+    generated within this context are stored in memory and can be either flushed
+    to the original handlers or discarded. Usefull for functions which produces
+    a lot of log records and it is not known in advanced if those log
+    entries will be kept or discarded (e.g. wait until functions).
+
+    Args:
+        None
+
+    Yields:
+        DeferredLogHandler: Temporary handler for storing buffered logs.
+    """
+    root_logger = logging.getLogger()
+    original_handlers = root_logger.handlers[:]
+    buffer_handler = DeferredLogHandler()
+    root_logger.handlers = [buffer_handler]
+    try:
+        yield buffer_handler
+    finally:
+        root_logger.handlers = original_handlers
+
+
+def truncate_long_text(text: str, max_size: int) -> str:
+    """Truncates long text if it exceeds the maximum size.
+
+    Args:
+        text (str): Text to be truncate.
+        max_size (int): Maximum allowed length. If -1, text is not truncated.
+
+    Returns:
+        str: Final truncated text.
+    """
+    if max_size == -1:
+        return text
+
+    if len(text) > max_size:
+            text = text[:max_size] + " ... (truncated long output)"
+
+    return text
+
+
 def verify_jsons_match(
     json1: str,
     json2: str,
@@ -170,11 +248,14 @@ def wait_until_function_returns_value_with_custom_value_validator(
         Any: Return value returend by last successful function call.
     """
     last_exception = None
+    logger_buffer = None
 
     for retry_num in range(retry_count):
         try:
-            result = function(*args, **kwargs)
+            with deferred_logging() as logger_buffer:
+                result = function(*args, **kwargs)
             if return_value_validator(result):
+                logger_buffer.flush_to_target(log)
                 return result
             else:
                 raise AssertionError(
@@ -190,6 +271,8 @@ def wait_until_function_returns_value_with_custom_value_validator(
             log.debug(f"failed with: {e}")
         time.sleep(interval)
     else:
+        if logger_buffer:
+            logger_buffer.flush_to_target(log)
         raise AssertionError(
             f"Failed to execute "
             f"{function.__name__}({','.join([str(arg) for arg in args])} "
@@ -325,17 +408,22 @@ def verify_function_returns_value_which_passes_custom_value_validator_for_some_t
     Returns:
         Any: Return value returend by last successful function call.
     """
+    logger_buffer = None
+
     for retry_num in range(retry_count):
         try:
-            result = function(*args, **kwargs)
+            with deferred_logging() as logger_buffer:
+                result = function(*args, **kwargs)
             passed_value_validator = return_value_validator(result)
         except Exception as e:
+            logger_buffer.flush_to_target(log)
             raise AssertionError(
                 f"Function {function.__name__}"
                 f"({','.join([str(arg) for arg in args])} {kwargs or ''}) "
                 f"failed with the following error {e}"
             )
         if not passed_value_validator:
+            logger_buffer.flush_to_target(log)
             raise AssertionError(
                 f"Function {function.__name__}"
                 f"({','.join([str(arg) for arg in args])} {kwargs or ''}) "
@@ -348,6 +436,7 @@ def verify_function_returns_value_which_passes_custom_value_validator_for_some_t
         )
         time.sleep(interval)
 
+    logger_buffer.flush_to_target(log)
     return result
 
 

tests/libraries/variables.py

@@ -40,6 +40,7 @@ def REST_API(self) -> str:
     TOPOLOGY_URL: str = "rests/data/network-topology:network-topology/topology"
     ENABLE_TCP_TW_REUSE: bool = False
     DEFAULT_PCEP_STATS_UPDATE_INTERVAL: int = 5
+    MAX_HTTP_RESPONSE_BODY_LOG_SIZE: int = 500
 
 
 variables = Variables()