Repository: getmoto/moto
Pull request: SQS: Improve deduplication logic
Dedupe according to logic in https://docs.aws.amazon.com/AWSSimpleQueueService/latest/APIReference/API_SendMessage.html
and
https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/using-messagededuplicationid-property.html

Closes https://github.com/getmoto/moto/issues/1438

Problem statement:
Fifo messages aren't deduped
Anaconda-IPython
`sys.version[:5] == '2.7.6'`
`boto3.__version__ == '1.5.17'`
`moto.__version__ == (u'1.2.0',)`

## Reproduce
```python
import boto3
from moto import mock_sqs

@mock_sqs
def dedupe_fifo():
    sqs = boto3.resource('sqs', region_name='us-west-2')
    queue = sqs.create_queue(
        QueueName='my-queue.fifo',
        Attributes={
            'FifoQueue': 'true',
        },
    )

    for _ in range(5):
        queue.send_message(
            MessageBody="test",
            MessageDeduplicationId="1",
            MessageGroupId="2",
        )
    assert queue.attributes['ApproximateNumberOfMessages'] == 1
```

## Expected Behavior
For the `assert` to pass, as messages have been deduped. Note that my code sample should work whether `Content-based deduplication` is enabled or not, since the `MessageBody` is identical

## Experienced Behavior
No de-duping occurs, so I get an `AssertionError`, `assert 5 == 1`, where 5 is length of my for-loop

Hints from issue discussion:
Possibly related, when I retrieve messages from a mocked FIFO queue, the `MessageGroupId` is not present.
Thanks for opening. Marking this as a feature request.
Also encountered this so just wanted to +1
@bblommers let me take a look at this
nvm, looks like it's already been tackled https://github.com/getmoto/moto/blob/master/moto/sqs/models.py#L533
@rafcio19 I'd have to verify this against AWS, but from my understanding of the docs and our implementation, two scenarios are not yet implemented in Moto:

 - with ContentBasedDeduplication can be set to False (like in the example above) - we should still deduplicate. (The if-statement you linked is only active if ContentBasedDeduplication is set to True)
 - the `deduplication_id`-attribute is set when ContentBasedDeduplication=True (https://github.com/getmoto/moto/pull/3538/files) - I assume that it only needs to be set if it isn't explicitly provided by the user
Gotcha, in that case let me take a look 👀

Commit message: tests: fix stray test

Patch to review:
diff --git a/moto/sqs/models.py b/moto/sqs/models.py
index ee93e1f48fb7..b7fa5c7eef4f 100644
--- a/moto/sqs/models.py
+++ b/moto/sqs/models.py
@@ -525,17 +525,22 @@ def messages(self):
         ]
 
     def add_message(self, message):
-        if (
-            self.fifo_queue
-            and self.attributes.get("ContentBasedDeduplication") == "true"
-        ):
-            for m in self._messages:
-                if m.deduplication_id == message.deduplication_id:
-                    diff = message.sent_timestamp - m.sent_timestamp
-                    # if a duplicate message is received within the deduplication time then it should
-                    # not be added to the queue
-                    if diff / 1000 < DEDUPLICATION_TIME_IN_SECONDS:
-                        return
+        if self.fifo_queue:
+
+            # the cases in which we dedupe fifo messages
+            # from https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/using-messagededuplicationid-property.html
+            # https://docs.aws.amazon.com/AWSSimpleQueueService/latest/APIReference/API_SendMessage.html
+            if (
+                self.attributes.get("ContentBasedDeduplication") == "true"
+                or message.deduplication_id
+            ):
+                for m in self._messages:
+                    if m.deduplication_id == message.deduplication_id:
+                        diff = message.sent_timestamp - m.sent_timestamp
+                        # if a duplicate message is received within the deduplication time then it should
+                        # not be added to the queue
+                        if diff / 1000 < DEDUPLICATION_TIME_IN_SECONDS:
+                            return
 
         with self._messages_lock:
             self._messages.append(message)
