Closed
Description
Expected Behaviour
No failed tests.
Current Behaviour
Some of the tests will fail at random.
Code snippet
import concurrent.futures
import pytest
from aws_lambda_powertools import Logger
logger = Logger()
class Entity(object):
__slots__ = [
"class_dict",
"id1",
"id2",
"extra",
]
def __init__(self, class_dict):
logger.remove_keys(self.__slots__)
# Default values
for key in self.__slots__:
setattr(self, key, None)
self.class_dict = class_dict
def parse(self):
self.id1 = self.class_dict.get("id1")
self.id2 = self.class_dict.get("id2")
self.__update_logger_keys()
return self
def __update_logger_keys(self):
extra = {}
for attribute in self.__slots__:
# Skip meaningless data
if attribute not in [
"class_dict",
"extra",
]:
extra[attribute] = getattr(self, attribute)
self.extra = extra
logger.append_keys(extra=extra)
@staticmethod
def print_ok():
logger.info("Success!")
def process(sample_dict):
sample_entity = Entity(sample_dict).parse()
sample_entity.print_ok()
id_value = sample_dict.get("id1")
# This never fails
assert sample_entity.id1 == id_value
# This fails randomly
assert logger.registered_formatter.log_format.get("extra").get("id1") == id_value
def logging_in_threads():
futures = []
with concurrent.futures.ThreadPoolExecutor() as pool_executor:
for i in range(10):
sample_dict = {"id1": i, "id2": i}
futures.append(pool_executor.submit(process, sample_dict))
[f.result() for f in concurrent.futures.as_completed(futures)]
@pytest.mark.parametrize("run_number", range(100))
def test_logging_in_threads(run_number):
logging_in_threads()Possible Solution
No response
Steps to Reproduce
Run pytest with the snippet above. Some of the tests will fail at random.
AWS Lambda Powertools for Python version
latest
AWS Lambda function runtime
3.9
Packaging format used
PyPi
Debugging logs
No response