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
|
import json
from unittest import SkipTest
import boto3
import requests
from moto import mock_aws, settings
from ..test_awslambda.utilities import get_role_name, get_test_zip_file1
LAMBDA_REGION = "us-west-2"
PYTHON_VERSION = "3.11"
FUNCTION_NAME = "test-function-123"
if settings.TEST_SERVER_MODE:
raise SkipTest("No point in testing batch_simple in ServerMode")
@mock_aws(config={"lambda": {"use_docker": False}})
def test_run_function():
# Setup
client = setup_lambda()
# Execute
result = client.invoke(
FunctionName=FUNCTION_NAME,
LogType="Tail",
)
# Verify
assert result["StatusCode"] == 200
assert result["Payload"].read().decode("utf-8") == "Simple Lambda happy path OK"
@mock_aws(config={"lambda": {"use_docker": False}})
def test_run_function_no_log():
# Setup
client = setup_lambda()
payload = {"results": "results"}
# Execute
result = client.invoke(FunctionName=FUNCTION_NAME, Payload=json.dumps(payload))
# Verify
assert result["StatusCode"] == 200
assert json.loads(result["Payload"].read().decode("utf-8")) == payload
# Execute
result = client.invoke(FunctionName=FUNCTION_NAME)
# Verify
assert result["StatusCode"] == 200
assert result["Payload"].read().decode("utf-8") == "Simple Lambda happy path OK"
@mock_aws(config={"lambda": {"use_docker": False}})
def test_set_lambda_simple_query_results():
# Setup
base_url = (
settings.test_server_mode_endpoint()
if settings.TEST_SERVER_MODE
else "http://motoapi.amazonaws.com"
)
results = {"results": ["test", "test 2"], "region": LAMBDA_REGION}
resp = requests.post(
f"{base_url}/moto-api/static/lambda-simple/response",
json=results,
)
assert resp.status_code == 201
client = setup_lambda()
# Execute & Verify
resp = client.invoke(
FunctionName=FUNCTION_NAME,
LogType="Tail",
)
assert resp["Payload"].read().decode() == results["results"][0]
resp = client.invoke(
FunctionName=FUNCTION_NAME,
LogType="Tail",
)
assert resp["Payload"].read().decode() == results["results"][1]
resp = client.invoke(
FunctionName=FUNCTION_NAME,
LogType="Tail",
)
assert resp["Payload"].read().decode() == "Simple Lambda happy path OK"
def setup_lambda():
client = boto3.client("lambda", LAMBDA_REGION)
zip_content = get_test_zip_file1()
function_name = FUNCTION_NAME
role = get_role_name()
client.create_function(
FunctionName=function_name,
Runtime=PYTHON_VERSION,
Role=role,
Handler="lambda_function.lambda_handler",
Code={"ZipFile": zip_content},
Description="test lambda function",
Timeout=3,
MemorySize=128,
Publish=True,
)
return client
|
