forked from aws-powertools/powertools-lambda-python
-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathtest_logger.py
143 lines (107 loc) · 4.79 KB
/
test_logger.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
import boto3
import pytest
from e2e import conftest
from tests.e2e.utils import data_fetcher
@pytest.fixture(scope="module")
def config() -> conftest.LambdaConfig:
return {
"parameters": {},
"environment_variables": {
"MESSAGE": "logger message test",
"LOG_LEVEL": "INFO",
"ADDITIONAL_KEY": "extra_info",
},
}
def test_basic_lambda_logs_visible(execute_lambda: conftest.InfrastructureOutput, config: conftest.LambdaConfig):
# GIVEN
lambda_name = execute_lambda.get_lambda_function_name(cf_output_name="basichandlerarn")
timestamp = execute_lambda.get_lambda_execution_time_timestamp()
cw_client = boto3.client("logs")
# WHEN
filtered_logs = data_fetcher.get_logs(lambda_function_name=lambda_name, start_time=timestamp, log_client=cw_client)
# THEN
assert any(
log.message == config["environment_variables"]["MESSAGE"]
and log.level == config["environment_variables"]["LOG_LEVEL"]
for log in filtered_logs
)
def test_basic_lambda_no_debug_logs_visible(
execute_lambda: conftest.InfrastructureOutput, config: conftest.LambdaConfig
):
# GIVEN
lambda_name = execute_lambda.get_lambda_function_name(cf_output_name="basichandlerarn")
timestamp = execute_lambda.get_lambda_execution_time_timestamp()
cw_client = boto3.client("logs")
# WHEN
filtered_logs = data_fetcher.get_logs(lambda_function_name=lambda_name, start_time=timestamp, log_client=cw_client)
# THEN
assert not any(
log.message == config["environment_variables"]["MESSAGE"] and log.level == "DEBUG" for log in filtered_logs
)
def test_basic_lambda_contextual_data_logged(execute_lambda: conftest.InfrastructureOutput):
# GIVEN
required_keys = (
"xray_trace_id",
"function_request_id",
"function_arn",
"function_memory_size",
"function_name",
"cold_start",
)
lambda_name = execute_lambda.get_lambda_function_name(cf_output_name="basichandlerarn")
timestamp = execute_lambda.get_lambda_execution_time_timestamp()
cw_client = boto3.client("logs")
# WHEN
filtered_logs = data_fetcher.get_logs(lambda_function_name=lambda_name, start_time=timestamp, log_client=cw_client)
# THEN
assert all(keys in logs.dict(exclude_unset=True) for logs in filtered_logs for keys in required_keys)
def test_basic_lambda_additional_key_persistence_basic_lambda(
execute_lambda: conftest.InfrastructureOutput, config: conftest.LambdaConfig
):
# GIVEN
lambda_name = execute_lambda.get_lambda_function_name(cf_output_name="basichandlerarn")
timestamp = execute_lambda.get_lambda_execution_time_timestamp()
cw_client = boto3.client("logs")
# WHEN
filtered_logs = data_fetcher.get_logs(lambda_function_name=lambda_name, start_time=timestamp, log_client=cw_client)
# THEN
assert any(
log.extra_info
and log.message == config["environment_variables"]["MESSAGE"]
and log.level == config["environment_variables"]["LOG_LEVEL"]
for log in filtered_logs
)
def test_basic_lambda_empty_event_logged(execute_lambda: conftest.InfrastructureOutput):
# GIVEN
lambda_name = execute_lambda.get_lambda_function_name(cf_output_name="basichandlerarn")
timestamp = execute_lambda.get_lambda_execution_time_timestamp()
cw_client = boto3.client("logs")
# WHEN
filtered_logs = data_fetcher.get_logs(lambda_function_name=lambda_name, start_time=timestamp, log_client=cw_client)
# THEN
assert any(log.message == {} for log in filtered_logs)
def test_no_context_lambda_contextual_data_not_logged(execute_lambda: conftest.InfrastructureOutput):
# GIVEN
required_missing_keys = (
"function_request_id",
"function_arn",
"function_memory_size",
"function_name",
"cold_start",
)
lambda_name = execute_lambda.get_lambda_function_name(cf_output_name="nocontexthandlerarn")
timestamp = execute_lambda.get_lambda_execution_time_timestamp()
cw_client = boto3.client("logs")
# WHEN
filtered_logs = data_fetcher.get_logs(lambda_function_name=lambda_name, start_time=timestamp, log_client=cw_client)
# THEN
assert not any(keys in logs.dict(exclude_unset=True) for logs in filtered_logs for keys in required_missing_keys)
def test_no_context_lambda_event_not_logged(execute_lambda: conftest.InfrastructureOutput):
# GIVEN
lambda_name = execute_lambda.get_lambda_function_name(cf_output_name="nocontexthandlerarn")
timestamp = execute_lambda.get_lambda_execution_time_timestamp()
cw_client = boto3.client("logs")
# WHEN
filtered_logs = data_fetcher.get_logs(lambda_function_name=lambda_name, start_time=timestamp, log_client=cw_client)
# THEN
assert not any(log.message == {} for log in filtered_logs)