Yasuke/moto
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

AWSLambda - refactor exception handling

Browse Source
This commit is contained in:
Bert Blommers 2019-11-17 10:59:20 +00:00
parent ad23b65784
commit c6dd3c96ea
4 changed files with 56 additions and 42 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

31
moto/awslambda/exceptions.py Normal file
View File

@ -0,0 +1,31 @@
from botocore.client import ClientError
class LambdaClientError(ClientError):
def __init__(self, error, message):
error_response = {"Error": {"Code": error, "Message": message}}
super(LambdaClientError, self).__init__(error_response, None)
class CrossAccountNotAllowed(LambdaClientError):
def __init__(self):
super(CrossAccountNotAllowed, self).__init__(
"AccessDeniedException", "Cross-account pass role is not allowed."
)
class InvalidParameterValueException(LambdaClientError):
def __init__(self, message):
super(InvalidParameterValueException, self).__init__(
"InvalidParameterValueException", message
)
class InvalidRoleFormat(LambdaClientError):
pattern = r"arn:(aws[a-zA-Z-]*)?:iam::(\d{12}):role/?[a-zA-Z_0-9+=,.@\-_/]+"
def __init__(self, role):
message = "1 validation error detected: Value '{0}' at 'role' failed to satisfy constraint: Member must satisfy regular expression pattern: {1}".format(
role, InvalidRoleFormat.pattern
)
super(InvalidRoleFormat, self).__init__("ValidationException", message)

29
moto/awslambda/models.py
View File

@ -33,6 +33,11 @@ from moto.s3.models import s3_backend
from moto.logs.models import logs_backends
from moto.s3.exceptions import MissingBucket, MissingKey
from moto import settings
from .exceptions import (
CrossAccountNotAllowed,
InvalidRoleFormat,
InvalidParameterValueException,
)
from .utils import make_function_arn, make_function_ver_arn
from moto.sqs import sqs_backends
from moto.dynamodb2 import dynamodb_backends2
@ -215,9 +220,8 @@ class LambdaFunction(BaseModel):
key = s3_backend.get_key(self.code["S3Bucket"], self.code["S3Key"])
except MissingBucket:
if do_validate_s3():
raise ValueError(
"InvalidParameterValueException",
"Error occurred while GetObject. S3 Error Code: NoSuchBucket. S3 Error Message: The specified bucket does not exist",
raise InvalidParameterValueException(
"Error occurred while GetObject. S3 Error Code: NoSuchBucket. S3 Error Message: The specified bucket does not exist"
)
except MissingKey:
if do_validate_s3():
@ -669,28 +673,19 @@ class LambdaStorage(object):
:param fn: Function
:type fn: LambdaFunction
"""
pattern = r"arn:(aws[a-zA-Z-]*)?:iam::(\d{12}):role/?[a-zA-Z_0-9+=,.@\-_/]+"
valid_role = re.match(pattern, fn.role)
valid_role = re.match(InvalidRoleFormat.pattern, fn.role)
if valid_role:
account = valid_role.group(2)
if account != ACCOUNT_ID:
raise ValueError(
"AccessDeniedException", "Cross-account pass role is not allowed."
)
raise CrossAccountNotAllowed()
try:
iam_backend.get_role_by_arn(fn.role)
except IAMNotFoundException:
raise ValueError(
"InvalidParameterValueException",
"The role defined for the function cannot be assumed by Lambda.",
raise InvalidParameterValueException(
"The role defined for the function cannot be assumed by Lambda."
)
else:
raise ValueError(
"ValidationException",
"1 validation error detected: Value '{0}' at 'role' failed to satisfy constraint: Member must satisfy regular expression pattern: {1}".format(
fn.role, pattern
),
)
raise InvalidRoleFormat(fn.role)
if fn.function_name in self._functions:
self._functions[fn.function_name]["latest"] = fn
else:

24
moto/awslambda/responses.py
View File

@ -211,26 +211,14 @@ class LambdaResponse(BaseResponse):
return 200, {}, json.dumps(result)
def _create_function(self, request, full_url, headers):
try:
fn = self.lambda_backend.create_function(self.json_body)
except ValueError as e:
return 400, {}, json.dumps({"__type": e.args[0], "message": e.args[1]})
else:
config = fn.get_configuration()
return 201, {}, json.dumps(config)
fn = self.lambda_backend.create_function(self.json_body)
config = fn.get_configuration()
return 201, {}, json.dumps(config)
def _create_event_source_mapping(self, request, full_url, headers):
try:
fn = self.lambda_backend.create_event_source_mapping(self.json_body)
except ValueError as e:
return (
400,
{},
json.dumps({"Error": {"Code": e.args[0], "Message": e.args[1]}}),
)
else:
config = fn.get_configuration()
return 201, {}, json.dumps(config)
fn = self.lambda_backend.create_event_source_mapping(self.json_body)
config = fn.get_configuration()
return 201, {}, json.dumps(config)
def _list_event_source_mappings(self, event_source_arn, function_name):
esms = self.lambda_backend.list_event_source_mappings(

14
tests/test_cloudformation/test_cloudformation_stack_crud.py
View File

@ -4,6 +4,7 @@ import os
import json
import boto
import boto3
import boto.iam
import boto.s3
import boto.s3.key
@ -19,9 +20,9 @@ from moto import (
mock_cloudformation_deprecated,
mock_s3_deprecated,
mock_route53_deprecated,
mock_iam,
)
from moto.cloudformation import cloudformation_backends
from moto.iam import mock_iam_deprecated
dummy_template = {
"AWSTemplateFormatVersion": "2010-09-09",
@ -596,9 +597,8 @@ def test_create_stack_kinesis():
def get_role_name():
with mock_iam_deprecated():
iam = boto.connect_iam()
role = iam.create_role("my-role")["create_role_response"]["create_role_result"][
"role"
]["arn"]
return role
with mock_iam():
iam = boto3.client("iam")
return iam.create_role(
RoleName="TestRole", AssumeRolePolicyDocument="some_policy"
)["Role"]["Arn"]