-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathtest_submitJob.py
More file actions
59 lines (43 loc) · 1.71 KB
/
Copy pathtest_submitJob.py
File metadata and controls
59 lines (43 loc) · 1.71 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
import json
import boto3
from moto import mock_sqs, mock_ecs
import run
import config
from tests.conftest import JOB_FILE
class TestJobQueue:
@mock_sqs
@mock_ecs
def test_create_job_queue_instance(self, run_setup):
run_setup()
job_queue = run.JobQueue()
expected_url = f"https://sqs.{config.AWS_REGION}.amazonaws.com/123456789012/{config.SQS_QUEUE_NAME}"
assert job_queue is not None
assert job_queue.queue.url == expected_url
class TestJobFile:
def test_job_file(self):
assert JOB_FILE.exists()
class TestSubmitJob:
@mock_sqs
@mock_ecs
def test_submit_job(self, run_submitJob):
run_submitJob()
sqs = boto3.resource('sqs')
queue = sqs.get_queue_by_name(QueueName=config.SQS_QUEUE_NAME)
job_info = json.loads(JOB_FILE.read_text())
templateMessage = {
eachkey:job_info[eachkey] for eachkey in job_info.keys() if eachkey != "groups" and "_comment" not in eachkey
}
expected_messages = []
for batch in job_info["groups"]:
msg = templateMessage.copy()
msg["group"] = batch
expected_messages.append(msg)
assert queue.attributes['ApproximateNumberOfMessages'] == str(len(expected_messages))
for _ in range(len(expected_messages)):
received_msg = queue.receive_messages(MaxNumberOfMessages=1)
assert len(received_msg) == 1
received_msg = json.loads(received_msg[0].body)
assert received_msg in expected_messages
# should have no more messages
final_received_msg = queue.receive_messages(MaxNumberOfMessages=1)
assert len(final_received_msg) == 0