don't use table of test cases and reduce duplication in get_policy method
This commit is contained in:
parent
2a2ff32dec
commit
cb07d17e13
@ -14,22 +14,10 @@ class Policy:
|
|||||||
self.statements = []
|
self.statements = []
|
||||||
self.parent = parent
|
self.parent = parent
|
||||||
|
|
||||||
def __repr__(self):
|
|
||||||
return json.dumps(self.get_policy())
|
|
||||||
|
|
||||||
def wire_format(self):
|
def wire_format(self):
|
||||||
return json.dumps(
|
p = self.get_policy()
|
||||||
{
|
p["Policy"] = json.dumps(p["Policy"])
|
||||||
"Policy": json.dumps(
|
return json.dumps(p)
|
||||||
{
|
|
||||||
"Version": "2012-10-17",
|
|
||||||
"Id": "default",
|
|
||||||
"Statement": self.statements,
|
|
||||||
}
|
|
||||||
),
|
|
||||||
"RevisionId": self.revision,
|
|
||||||
}
|
|
||||||
)
|
|
||||||
|
|
||||||
def get_policy(self):
|
def get_policy(self):
|
||||||
return {
|
return {
|
||||||
@ -81,6 +69,7 @@ class Policy:
|
|||||||
# transform field names and values
|
# transform field names and values
|
||||||
self.transform_property(obj, "StatementId", "Sid", self.nop_formatter)
|
self.transform_property(obj, "StatementId", "Sid", self.nop_formatter)
|
||||||
self.transform_property(obj, "Principal", "Principal", self.principal_formatter)
|
self.transform_property(obj, "Principal", "Principal", self.principal_formatter)
|
||||||
|
|
||||||
self.transform_property(
|
self.transform_property(
|
||||||
obj, "SourceArn", "SourceArn", self.source_arn_formatter
|
obj, "SourceArn", "SourceArn", self.source_arn_formatter
|
||||||
)
|
)
|
||||||
|
@ -12,93 +12,39 @@ class MockLambdaFunction:
|
|||||||
self.policy = None
|
self.policy = None
|
||||||
|
|
||||||
|
|
||||||
class TC:
|
class TestPolicy(unittest.TestCase):
|
||||||
def __init__(self, lambda_arn, statement, expected):
|
def test_policy(self):
|
||||||
self.statement = statement
|
policy = Policy(MockLambdaFunction("arn"))
|
||||||
self.expected = expected
|
statement = {
|
||||||
self.fn = MockLambdaFunction(lambda_arn)
|
"StatementId": "statement0",
|
||||||
self.policy = Policy(self.fn)
|
"Action": "lambda:InvokeFunction",
|
||||||
|
"FunctionName": "function_name",
|
||||||
|
"Principal": "events.amazonaws.com",
|
||||||
|
"SourceArn": "arn:aws:events:us-east-1:111111111111:rule/rule_name",
|
||||||
|
"SourceAccount": "111111111111",
|
||||||
|
}
|
||||||
|
|
||||||
def Run(self, parent):
|
expected = {
|
||||||
self.policy.add_statement(json.dumps(self.statement))
|
"Action": "lambda:InvokeFunction",
|
||||||
parent.assertDictEqual(self.expected, self.policy.statements[0])
|
"FunctionName": "function_name",
|
||||||
|
"Principal": {"Service": "events.amazonaws.com"},
|
||||||
|
"Effect": "Allow",
|
||||||
|
"Resource": "arn:$LATEST",
|
||||||
|
"Sid": "statement0",
|
||||||
|
"Condition": {
|
||||||
|
"ArnLike": {
|
||||||
|
"AWS:SourceArn": "arn:aws:events:us-east-1:111111111111:rule/rule_name",
|
||||||
|
},
|
||||||
|
"StringEquals": {"AWS:SourceAccount": "111111111111"},
|
||||||
|
},
|
||||||
|
}
|
||||||
|
|
||||||
sid = self.statement.get("StatementId", None)
|
policy.add_statement(json.dumps(statement))
|
||||||
|
self.assertDictEqual(expected, policy.statements[0])
|
||||||
|
|
||||||
|
sid = statement.get("StatementId", None)
|
||||||
if sid == None:
|
if sid == None:
|
||||||
raise "TestCase.statement does not contain StatementId"
|
raise "TestCase.statement does not contain StatementId"
|
||||||
|
|
||||||
self.policy.del_statement(sid)
|
policy.del_statement(sid)
|
||||||
parent.assertEqual([], self.policy.statements)
|
self.assertEqual([], 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)
|
|
||||||
|
Loading…
Reference in New Issue
Block a user