add test for sqs fifo source order/parallelism

dominikschubert · dominikschubert · commit 9250f1cac51d · 2022-10-18T00:09:22.000+02:00
diff --git a/tests/integration/awslambda/functions/lambda_sleep.py b/tests/integration/awslambda/functions/lambda_sleep.py
@@ -0,0 +1,11 @@
+import os
+import time
+
+sleep_duration = int(os.getenv("TEST_SLEEP_S", "0"))
+
+
+def handler(event, context):
+    print(f"sleeping for {sleep_duration}")
+    time.sleep(sleep_duration)
+    print("done sleeping")
+    return {"status": "ok"}
diff --git a/tests/integration/awslambda/test_lambda_integration_sqs.py b/tests/integration/awslambda/test_lambda_integration_sqs.py
@@ -5,6 +5,7 @@
 import pytest
 from botocore.exceptions import ClientError
 
+from localstack.aws.api.lambda_ import Runtime
 from localstack.services.awslambda.lambda_api import (
     BATCH_SIZE_RANGES,
     INVALID_PARAMETER_VALUE_EXCEPTION,
@@ -1087,3 +1088,71 @@ def test_sqs_invalid_event_filter(
 
 
 # TODO: test integration with lambda logs
+
+
+TEST_LAMBDA_SLEEP = os.path.join(THIS_FOLDER, "functions/lambda_sleep.py")
+
+
+class TestFifoQueueMapping:
+    def test_fifo_behavior(
+        self,
+        sqs_client,
+        create_lambda_function,
+        lambda_client,
+        lambda_su_role,
+        logs_client,
+        cleanups,
+    ):
+        # create FIFO queue
+        queue_name = f"test-queue-{short_uid()}.fifo"
+        create_queue_result = sqs_client.create_queue(
+            QueueName=queue_name,
+            Attributes={
+                "FifoQueue": "true",
+                "ContentBasedDeduplication": "true",
+                "VisibilityTimeout": str(10 * 6),
+            },
+        )
+        queue_url = create_queue_result["QueueUrl"]
+        queue_arn = sqs_client.get_queue_attributes(
+            QueueUrl=queue_url, AttributeNames=["QueueArn"]
+        )["Attributes"]["QueueArn"]
+
+        message_group_id = "fixed-message-group-id-test"
+
+        # create a lambda to process messages
+        function_name = f"function-name-{short_uid()}"
+
+        create_lambda_function(
+            func_name=function_name,
+            handler_file=TEST_LAMBDA_SLEEP,
+            runtime=Runtime.python3_9,
+            role=lambda_su_role,
+            timeout=10,
+            Environment={"Variables": {"TEST_SLEEP_S": "5"}},
+        )
+
+        # create event source mapping
+        create_esm_result = lambda_client.create_event_source_mapping(
+            FunctionName=function_name, EventSourceArn=queue_arn, Enabled=False, BatchSize=1
+        )
+        esm_uuid = create_esm_result["UUID"]
+        cleanups.append(lambda: lambda_client.delete_event_source_mapping(UUID=esm_uuid))
+
+        # send messages
+        for i in range(5):
+            sqs_client.send_message(
+                QueueUrl=queue_url, MessageBody=f"message-{i}", MessageGroupId=message_group_id
+            )
+
+        # enable event source mapping
+        lambda_client.update_event_source_mapping(UUID=esm_uuid, Enabled=True)
+        _await_event_source_mapping_enabled(lambda_client, esm_uuid)
+
+        # since the lambda has to be called in-order anyway, there shouldn't be any parallel executions
+        log_group_name = f"/aws/lambda/{function_name}"
+
+        time.sleep(60)
+
+        log_streams = logs_client.describe_log_streams(logGroupName=log_group_name)
+        assert len(log_streams["logStreams"]) == 1
