-
Notifications
You must be signed in to change notification settings - Fork 1.2k
/
test_validate_command.py
211 lines (189 loc) · 8.82 KB
/
test_validate_command.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
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
"""
Integration tests for sam validate
"""
import json
import os
import re
import tempfile
from enum import Enum, auto
from pathlib import Path
from typing import List, Optional
from unittest import TestCase
from unittest.case import skipIf
from parameterized import parameterized
from tests.testing_utils import (
RUN_BY_CANARY,
RUNNING_ON_CI,
RUNNING_TEST_FOR_MASTER_ON_CI,
run_command,
get_sam_command,
)
# Validate 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_VALIDATE_TESTS = RUNNING_ON_CI and RUNNING_TEST_FOR_MASTER_ON_CI and not RUN_BY_CANARY
class TemplateFileTypes(Enum):
JSON = auto()
YAML = auto()
@skipIf(SKIP_VALIDATE_TESTS, "Skip validate tests in CI/CD only")
class TestValidate(TestCase):
@classmethod
def setUpClass(cls):
cls.patterns = {
TemplateFileTypes.JSON: re.compile(
r"template\.json is a valid SAM Template. This is according to basic SAM Validation, "
'for additional validation, please run with "--lint" option(\r\n)?$'
),
TemplateFileTypes.YAML: re.compile(
r"template\.yaml is a valid SAM Template. This is according to basic SAM Validation, "
'for additional validation, please run with "--lint" option(\r\n)?$'
),
}
cls.lint_patterns = {
TemplateFileTypes.JSON: re.compile(r"template\.json is a valid SAM Template(\r\n)?$"),
TemplateFileTypes.YAML: re.compile(r"template\.yaml is a valid SAM Template(\r\n)?$"),
}
def command_list(
self,
template_file: Optional[Path] = None,
profile: Optional[str] = None,
region: Optional[str] = None,
config_file: Optional[Path] = None,
lint: Optional[bool] = None,
) -> List[str]:
command_list = [get_sam_command(), "validate"]
if template_file:
command_list += ["--template-file", str(template_file)]
if profile:
command_list += ["--profile", profile]
if region:
command_list += ["--region", region]
if config_file:
command_list += ["--config_file", str(config_file)]
if lint:
command_list += ["--lint"]
return command_list
@parameterized.expand(
[
("default_yaml", TemplateFileTypes.YAML), # project with template.yaml
("default_json", TemplateFileTypes.JSON), # project with template.json
("multiple_files", TemplateFileTypes.YAML), # project with both template.yaml and template.json
(
"with_build",
TemplateFileTypes.JSON,
), # project with template.json and standard build directory .aws-sam/build/template.yaml
]
)
def test_default_template_file_choice(self, relative_folder: str, expected_file: TemplateFileTypes):
test_data_path = Path(__file__).resolve().parents[2] / "integration" / "testdata" / "validate"
process_dir = test_data_path / relative_folder
command_result = run_command(self.command_list(), cwd=str(process_dir))
pattern = self.patterns[expected_file] # type: ignore
output = command_result.stdout.decode("utf-8")
self.assertEqual(command_result.process.returncode, 0)
self.assertRegex(output, pattern)
def test_validate_logs_warning_for_cdk_project(self):
test_data_path = Path(__file__).resolve().parents[2] / "integration" / "testdata" / "package"
template_file = "aws-serverless-function-cdk.yaml"
template_path = test_data_path.joinpath(template_file)
command_result = run_command(self.command_list(template_file=template_path))
output = command_result.stdout.decode("utf-8")
warning_message = (
f"Warning: CDK apps are not officially supported with this command.{os.linesep}"
"We recommend you use this alternative command: cdk doctor"
)
self.assertIn(warning_message, output)
@parameterized.expand(
[
("default_yaml", TemplateFileTypes.YAML), # project with template.yaml
("default_json", TemplateFileTypes.JSON), # project with template.json
("multiple_files", TemplateFileTypes.YAML), # project with both template.yaml and template.json
(
"with_build",
TemplateFileTypes.JSON,
), # project with template.json and standard build directory .aws-sam/build/template.yaml
]
)
def test_lint_template(self, relative_folder: str, expected_file: TemplateFileTypes):
test_data_path = Path(__file__).resolve().parents[2] / "integration" / "testdata" / "validate"
process_dir = test_data_path / relative_folder
command_result = run_command(self.command_list(lint=True), cwd=str(process_dir))
pattern = self.lint_patterns[expected_file] # type: ignore
output = command_result.stdout.decode("utf-8")
self.assertEqual(command_result.process.returncode, 0)
self.assertRegex(output, pattern)
def test_lint_supported_runtimes(self):
template = {
"AWSTemplateFormatVersion": "2010-09-09",
"Transform": "AWS::Serverless-2016-10-31",
"Resources": {},
}
supported_runtimes = [
"dotnet6",
"go1.x",
"java17",
"java11",
"java8",
"java8.al2",
"nodejs14.x",
"nodejs16.x",
"nodejs18.x",
"provided",
"provided.al2",
"python3.7",
"python3.8",
"python3.9",
"python3.10",
"ruby2.7",
]
i = 0
for runtime in supported_runtimes:
i += 1
template["Resources"][f"HelloWorldFunction{i}"] = {
"Type": "AWS::Serverless::Function",
"Properties": {
"CodeUri": "HelloWorldFunction",
"Handler": "app.lambdaHandler",
"Runtime": runtime,
},
}
with tempfile.TemporaryDirectory() as temp:
template_file = Path(temp, "template.json")
with open(template_file, "w") as f:
f.write(json.dumps(template, indent=4) + "\n")
command_result = run_command(self.command_list(lint=True), cwd=str(temp))
pattern = self.lint_patterns[TemplateFileTypes.JSON]
output = command_result.stdout.decode("utf-8")
self.assertEqual(command_result.process.returncode, 0)
self.assertRegex(output, pattern)
def test_lint_error_no_region(self):
test_data_path = Path(__file__).resolve().parents[2] / "integration" / "testdata" / "validate" / "default_json"
template_file = "template.json"
template_path = test_data_path.joinpath(template_file)
command_result = run_command(self.command_list(lint=True, region="--debug", template_file=template_path))
output = command_result.stderr.decode("utf-8")
error_message = f"Error: Provided region: --debug doesn't match a supported format"
self.assertIn(error_message, output)
def test_lint_error_invalid_region(self):
test_data_path = Path(__file__).resolve().parents[2] / "integration" / "testdata" / "validate" / "default_json"
template_file = "template.json"
template_path = test_data_path.joinpath(template_file)
command_result = run_command(self.command_list(lint=True, region="us-north-5", template_file=template_path))
output = command_result.stderr.decode("utf-8")
error_message = f"Error: AWS Region was not found. Please configure your region through the --region option"
self.assertIn(error_message, output)
def test_lint_invalid_template(self):
test_data_path = Path(__file__).resolve().parents[2] / "integration" / "testdata" / "validate" / "default_yaml"
template_file = "templateError.yaml"
template_path = test_data_path.joinpath(template_file)
command_result = run_command(self.command_list(lint=True, template_file=template_path))
output = command_result.stdout.decode("utf-8")
# Remove Windows Line Endings for comparison.
output = output.replace("\r", "")
warning_message = (
'E0000 Duplicate found "HelloWorldFunction" (line 5)\n'
f'{os.path.join(test_data_path, "templateError.yaml")}:5:3\n\n'
'E0000 Duplicate found "HelloWorldFunction" (line 12)\n'
f'{os.path.join(test_data_path, "templateError.yaml")}:12:3\n\n'
)
self.assertIn(warning_message, output)
self.assertEqual(command_result.process.returncode, 1)