@@ -758,6 +763,24 @@ def send_message(
 
         queue = self.get_queue(queue_name)
 
+        if (
+            queue.attributes.get("ContentBasedDeduplication") == "false"
+            and not group_id
+        ):
+            msg = "MessageGroupId"
+            raise MissingParameter(msg)
+
+        if (
+            queue.attributes.get("ContentBasedDeduplication") == "false"
+            and group_id
+            and not deduplication_id
+        ):
+            msg = (
+                "The queue should either have ContentBasedDeduplication enabled or "
+                "MessageDeduplicationId provided explicitly"
+            )
+            raise InvalidParameterValue(msg)
+
         if len(message_body) > queue.maximum_message_size:
             msg = f"One or more parameters are invalid. Reason: Message must be shorter than {queue.maximum_message_size} bytes."
             raise InvalidParameterValue(msg)
diff --git a/tests/test_sqs/test_sqs.py b/tests/test_sqs/test_sqs.py
index 72f167e731df..60f8fb480363 100644
--- a/tests/test_sqs/test_sqs.py
+++ b/tests/test_sqs/test_sqs.py
@@ -2090,7 +2090,8 @@ def test_batch_change_message_visibility():
     with freeze_time("2015-01-01 12:00:00"):
         sqs = boto3.client("sqs", region_name="us-east-1")
         resp = sqs.create_queue(
-            QueueName="test-dlr-queue.fifo", Attributes={"FifoQueue": "true"}
+            QueueName="test-dlr-queue.fifo",
+            Attributes={"FifoQueue": "true", "ContentBasedDeduplication": "true"},
         )
         queue_url = resp["QueueUrl"]
 
@@ -2524,7 +2525,8 @@ def test_queue_with_dlq():
 
     with freeze_time("2015-01-01 12:00:00"):
         resp = sqs.create_queue(
-            QueueName=f"{str(uuid4())[0:6]}.fifo", Attributes={"FifoQueue": "true"}
+            QueueName=f"{str(uuid4())[0:6]}.fifo",
+            Attributes={"FifoQueue": "true", "ContentBasedDeduplication": "true"},
         )
         queue_url1 = resp["QueueUrl"]
         queue_arn1 = sqs.get_queue_attributes(
@@ -2535,6 +2537,7 @@ def test_queue_with_dlq():
             QueueName=f"{str(uuid4())[0:6]}.fifo",
             Attributes={
                 "FifoQueue": "true",
+                "ContentBasedDeduplication": "true",
                 "RedrivePolicy": json.dumps(
                     {"deadLetterTargetArn": queue_arn1, "maxReceiveCount": 2}
                 ),
@@ -2681,7 +2684,8 @@ def test_redrive_policy_set_attributes_with_string_value():
 def test_receive_messages_with_message_group_id():
     sqs = boto3.resource("sqs", region_name="us-east-1")
     queue = sqs.create_queue(
-        QueueName=f"{str(uuid4())[0:6]}.fifo", Attributes={"FifoQueue": "true"}
+        QueueName=f"{str(uuid4())[0:6]}.fifo",
+        Attributes={"FifoQueue": "true", "ContentBasedDeduplication": "true"},
     )
     queue.set_attributes(Attributes={"VisibilityTimeout": "3600"})
     queue.send_message(MessageBody="message-1", MessageGroupId="group")
@@ -2712,7 +2716,8 @@ def test_receive_messages_with_message_group_id():
 def test_receive_messages_with_message_group_id_on_requeue():
     sqs = boto3.resource("sqs", region_name="us-east-1")
     queue = sqs.create_queue(
-        QueueName=f"{str(uuid4())[0:6]}.fifo", Attributes={"FifoQueue": "true"}
+        QueueName=f"{str(uuid4())[0:6]}.fifo",
+        Attributes={"FifoQueue": "true", "ContentBasedDeduplication": "true"},
     )
     queue.set_attributes(Attributes={"VisibilityTimeout": "3600"})
     queue.send_message(MessageBody="message-1", MessageGroupId="group")
@@ -2742,7 +2747,8 @@ def test_receive_messages_with_message_group_id_on_visibility_timeout():
     with freeze_time("2015-01-01 12:00:00"):
         sqs = boto3.resource("sqs", region_name="us-east-1")
         queue = sqs.create_queue(
-            QueueName="test-queue.fifo", Attributes={"FifoQueue": "true"}
+            QueueName="test-queue.fifo",
+            Attributes={"FifoQueue": "true", "ContentBasedDeduplication": "true"},
         )
         queue.set_attributes(Attributes={"VisibilityTimeout": "3600"})
         queue.send_message(MessageBody="message-1", MessageGroupId="group")
@@ -2989,7 +2995,10 @@ def test_fifo_send_message_when_same_group_id_is_in_dlq():
 
     sqs = boto3.resource("sqs", region_name="us-east-1")
     q_name = f"{str(uuid4())[0:6]}-dlq.fifo"
-    dlq = sqs.create_queue(QueueName=q_name, Attributes={"FifoQueue": "true"})
+    dlq = sqs.create_queue(
+        QueueName=q_name,
+        Attributes={"FifoQueue": "true", "ContentBasedDeduplication": "true"},
+    )
 
     queue = sqs.get_queue_by_name(QueueName=q_name)
     dead_letter_queue_arn = queue.attributes.get("QueueArn")
@@ -2998,6 +3007,7 @@ def test_fifo_send_message_when_same_group_id_is_in_dlq():
         QueueName=f"{str(uuid4())[0:6]}.fifo",
         Attributes={
             "FifoQueue": "true",
+            "ContentBasedDeduplication": "true",
             "RedrivePolicy": json.dumps(
                 {"deadLetterTargetArn": dead_letter_queue_arn, "maxReceiveCount": 1}
             ),
@@ -3197,3 +3207,60 @@ def test_receive_message_that_becomes_visible_while_long_polling():
     assert len(messages) == 1
     assert messages[0].body == msg_body
     assert end - begin < time_to_wait
+
+
+@mock_sqs
+def test_dedupe_fifo():
+    sqs = boto3.resource("sqs", region_name="us-east-1")
+    queue = sqs.create_queue(
+        QueueName="my-queue.fifo",
+        Attributes={
+            "FifoQueue": "true",
+        },
+    )
+
+    for _ in range(5):
+        queue.send_message(
+            MessageBody="test",
+            MessageDeduplicationId="1",
+            MessageGroupId="2",
+        )
+    assert int(queue.attributes["ApproximateNumberOfMessages"]) == 1
+
+
+@mock_sqs
+def test_fifo_dedupe_error_no_message_group_id():
+    sqs = boto3.resource("sqs", region_name="us-east-1")
+    queue = sqs.create_queue(
+        QueueName="my-queue.fifo",
+        Attributes={"FifoQueue": "true"},
+    )
+    with pytest.raises(ClientError) as exc:
+        queue.send_message(
+            MessageBody="test",
+            MessageDeduplicationId="1",
+        )
+
+    exc.value.response["Error"]["Code"].should.equal("MissingParameter")
+    exc.value.response["Error"]["Message"].should.equal(
+        "The request must contain the parameter MessageGroupId."
+    )
+
+
+@mock_sqs
+def test_fifo_dedupe_error_no_message_dedupe_id():
+    sqs = boto3.resource("sqs", region_name="us-east-1")
+    queue = sqs.create_queue(
+        QueueName="my-queue.fifo",
+        Attributes={"FifoQueue": "true"},
+    )
+    with pytest.raises(ClientError) as exc:
+        queue.send_message(
+            MessageBody="test",
+            MessageGroupId="1",
+        )
+
+    exc.value.response["Error"]["Code"].should.equal("InvalidParameterValue")
+    exc.value.response["Error"]["Message"].should.equal(
+        "The queue should either have ContentBasedDeduplication enabled or MessageDeduplicationId provided explicitly"
+    )
diff --git a/tests/test_sqs/test_sqs_integration.py b/tests/test_sqs/test_sqs_integration.py
index 1e89f71376a8..15e2c1698b0d 100644
--- a/tests/test_sqs/test_sqs_integration.py
+++ b/tests/test_sqs/test_sqs_integration.py
@@ -85,7 +85,10 @@ def test_invoke_function_from_sqs_fifo_queue():
     logs_conn = boto3.client("logs", region_name="us-east-1")
     sqs = boto3.resource("sqs", region_name="us-east-1")
     queue_name = str(uuid.uuid4())[0:6] + ".fifo"
-    queue = sqs.create_queue(QueueName=queue_name, Attributes={"FifoQueue": "true"})
+    queue = sqs.create_queue(
+        QueueName=queue_name,
+        Attributes={"FifoQueue": "true", "ContentBasedDeduplication": "true"},
+    )
 
     fn_name = str(uuid.uuid4())[0:6]
     conn = boto3.client("lambda", region_name="us-east-1")


Code review comments:
diff_hunk: @@ -758,6 +763,24 @@ def send_message(
 
         queue = self.get_queue(queue_name)
 
+        if (
path: moto/sqs/models.py
text: @user1:
Just a nitpick - I think these two checks should be behind an `is queue.fifo_queue` check.
The current approach works, because `ContentBasedDeduplication` is only set for `fifo`-queues. In other words: `queue.attributes.get("ContentBasedDeduplication")` always returns None for a non-fifo queue. 

But it's more readable IMO if we make it explicit that these checks are only applicable for FIFO queues.

@author:
Makes sense

@author:
Updated the PR