forked from aws-powertools/powertools-lambda-python
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathtest_apigwv2.py
More file actions
113 lines (90 loc) · 5.24 KB
/
test_apigwv2.py
File metadata and controls
113 lines (90 loc) · 5.24 KB
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
from aws_lambda_powertools.utilities.parser import envelopes, parse
from aws_lambda_powertools.utilities.parser.models import (
APIGatewayProxyEventV2Model,
RequestContextV2,
RequestContextV2Authorizer,
)
from tests.functional.utils import load_event
from tests.unit.parser.schemas import MyApiGatewayBusiness
def test_apigw_v2_event_with_envelope():
raw_event = load_event("apiGatewayProxyV2Event.json")
raw_event["body"] = '{"message": "Hello", "username": "Ran"}'
parsed_event: MyApiGatewayBusiness = parse(
event=raw_event,
model=MyApiGatewayBusiness,
envelope=envelopes.ApiGatewayV2Envelope,
)
assert parsed_event.message == "Hello"
assert parsed_event.username == "Ran"
def test_apigw_v2_event_jwt_authorizer():
raw_event = load_event("apiGatewayProxyV2Event.json")
parsed_event: APIGatewayProxyEventV2Model = APIGatewayProxyEventV2Model(**raw_event)
assert parsed_event.version == raw_event["version"]
assert parsed_event.routeKey == raw_event["routeKey"]
assert parsed_event.rawPath == raw_event["rawPath"]
assert parsed_event.rawQueryString == raw_event["rawQueryString"]
assert parsed_event.cookies == raw_event["cookies"]
assert parsed_event.cookies[0] == "cookie1"
assert parsed_event.headers == raw_event["headers"]
assert parsed_event.queryStringParameters == raw_event["queryStringParameters"]
assert parsed_event.queryStringParameters.get("parameter2") == raw_event["queryStringParameters"]["parameter2"]
request_context = parsed_event.requestContext
assert request_context.accountId == raw_event["requestContext"]["accountId"]
assert request_context.apiId == raw_event["requestContext"]["apiId"]
assert request_context.authorizer.jwt.claims == raw_event["requestContext"]["authorizer"]["jwt"]["claims"]
assert request_context.authorizer.jwt.scopes == raw_event["requestContext"]["authorizer"]["jwt"]["scopes"]
assert request_context.domainName == raw_event["requestContext"]["domainName"]
assert request_context.domainPrefix == raw_event["requestContext"]["domainPrefix"]
http = request_context.http
raw_http = raw_event["requestContext"]["http"]
assert http.method == raw_http["method"]
assert http.path == raw_http["path"]
assert http.protocol == raw_http["protocol"]
assert str(http.sourceIp) == raw_http["sourceIp"]
assert http.userAgent == raw_http["userAgent"]
assert request_context.requestId == raw_event["requestContext"]["requestId"]
assert request_context.routeKey == raw_event["requestContext"]["routeKey"]
assert request_context.stage == raw_event["requestContext"]["stage"]
assert request_context.time == raw_event["requestContext"]["time"]
convert_time = int(round(request_context.timeEpoch.timestamp() * 1000))
assert convert_time == raw_event["requestContext"]["timeEpoch"]
assert parsed_event.body == raw_event["body"]
assert parsed_event.pathParameters == raw_event["pathParameters"]
assert parsed_event.isBase64Encoded == raw_event["isBase64Encoded"]
assert parsed_event.stageVariables == raw_event["stageVariables"]
def test_apigw_v2_event_empty_jwt_scopes():
raw_event = load_event("apiGatewayProxyV2Event.json")
raw_event["requestContext"]["authorizer"]["jwt"]["scopes"] = None
APIGatewayProxyEventV2Model(**raw_event)
def test_api_gateway_proxy_v2_event_lambda_authorizer():
raw_event = load_event("apiGatewayProxyV2LambdaAuthorizerEvent.json")
parsed_event: APIGatewayProxyEventV2Model = APIGatewayProxyEventV2Model(**raw_event)
request_context: RequestContextV2 = parsed_event.requestContext
assert request_context is not None
lambda_props: RequestContextV2Authorizer = request_context.authorizer.lambda_value
assert lambda_props is not None
assert lambda_props["key"] == raw_event["requestContext"]["authorizer"]["lambda"]["key"]
def test_api_gateway_proxy_v2_event_iam_authorizer():
raw_event = load_event("apiGatewayProxyV2IamEvent.json")
parsed_event: APIGatewayProxyEventV2Model = APIGatewayProxyEventV2Model(**raw_event)
iam = parsed_event.requestContext.authorizer.iam
raw_iam = raw_event["requestContext"]["authorizer"]["iam"]
assert iam is not None
assert iam.accessKey == raw_iam["accessKey"]
assert iam.accountId == raw_iam["accountId"]
assert iam.callerId == raw_iam["callerId"]
assert iam.cognitoIdentity.amr == raw_iam["cognitoIdentity"]["amr"]
assert iam.cognitoIdentity.identityId == raw_iam["cognitoIdentity"]["identityId"]
assert iam.cognitoIdentity.identityPoolId == raw_iam["cognitoIdentity"]["identityPoolId"]
assert iam.principalOrgId == raw_iam["principalOrgId"]
assert iam.userArn == raw_iam["userArn"]
assert iam.userId == raw_iam["userId"]
def test_apigw_event_empty_body():
raw_event = load_event("apiGatewayProxyV2Event.json")
raw_event.pop("body") # API GW v2 removes certain keys when no data is passed
parse(event=raw_event, model=APIGatewayProxyEventV2Model)
def test_apigw_event_empty_query_strings():
raw_event = load_event("apiGatewayProxyV2Event.json")
raw_event["rawQueryString"] = ""
raw_event.pop("queryStringParameters") # API GW v2 removes certain keys when no data is passed
parse(event=raw_event, model=APIGatewayProxyEventV2Model)