-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathtest_setup.py
More file actions
240 lines (165 loc) · 8.18 KB
/
Copy pathtest_setup.py
File metadata and controls
240 lines (165 loc) · 8.18 KB
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
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
import json
from textwrap import dedent
from moto import mock_sqs, mock_ecs
import config
import run
from tests.conftest import FAKE_AWS_ACCESS_KEY_ID, FAKE_AWS_SECRET_ACCESS_KEY
ECS_TASK_NAME = config.APP_NAME + 'Task'
ECS_SERVICE_NAME = config.APP_NAME + 'Service'
class TestGetQueueURL:
queue_name = "test_queue"
def test_get_nonexistent_queue_url(self, sqs):
url = run.get_queue_url(sqs, self.queue_name)
assert url is None
def test_get_existing_queue_url(self, sqs):
sqs.create_queue(QueueName=self.queue_name)
url = run.get_queue_url(sqs,self.queue_name)
assert url is not None
assert url == f"https://sqs.{config.AWS_REGION}.amazonaws.com/123456789012/{self.queue_name}"
class TestGetOrCreateQueue:
def test_create_nonexistent_dead_queue(self, sqs):
run.get_or_create_queue(sqs)
res = sqs.list_queues()
res_urls = sorted(res['QueueUrls'])
expected_urls = sorted([
f"https://sqs.{config.AWS_REGION}.amazonaws.com/123456789012/{config.SQS_DEAD_LETTER_QUEUE}",
f"https://sqs.{config.AWS_REGION}.amazonaws.com/123456789012/{config.SQS_QUEUE_NAME}"
])
assert res_urls == expected_urls
dead_url = run.get_queue_url(sqs, config.SQS_DEAD_LETTER_QUEUE)
dead_queue_arn = sqs.get_queue_attributes(
QueueUrl=dead_url, AttributeNames=["All"]
)
queue_url = run.get_queue_url(sqs, config.SQS_QUEUE_NAME)
queue_arn = sqs.get_queue_attributes(
QueueUrl=queue_url, AttributeNames=["All"]
)
assert "QueueArn" in dead_queue_arn["Attributes"]
assert "RedrivePolicy" in queue_arn["Attributes"]
redrive_policy = json.loads(queue_arn["Attributes"]["RedrivePolicy"])
redrive_policy_arn = redrive_policy["deadLetterTargetArn"]
assert redrive_policy_arn == dead_queue_arn["Attributes"]["QueueArn"]
def test_create_existing_dead_queue(self, sqs):
sqs.create_queue(QueueName=config.SQS_DEAD_LETTER_QUEUE)
run.get_or_create_queue(sqs)
res = sqs.list_queues()
res_urls = sorted(res['QueueUrls'])
expected_urls = sorted([
f"https://sqs.{config.AWS_REGION}.amazonaws.com/123456789012/{config.SQS_DEAD_LETTER_QUEUE}",
f"https://sqs.{config.AWS_REGION}.amazonaws.com/123456789012/{config.SQS_QUEUE_NAME}"
])
assert res_urls == expected_urls
def test_create_existing_queue(self, sqs):
sqs.create_queue(QueueName=config.SQS_QUEUE_NAME)
run.get_or_create_queue(sqs)
res = sqs.list_queues()
res_urls = sorted(res['QueueUrls'])
expected_urls = sorted([
f"https://sqs.{config.AWS_REGION}.amazonaws.com/123456789012/{config.SQS_DEAD_LETTER_QUEUE}",
f"https://sqs.{config.AWS_REGION}.amazonaws.com/123456789012/{config.SQS_QUEUE_NAME}"
])
assert res_urls == expected_urls
class TestGetOrCreateCluster:
def test_create_nonexistent_cluster(self, ecs):
run.get_or_create_cluster(ecs)
res = ecs.list_clusters()
assert res["clusterArns"] == [f"arn:aws:ecs:{config.AWS_REGION}:123456789012:cluster/{config.ECS_CLUSTER}"]
def test_create_existing_cluster(self, ecs):
ecs.create_cluster(clusterName=config.ECS_CLUSTER)
run.get_or_create_cluster(ecs)
res = ecs.list_clusters()
assert res["clusterArns"] == [f"arn:aws:ecs:{config.AWS_REGION}:123456789012:cluster/{config.ECS_CLUSTER}"]
# for constructing expected results, see:
# https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/ecs/client/register_task_definition.html
class TestGenerateTaskDefinition:
def test_generate_task_definition(self, aws_config, sqs, ecs):
run.get_or_create_queue(sqs)
run.get_or_create_cluster(ecs)
task_definition, taskRoleArn = run.generate_task_definition(config.AWS_PROFILE)
assert taskRoleArn == False
task_def_env = task_definition["containerDefinitions"][0]["environment"]
aws_access_key_id_res = list(filter(
lambda x: x["name"] == "AWS_ACCESS_KEY_ID",
task_def_env
))
aws_secret_access_key_res = list(filter(
lambda x: x["name"] == "AWS_SECRET_ACCESS_KEY", task_def_env
))
queue_name_res = list(filter(
lambda x: x["name"] == "SQS_QUEUE_URL",
task_def_env
))
assert len(aws_access_key_id_res) == 1
assert len(aws_secret_access_key_res) == 1
assert len(queue_name_res) == 1
assert aws_access_key_id_res[0]["value"] == FAKE_AWS_ACCESS_KEY_ID
assert aws_secret_access_key_res[0]["value"] == FAKE_AWS_SECRET_ACCESS_KEY
assert queue_name_res[0]["value"] == run.get_queue_url(sqs, config.SQS_QUEUE_NAME)
def test_generate_task_definition_role_arn(self, aws_config, sqs, ecs):
dummy_role_arn = "arn:aws:iam::123456789012:role/ecsTaskExecutionRole"
aws_config['aws_config_file'].write_text(dedent(
f"""
[default]
aws_access_key_id = testing
aws_secret_access_key = testing
role_arn = {dummy_role_arn}
"""
))
run.get_or_create_queue(sqs)
run.get_or_create_cluster(ecs)
_, taskRoleArn = run.generate_task_definition(config.AWS_PROFILE)
assert taskRoleArn == dummy_role_arn
class TestUpdateECSTaskDefinition:
def test_update_ecs_task_definition(self, aws_config, sqs, ecs):
run.get_or_create_queue(sqs)
run.get_or_create_cluster(ecs)
run.update_ecs_task_definition(ecs, ECS_TASK_NAME, config.AWS_PROFILE)
res = ecs.list_task_definitions()
assert res["ResponseMetadata"]["HTTPStatusCode"] == 200
assert res["taskDefinitionArns"] == [f"arn:aws:ecs:{config.AWS_REGION}:123456789012:task-definition/{ECS_TASK_NAME}:1"]
class TestCreateUpdateECSService:
def test_create_ecs_service(self, aws_config, sqs, ecs):
run.get_or_create_queue(sqs)
run.get_or_create_cluster(ecs)
run.update_ecs_task_definition(ecs, config.APP_NAME + 'Task', config.AWS_PROFILE)
run.create_or_update_ecs_service(ecs, ECS_SERVICE_NAME, ECS_TASK_NAME)
res = ecs.list_services(cluster=config.ECS_CLUSTER)
assert res["ResponseMetadata"]["HTTPStatusCode"] == 200
assert res["serviceArns"] == [f"arn:aws:ecs:{config.AWS_REGION}:123456789012:service/{config.AWS_PROFILE}/{ECS_SERVICE_NAME}"]
def test_update_ecs_service(self, aws_config, sqs, ecs, capsys):
run.get_or_create_queue(sqs)
run.get_or_create_cluster(ecs)
run.update_ecs_task_definition(ecs, config.APP_NAME + 'Task', config.AWS_PROFILE)
run.create_or_update_ecs_service(ecs, ECS_SERVICE_NAME, ECS_TASK_NAME)
run.create_or_update_ecs_service(ecs, ECS_SERVICE_NAME, ECS_TASK_NAME)
captured = capsys.readouterr().out.split('\n')
service_already_exists = False
for line in captured:
if "service exists" in line.lower():
service_already_exists = True
break
assert service_already_exists
# if all of the above pass, this should pass without error
class TestSetup:
@mock_sqs
@mock_ecs
def test_setup(self, capsys, run_setup):
run_setup()
res = capsys.readouterr().out.split('\n')
dead_letter_queue_created = False
queue_created = False
cluster_created = False
task_definition_registered = False
service_created = False
for line in res:
if "creating deadletter queue" in line.lower():
dead_letter_queue_created = True
if "creating queue" in line.lower():
queue_created = True
if f"cluster {config.AWS_PROFILE} created" in line.lower():
cluster_created = True
if "task definition registered" in line.lower():
task_definition_registered = True
if "service created" in line.lower():
service_created = True
assert all([dead_letter_queue_created, queue_created, cluster_created, task_definition_registered, service_created])