|
| 1 | +from aws_lambda_powertools.utilities.parser import envelopes, event_parser |
| 2 | +from aws_lambda_powertools.utilities.parser.models import ( |
| 3 | + APIGatewayProxyEventV2Model, |
| 4 | + RequestContextV2, |
| 5 | + RequestContextV2Authorizer, |
| 6 | +) |
| 7 | +from aws_lambda_powertools.utilities.typing import LambdaContext |
| 8 | +from tests.functional.parser.schemas import MyApiGatewayBusiness |
| 9 | +from tests.functional.utils import load_event |
| 10 | + |
| 11 | + |
| 12 | +@event_parser(model=MyApiGatewayBusiness, envelope=envelopes.ApiGatewayV2Envelope) |
| 13 | +def handle_apigw_with_envelope(event: MyApiGatewayBusiness, _: LambdaContext): |
| 14 | + assert event.message == "Hello" |
| 15 | + assert event.username == "Ran" |
| 16 | + |
| 17 | + |
| 18 | +@event_parser(model=APIGatewayProxyEventV2Model) |
| 19 | +def handle_apigw_event(event: APIGatewayProxyEventV2Model, _: LambdaContext): |
| 20 | + return event |
| 21 | + |
| 22 | + |
| 23 | +def test_apigw_v2_event_with_envelope(): |
| 24 | + event = load_event("apiGatewayProxyV2Event.json") |
| 25 | + event["body"] = '{"message": "Hello", "username": "Ran"}' |
| 26 | + handle_apigw_with_envelope(event, LambdaContext()) |
| 27 | + |
| 28 | + |
| 29 | +def test_apigw_v2_event_jwt_authorizer(): |
| 30 | + event = load_event("apiGatewayProxyV2Event.json") |
| 31 | + parsed_event: APIGatewayProxyEventV2Model = handle_apigw_event(event, LambdaContext()) |
| 32 | + assert parsed_event.version == event["version"] |
| 33 | + assert parsed_event.routeKey == event["routeKey"] |
| 34 | + assert parsed_event.rawPath == event["rawPath"] |
| 35 | + assert parsed_event.rawQueryString == event["rawQueryString"] |
| 36 | + assert parsed_event.cookies == event["cookies"] |
| 37 | + assert parsed_event.cookies[0] == "cookie1" |
| 38 | + assert parsed_event.headers == event["headers"] |
| 39 | + assert parsed_event.queryStringParameters == event["queryStringParameters"] |
| 40 | + assert parsed_event.queryStringParameters["parameter2"] == "value" |
| 41 | + |
| 42 | + request_context = parsed_event.requestContext |
| 43 | + assert request_context.accountId == event["requestContext"]["accountId"] |
| 44 | + assert request_context.apiId == event["requestContext"]["apiId"] |
| 45 | + assert request_context.authorizer.jwt.claims == event["requestContext"]["authorizer"]["jwt"]["claims"] |
| 46 | + assert request_context.authorizer.jwt.scopes == event["requestContext"]["authorizer"]["jwt"]["scopes"] |
| 47 | + assert request_context.domainName == event["requestContext"]["domainName"] |
| 48 | + assert request_context.domainPrefix == event["requestContext"]["domainPrefix"] |
| 49 | + |
| 50 | + http = request_context.http |
| 51 | + assert http.method == "POST" |
| 52 | + assert http.path == "/my/path" |
| 53 | + assert http.protocol == "HTTP/1.1" |
| 54 | + assert str(http.sourceIp) == "192.168.0.1/32" |
| 55 | + assert http.userAgent == "agent" |
| 56 | + |
| 57 | + assert request_context.requestId == event["requestContext"]["requestId"] |
| 58 | + assert request_context.routeKey == event["requestContext"]["routeKey"] |
| 59 | + assert request_context.stage == event["requestContext"]["stage"] |
| 60 | + assert request_context.time == event["requestContext"]["time"] |
| 61 | + convert_time = int(round(request_context.timeEpoch.timestamp() * 1000)) |
| 62 | + assert convert_time == event["requestContext"]["timeEpoch"] |
| 63 | + assert parsed_event.body == event["body"] |
| 64 | + assert parsed_event.pathParameters == event["pathParameters"] |
| 65 | + assert parsed_event.isBase64Encoded == event["isBase64Encoded"] |
| 66 | + assert parsed_event.stageVariables == event["stageVariables"] |
| 67 | + |
| 68 | + |
| 69 | +def test_api_gateway_proxy_v2_event_lambda_authorizer(): |
| 70 | + event = load_event("apiGatewayProxyV2LambdaAuthorizerEvent.json") |
| 71 | + parsed_event: APIGatewayProxyEventV2Model = handle_apigw_event(event, LambdaContext()) |
| 72 | + request_context: RequestContextV2 = parsed_event.requestContext |
| 73 | + assert request_context is not None |
| 74 | + lambda_props: RequestContextV2Authorizer = request_context.authorizer.lambda_value |
| 75 | + assert lambda_props is not None |
| 76 | + assert lambda_props["key"] == "value" |
| 77 | + |
| 78 | + |
| 79 | +def test_api_gateway_proxy_v2_event_iam_authorizer(): |
| 80 | + event = load_event("apiGatewayProxyV2IamEvent.json") |
| 81 | + parsed_event: APIGatewayProxyEventV2Model = handle_apigw_event(event, LambdaContext()) |
| 82 | + iam = parsed_event.requestContext.authorizer.iam |
| 83 | + assert iam is not None |
| 84 | + assert iam.accessKey == "ARIA2ZJZYVUEREEIHAKY" |
| 85 | + assert iam.accountId == "1234567890" |
| 86 | + assert iam.callerId == "AROA7ZJZYVRE7C3DUXHH6:CognitoIdentityCredentials" |
| 87 | + assert iam.cognitoIdentity.amr == ["foo"] |
| 88 | + assert iam.cognitoIdentity.identityId == "us-east-1:3f291106-8703-466b-8f2b-3ecee1ca56ce" |
| 89 | + assert iam.cognitoIdentity.identityPoolId == "us-east-1:4f291106-8703-466b-8f2b-3ecee1ca56ce" |
| 90 | + assert iam.principalOrgId == "AwsOrgId" |
| 91 | + assert iam.userArn == "arn:aws:iam::1234567890:user/Admin" |
| 92 | + assert iam.userId == "AROA2ZJZYVRE7Y3TUXHH6" |
0 commit comments