-
Notifications
You must be signed in to change notification settings - Fork 1.2k
/
test_runtimes_e2e.py
192 lines (165 loc) · 8.81 KB
/
test_runtimes_e2e.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
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
from distutils.dir_util import copy_tree
from unittest import skipIf
import json
from pathlib import Path
import os
from parameterized import parameterized_class
from tests.end_to_end.end_to_end_base import EndToEndBase
from tests.end_to_end.end_to_end_context import EndToEndTestContext
from tests.end_to_end.test_stages import (
DefaultInitStage,
PackageDownloadZipFunctionStage,
DefaultDeleteStage,
EndToEndBaseStage,
DefaultSyncStage,
BaseValidator,
)
from tests.testing_utils import RUNNING_ON_CI, RUNNING_TEST_FOR_MASTER_ON_CI, RUN_BY_CANARY
# Deploy tests require credentials and CI/CD will only add credentials to the env if the PR is from the same repo.
# This is to restrict package tests to run outside of CI/CD, when the branch is not master or tests are not run by Canary
SKIP_E2E_TESTS = RUNNING_ON_CI and RUNNING_TEST_FOR_MASTER_ON_CI and not RUN_BY_CANARY
from tests.testing_utils import CommandResult
class InitValidator(BaseValidator):
def validate(self, command_result: CommandResult):
self.assertEqual(command_result.process.returncode, 0)
self.assertTrue(Path(self.test_context.working_directory).is_dir())
self.assertTrue(Path(self.test_context.project_directory).is_dir())
class BuildValidator(BaseValidator):
def validate(self, command_result: CommandResult):
self.assertEqual(command_result.process.returncode, 0)
build_dir = Path(self.test_context.project_directory) / ".aws-sam"
self.assertTrue(build_dir.is_dir())
class LocalInvokeValidator(BaseValidator):
def validate(self, command_result: CommandResult):
response = json.loads(command_result.stdout.decode("utf-8").split("\n")[-1])
self.assertEqual(command_result.process.returncode, 0)
self.assertEqual(response["statusCode"], 200)
class RemoteInvokeValidator(BaseValidator):
def validate(self, command_result: CommandResult):
response = json.loads(command_result.stdout.decode("utf-8"))
self.assertEqual(command_result.process.returncode, 0)
self.assertEqual(response["StatusCode"], 200)
self.assertEqual(response.get("FunctionError", ""), "")
class StackOutputsValidator(BaseValidator):
def validate(self, command_result: CommandResult):
self.assertEqual(command_result.process.returncode, 0)
stack_outputs = json.loads(command_result.stdout.decode())
self.assertEqual(len(stack_outputs), 3)
for output in stack_outputs:
self.assertIn("OutputKey", output)
self.assertIn("OutputValue", output)
self.assertIn("Description", output)
@skipIf(SKIP_E2E_TESTS, "Skip E2E tests in CI/CD only")
@parameterized_class(
("runtime", "dependency_manager"),
[
("go1.x", "mod"),
("python3.7", "pip"),
],
)
class TestHelloWorldDefaultEndToEnd(EndToEndBase):
app_template = "hello-world"
def test_hello_world_default_workflow(self):
stack_name = self._method_to_stack_name(self.id())
function_name = "HelloWorldFunction"
event = '{"hello": "world"}'
with EndToEndTestContext(self.app_name) as e2e_context:
self.template_path = e2e_context.template_path
init_command_list = self._get_init_command(e2e_context.working_directory)
build_command_list = self.get_command_list()
deploy_command_list = self._get_deploy_command(stack_name)
stack_outputs_command_list = self._get_stack_outputs_command(stack_name)
remote_invoke_command_list = self._get_remote_invoke_command(stack_name, function_name, event, "json")
delete_command_list = self._get_delete_command(stack_name)
stages = [
DefaultInitStage(InitValidator(e2e_context), e2e_context, init_command_list, self.app_name),
EndToEndBaseStage(BuildValidator(e2e_context), e2e_context, build_command_list),
EndToEndBaseStage(BaseValidator(e2e_context), e2e_context, deploy_command_list),
EndToEndBaseStage(RemoteInvokeValidator(e2e_context), e2e_context, remote_invoke_command_list),
EndToEndBaseStage(BaseValidator(e2e_context), e2e_context, stack_outputs_command_list),
DefaultDeleteStage(BaseValidator(e2e_context), e2e_context, delete_command_list, stack_name),
]
self._run_tests(stages)
@skipIf(SKIP_E2E_TESTS, "Skip E2E tests in CI/CD only")
@parameterized_class(
("runtime", "dependency_manager"),
[
("go1.x", "mod"),
("python3.7", "pip"),
],
)
class TestHelloWorldZipPackagePermissionsEndToEnd(EndToEndBase):
"""This end to end test is to ensure the zip file created using sam package
has the required permissions to invoke the Function.
"""
app_template = "hello-world"
def test_hello_world_workflow(self):
function_name = "HelloWorldFunction"
with EndToEndTestContext(self.app_name) as e2e_context:
self.template_path = e2e_context.template_path
init_command_list = self._get_init_command(e2e_context.working_directory)
build_command_list = self.get_command_list()
package_command_list = self._get_package_command(
s3_prefix="end-to-end-package-test", use_json=True, output_template_file="packaged_template.json"
)
local_command_list = self._get_local_command(function_name)
stages = [
DefaultInitStage(InitValidator(e2e_context), e2e_context, init_command_list, self.app_name),
EndToEndBaseStage(BuildValidator(e2e_context), e2e_context, build_command_list),
PackageDownloadZipFunctionStage(
BaseValidator(e2e_context), e2e_context, package_command_list, function_name
),
EndToEndBaseStage(LocalInvokeValidator(e2e_context), e2e_context, local_command_list),
]
self._run_tests(stages)
@skipIf(SKIP_E2E_TESTS, "Skip E2E tests in CI/CD only")
@parameterized_class(
("runtime", "dependency_manager"),
[
("go1.x", "mod"),
("python3.7", "pip"),
],
)
class TestHelloWorldDefaultSyncEndToEnd(EndToEndBase):
app_template = "hello-world"
def test_go_hello_world_default_workflow(self):
function_name = "HelloWorldFunction"
event = '{"hello": "world"}'
stack_name = self._method_to_stack_name(self.id())
with EndToEndTestContext(self.app_name) as e2e_context:
self.template_path = e2e_context.template_path
init_command_list = self._get_init_command(e2e_context.working_directory)
sync_command_list = self._get_sync_command(stack_name)
stack_outputs_command_list = self._get_stack_outputs_command(stack_name)
remote_invoke_command_list = self._get_remote_invoke_command(stack_name, function_name, event, "json")
delete_command_list = self._get_delete_command(stack_name)
stages = [
DefaultInitStage(InitValidator(e2e_context), e2e_context, init_command_list, self.app_name),
DefaultSyncStage(BaseValidator(e2e_context), e2e_context, sync_command_list),
EndToEndBaseStage(RemoteInvokeValidator(e2e_context), e2e_context, remote_invoke_command_list),
EndToEndBaseStage(BaseValidator(e2e_context), e2e_context, stack_outputs_command_list),
DefaultDeleteStage(BaseValidator(e2e_context), e2e_context, delete_command_list, stack_name),
]
self._run_tests(stages)
class TestEsbuildDatadogLayerIntegration(EndToEndBase):
app_template = ""
def test_integration(self):
function_name = "HelloWorldFunction"
event = '{"hello": "world"}'
stack_name = self._method_to_stack_name(self.id())
with EndToEndTestContext(self.app_name) as e2e_context:
project_path = str(self.e2e_test_data_path / "esbuild-datadog-integration")
os.mkdir(e2e_context.project_directory)
copy_tree(project_path, e2e_context.project_directory)
self.template_path = e2e_context.template_path
build_command_list = self.get_command_list()
deploy_command_list = self._get_deploy_command(stack_name)
remote_invoke_command_list = self._get_remote_invoke_command(stack_name, function_name, event, "json")
delete_command_list = self._get_delete_command(stack_name)
stages = [
EndToEndBaseStage(BuildValidator(e2e_context), e2e_context, build_command_list),
EndToEndBaseStage(BaseValidator(e2e_context), e2e_context, deploy_command_list),
EndToEndBaseStage(RemoteInvokeValidator(e2e_context), e2e_context, remote_invoke_command_list),
DefaultDeleteStage(BaseValidator(e2e_context), e2e_context, delete_command_list, stack_name),
]
self._run_tests(stages)