catalin/moto
1
0
Fork 0
Code Issues Pull requests Projects Releases Wiki Activity

improves support for AWS lambda policy management

Browse source
This commit is contained in:
Brady 2020-01-23 12:46:24 -06:00
commit 2a2ff32dec
7 changed files with 310 additions and 18 deletions
Download patch file Download diff file Expand all files Collapse all files

6
tests/test_awslambda/test_lambda.py
View file

@ -324,6 +324,7 @@ def test_create_function_from_aws_bucket():
"VpcId": "vpc-123abc",
},
"ResponseMetadata": {"HTTPStatusCode": 201},
"State": "Active",
}
)
@ -367,6 +368,7 @@ def test_create_function_from_zipfile():
"Version": "1",
"VpcConfig": {"SecurityGroupIds": [], "SubnetIds": []},
"ResponseMetadata": {"HTTPStatusCode": 201},
"State": "Active",
}
)
@ -631,6 +633,7 @@ def test_list_create_list_get_delete_list():
"Timeout": 3,
"Version": "$LATEST",
"VpcConfig": {"SecurityGroupIds": [], "SubnetIds": []},
"State": "Active",
},
"ResponseMetadata": {"HTTPStatusCode": 200},
}
@ -827,6 +830,7 @@ def test_get_function_created_with_zipfile():
"Timeout": 3,
"Version": "$LATEST",
"VpcConfig": {"SecurityGroupIds": [], "SubnetIds": []},
"State": "Active",
}
)
@ -1436,6 +1440,7 @@ def test_update_function_zip():
"Timeout": 3,
"Version": "2",
"VpcConfig": {"SecurityGroupIds": [], "SubnetIds": []},
"State": "Active",
}
)
@ -1498,6 +1503,7 @@ def test_update_function_s3():
"Timeout": 3,
"Version": "2",
"VpcConfig": {"SecurityGroupIds": [], "SubnetIds": []},
"State": "Active",
}
)

104
tests/test_awslambda/test_policy.py Normal file
View file

@ -0,0 +1,104 @@
from __future__ import unicode_literals
import unittest
import json
from moto.awslambda.policy import Policy
class MockLambdaFunction:
def __init__(self, arn):
self.function_arn = arn
self.policy = None
class TC:
def __init__(self, lambda_arn, statement, expected):
self.statement = statement
self.expected = expected
self.fn = MockLambdaFunction(lambda_arn)
self.policy = Policy(self.fn)
def Run(self, parent):
self.policy.add_statement(json.dumps(self.statement))
parent.assertDictEqual(self.expected, self.policy.statements[0])
sid = self.statement.get("StatementId", None)
if sid == None:
raise "TestCase.statement does not contain StatementId"
self.policy.del_statement(sid)
parent.assertEqual([], self.policy.statements)
class TestPolicy(unittest.TestCase):
def test(self):
tt = [
TC(
# lambda_arn
"arn",
{ # statement
"StatementId": "statement0",
"Action": "lambda:InvokeFunction",
"FunctionName": "function_name",
"Principal": "events.amazonaws.com",
},
{ # expected
"Action": "lambda:InvokeFunction",
"FunctionName": "function_name",
"Principal": {"Service": "events.amazonaws.com"},
"Effect": "Allow",
"Resource": "arn:$LATEST",
"Sid": "statement0",
},
),
TC(
# lambda_arn
"arn",
{ # statement
"StatementId": "statement1",
"Action": "lambda:InvokeFunction",
"FunctionName": "function_name",
"Principal": "events.amazonaws.com",
"SourceArn": "arn:aws:events:us-east-1:111111111111:rule/rule_name",
},
{
"Action": "lambda:InvokeFunction",
"FunctionName": "function_name",
"Principal": {"Service": "events.amazonaws.com"},
"Effect": "Allow",
"Resource": "arn:$LATEST",
"Sid": "statement1",
"Condition": {
"ArnLike": {
"AWS:SourceArn": "arn:aws:events:us-east-1:111111111111:rule/rule_name"
}
},
},
),
TC(
# lambda_arn
"arn",
{ # statement
"StatementId": "statement2",
"Action": "lambda:InvokeFunction",
"FunctionName": "function_name",
"Principal": "events.amazonaws.com",
"SourceAccount": "111111111111",
},
{ # expected
"Action": "lambda:InvokeFunction",
"FunctionName": "function_name",
"Principal": {"Service": "events.amazonaws.com"},
"Effect": "Allow",
"Resource": "arn:$LATEST",
"Sid": "statement2",
"Condition": {
"StringEquals": {"AWS:SourceAccount": "111111111111"}
},
},
),
]
for tc in tt:
tc.Run(self)