-
Notifications
You must be signed in to change notification settings - Fork 2.4k
/
test_api_resource.py
143 lines (117 loc) · 6.31 KB
/
test_api_resource.py
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
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
import json
import os
from unittest import TestCase
from unittest.mock import MagicMock, patch
from tests.translator.helpers import get_template_parameter_values
from samtranslator.translator.transform import transform
from samtranslator.model.apigateway import ApiGatewayDeployment
from tests.plugins.application.test_serverless_app_plugin import mock_get_region
mock_policy_loader = MagicMock()
mock_policy_loader.load.return_value = {
"AmazonDynamoDBFullAccess": "arn:aws:iam::aws:policy/AmazonDynamoDBFullAccess",
"AmazonDynamoDBReadOnlyAccess": "arn:aws:iam::aws:policy/AmazonDynamoDBReadOnlyAccess",
"AWSLambdaRole": "arn:aws:iam::aws:policy/service-role/AWSLambdaRole",
}
@patch("botocore.client.ClientEndpointBridge._check_default_region", mock_get_region)
def test_redeploy_explicit_api():
"""
Test to verify that we will redeploy an API when Swagger document changes
:return:
"""
manifest = {
"Transform": "AWS::Serverless-2016-10-31",
"Resources": {
"ExplicitApi": {
"Type": "AWS::Serverless::Api",
"Properties": {"StageName": "prod", "DefinitionUri": "s3://mybucket/swagger.json?versionId=123"},
}
},
}
original_deployment_ids = translate_and_find_deployment_ids(manifest)
# Now update the API specification. This should redeploy the API by creating a new deployment resource
manifest["Resources"]["ExplicitApi"]["Properties"]["DefinitionUri"] = "s3://mybucket/swagger.json?versionId=456"
updated_deployment_ids = translate_and_find_deployment_ids(manifest)
assert original_deployment_ids != updated_deployment_ids
# Now, update an unrelated property. This should NOT generate a new deploymentId
manifest["Resources"]["ExplicitApi"]["Properties"]["StageName"] = "newStageName"
assert updated_deployment_ids == translate_and_find_deployment_ids(manifest)
@patch("botocore.client.ClientEndpointBridge._check_default_region", mock_get_region)
def test_redeploy_implicit_api():
manifest = {
"Transform": "AWS::Serverless-2016-10-31",
"Resources": {
"FirstLambdaFunction": {
"Type": "AWS::Serverless::Function",
"Properties": {
"FunctionName": "InitialFunction",
"CodeUri": "s3://bucket/code.zip",
"Handler": "index.handler",
"Runtime": "nodejs12.x",
"Events": {"MyApi": {"Type": "Api", "Properties": {"Path": "/first", "Method": "get"}}},
},
},
"SecondLambdaFunction": {
"Type": "AWS::Serverless::Function",
"Properties": {
"CodeUri": "s3://bucket/code.zip",
"Handler": "index.handler",
"Runtime": "nodejs12.x",
"Events": {"MyApi": {"Type": "Api", "Properties": {"Path": "/second", "Method": "get"}}},
},
},
},
}
original_deployment_ids = translate_and_find_deployment_ids(manifest)
# Update API of one Lambda function should redeploy API
manifest["Resources"]["FirstLambdaFunction"]["Properties"]["Events"]["MyApi"]["Properties"]["Method"] = "post"
first_updated_deployment_ids = translate_and_find_deployment_ids(manifest)
assert original_deployment_ids != first_updated_deployment_ids
# Update API of second Lambda function should redeploy API
manifest["Resources"]["SecondLambdaFunction"]["Properties"]["Events"]["MyApi"]["Properties"]["Method"] = "post"
second_updated_deployment_ids = translate_and_find_deployment_ids(manifest)
assert first_updated_deployment_ids != second_updated_deployment_ids
# Now, update an unrelated property. This should NOT generate a new deploymentId
manifest["Resources"]["SecondLambdaFunction"]["Properties"]["Runtime"] = "java"
assert second_updated_deployment_ids == translate_and_find_deployment_ids(manifest)
manifest["Resources"]["FirstLambdaFunction"]["Properties"]["FunctionName"] = "ChangedFunctionName"
fourth_updated_deployment_ids = translate_and_find_deployment_ids(manifest)
assert fourth_updated_deployment_ids != second_updated_deployment_ids
@patch("boto3.session.Session.region_name", "ap-southeast-1")
def translate_and_find_deployment_ids(manifest):
parameter_values = get_template_parameter_values()
output_fragment = transform(manifest, parameter_values, mock_policy_loader)
print(json.dumps(output_fragment, indent=2))
deployment_ids = set()
for key, value in output_fragment["Resources"].items():
if value["Type"] == "AWS::ApiGateway::Deployment":
deployment_ids.add(key)
return deployment_ids
class TestApiGatewayDeploymentResource(TestCase):
@patch("samtranslator.translator.logical_id_generator.LogicalIdGenerator")
def test_make_auto_deployable_with_swagger_dict(self, LogicalIdGeneratorMock):
prefix = "prefix"
generator_mock = LogicalIdGeneratorMock.return_value
stage = MagicMock()
id_val = "SomeLogicalId"
full_hash = "127e3fb91142ab1ddc5f5446adb094442581a90d"
generator_mock.gen.return_value = id_val
generator_mock.get_hash.return_value = full_hash
swagger = {"a": "b"}
deployment = ApiGatewayDeployment(logical_id=prefix)
deployment.make_auto_deployable(stage, swagger=swagger)
self.assertEqual(deployment.logical_id, id_val)
self.assertEqual(deployment.Description, "RestApi deployment id: {}".format(full_hash))
LogicalIdGeneratorMock.assert_called_once_with(prefix, str(swagger))
generator_mock.gen.assert_called_once_with()
generator_mock.get_hash.assert_called_once_with(length=40) # getting full SHA
stage.update_deployment_ref.assert_called_once_with(id_val)
@patch("samtranslator.translator.logical_id_generator.LogicalIdGenerator")
def test_make_auto_deployable_no_swagger(self, LogicalIdGeneratorMock):
prefix = "prefix"
stage = MagicMock()
deployment = ApiGatewayDeployment(logical_id=prefix)
deployment.make_auto_deployable(stage, swagger=None)
self.assertEqual(deployment.logical_id, prefix)
self.assertEqual(deployment.Description, None)
LogicalIdGeneratorMock.assert_not_called()
stage.update_deployment_ref.assert_not_called()