SQS: Implement DelaySeconds validation (#5703)

This commit is contained in:
Rahi Bhensdadia 2022-11-28 03:17:18 +05:30 committed by GitHub
parent ff1f420689
commit 97b5e8b3ab
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 114 additions and 15 deletions

View File

@ -44,6 +44,8 @@ MAXIMUM_MESSAGE_LENGTH = 262144 # 256 KiB
MAXIMUM_MESSAGE_SIZE_ATTR_LOWER_BOUND = 1024
MAXIMUM_MESSAGE_SIZE_ATTR_UPPER_BOUND = MAXIMUM_MESSAGE_LENGTH
MAXIMUM_MESSAGE_DELAY = 900
TRANSPORT_TYPE_ENCODINGS = {
"String": b"\x01",
"Binary": b"\x02",
@ -799,6 +801,13 @@ class SQSBackend(BaseBackend):
if message_attributes:
message.message_attributes = message_attributes
if delay_seconds > MAXIMUM_MESSAGE_DELAY:
msg = (
f"Value {delay_seconds} for parameter DelaySeconds is invalid. "
"Reason: DelaySeconds must be >= 0 and <= 900."
)
raise InvalidParameterValue(msg)
message.mark_sent(delay_seconds=delay_seconds)
queue.add_message(message)
@ -834,21 +843,24 @@ class SQSBackend(BaseBackend):
raise TooManyEntriesInBatchRequest(len(entries))
messages = []
failedInvalidDelay = []
for entry in entries.values():
# Loop through looking for messages
message = self.send_message(
queue_name,
entry["MessageBody"],
message_attributes=entry["MessageAttributes"],
delay_seconds=entry["DelaySeconds"],
group_id=entry.get("MessageGroupId"),
deduplication_id=entry.get("MessageDeduplicationId"),
)
message.user_id = entry["Id"]
try:
# Loop through looking for messages
message = self.send_message(
queue_name,
entry["MessageBody"],
message_attributes=entry["MessageAttributes"],
delay_seconds=entry["DelaySeconds"],
group_id=entry.get("MessageGroupId"),
deduplication_id=entry.get("MessageDeduplicationId"),
)
message.user_id = entry["Id"]
messages.append(message)
except InvalidParameterValue:
failedInvalidDelay.append(entry)
messages.append(message)
return messages
return messages, failedInvalidDelay
def _get_first_duplicate_id(self, ids):
unique_ids = set()

View File

@ -299,10 +299,23 @@ class SQSResponse(BaseResponse):
if entries == {}:
raise EmptyBatchRequest()
messages = self.sqs_backend.send_message_batch(queue_name, entries)
messages, failedInvalidDelay = self.sqs_backend.send_message_batch(
queue_name, entries
)
errors = []
for entry in failedInvalidDelay:
errors.append(
{
"Id": entry["Id"],
"SenderFault": "true",
"Code": "InvalidParameterValue",
"Message": "Value 1800 for parameter DelaySeconds is invalid. Reason: DelaySeconds must be &gt;= 0 and &lt;= 900.",
}
)
template = self.response_template(SEND_MESSAGE_BATCH_RESPONSE)
return template.render(messages=messages)
return template.render(messages=messages, errors=errors)
def delete_message(self):
queue_name = self._get_queue_name()
@ -650,6 +663,7 @@ RECEIVE_MESSAGE_RESPONSE = """<ReceiveMessageResponse>
</ResponseMetadata>
</ReceiveMessageResponse>"""
# UPDATED Line 681-688
SEND_MESSAGE_BATCH_RESPONSE = """<SendMessageBatchResponse>
<SendMessageBatchResult>
{% for message in messages %}
@ -662,6 +676,14 @@ SEND_MESSAGE_BATCH_RESPONSE = """<SendMessageBatchResponse>
{% endif %}
</SendMessageBatchResultEntry>
{% endfor %}
{% for error_dict in errors %}
<BatchResultErrorEntry>
<Id>{{ error_dict['Id'] }}</Id>
<Code>{{ error_dict['Code'] }}</Code>
<Message>{{ error_dict['Message'] }}</Message>
<SenderFault>{{ error_dict['SenderFault'] }}</SenderFault>
</BatchResultErrorEntry>
{% endfor %}
</SendMessageBatchResult>
<ResponseMetadata>
<RequestId></RequestId>

View File

@ -3115,3 +3115,68 @@ def test_message_has_windows_return():
messages = queue.receive_messages()
messages.should.have.length_of(1)
messages[0].body.should.match(message)
@mock_sqs
def test_message_delay_is_more_than_15_minutes():
client = boto3.client("sqs", region_name="us-east-1")
response = client.create_queue(
QueueName=f"{str(uuid4())[0:6]}.fifo", Attributes={"FifoQueue": "true"}
)
queue_url = response["QueueUrl"]
response = client.send_message_batch(
QueueUrl=queue_url,
Entries=[
{
"Id": "id_1",
"MessageBody": "body_1",
"DelaySeconds": 3,
"MessageAttributes": {
"attribute_name_1": {
"StringValue": "attribute_value_1",
"DataType": "String",
}
},
"MessageGroupId": "message_group_id_1",
"MessageDeduplicationId": "message_deduplication_id_1",
},
{
"Id": "id_2",
"MessageBody": "body_2",
"DelaySeconds": 1800,
"MessageAttributes": {
"attribute_name_2": {"StringValue": "123", "DataType": "Number"}
},
"MessageGroupId": "message_group_id_2",
"MessageDeduplicationId": "message_deduplication_id_2",
},
],
)
sorted([entry["Id"] for entry in response["Successful"]]).should.equal(["id_1"])
sorted([entry["Id"] for entry in response["Failed"]]).should.equal(["id_2"])
# print(response)
time.sleep(4)
response = client.receive_message(
QueueUrl=queue_url,
MaxNumberOfMessages=10,
MessageAttributeNames=["attribute_name_1", "attribute_name_2"],
AttributeNames=["MessageDeduplicationId", "MessageGroupId"],
)
response["Messages"].should.have.length_of(1)
response["Messages"][0]["Body"].should.equal("body_1")
response["Messages"][0]["MessageAttributes"].should.equal(
{"attribute_name_1": {"StringValue": "attribute_value_1", "DataType": "String"}}
)
response["Messages"][0]["Attributes"]["MessageGroupId"].should.equal(
"message_group_id_1"
)
response["Messages"][0]["Attributes"]["MessageDeduplicationId"].should.equal(
"message_deduplication_id_1"
)