text: import datetime
  import json
  import os
  import time
  import unittest
  
  import requests
  from botocore.auth import SIGV4_TIMESTAMP, SigV4Auth
  from botocore.awsrequest import AWSRequest
  from botocore.credentials import Credentials
  from botocore.exceptions import ClientError
  from six.moves.urllib.parse import urlencode
  
  from localstack import config
  from localstack.constants import (
      TEST_AWS_ACCESS_KEY_ID,
      TEST_AWS_ACCOUNT_ID,
      TEST_AWS_SECRET_ACCESS_KEY,
  )
  from localstack.utils import testutil
  from localstack.utils.aws import aws_stack
  from localstack.utils.common import get_service_protocol, retry, short_uid, to_str
  from localstack.utils.testutil import get_lambda_log_events
  
  from .lambdas import lambda_integration
  from .test_lambda import (
      LAMBDA_RUNTIME_DOTNETCORE2,
      LAMBDA_RUNTIME_DOTNETCORE31,
      LAMBDA_RUNTIME_PYTHON36,
      TEST_LAMBDA_DOTNETCORE2,
      TEST_LAMBDA_DOTNETCORE31,
      TEST_LAMBDA_LIBS,
      TEST_LAMBDA_PYTHON,
      load_file,
      use_docker,
  )
  
  TEST_QUEUE_NAME = "TestQueue"
  
  TEST_POLICY = """
  {
    "Version":"2012-10-17",
    "Statement":[
      {
        "Effect": "Allow",
        "Principal": { "AWS": "*" },
        "Action": "sqs:SendMessage",
        "Resource": "'$sqs_queue_arn'",
        "Condition":{
          "ArnEquals":{
          "aws:SourceArn":"'$sns_topic_arn'"
          }
        }
      }
    ]
  }
  """
  
  THIS_FOLDER = os.path.dirname(os.path.realpath(__file__))
  TEST_LAMBDA_ECHO_FILE = os.path.join(THIS_FOLDER, "lambdas", "lambda_echo.py")
  
  TEST_MESSAGE_ATTRIBUTES = {
      "City": {
          "DataType": "String",
          "StringValue": "Any City - with special characters: <?`",
      },
      "Population": {"DataType": "Number", "StringValue": "1250800"},
  }
  
  
  class SQSTest(unittest.TestCase):
      @classmethod
      def setUpClass(cls):
          cls.client = aws_stack.connect_to_service("sqs")
  
      def test_list_queue_tags(self):
          queue_info = self.client.create_queue(QueueName=TEST_QUEUE_NAME)
          queue_url = queue_info["QueueUrl"]
  
          # list queues with name prefix
          result = self.client.list_queues(QueueNamePrefix=TEST_QUEUE_NAME[0:-2])
          self.assertIn("QueueUrls", result)
          self.assertEqual(1, len(result.get("QueueUrls")))
  
          result = self.client.list_queue_tags(QueueUrl=queue_url)
          # Apparently, if there are no tags, then `Tags` should NOT appear in the response.
          self.assertNotIn("Tags", result)
  
          # try to request details from queue URL directly via GET request
          response = requests.get(queue_url)
          content = to_str(response.content)
          self.assertIn(queue_url, content)
  
          # clean up
          self.client.delete_queue(QueueUrl=queue_url)
          result = self.client.list_queues(QueueNamePrefix=TEST_QUEUE_NAME)
          self.assertEqual(0, len(result.get("QueueUrls", [])))
  
      def test_publish_get_delete_message(self):
          queue_name = "queue-%s" % short_uid()
          queue_info = self.client.create_queue(QueueName=queue_name)
          queue_url = queue_info["QueueUrl"]
          self.assertIn(queue_name, queue_url)
  
          # publish/receive message
          self.client.send_message(QueueUrl=queue_url, MessageBody="msg123")
          for i in range(2):
              messages = self.client.receive_message(QueueUrl=queue_url, VisibilityTimeout=0)[
                  "Messages"
              ]
              self.assertEqual(len(messages), 1)
              self.assertEqual(messages[0]["Body"], "msg123")
  
          # delete/receive message
          self.client.delete_message(QueueUrl=queue_url, ReceiptHandle=messages[0]["ReceiptHandle"])
          response = self.client.receive_message(QueueUrl=queue_url)
          self.assertFalse(response.get("Messages"))
  
          # publish/receive message with change_message_visibility
          self.client.send_message(QueueUrl=queue_url, MessageBody="msg234")
          messages = self.client.receive_message(QueueUrl=queue_url)["Messages"]
          response = self.client.receive_message(QueueUrl=queue_url)
          self.assertFalse(response.get("Messages"))
          self.client.change_message_visibility(
              QueueUrl=queue_url,
              ReceiptHandle=messages[0]["ReceiptHandle"],
              VisibilityTimeout=0,
          )
          for i in range(2):
              messages = self.client.receive_message(QueueUrl=queue_url, VisibilityTimeout=0)[
                  "Messages"
              ]
              self.assertEqual(len(messages), 1)
              self.assertEqual(messages[0]["Body"], "msg234")
  
          # clean up
          self.client.delete_queue(QueueUrl=queue_url)
  
      def test_publish_get_delete_message_batch(self):
          queue_name = "queue-%s" % short_uid()
          queue_info = self.client.create_queue(QueueName=queue_name)
          queue_url = queue_info["QueueUrl"]
          self.assertIn(queue_name, queue_url)
  
          def receive_messages(**kwargs):
              kwds = dict(
                  QueueUrl=queue_url,
                  MaxNumberOfMessages=10,
                  MessageAttributeNames=["All"],
              )
              kwds.update(kwargs)
              messages = self.client.receive_message(**kwds)
              return messages
  
          def get_hashes(messages, outgoing=False):
              body_key = "MD5OfMessageBody" if outgoing else "MD5OfBody"
              return set([(m[body_key], m["MD5OfMessageAttributes"]) for m in messages])
  
          messages_to_send = [
              {
                  "Id": "message{:02d}".format(i),
                  "MessageBody": "msgBody{:02d}".format(i),
                  "MessageAttributes": {
                      "CustomAttribute": {
                          "DataType": "String",
                          "StringValue": "CustomAttributeValue{:02d}".format(i),
                      }
                  },
              }
              for i in range(1, 11)
          ]
  
          resp = self.client.send_message_batch(QueueUrl=queue_url, Entries=messages_to_send)
          sent_hashes = get_hashes(resp.get("Successful", []), outgoing=True)
          self.assertEqual(len(sent_hashes), len(messages_to_send))
  
          for i in range(2):
              messages = receive_messages(VisibilityTimeout=0)["Messages"]
              received_hashes = get_hashes(messages)
              self.assertEqual(received_hashes, sent_hashes)
  
          self.client.delete_message_batch(
              QueueUrl=queue_url,
              Entries=[
                  {"Id": "{:02d}".format(i), "ReceiptHandle": m["ReceiptHandle"]}
                  for i, m in enumerate(messages)
              ],
          )
  
          response = receive_messages()
          self.assertFalse(response.get("Messages"))
  
          # publish/receive message with change_message_visibility
          self.client.send_message_batch(QueueUrl=queue_url, Entries=messages_to_send)
          messages = receive_messages()["Messages"]
          response = receive_messages()
          self.assertFalse(response.get("Messages"))
  
          reset_hashes = get_hashes(messages[:5])
          self.client.change_message_visibility_batch(
              QueueUrl=queue_url,
              Entries=[
                  {
                      "Id": "{:02d}".format(i),
                      "ReceiptHandle": msg["ReceiptHandle"],
                      "VisibilityTimeout": 0,
                  }
                  for i, msg in enumerate(messages[:5])
              ],
          )
          for i in range(2):
              messages = receive_messages(VisibilityTimeout=0)["Messages"]
              received_hashes = get_hashes(messages)
              self.assertEqual(reset_hashes, received_hashes)
  
          # clean up
          self.client.delete_queue(QueueUrl=queue_url)
  
      def test_create_fifo_queue(self):
          fifo_queue = "my-queue.fifo"
          queue_info = self.client.create_queue(
              QueueName=fifo_queue, Attributes={"FifoQueue": "true"}
          )
          queue_url = queue_info["QueueUrl"]
  
          # it should preserve .fifo in the queue name
          self.assertIn(fifo_queue, queue_url)
  
          # try sending a message with message group ID and deduplication ID
          response = self.client.send_message(
              QueueUrl=queue_url,
              MessageBody="test msg 123",
              MessageDeduplicationId="dedup-1",
              MessageGroupId="group-1",
          )
          self.assertEqual(200, response["ResponseMetadata"]["HTTPStatusCode"])
          self.assertIn("MessageId", response)
          self.assertIn("MD5OfMessageBody", response)
  
          # clean up
          self.client.delete_queue(QueueUrl=queue_url)
  
      def test_set_queue_policy(self):
          queue_name = "queue-%s" % short_uid()
          queue_info = self.client.create_queue(QueueName=queue_name)
          queue_url = queue_info["QueueUrl"]
  
          attributes = {"Policy": TEST_POLICY}
          self.client.set_queue_attributes(QueueUrl=queue_url, Attributes=attributes)
  
          attrs = self.client.get_queue_attributes(QueueUrl=queue_url, AttributeNames=["All"])[
              "Attributes"
          ]
          self.assertIn("sqs:SendMessage", attrs["Policy"])
          attrs = self.client.get_queue_attributes(QueueUrl=queue_url, AttributeNames=["Policy"])[
              "Attributes"
          ]
          self.assertIn("sqs:SendMessage", attrs["Policy"])
  
          # clean up
          self.client.delete_queue(QueueUrl=queue_url)
  
      def test_send_message_attributes(self):
          queue_name = "queue-%s" % short_uid()
  
          queue_url = self.client.create_queue(QueueName=queue_name)["QueueUrl"]
  
          payload = {}
          attrs = {"attr1": {"StringValue": "val1", "DataType": "String"}}
          self.client.send_message(
              QueueUrl=queue_url, MessageBody=json.dumps(payload), MessageAttributes=attrs
          )
  
          result = self.client.receive_message(QueueUrl=queue_url, MessageAttributeNames=["All"])
          messages = result["Messages"]
          self.assertEqual(messages[0]["MessageAttributes"], attrs)
  
          # clean up
          self.client.delete_queue(QueueUrl=queue_url)
  
      def test_send_message_with_invalid_string_attributes(self):
          queue_name = "queue-%s" % short_uid()
  
          queue_url = self.client.create_queue(QueueName=queue_name)["QueueUrl"]
  
          payload = {}
          # String Attributes must not contain non-printable characters
          # See: https://docs.aws.amazon.com/AWSSimpleQueueService/latest/APIReference/API_SendMessage.html
          attrs = {
              "attr1": {
                  "StringValue": "invalid characters, %s, %s, %s" % (chr(8), chr(11), chr(12)),
                  "DataType": "String",
              }
          }
          with self.assertRaises(Exception):
              self.client.send_message(
                  QueueUrl=queue_url,
                  MessageBody=json.dumps(payload),
                  MessageAttributes=attrs,
              )
  
          # clean up
          self.client.delete_queue(QueueUrl=queue_url)
  
      def test_send_message_with_invalid_payload_characters(self):
          queue_name = "queue-%s" % short_uid()
  
          queue_url = self.client.create_queue(QueueName=queue_name)["QueueUrl"]
  
          # Some common control characters and some code points just outside of a permitted range
          for invalid_char in ["\0", "\v", "\f", "\u0019", "\uFFFE", "\uFFFF"]:
              raw_payload = "invalid character: " + invalid_char
              with self.assertRaisesRegex(Exception, "invalid characters"):
                  self.client.send_message(QueueUrl=queue_url, MessageBody=raw_payload)
  
          raw_payload = "valid characters: \t\n\r\uD7FF\uE000\uFFFD\u10000\u10FFFF"
          self.client.send_message(QueueUrl=queue_url, MessageBody=raw_payload)
  
          # clean up
          self.client.delete_queue(QueueUrl=queue_url)
  
      def test_dead_letter_queue_config(self):
          queue_name = "queue-%s" % short_uid()
          dlq_name = "queue-%s" % short_uid()
  
          dlq_info = self.client.create_queue(QueueName=dlq_name)
          dlq_arn = aws_stack.sqs_queue_arn(dlq_name)
  
          attributes = {
              "RedrivePolicy": json.dumps({"deadLetterTargetArn": dlq_arn, "maxReceiveCount": 100})
          }
          queue_url = self.client.create_queue(QueueName=queue_name, Attributes=attributes)[
              "QueueUrl"
          ]
  
          # clean up
          self.client.delete_queue(QueueUrl=queue_url)
          self.client.delete_queue(QueueUrl=dlq_info["QueueUrl"])
  
      def test_dead_letter_queue_execution(self):
          lambda_client = aws_stack.connect_to_service("lambda")
  
          # create SQS queue with DLQ redrive policy
          queue_name1 = "dlq-%s" % short_uid()
          queue_name2 = "test-dlq-%s" % short_uid()
          queue_url1 = self.client.create_queue(QueueName=queue_name1)["QueueUrl"]
          queue_arn1 = aws_stack.sqs_queue_arn(queue_name1)
          policy = {"deadLetterTargetArn": queue_arn1, "maxReceiveCount": 1}
          queue_url2 = self.client.create_queue(
              QueueName=queue_name2, Attributes={"RedrivePolicy": json.dumps(policy)}
          )["QueueUrl"]
          queue_arn2 = aws_stack.sqs_queue_arn(queue_name2)
  
          # create Lambda and add source mapping
          lambda_name = "test-%s" % short_uid()
          testutil.create_lambda_function(
              func_name=lambda_name,
              libs=TEST_LAMBDA_LIBS,
              handler_file=TEST_LAMBDA_PYTHON,
              runtime=LAMBDA_RUNTIME_PYTHON36,
          )
          lambda_client.create_event_source_mapping(
              EventSourceArn=queue_arn2, FunctionName=lambda_name
          )
  
          # add message to SQS, which will trigger the Lambda, resulting in an error
          payload = {lambda_integration.MSG_BODY_RAISE_ERROR_FLAG: 1}
          self.client.send_message(QueueUrl=queue_url2, MessageBody=json.dumps(payload))
  
          retry(lambda: self.receive_dlq(queue_url1), retries=8, sleep=2)
  
      def test_dead_letter_queue_max_receive_count(self):
  
          # create SQS queue with DLQ redrive policy using "maxReceiveCount"
          queue_name1 = "dlq-%s" % short_uid()
          queue_name2 = "test-dlq-%s" % short_uid()
          queue_url1 = self.client.create_queue(QueueName=queue_name1)["QueueUrl"]
          queue_arn1 = aws_stack.sqs_queue_arn(queue_name1)
          policy = {"deadLetterTargetArn": queue_arn1, "maxReceiveCount": "1"}
          queue_url2 = self.client.create_queue(
              QueueName=queue_name2,
              Attributes={"VisibilityTimeout": "1", "RedrivePolicy": json.dumps(policy)},
          )["QueueUrl"]
  
          # add message to SQS, then retrieve the message
          payload = {}
          self.client.send_message(QueueUrl=queue_url2, MessageBody=json.dumps(payload))
          rs = self.client.receive_message(QueueUrl=queue_url2)
          self.assertEqual(len(rs.get("Messages", [])), 1)
          # wait some time, then try to receive the message again - should be empty
          time.sleep(1.01)
          rs = self.client.receive_message(QueueUrl=queue_url2)
          self.assertEqual(len(rs.get("Messages", [])), 0)
  
          # assert that message has been put on the DLQ
          retry(
              lambda: self.receive_dlq(queue_url1, assert_receive_count=2),
              retries=8,
              sleep=2,
          )
  
      def test_set_queue_attribute_at_creation(self):
          queue_name = "queue-%s" % short_uid()
  
          attributes = {
              "MessageRetentionPeriod": "604800",  # Unsupported by ElasticMq, should be saved in memory
              "ReceiveMessageWaitTimeSeconds": "10",
              "VisibilityTimeout": "30",
          }
  
          queue_url = self.client.create_queue(QueueName=queue_name, Attributes=attributes)[
              "QueueUrl"
          ]
          creation_attributes = self.client.get_queue_attributes(
              QueueUrl=queue_url, AttributeNames=["All"]
          )
  
          # assertion
          self.assertIn("MessageRetentionPeriod", creation_attributes["Attributes"].keys())
          self.assertEqual("604800", creation_attributes["Attributes"]["MessageRetentionPeriod"])
  
          # cleanup
          self.client.delete_queue(QueueUrl=queue_url)
  
      def test_get_specific_queue_attribute_response(self):
          queue_name = "queue-%s" % short_uid()
  
          dead_queue_url = self.client.create_queue(QueueName="newQueue")["QueueUrl"]
          supported_attribute_get = self.client.get_queue_attributes(
              QueueUrl=dead_queue_url, AttributeNames=["QueueArn"]
          )
  
          self.assertTrue("QueueArn" in supported_attribute_get["Attributes"].keys())
          dead_queue_arn = supported_attribute_get["Attributes"]["QueueArn"]
  
          _redrive_policy = {
              "deadLetterTargetArn": dead_queue_arn,
              "maxReceiveCount": "10",
          }
  
          attributes = {
              "MessageRetentionPeriod": "604800",
              "DelaySeconds": "10",
              "RedrivePolicy": json.dumps(_redrive_policy),
          }
  
          queue_url = self.client.create_queue(QueueName=queue_name, Attributes=attributes)[
              "QueueUrl"
          ]
          unsupported_attribute_get = self.client.get_queue_attributes(
              QueueUrl=queue_url,
              AttributeNames=["MessageRetentionPeriod", "RedrivePolicy"],
          )
          supported_attribute_get = self.client.get_queue_attributes(
              QueueUrl=queue_url, AttributeNames=["QueueArn"]
          )
          # assertion
          self.assertTrue("MessageRetentionPeriod" in unsupported_attribute_get["Attributes"].keys())
          self.assertEqual(
              "604800", unsupported_attribute_get["Attributes"]["MessageRetentionPeriod"]
          )
          self.assertTrue("QueueArn" in supported_attribute_get["Attributes"].keys())
          self.assertTrue("RedrivePolicy" in unsupported_attribute_get["Attributes"].keys())
  
          redrive_policy = json.loads(unsupported_attribute_get["Attributes"]["RedrivePolicy"])
          self.assertTrue(isinstance(redrive_policy["maxReceiveCount"], int))
  
          # cleanup
          self.client.delete_queue(QueueUrl=queue_url)
          self.client.delete_queue(QueueUrl=dead_queue_url)
  
      def test_set_unsupported_attributes(self):
          queue_name = "queue-%s" % short_uid()
          queue_url = self.client.create_queue(QueueName=queue_name)["QueueUrl"]
  
          self.client.set_queue_attributes(QueueUrl=queue_url, Attributes={"FifoQueue": "true"})
  
          rs = self.client.get_queue_attributes(QueueUrl=queue_url, AttributeNames=["All"])
  
          self.assertEqual(rs["ResponseMetadata"]["HTTPStatusCode"], 200)
          self.assertIn("FifoQueue", rs["Attributes"])
          self.assertEqual(rs["Attributes"]["FifoQueue"], "true")
  
          # cleanup
          self.client.delete_queue(QueueUrl=queue_url)
  
      def test_list_dead_letter_source_queues(self):
          normal_queue_name = "queue-%s" % short_uid()
          dlq_name = "queue-%s" % short_uid()
  
          dlq = self.client.create_queue(QueueName=dlq_name)
  
          dlq_arn = aws_stack.sqs_queue_arn(dlq_name)
  
          attributes = {
              "RedrivePolicy": json.dumps({"deadLetterTargetArn": dlq_arn, "maxReceiveCount": 100})
          }
          nq = self.client.create_queue(QueueName=normal_queue_name, Attributes=attributes)[
              "QueueUrl"
          ]
  
          res = self.client.list_dead_letter_source_queues(QueueUrl=dlq["QueueUrl"])
  
          self.assertEqual(res["queueUrls"][0], nq)
          self.assertEqual(res["ResponseMetadata"]["HTTPStatusCode"], 200)
  
          self.assertEqual(res["queueUrls"][0], nq)
          self.assertEqual(len(res["queueUrls"]), 1)
          self.assertEqual(res["ResponseMetadata"]["HTTPStatusCode"], 200)
  
          # clean up
          self.client.delete_queue(QueueUrl=nq)
          self.client.delete_queue(QueueUrl=dlq["QueueUrl"])
  
      def test_lambda_invoked_by_sqs_message_with_attributes(self):
          function_name = "lambda_func-{}".format(short_uid())
          queue_name = "queue-{}".format(short_uid())
  
          queue_url = self.client.create_queue(QueueName=queue_name)["QueueUrl"]
  
          testutil.create_lambda_function(
              handler_file=TEST_LAMBDA_ECHO_FILE,
              func_name=function_name,
              runtime=LAMBDA_RUNTIME_PYTHON36,
          )
  
          lambda_client = aws_stack.connect_to_service("lambda")
          lambda_client.create_event_source_mapping(
              EventSourceArn=aws_stack.sqs_queue_arn(queue_name),
              FunctionName=function_name,
          )
  
          self.client.send_message(
              QueueUrl=queue_url,
              MessageBody="hello world.",
              MessageAttributes=TEST_MESSAGE_ATTRIBUTES,
          )
  
          events = retry(get_lambda_log_events, sleep_before=3, function_name=function_name)
          self.assertEqual(len(events), 1)
  
          sqs_msg = events[0]["Records"][0]
          self.assertEqual(sqs_msg["body"], "hello world.")
  
          self.assertIn("messageAttributes", sqs_msg)
          self.assertIn("City", sqs_msg["messageAttributes"])
          attr_lower = TEST_MESSAGE_ATTRIBUTES["City"]
          attr_lower = {
              "dataType": attr_lower["DataType"],
              "stringValue": attr_lower["StringValue"],
          }
          self.assertEqual(sqs_msg["messageAttributes"]["City"], attr_lower)
  
          # clean up
          self.client.delete_queue(QueueUrl=queue_url)
          lambda_client.delete_function(FunctionName=function_name)
  
      def test_send_message_with_delay_seconds(self):
          queue_name = "queue-{}".format(short_uid())
          queue_url = self.client.create_queue(QueueName=queue_name)["QueueUrl"]
  
          # send message with DelaySeconds = 0
          self.client.send_message(QueueUrl=queue_url, MessageBody="hello world.", DelaySeconds=0)
  
          rs = self.client.receive_message(QueueUrl=queue_url)
          self.assertIn("Messages", rs)
          self.assertEqual(len(rs["Messages"]), 1)
  
          message = rs["Messages"][0]
          self.assertEqual(message["Body"], "hello world.")
          self.client.delete_message_batch(
              QueueUrl=queue_url,
              Entries=[{"Id": short_uid(), "ReceiptHandle": message["ReceiptHandle"]}],
          )
  
          # send message with DelaySeconds = 10
          self.client.send_message(QueueUrl=queue_url, MessageBody="test_message_2", DelaySeconds=10)
  
          rs = self.client.receive_message(QueueUrl=queue_url)
          self.assertEqual(rs["ResponseMetadata"]["HTTPStatusCode"], 200)
          self.assertNotIn("Messages", rs)
  
          def get_message(q_url):
              resp = self.client.receive_message(QueueUrl=q_url)
              return resp["Messages"]
  
          messages = retry(get_message, retries=3, sleep=10, q_url=queue_url)
          self.assertEqual(len(messages), 1)
          self.assertEqual(messages[0]["Body"], "test_message_2")
  
          # clean up
          self.client.delete_queue(QueueUrl=queue_url)
  
      def test_get_multiple_messages(self):
          queue_name = "queue-{}".format(short_uid())
          queue_url = self.client.create_queue(QueueName=queue_name)["QueueUrl"]
          number_of_messages = 3
  
          for i in range(number_of_messages):
              self.client.send_message(
                  QueueUrl=queue_url,
                  MessageBody="hello world. {}".format(i),
                  DelaySeconds=0,
              )
  
          messages = {}
          for i in range(number_of_messages):
              rs = self.client.receive_message(QueueUrl=queue_url)
              m = rs["Messages"][0]
              messages[m["MessageId"]] = m["Body"]
  
          self.assertEqual(len(messages.keys()), number_of_messages)
  
          # clean up
          self.client.delete_queue(QueueUrl=queue_url)
  
      def test_lambda_invoked_by_sqs_message_with_delay_seconds(self):
          function_name = "lambda_func-{}".format(short_uid())
          queue_name = "queue-{}".format(short_uid())
          delay_time = 6
  
          queue_url = self.client.create_queue(QueueName=queue_name)["QueueUrl"]
          queue_arn = aws_stack.sqs_queue_arn(queue_name)
  
          testutil.create_lambda_function(
              handler_file=TEST_LAMBDA_ECHO_FILE,
              func_name=function_name,
              runtime=LAMBDA_RUNTIME_PYTHON36,
          )
  
          lambda_client = aws_stack.connect_to_service("lambda")
          lambda_client.create_event_source_mapping(
              EventSourceArn=queue_arn, FunctionName=function_name
          )
  
          rs = self.client.send_message(
              QueueUrl=queue_url, MessageBody="hello world.", DelaySeconds=delay_time
          )
          message_id = rs["MessageId"]
  
          time.sleep(delay_time / 2)
  
          # There is no log group for this lambda (lambda not invoked yet)
          log_events = get_lambda_log_events(function_name)
          self.assertEqual(len(log_events), 0)
  
          # After delay time, lambda invoked by sqs
          events = get_lambda_log_events(function_name, delay_time * 1.5)
          # Lambda just invoked 1 time
          self.assertEqual(len(events), 1)
  
          message = events[0]["Records"][0]
          self.assertEqual(message["eventSourceARN"], queue_arn)
          self.assertEqual(message["messageId"], message_id)
  
          # clean up
          self.client.delete_queue(QueueUrl=queue_url)
          lambda_client.delete_function(FunctionName=function_name)
  
      def test_get_queue_attributes(self):
  
          sqs = self.client
          queue_name1 = "test-%s" % short_uid()
          queue_name2 = "test-%s" % short_uid()
  
          sqs.create_queue(QueueName=queue_name1)
          queue_arn1 = aws_stack.sqs_queue_arn(queue_name1)
          policy = {"deadLetterTargetArn": queue_arn1, "maxReceiveCount": 1}
  
          queue_url2 = self.client.create_queue(
              QueueName=queue_name2, Attributes={"RedrivePolicy": json.dumps(policy)}
          )["QueueUrl"]
  
          response = sqs.get_queue_attributes(QueueUrl=queue_url2)
  
          redrive_policy = json.loads(response["Attributes"]["RedrivePolicy"])
          self.assertEqual(redrive_policy["maxReceiveCount"], 1)
          self.assertIn(redrive_policy["deadLetterTargetArn"], queue_arn1)
  
      def test_send_message_batch_with_empty_list(self):
          client = self.client
          response = client.create_queue(QueueName="test-queue")
          queue_url = response["QueueUrl"]
  
          try:
              client.send_message_batch(QueueUrl=queue_url, Entries=[])
          except ClientError as e:
              self.assertEqual(e.response["Error"]["Code"], "EmptyBatchRequest")
              self.assertIn(e.response["ResponseMetadata"]["HTTPStatusCode"], [400, 404])
  
          entries = [
              {
                  "Id": "message{:02d}".format(0),
                  "MessageBody": "msgBody{:02d}".format(0),
                  "MessageAttributes": {
                      "CustomAttribute": {
                          "DataType": "String",
                          "StringValue": "CustomAttributeValue{:02d}".format(0),
                      }
                  },
              }
          ]
  
          result = client.send_message_batch(QueueUrl=queue_url, Entries=entries)
          self.assertEqual(result["ResponseMetadata"]["HTTPStatusCode"], 200)
          # clean up
          client.delete_queue(QueueUrl=queue_url)
  
      def _run_test_lambda_invoked_by_sqs_message_with_delay_seconds_dotnet(
          self, zip_file, handler, runtime
      ):
          if not use_docker():
              return
  
          func_name = "dotnet-sqs-{}".format(short_uid())
          queue_name = "queue-%s" % short_uid()
          queue_url = self.client.create_queue(QueueName=queue_name)["QueueUrl"]
          queue_arn = aws_stack.sqs_queue_arn(queue_name)
          delay_time = 1
  
          testutil.create_lambda_function(
              func_name=func_name, zip_file=zip_file, handler=handler, runtime=runtime
          )
  
          lambda_client = aws_stack.connect_to_service("lambda")
          lambda_client.create_event_source_mapping(EventSourceArn=queue_arn, FunctionName=func_name)
  
          self.client.send_message(
              QueueUrl=queue_url, MessageBody="hello world.", DelaySeconds=delay_time
          )
  
          # assert that the Lambda has been invoked
          def get_logs():
              logs = get_lambda_log_events(func_name)
              self.assertGreater(len(logs), 0)
  
          retry(get_logs, retries=5, sleep=3)
  
          # assert that the message has been deleted from the queue
          resp = self.client.receive_message(QueueUrl=queue_url, MessageAttributeNames=["All"])
          self.assertEqual(200, resp["ResponseMetadata"]["HTTPStatusCode"])
          self.assertEqual(None, resp.get("Messages", None))
  
          # clean up
          self.client.delete_queue(QueueUrl=queue_url)
          testutil.delete_lambda_function(func_name)
  
      def test_lambda_invoked_by_sqs_message_with_delay_seconds_dotnetcore2(self):
          zip_file = load_file(TEST_LAMBDA_DOTNETCORE2, mode="rb")
          handler = "DotNetCore2::DotNetCore2.Lambda.Function::SimpleFunctionHandler"
  
          self._run_test_lambda_invoked_by_sqs_message_with_delay_seconds_dotnet(
              zip_file, handler, LAMBDA_RUNTIME_DOTNETCORE2
          )
  
      def test_lambda_invoked_by_sqs_message_with_delay_seconds_dotnetcore31(self):
          zip_file = load_file(TEST_LAMBDA_DOTNETCORE31, mode="rb")
          handler = "dotnetcore31::dotnetcore31.Function::FunctionHandler"
  
          self._run_test_lambda_invoked_by_sqs_message_with_delay_seconds_dotnet(
              zip_file, handler, LAMBDA_RUNTIME_DOTNETCORE31
          )
  
      def _run_test_fifo_queue_send_multiple_messages(self):
          fifo_queue = "queue-{}.fifo".format(short_uid())
  
          message_group = "group-%s" % short_uid()
          results = []
          number_of_messages = 5
  
          queue_url = self.client.create_queue(
              QueueName=fifo_queue, Attributes={"FifoQueue": "true"}
          )["QueueUrl"]
  
          # it should preserve .fifo in the queue name
          self.assertIn(fifo_queue, queue_url)
  
          # try sending multiple message with message group ID and deduplication ID
          for i in range(number_of_messages):
              rs = self.client.send_message(
                  QueueUrl=queue_url,
                  MessageBody="message-{}".format(i),
                  MessageDeduplicationId="deduplication-{}".format(i),
                  MessageGroupId=message_group,
              )
              results.append(rs)
  
          return queue_url, number_of_messages, results
  
      def test_fifo_queue_send_multiple_messages_single_receive(self):
          (
              queue_url,
              number_of_messages,
              results,
          ) = self._run_test_fifo_queue_send_multiple_messages()
  
          # receive multiple message in the same time
          messages = self.client.receive_message(
              QueueUrl=queue_url,
              MessageAttributeNames=["All"],
              MaxNumberOfMessages=number_of_messages,
          )
  
          # asset the received messages data
          self.assertEqual(number_of_messages, len(messages["Messages"]))
          for i in range(number_of_messages):
              self.assertEqual("message-{}".format(i), messages["Messages"][i]["Body"])
              self.assertEqual(results[i]["MD5OfMessageBody"], messages["Messages"][i]["MD5OfBody"])
              self.assertEqual(results[i]["MessageId"], messages["Messages"][i]["MessageId"])
  
          # clean up
          self.client.delete_queue(QueueUrl=queue_url)
  
      def test_fifo_queue_send_multiple_messages_multiple_receives(self):
          (
              queue_url,
              number_of_messages,
              results,
          ) = self._run_test_fifo_queue_send_multiple_messages()
          first_receives = number_of_messages // 2
  
          # receive multiple message first time
          messages = self.client.receive_message(
              QueueUrl=queue_url,
              MessageAttributeNames=["All"],
              MaxNumberOfMessages=first_receives,
          )
  
          # asset the received messages data first time
          self.assertEqual(first_receives, len(messages["Messages"]))
          for i in range(first_receives):
              self.assertEqual("message-{}".format(i), messages["Messages"][i]["Body"])
              self.assertEqual(results[i]["MD5OfMessageBody"], messages["Messages"][i]["MD5OfBody"])
              self.assertEqual(results[i]["MessageId"], messages["Messages"][i]["MessageId"])
  
              # delete message to receive next message in queue
              self.client.delete_message(
                  QueueUrl=queue_url,
                  ReceiptHandle=messages["Messages"][i]["ReceiptHandle"],
              )
  
          second_receives = number_of_messages - first_receives
  
          # try to get one by one message in the second time
          for i in range(second_receives):
              message = self.client.receive_message(QueueUrl=queue_url)
              self.assertEqual(
                  "message-{}".format(first_receives + i), message["Messages"][0]["Body"]
              )
              self.assertEqual(
                  results[first_receives + i]["MD5OfMessageBody"],
                  message["Messages"][0]["MD5OfBody"],
              )
              self.assertEqual(
                  results[first_receives + i]["MessageId"],
                  message["Messages"][0]["MessageId"],
              )
  
              # delete message to receive next message in queue
              self.client.delete_message(
                  QueueUrl=queue_url,
                  ReceiptHandle=message["Messages"][0]["ReceiptHandle"],
              )
  
          # clean up
          self.client.delete_queue(QueueUrl=queue_url)
  
      def test_fifo_queue_send_multiple_messages_multiple_single_receives(self):
          (
              queue_url,
              number_of_messages,
              results,
          ) = self._run_test_fifo_queue_send_multiple_messages()
  
          for i in range(number_of_messages):
              resp = self.client.receive_message(QueueUrl=queue_url)
              self.assertEqual("message-{}".format(i), resp["Messages"][0]["Body"])
              self.assertEqual(results[i]["MD5OfMessageBody"], resp["Messages"][0]["MD5OfBody"])
              self.assertEqual(results[i]["MessageId"], resp["Messages"][0]["MessageId"])
  
              # delete message to receive next message in queue
              self.client.delete_message(
                  QueueUrl=queue_url, ReceiptHandle=resp["Messages"][0]["ReceiptHandle"]
              )
  
          # clean up
          self.client.delete_queue(QueueUrl=queue_url)
  
      def test_tag_untag_queue(self):
          queue_name = "queue-{}".format(short_uid())
          queue_url = self.client.create_queue(QueueName=queue_name)["QueueUrl"]
  
          response = self.client.tag_queue(
              QueueUrl=queue_url, Tags={"tag1": "value1", "tag2": "value2"}
          )
          self.assertEqual(200, response["ResponseMetadata"]["HTTPStatusCode"])
  
          response = self.client.list_queue_tags(QueueUrl=queue_url)
          self.assertEqual(200, response["ResponseMetadata"]["HTTPStatusCode"])
          self.assertIn("tag1", response["Tags"])
          self.assertIn("tag2", response["Tags"])
  
          response = self.client.untag_queue(QueueUrl=queue_url, TagKeys=["tag2"])
          self.assertEqual(200, response["ResponseMetadata"]["HTTPStatusCode"])
  
          response = self.client.list_queue_tags(QueueUrl=queue_url)
          self.assertIn("tag1", response["Tags"])
          self.assertNotIn("tag2", response["Tags"])
          self.assertEqual(200, response["ResponseMetadata"]["HTTPStatusCode"])
  
          # clean up
          self.client.untag_queue(QueueUrl=queue_url, TagKeys=["tag1"])
          self.client.delete_queue(QueueUrl=queue_url)
  
      def test_posting_to_queue_with_trailing_slash(self):
          queue_name = "queue-{}".format(short_uid())
          queue_url = self.client.create_queue(QueueName=queue_name)["QueueUrl"]
  
          base_url = "{}://{}:{}".format(
              get_service_protocol(), config.LOCALSTACK_HOSTNAME, config.PORT_SQS
          )
          encoded_url = urlencode(
              {
                  "Action": "SendMessage",
                  "Version": "2012-11-05",
                  "QueueUrl": "{}/{}/{}/".format(base_url, TEST_AWS_ACCOUNT_ID, queue_name),
                  "MessageBody": "test body",
              }
          )
          r = requests.post(url=base_url, data=encoded_url)
          self.assertEqual(r.status_code, 200)
  
          # We can get the message back
          resp = self.client.receive_message(QueueUrl=queue_url)
          self.assertEqual(resp["Messages"][0]["Body"], "test body")
  
          # clean up
          self.client.delete_queue(QueueUrl=queue_url)
  
      def test_create_queue_with_slashes(self):
          queue_name = "queue/%s" % short_uid()
          queue_url = self.client.create_queue(QueueName=queue_name)
  
          result = self.client.list_queues()
          self.assertIn(queue_url.get("QueueUrl"), result.get("QueueUrls"))
  
          # clean up
          self.client.delete_queue(QueueUrl=queue_url.get("QueueUrl"))
  
          result = self.client.list_queues()
          self.assertNotIn(queue_url.get("QueueUrl"), result.get("QueueUrls", []))
  
      def list_queues_with_auth_in_presigned_url(self, method):
          base_url = "{}://{}:{}".format(
              get_service_protocol(), config.LOCALSTACK_HOSTNAME, config.PORT_SQS
          )
  
          req = AWSRequest(
              method=method,
              url=base_url,
              data={"Action": "ListQueues", "Version": "2012-11-05"},
          )
  
          # boto doesn't support querystring-style auth, so we have to do some
          # weird logic to use boto's signing functions, to understand what's
          # going on here look at the internals of the SigV4Auth.add_auth
          # method.
          datetime_now = datetime.datetime.utcnow()
          req.context["timestamp"] = datetime_now.strftime(SIGV4_TIMESTAMP)
          signer = SigV4Auth(
              Credentials(TEST_AWS_ACCESS_KEY_ID, TEST_AWS_SECRET_ACCESS_KEY),
              "sqs",
              aws_stack.get_region(),
          )
          canonical_request = signer.canonical_request(req)
          string_to_sign = signer.string_to_sign(req, canonical_request)
  
          payload = {
              "Action": "ListQueues",
              "Version": "2012-11-05",
              "X-Amz-Algorithm": "AWS4-HMAC-SHA256",
              "X-Amz-Credential": signer.scope(req),
              "X-Amz-SignedHeaders": ";".join(signer.headers_to_sign(req).keys()),
              "X-Amz-Signature": signer.signature(string_to_sign, req),
          }
  
          if method == "GET":
              return requests.get(url=base_url, params=payload)
          else:
              return requests.post(url=base_url, data=urlencode(payload))
  
      def test_post_list_queues_with_auth_in_presigned_url(self):
          res = self.list_queues_with_auth_in_presigned_url(method="POST")
          self.assertEqual(res.status_code, 200)
          self.assertTrue(b"<ListQueuesResponse>" in res.content)
  
      def test_get_list_queues_with_auth_in_presigned_url(self):
          res = self.list_queues_with_auth_in_presigned_url(method="GET")
          self.assertEqual(res.status_code, 200)
          self.assertTrue(b"<ListQueuesResponse>" in res.content)
  
      # ---------------
      # HELPER METHODS
      # ---------------
  
      def receive_dlq(self, queue_url, assert_error_details=False, assert_receive_count=None):
          """Assert that a message has been received on the given DLQ"""
          result = self.client.receive_message(QueueUrl=queue_url, MessageAttributeNames=["All"])
          self.assertGreater(len(result["Messages"]), 0)
          msg = result["Messages"][0]
          msg_attrs = msg.get("MessageAttributes") or msg.get("Attributes")
          if assert_error_details:
              self.assertIn("RequestID", msg_attrs)
              self.assertIn("ErrorCode", msg_attrs)
              self.assertIn("ErrorMessage", msg_attrs)
          else:
              if assert_receive_count is not None:
                  pass
                  # TODO: this started failing with latest moto upgrade,
                  # probably in or around this commit:
                  # https://github.com/spulec/moto/commit/6da4905da940e25e317db60b7657ea632f58ef1d
                  # self.assertEqual(str(assert_receive_count), msg_attrs.get('ApproximateReceiveCount'))
  