Adding e2e tests

Author: leandrodamascena

tests/e2e/logger/handlers/buffer_logs_with_flush.py

@@ -0,0 +1,14 @@
+from aws_lambda_powertools import Logger
+from aws_lambda_powertools.logging.buffer import LoggerBufferConfig
+
+logger_buffer_config = LoggerBufferConfig(max_size=10240)
+
+logger = Logger(level="INFO", logger_buffer=logger_buffer_config)
+
+
+def lambda_handler(event, context):
+ message_visible, message_buffered = event.get("message_visible", ""), event.get("message_buffered", {})
+ logger.info(message_visible)
+ logger.debug(message_buffered)
+ logger.flush_buffer()
+ return "success"

tests/e2e/logger/handlers/buffer_logs_without_flush.py

@@ -0,0 +1,13 @@
+from aws_lambda_powertools import Logger
+from aws_lambda_powertools.logging.buffer import LoggerBufferConfig
+
+logger_buffer_config = LoggerBufferConfig(max_size=10240)
+
+logger = Logger(level="INFO", logger_buffer=logger_buffer_config)
+
+
+def lambda_handler(event, context):
+ message_visible, message_buffered = event.get("message_visible", ""), event.get("message_buffered", {})
+ logger.info(message_visible)
+ logger.debug(message_buffered)
+ return "success"

tests/e2e/logger/test_logger.py

@@ -39,6 +39,68 @@ def multiple_logger_instances_arn(infrastructure: dict) -> str:
  return infrastructure.get("MultipleLoggerInstancesArn", "")
 
 
+@pytest.fixture
+def buffer_logs_without_flush_fn(infrastructure: dict) -> str:
+ return infrastructure.get("BufferLogsWithoutFlush", "")
+
+
+@pytest.fixture
+def buffer_logs_without_flush_arn(infrastructure: dict) -> str:
+ return infrastructure.get("BufferLogsWithoutFlushArn", "")
+
+
+@pytest.fixture
+def buffer_logs_with_flush_fn(infrastructure: dict) -> str:
+ return infrastructure.get("BufferLogsWithFlush", "")
+
+
+@pytest.fixture
+def buffer_logs_with_flush_arn(infrastructure: dict) -> str:
+ return infrastructure.get("BufferLogsWithFlushArn", "")
+
+
+@pytest.mark.xdist_group(name="logger")
+def test_buffer_logs_without_flush(buffer_logs_without_flush_fn, buffer_logs_without_flush_arn):
+ # GIVEN
+ message = "logs should be visible with default settings"
+ message_buffer = "not visible message"
+ payload = json.dumps({"message_visible": message, "message_buffered": message_buffer})
+
+ # WHEN
+ _, execution_time = data_fetcher.get_lambda_response(lambda_arn=buffer_logs_without_flush_arn, payload=payload)
+ data_fetcher.get_lambda_response(lambda_arn=buffer_logs_without_flush_arn, payload=payload)
+
+ # THEN
+ logs = data_fetcher.get_logs(
+ function_name=buffer_logs_without_flush_fn,
+ start_time=execution_time,
+ minimum_log_entries=2,
+ )
+
+ assert len(logs) == 2
+
+
+@pytest.mark.xdist_group(name="logger")
+def test_buffer_logs_with_flush(buffer_logs_with_flush_fn, buffer_logs_with_flush_arn):
+ # GIVEN
+ message = "logs should be visible with default settings"
+ message_buffer = "not visible message"
+ payload = json.dumps({"message_visible": message, "message_buffered": message_buffer})
+
+ # WHEN
+ _, execution_time = data_fetcher.get_lambda_response(lambda_arn=buffer_logs_with_flush_arn, payload=payload)
+ data_fetcher.get_lambda_response(lambda_arn=buffer_logs_with_flush_arn, payload=payload)
+
+ # THEN
+ logs = data_fetcher.get_logs(
+ function_name=buffer_logs_with_flush_fn,
+ start_time=execution_time,
+ minimum_log_entries=4,
+ )
+
+ assert len(logs) == 4
+
+
 @pytest.mark.xdist_group(name="logger")
 def test_basic_lambda_logs_visible(basic_handler_fn, basic_handler_fn_arn):
  # GIVEN