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
|
import pkgutil
from uuid import uuid4
import boto3
import pytest
from moto import mock_aws
from tests.markers import requires_docker
from .utilities import _process_lambda, get_role_name
PYTHON_VERSION = "python3.11"
_lambda_region = "us-west-2"
def get_requests_zip_file():
pfunc = """
import requests
def lambda_handler(event, context):
return requests.__version__
"""
return _process_lambda(pfunc)
@requires_docker
@mock_aws
@pytest.mark.filterwarnings("ignore:Error extracting layer to Lambda")
def test_invoke_local_lambda_layers():
conn = boto3.client("lambda", _lambda_region)
lambda_name = str(uuid4())[0:6]
#
# Info about KLayers, including ARNs
# https://api.klayers.cloud/api/v2/p3.11/layers/latest/us-east-1/json
#
# Get all info about a specific layer
# aws lambda get-layer-version-by-arn --arn "..."
#
# Download layer as a ZIP file
# curl $(aws lambda get-layer-version-by-arn --arn "..." --query "Content.Location" --output text) --output klayer.zip
requests_location = "resources/klayer_311_19.zip"
requests_layer = pkgutil.get_data(__name__, requests_location)
layer_arn = conn.publish_layer_version(
LayerName=str(uuid4())[0:6],
Content={"ZipFile": requests_layer},
CompatibleRuntimes=["python3.11"],
LicenseInfo="MIT",
)["LayerArn"]
bogus_layer_arn = conn.publish_layer_version(
LayerName=str(uuid4())[0:6],
Content={"ZipFile": b"zipfile"},
CompatibleRuntimes=["python3.11"],
LicenseInfo="MIT",
)["LayerArn"]
function_arn = conn.create_function(
FunctionName=lambda_name,
Runtime="python3.11",
Role=get_role_name(),
Handler="lambda_function.lambda_handler",
Code={"ZipFile": get_requests_zip_file()},
Timeout=3,
MemorySize=128,
Publish=True,
Layers=[f"{layer_arn}:1", f"{bogus_layer_arn}:1"],
)["FunctionArn"]
success_result = conn.invoke(
FunctionName=function_arn, Payload="{}", LogType="Tail"
)
msg = success_result["Payload"].read().decode("utf-8")
assert msg == '"2.32.4"'
|
