Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use aws_http_client_factory method in localstack

Best Python code snippet using localstack_python

test_sqs.py

Source:test_sqs.py

...1502            sqs_create_queue(QueueName=queue_name)1503        e.match("InvalidParameterValue")1504    @pytest.mark.aws_validated1505    def test_get_list_queues_with_query_auth(self, aws_http_client_factory):1506        client = aws_http_client_factory("sqs", region="us-east-1")1507        if os.environ.get("TEST_TARGET") == "AWS_CLOUD":1508            endpoint_url = "https://queue.amazonaws.com"1509        else:1510            endpoint_url = config.get_edge_url()1511        response = client.get(1512            endpoint_url, params={"Action": "ListQueues", "Version": "2012-11-05"}1513        )1514        assert response.status_code == 2001515        assert b"<ListQueuesResponse" in response.content1516    @pytest.mark.aws_validated1517    def test_system_attributes_have_no_effect_on_attr_md5(self, sqs_create_queue, sqs_client):1518        queue_url = sqs_create_queue()1519        msg_attrs_provider = {"timestamp": {"StringValue": "1493147359900", "DataType": "Number"}}1520        aws_trace_header = {1521            "AWSTraceHeader": {1522                "StringValue": "Root=1-5759e988-bd862e3fe1be46a994272793;Parent=53995c3f42cd8ad8;Sampled=1",1523                "DataType": "String",1524            }1525        }1526        response_send = sqs_client.send_message(1527            QueueUrl=queue_url, MessageBody="test", MessageAttributes=msg_attrs_provider1528        )1529        response_send_system_attr = sqs_client.send_message(1530            QueueUrl=queue_url,1531            MessageBody="test",1532            MessageAttributes=msg_attrs_provider,1533            MessageSystemAttributes=aws_trace_header,1534        )1535        assert (1536            response_send["MD5OfMessageAttributes"]1537            == response_send_system_attr["MD5OfMessageAttributes"]1538        )1539        assert response_send.get("MD5OfMessageSystemAttributes") is None1540        assert (1541            response_send_system_attr.get("MD5OfMessageSystemAttributes")1542            == "5ae4d5d7636402d80f4eb6d213245a88"1543        )1544    def test_inflight_message_requeue(self, sqs_client, sqs_create_queue):1545        visibility_timeout = 31546        queue_name = f"queue-{short_uid()}"1547        queue_url = sqs_create_queue(1548            QueueName=queue_name1549        )  # , Attributes={"VisibilityTimeout": str(visibility_timeout)})1550        sqs_client.send_message(QueueUrl=queue_url, MessageBody="test1")1551        result_receive1 = sqs_client.receive_message(1552            QueueUrl=queue_url, VisibilityTimeout=visibility_timeout1553        )1554        time.sleep(visibility_timeout / 2)1555        sqs_client.send_message(QueueUrl=queue_url, MessageBody="test2")1556        time.sleep(visibility_timeout)1557        result_receive2 = sqs_client.receive_message(1558            QueueUrl=queue_url, VisibilityTimeout=visibility_timeout1559        )1560        assert result_receive1["Messages"][0]["Body"] == result_receive2["Messages"][0]["Body"]1561    @pytest.mark.aws_validated1562    def test_sequence_number(self, sqs_client, sqs_create_queue):1563        fifo_queue_name = f"queue-{short_uid()}.fifo"1564        fifo_queue_url = sqs_create_queue(1565            QueueName=fifo_queue_name, Attributes={"FifoQueue": "true"}1566        )1567        message_content = f"test{short_uid()}"1568        dedup_id = f"fifo_dedup-{short_uid()}"1569        group_id = f"fifo_group-{short_uid()}"1570        send_result_fifo = sqs_client.send_message(1571            QueueUrl=fifo_queue_url,1572            MessageBody=message_content,1573            MessageGroupId=group_id,1574            MessageDeduplicationId=dedup_id,1575        )1576        assert "SequenceNumber" in send_result_fifo.keys()1577        queue_name = f"queue-{short_uid()}"1578        queue_url = sqs_create_queue(QueueName=queue_name)1579        send_result = sqs_client.send_message(QueueUrl=queue_url, MessageBody=message_content)1580        assert "SequenceNumber" not in send_result1581    @pytest.mark.aws_validated1582    def test_posting_to_fifo_requires_deduplicationid_group_id(self, sqs_client, sqs_create_queue):1583        fifo_queue_name = f"queue-{short_uid()}.fifo"1584        queue_url = sqs_create_queue(QueueName=fifo_queue_name, Attributes={"FifoQueue": "true"})1585        message_content = f"test{short_uid()}"1586        dedup_id = f"fifo_dedup-{short_uid()}"1587        group_id = f"fifo_group-{short_uid()}"1588        with pytest.raises(Exception) as e:1589            sqs_client.send_message(1590                QueueUrl=queue_url, MessageBody=message_content, MessageGroupId=group_id1591            )1592        e.match("InvalidParameterValue")1593        with pytest.raises(Exception) as e:1594            sqs_client.send_message(1595                QueueUrl=queue_url, MessageBody=message_content, MessageDeduplicationId=dedup_id1596            )1597        e.match("MissingParameter")1598    @pytest.mark.aws_validated1599    def test_posting_to_queue_via_queue_name(self, sqs_client, sqs_create_queue):1600        # TODO: behaviour diverges from AWS1601        queue_name = f"queue-{short_uid()}"1602        sqs_create_queue(QueueName=queue_name)1603        result_send = sqs_client.send_message(1604            QueueUrl=queue_name, MessageBody="Using name instead of URL"1605        )1606        assert result_send["MD5OfMessageBody"] == "86a83f96652a1bfad3891e7d523750cb"1607        assert result_send["ResponseMetadata"]["HTTPStatusCode"] == 2001608    @pytest.mark.aws_validated1609    def test_invalid_string_attributes_cause_invalid_parameter_value_error(1610        self, sqs_client, sqs_create_queue1611    ):1612        queue_name = f"queue-{short_uid()}"1613        queue_url = sqs_create_queue(QueueName=queue_name)1614        invalid_attribute = {1615            "attr1": {"StringValue": f"Invalid-{chr(8)},{chr(11)}", "DataType": "String"}1616        }1617        with pytest.raises(Exception) as e:1618            sqs_client.send_message(1619                QueueUrl=queue_url, MessageBody="test", MessageAttributes=invalid_attribute1620            )1621        e.match("InvalidParameterValue")1622    @pytest.mark.aws_validated1623    def test_change_message_visibility_not_permanent(self, sqs_client, sqs_create_queue):1624        queue_name = f"queue-{short_uid()}"1625        queue_url = sqs_create_queue(QueueName=queue_name)1626        sqs_client.send_message(QueueUrl=queue_url, MessageBody="test")1627        result_receive = sqs_client.receive_message(QueueUrl=queue_url)1628        receipt_handle = result_receive.get("Messages")[0]["ReceiptHandle"]1629        sqs_client.change_message_visibility(1630            QueueUrl=queue_url, ReceiptHandle=receipt_handle, VisibilityTimeout=01631        )1632        result_recv_1 = sqs_client.receive_message(QueueUrl=queue_url)1633        result_recv_2 = sqs_client.receive_message(QueueUrl=queue_url)1634        assert (1635            result_recv_1.get("Messages")[0]["MessageId"]1636            == result_receive.get("Messages")[0]["MessageId"]1637        )1638        assert "Messages" not in result_recv_2.keys()1639    @pytest.mark.skip1640    def test_dead_letter_queue_execution_lambda_mapping_preserves_id(1641        self, sqs_client, sqs_create_queue, lambda_client, create_lambda_function1642    ):1643        # TODO: lambda triggered dead letter delivery does not preserve the message id1644        # https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/sqs-dead-letter-queues.html1645        queue_name = f"queue-{short_uid()}"1646        dead_letter_queue_name = "dl-queue-{}".format(short_uid())1647        dl_queue_url = sqs_create_queue(QueueName=dead_letter_queue_name)1648        # create arn1649        url_parts = dl_queue_url.split("/")1650        region = get_region()1651        dl_target_arn = "arn:aws:sqs:{}:{}:{}".format(1652            region, url_parts[len(url_parts) - 2], url_parts[-1]1653        )1654        policy = {"deadLetterTargetArn": dl_target_arn, "maxReceiveCount": 1}1655        queue_url = sqs_create_queue(1656            QueueName=queue_name, Attributes={"RedrivePolicy": json.dumps(policy)}1657        )1658        lambda_name = "lambda-{}".format(short_uid())1659        create_lambda_function(1660            func_name=lambda_name,1661            libs=TEST_LAMBDA_LIBS,1662            handler_file=TEST_LAMBDA_PYTHON,1663            runtime=LAMBDA_RUNTIME_PYTHON36,1664        )1665        # create arn1666        url_parts = queue_url.split("/")1667        queue_arn = "arn:aws:sqs:{}:{}:{}".format(1668            region, url_parts[len(url_parts) - 2], url_parts[-1]1669        )1670        lambda_client.create_event_source_mapping(1671            EventSourceArn=queue_arn, FunctionName=lambda_name1672        )1673        # add message to SQS, which will trigger the Lambda, resulting in an error1674        payload = {lambda_integration.MSG_BODY_RAISE_ERROR_FLAG: 1}1675        result_send = sqs_client.send_message(QueueUrl=queue_url, MessageBody=json.dumps(payload))1676        assert poll_condition(1677            lambda: "Messages"1678            in sqs_client.receive_message(QueueUrl=dl_queue_url, VisibilityTimeout=0),1679            5.0,1680            1.0,1681        )1682        result_recv = sqs_client.receive_message(QueueUrl=dl_queue_url, VisibilityTimeout=0)1683        assert result_recv["Messages"][0]["MessageId"] == result_send["MessageId"]1684    # verification of community posted issue1685    # FIXME: \r gets lost1686    @pytest.mark.skip1687    def test_message_with_carriage_return(self, sqs_client, sqs_create_queue):1688        queue_name = f"queue-{short_uid()}"1689        queue_url = sqs_create_queue(QueueName=queue_name)1690        message_content = "{\r\n" + '"machineID" : "d357006e26ff47439e1ef894225d4307"' + "}"1691        result_send = sqs_client.send_message(QueueUrl=queue_url, MessageBody=message_content)1692        result_receive = sqs_client.receive_message(QueueUrl=queue_url)1693        assert result_send["MD5OfMessageBody"] == result_receive["Messages"][0]["MD5OfBody"]1694        assert message_content == result_receive["Messages"][0]["Body"]1695    @pytest.mark.aws_validated1696    def test_purge_queue(self, sqs_client, sqs_create_queue):1697        queue_name = f"queue-{short_uid()}"1698        queue_url = sqs_create_queue(QueueName=queue_name)1699        for i in range(3):1700            message_content = f"test-{i}"1701            sqs_client.send_message(QueueUrl=queue_url, MessageBody=message_content)1702        approx_nr_of_messages = sqs_client.get_queue_attributes(1703            QueueUrl=queue_url, AttributeNames=["ApproximateNumberOfMessages"]1704        )1705        assert int(approx_nr_of_messages["Attributes"]["ApproximateNumberOfMessages"]) > 11706        sqs_client.purge_queue(QueueUrl=queue_url)1707        receive_result = sqs_client.receive_message(QueueUrl=queue_url)1708        assert "Messages" not in receive_result.keys()1709    @pytest.mark.aws_validated1710    def test_remove_message_with_old_receipt_handle(self, sqs_client, sqs_create_queue):1711        queue_name = f"queue-{short_uid()}"1712        queue_url = sqs_create_queue(QueueName=queue_name)1713        sqs_client.send_message(QueueUrl=queue_url, MessageBody="test")1714        result_receive = sqs_client.receive_message(QueueUrl=queue_url, VisibilityTimeout=1)1715        time.sleep(2)1716        receipt_handle = result_receive["Messages"][0]["ReceiptHandle"]1717        sqs_client.delete_message(QueueUrl=queue_url, ReceiptHandle=receipt_handle)1718        # This is more suited to the check than receiving because it simply1719        # returns the number of elements in the queue, without further logic1720        approx_nr_of_messages = sqs_client.get_queue_attributes(1721            QueueUrl=queue_url, AttributeNames=["ApproximateNumberOfMessages"]1722        )1723        assert int(approx_nr_of_messages["Attributes"]["ApproximateNumberOfMessages"]) == 01724    @pytest.mark.only_localstack1725    def test_list_queues_multi_region_without_endpoint_strategy(1726        self, create_boto_client, cleanups, monkeypatch1727    ):1728        monkeypatch.setattr(config, "SQS_ENDPOINT_STRATEGY", "off")1729        region1 = "us-east-1"1730        region2 = "eu-central-1"1731        region1_client = create_boto_client("sqs", region_name=region1)1732        region2_client = create_boto_client("sqs", region_name=region2)1733        queue1_name = f"queue-region1-{short_uid()}"1734        queue2_name = f"queue-region2-{short_uid()}"1735        queue1_url = region1_client.create_queue(QueueName=queue1_name)["QueueUrl"]1736        cleanups.append(lambda: region1_client.delete_queue(QueueUrl=queue1_url))1737        queue2_url = region2_client.create_queue(QueueName=queue2_name)["QueueUrl"]1738        cleanups.append(lambda: region2_client.delete_queue(QueueUrl=queue2_url))1739        # region should not be in the queue url with endpoint strategy "off"1740        assert region1 not in queue1_url1741        assert region1 not in queue2_url1742        assert queue1_url in region1_client.list_queues().get("QueueUrls", [])1743        assert queue2_url not in region1_client.list_queues().get("QueueUrls", [])1744        assert queue1_url not in region2_client.list_queues().get("QueueUrls", [])1745        assert queue2_url in region2_client.list_queues().get("QueueUrls", [])1746    @pytest.mark.aws_validated1747    def test_list_queues_multi_region_with_endpoint_strategy_domain(1748        self, create_boto_client, cleanups, monkeypatch1749    ):1750        monkeypatch.setattr(config, "SQS_ENDPOINT_STRATEGY", "domain")1751        region1 = "us-east-1"1752        region2 = "eu-central-1"1753        region1_client = create_boto_client("sqs", region_name=region1)1754        region2_client = create_boto_client("sqs", region_name=region2)1755        queue_name = f"queue-{short_uid()}"1756        queue1_url = region1_client.create_queue(QueueName=queue_name)["QueueUrl"]1757        cleanups.append(lambda: region1_client.delete_queue(QueueUrl=queue1_url))1758        queue2_url = region2_client.create_queue(QueueName=queue_name)["QueueUrl"]1759        cleanups.append(lambda: region2_client.delete_queue(QueueUrl=queue2_url))1760        assert region1 not in queue1_url  # us-east-1 is not included in the default region1761        assert region1 not in queue2_url1762        assert f"{region2}.queue" in queue2_url  # all other regions are part of the endpoint-url1763        assert queue1_url in region1_client.list_queues().get("QueueUrls", [])1764        assert queue2_url not in region1_client.list_queues().get("QueueUrls", [])1765        assert queue1_url not in region2_client.list_queues().get("QueueUrls", [])1766        assert queue2_url in region2_client.list_queues().get("QueueUrls", [])1767    @pytest.mark.aws_validated1768    def test_get_queue_url_multi_region(self, create_boto_client, cleanups, monkeypatch):1769        monkeypatch.setattr(config, "SQS_ENDPOINT_STRATEGY", "domain")1770        region1_client = create_boto_client("sqs", region_name="us-east-1")1771        region2_client = create_boto_client("sqs", region_name="eu-central-1")1772        queue_name = f"queue-{short_uid()}"1773        queue1_url = region1_client.create_queue(QueueName=queue_name)["QueueUrl"]1774        cleanups.append(lambda: region1_client.delete_queue(QueueUrl=queue1_url))1775        queue2_url = region2_client.create_queue(QueueName=queue_name)["QueueUrl"]1776        cleanups.append(lambda: region2_client.delete_queue(QueueUrl=queue2_url))1777        assert queue1_url != queue2_url1778        assert queue1_url == region1_client.get_queue_url(QueueName=queue_name)["QueueUrl"]1779        assert queue2_url == region2_client.get_queue_url(QueueName=queue_name)["QueueUrl"]1780    @pytest.mark.skip(1781        reason="this is an AWS behaviour test that requires 5 minutes to run. Only execute manually"1782    )1783    def test_deduplication_interval(self, sqs_client, sqs_create_queue):1784        # TODO: AWS behaviour here "seems" inconsistent -> current code might need adaption1785        fifo_queue_name = f"queue-{short_uid()}.fifo"1786        queue_url = sqs_create_queue(QueueName=fifo_queue_name, Attributes={"FifoQueue": "true"})1787        message_content = f"test{short_uid()}"1788        message_content_duplicate = f"{message_content}-duplicate"1789        message_content_half_time = f"{message_content}-half_time"1790        dedup_id = f"fifo_dedup-{short_uid()}"1791        group_id = f"fifo_group-{short_uid()}"1792        result_send = sqs_client.send_message(1793            QueueUrl=queue_url,1794            MessageBody=message_content,1795            MessageGroupId=group_id,1796            MessageDeduplicationId=dedup_id,1797        )1798        time.sleep(3)1799        sqs_client.send_message(1800            QueueUrl=queue_url,1801            MessageBody=message_content_duplicate,1802            MessageGroupId=group_id,1803            MessageDeduplicationId=dedup_id,1804        )1805        result_receive = sqs_client.receive_message(QueueUrl=queue_url)1806        sqs_client.delete_message(1807            QueueUrl=queue_url, ReceiptHandle=result_receive["Messages"][0]["ReceiptHandle"]1808        )1809        result_receive_duplicate = sqs_client.receive_message(QueueUrl=queue_url)1810        assert result_send.get("MessageId") == result_receive.get("Messages")[0].get("MessageId")1811        assert result_send.get("MD5OfMessageBody") == result_receive.get("Messages")[0].get(1812            "MD5OfBody"1813        )1814        assert "Messages" not in result_receive_duplicate.keys()1815        result_send = sqs_client.send_message(1816            QueueUrl=queue_url,1817            MessageBody=message_content,1818            MessageGroupId=group_id,1819            MessageDeduplicationId=dedup_id,1820        )1821        # ZZZZzzz...1822        # Fifo Deduplication Interval is 5 minutes at minimum, + there seems no way to change it.1823        # We give it a bit of leeway to avoid timing issues1824        # https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/using-messagededuplicationid-property.html1825        time.sleep(2)1826        sqs_client.send_message(1827            QueueUrl=queue_url,1828            MessageBody=message_content_half_time,1829            MessageGroupId=group_id,1830            MessageDeduplicationId=dedup_id,1831        )1832        time.sleep(6 * 60)1833        result_send_duplicate = sqs_client.send_message(1834            QueueUrl=queue_url,1835            MessageBody=message_content_duplicate,1836            MessageGroupId=group_id,1837            MessageDeduplicationId=dedup_id,1838        )1839        result_receive = sqs_client.receive_message(QueueUrl=queue_url)1840        sqs_client.delete_message(1841            QueueUrl=queue_url, ReceiptHandle=result_receive["Messages"][0]["ReceiptHandle"]1842        )1843        result_receive_duplicate = sqs_client.receive_message(QueueUrl=queue_url)1844        assert result_send.get("MessageId") == result_receive.get("Messages")[0].get("MessageId")1845        assert result_send.get("MD5OfMessageBody") == result_receive.get("Messages")[0].get(1846            "MD5OfBody"1847        )1848        assert result_send_duplicate.get("MessageId") == result_receive_duplicate.get("Messages")[1849            01850        ].get("MessageId")1851        assert result_send_duplicate.get("MD5OfMessageBody") == result_receive_duplicate.get(1852            "Messages"1853        )[0].get("MD5OfBody")1854    @pytest.mark.aws_validated1855    def test_sse_queue_attributes(self, sqs_client, sqs_create_queue, snapshot):1856        # KMS server-side encryption (SSE)1857        queue_url = sqs_create_queue()1858        attributes = {1859            "KmsMasterKeyId": "testKeyId",1860            "KmsDataKeyReusePeriodSeconds": "6000",1861            "SqsManagedSseEnabled": "false",1862        }1863        sqs_client.set_queue_attributes(QueueUrl=queue_url, Attributes=attributes)1864        response = sqs_client.get_queue_attributes(1865            QueueUrl=queue_url,1866            AttributeNames=[1867                "KmsMasterKeyId",1868                "KmsDataKeyReusePeriodSeconds",1869                "SqsManagedSseEnabled",1870            ],1871        )1872        snapshot.match("sse_kms_attributes", response)1873        # SQS SSE1874        queue_url = sqs_create_queue()1875        attributes = {1876            "SqsManagedSseEnabled": "true",1877        }1878        sqs_client.set_queue_attributes(QueueUrl=queue_url, Attributes=attributes)1879        response = sqs_client.get_queue_attributes(1880            QueueUrl=queue_url,1881            AttributeNames=[1882                "KmsMasterKeyId",1883                "KmsDataKeyReusePeriodSeconds",1884                "SqsManagedSseEnabled",1885            ],1886        )1887        snapshot.match("sse_sqs_attributes", response)1888    @pytest.mark.xfail(reason="validation currently not implemented in localstack")1889    @pytest.mark.aws_validated1890    def test_sse_kms_and_sqs_are_mutually_exclusive(self, sqs_client, sqs_create_queue, snapshot):1891        queue_url = sqs_create_queue()1892        attributes = {1893            "KmsMasterKeyId": "testKeyId",1894            "SqsManagedSseEnabled": "true",1895        }1896        with pytest.raises(ClientError) as e:1897            sqs_client.set_queue_attributes(QueueUrl=queue_url, Attributes=attributes)1898        snapshot.match("error", e.value)1899    @pytest.mark.aws_validated1900    def test_receive_message_message_attribute_names_filters(1901        self, sqs_client, sqs_create_queue, snapshot1902    ):1903        """1904        Receive message allows a list of filters to be passed with MessageAttributeNames. See:1905        https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/sqs.html#SQS.Client.receive_message1906        """1907        queue_url = sqs_create_queue(Attributes={"VisibilityTimeout": "0"})1908        response = sqs_client.send_message(1909            QueueUrl=queue_url,1910            MessageBody="msg",1911            MessageAttributes={1912                "Help.Me": {"DataType": "String", "StringValue": "Me"},1913                "Hello": {"DataType": "String", "StringValue": "There"},1914                "General": {"DataType": "String", "StringValue": "Kenobi"},1915            },1916        )1917        assert snapshot.match("send_message_response", response)1918        def receive_message(message_attribute_names):1919            return sqs_client.receive_message(1920                QueueUrl=queue_url,1921                WaitTimeSeconds=5,1922                MessageAttributeNames=message_attribute_names,1923            )1924        # test empty filter1925        response = receive_message([])1926        # do the first check with the entire response1927        assert snapshot.match("empty_filter", response)1928        # test "All"1929        response = receive_message(["All"])1930        assert snapshot.match("all_name", response)1931        # test ".*"1932        response = receive_message([".*"])1933        assert snapshot.match("all_wildcard", response["Messages"][0])1934        # test only non-existent names1935        response = receive_message(["Foo", "Help"])1936        assert snapshot.match("only_non_existing_names", response["Messages"][0])1937        # test all existing1938        response = receive_message(["Hello", "General"])1939        assert snapshot.match("only_existing", response["Messages"][0])1940        # test existing and non-existing1941        response = receive_message(["Foo", "Hello"])1942        assert snapshot.match("existing_and_non_existing", response["Messages"][0])1943        # test prefix filters1944        response = receive_message(["Hel.*"])1945        assert snapshot.match("prefix_filter", response["Messages"][0])1946        # test illegal names1947        response = receive_message(["AWS."])1948        assert snapshot.match("illegal_name_1", response)1949        response = receive_message(["..foo"])1950        assert snapshot.match("illegal_name_2", response)1951    @pytest.mark.aws_validated1952    @pytest.mark.skip_snapshot_verify(paths=["$..Attributes.SenderId"])1953    def test_receive_message_attribute_names_filters(self, sqs_client, sqs_create_queue, snapshot):1954        # TODO -> senderId in LS == account ID, but on AWS it looks quite different: [A-Z]{21}:<email>1955        # account id is replaced with higher priority1956        queue_url = sqs_create_queue(Attributes={"VisibilityTimeout": "0"})1957        sqs_client.send_message(1958            QueueUrl=queue_url,1959            MessageBody="msg",1960            MessageAttributes={1961                "Foo": {"DataType": "String", "StringValue": "Bar"},1962            },1963        )1964        def receive_message(attribute_names, message_attribute_names=None):1965            return sqs_client.receive_message(1966                QueueUrl=queue_url,1967                WaitTimeSeconds=5,1968                AttributeNames=attribute_names,1969                MessageAttributeNames=message_attribute_names or [],1970            )1971        response = receive_message(["All"])1972        assert snapshot.match("all_attributes", response)1973        response = receive_message(["All"], ["All"])1974        assert snapshot.match("all_system_and_message_attributes", response)1975        response = receive_message(["SenderId"])1976        assert snapshot.match("single_attribute", response)1977        response = receive_message(["SenderId", "SequenceNumber"])1978        assert snapshot.match("multiple_attributes", response)1979    @pytest.mark.aws_validated1980    def test_change_visibility_on_deleted_message_raises_invalid_parameter_value(1981        self, sqs_client, sqs_queue1982    ):1983        # prepare the fixture1984        sqs_client.send_message(QueueUrl=sqs_queue, MessageBody="foo")1985        response = sqs_client.receive_message(QueueUrl=sqs_queue, WaitTimeSeconds=5)1986        handle = response["Messages"][0]["ReceiptHandle"]1987        # check that it works as expected1988        sqs_client.change_message_visibility(1989            QueueUrl=sqs_queue, ReceiptHandle=handle, VisibilityTimeout=421990        )1991        # delete the message, the handle becomes invalid1992        sqs_client.delete_message(QueueUrl=sqs_queue, ReceiptHandle=handle)1993        with pytest.raises(ClientError) as e:1994            sqs_client.change_message_visibility(1995                QueueUrl=sqs_queue, ReceiptHandle=handle, VisibilityTimeout=421996            )1997        err = e.value.response["Error"]1998        assert err["Code"] == "InvalidParameterValue"1999        assert (2000            err["Message"]2001            == f"Value {handle} for parameter ReceiptHandle is invalid. Reason: Message does not exist or is not "2002            f"available for visibility timeout change."2003        )2004    @pytest.mark.aws_validated2005    def test_delete_message_with_illegal_receipt_handle(self, sqs_client, sqs_queue):2006        with pytest.raises(ClientError) as e:2007            sqs_client.delete_message(QueueUrl=sqs_queue, ReceiptHandle="garbage")2008        err = e.value.response["Error"]2009        assert err["Code"] == "ReceiptHandleIsInvalid"2010        assert err["Message"] == 'The input receipt handle "garbage" is not a valid receipt handle.'2011    @pytest.mark.aws_validated2012    def test_delete_message_with_deleted_receipt_handle(self, sqs_client, sqs_queue):2013        sqs_client.send_message(QueueUrl=sqs_queue, MessageBody="foo")2014        response = sqs_client.receive_message(QueueUrl=sqs_queue, WaitTimeSeconds=5)2015        handle = response["Messages"][0]["ReceiptHandle"]2016        # does not raise errors even after successive calls2017        sqs_client.delete_message(QueueUrl=sqs_queue, ReceiptHandle=handle)2018        sqs_client.delete_message(QueueUrl=sqs_queue, ReceiptHandle=handle)2019        sqs_client.delete_message(QueueUrl=sqs_queue, ReceiptHandle=handle)2020def get_region():2021    return os.environ.get("AWS_DEFAULT_REGION") or TEST_REGION2022# TODO: test visibility timeout (with various ways to set them: queue attributes, receive parameter, update call)2023# TODO: test message attributes and message system attributes2024class TestSqsLambdaIntegration:2025    pass2026    # TODO: move tests here2027@pytest.fixture()2028def sqs_http_client(aws_http_client_factory):2029    yield aws_http_client_factory("sqs")2030class TestSqsQueryApi:2031    @pytest.mark.xfail(2032        reason="this behaviour is deprecated (see https://github.com/localstack/localstack/pull/5928)",2033    )2034    def test_call_fifo_queue_url(self, sqs_client, sqs_create_queue):2035        # TODO: remove once query API has been documented2036        queue_name = f"queue-{short_uid()}.fifo"2037        queue_url = sqs_create_queue(QueueName=queue_name, Attributes={"FifoQueue": "true"})2038        assert queue_url.endswith(".fifo")2039        response = requests.get(queue_url)2040        assert response.ok2041        assert queue_url in response.text2042    @pytest.mark.xfail(2043        reason="this behaviour is deprecated (see https://github.com/localstack/localstack/pull/5928)",2044    )2045    def test_request_via_url(self, sqs_create_queue):2046        # TODO: remove once query API has been documented2047        queue_url = sqs_create_queue()2048        response = requests.get(url=queue_url, params={"Action": "ListQueues"})2049        assert response.ok2050        assert queue_url in response.text2051    @pytest.mark.aws_validated2052    def test_get_queue_attributes_all(self, sqs_create_queue, sqs_http_client):2053        queue_url = sqs_create_queue()2054        response = sqs_http_client.get(2055            queue_url,2056            params={2057                "Action": "GetQueueAttributes",2058                "AttributeName.1": "All",2059            },2060        )2061        assert response.ok2062        assert "<GetQueueAttributesResponse" in response.text2063        assert "<Attribute><Name>QueueArn</Name><Value>arn:aws:sqs:" in response.text2064        assert "<Attribute><Name>VisibilityTimeout</Name><Value>30" in response.text2065        assert queue_url.split("/")[-1] in response.text2066    @pytest.mark.only_localstack2067    def test_get_queue_attributes_works_without_authparams(self, sqs_create_queue):2068        queue_url = sqs_create_queue()2069        response = requests.get(2070            queue_url,2071            params={2072                "Action": "GetQueueAttributes",2073                "AttributeName.1": "All",2074            },2075        )2076        assert response.ok2077        assert "<GetQueueAttributesResponse" in response.text2078        assert "<Attribute><Name>QueueArn</Name><Value>arn:aws:sqs:" in response.text2079        assert "<Attribute><Name>VisibilityTimeout</Name><Value>30" in response.text2080        assert queue_url.split("/")[-1] in response.text2081    @pytest.mark.aws_validated2082    def test_get_queue_attributes_with_query_args(self, sqs_create_queue, sqs_http_client):2083        queue_url = sqs_create_queue()2084        response = sqs_http_client.get(2085            queue_url,2086            params={2087                "Action": "GetQueueAttributes",2088                "AttributeName.1": "QueueArn",2089            },2090        )2091        assert response.ok2092        assert "<GetQueueAttributesResponse" in response.text2093        assert "<Attribute><Name>QueueArn</Name><Value>arn:aws:sqs:" in response.text2094        assert "<Attribute><Name>VisibilityTimeout</Name>" not in response.text2095        assert queue_url.split("/")[-1] in response.text2096    @pytest.mark.aws_validated2097    def test_invalid_action_raises_exception(self, sqs_create_queue, sqs_http_client):2098        queue_url = sqs_create_queue()2099        response = sqs_http_client.get(2100            queue_url,2101            params={2102                "Action": "FooBar",2103                "Version": "2012-11-05",2104            },2105        )2106        assert not response.ok2107        assert "<Code>InvalidAction</Code>" in response.text2108        assert "<Type>Sender</Type>" in response.text2109        assert (2110            "<Message>The action FooBar is not valid for this endpoint.</Message>" in response.text2111        )2112    @pytest.mark.aws_validated2113    def test_valid_action_with_missing_parameter_raises_exception(2114        self, sqs_create_queue, sqs_http_client2115    ):2116        queue_url = sqs_create_queue()2117        response = sqs_http_client.get(2118            queue_url,2119            params={2120                "Action": "SendMessage",2121            },2122        )2123        assert not response.ok2124        assert "<Code>MissingParameter</Code>" in response.text2125        assert "<Type>Sender</Type>" in response.text2126        assert (2127            "<Message>The request must contain the parameter MessageBody.</Message>"2128            in response.text2129        )2130    @pytest.mark.aws_validated2131    def test_get_queue_attributes_of_fifo_queue(self, sqs_create_queue, sqs_http_client):2132        queue_name = f"queue-{short_uid()}.fifo"2133        queue_url = sqs_create_queue(QueueName=queue_name, Attributes={"FifoQueue": "true"})2134        assert ".fifo" in queue_url2135        response = sqs_http_client.get(2136            queue_url,2137            params={2138                "Action": "GetQueueAttributes",2139                "AttributeName.1": "All",2140            },2141        )2142        assert response.ok2143        assert "<Name>FifoQueue</Name><Value>true</Value>" in response.text2144        assert queue_name in response.text2145    @pytest.mark.aws_validated2146    def test_get_queue_attributes_with_invalid_arg_returns_error(2147        self, sqs_create_queue, sqs_http_client2148    ):2149        queue_url = sqs_create_queue()2150        response = sqs_http_client.get(2151            queue_url,2152            params={2153                "Action": "GetQueueAttributes",2154                "AttributeName.1": "Foobar",2155            },2156        )2157        assert not response.ok2158        assert "<Type>Sender</Type>" in response.text2159        assert "<Code>InvalidAttributeName</Code>" in response.text2160        assert "<Message>Unknown Attribute Foobar.</Message>" in response.text2161    @pytest.mark.aws_validated2162    def test_get_delete_queue(2163        self, sqs_create_queue, sqs_client, sqs_http_client, sqs_queue_exists2164    ):2165        queue_url = sqs_create_queue()2166        response = sqs_http_client.get(2167            queue_url,2168            params={2169                "Action": "DeleteQueue",2170            },2171        )2172        assert response.ok2173        assert "<DeleteQueueResponse " in response.text2174        assert poll_condition(lambda: not sqs_queue_exists(queue_url), timeout=5)2175    @pytest.mark.aws_validated2176    def test_get_send_and_receive_messages(self, sqs_create_queue, sqs_http_client):2177        queue1_url = sqs_create_queue()2178        queue2_url = sqs_create_queue()2179        # items in queue 12180        response = sqs_http_client.get(2181            queue1_url,2182            params={2183                "Action": "SendMessage",2184                "MessageBody": "foobar",2185            },2186        )2187        assert response.ok2188        # no items in queue 22189        response = sqs_http_client.get(2190            queue2_url,2191            params={2192                "Action": "ReceiveMessage",2193            },2194        )2195        assert response.ok2196        assert "foobar" not in response.text2197        assert "<ReceiveMessageResult/>" in response.text.replace(2198            " />", "/>"2199        )  # expect response to be empty2200        # get items from queue 12201        response = sqs_http_client.get(2202            queue1_url,2203            params={2204                "Action": "ReceiveMessage",2205            },2206        )2207        assert response.ok2208        assert "<Body>foobar</Body>" in response.text2209        assert "<MD5OfBody>" in response.text2210    @pytest.mark.aws_validated2211    def test_get_on_deleted_queue_fails(2212        self, sqs_client, sqs_create_queue, sqs_http_client, sqs_queue_exists2213    ):2214        queue_url = sqs_create_queue()2215        sqs_client.delete_queue(QueueUrl=queue_url)2216        assert poll_condition(lambda: not sqs_queue_exists(queue_url), timeout=5)2217        response = sqs_http_client.get(2218            queue_url,2219            params={2220                "Action": "GetQueueAttributes",2221                "AttributeName.1": "QueueArn",2222            },2223        )2224        assert "<Code>AWS.SimpleQueueService.NonExistentQueue</Code>" in response.text2225        assert "<Message>The specified queue does not exist for this wsdl version" in response.text2226        assert response.status_code == 4002227    @pytest.mark.aws_validated2228    def test_get_without_query_returns_unknown_operation(self, sqs_create_queue, sqs_http_client):2229        queue_name = f"test-queue-{short_uid()}"2230        queue_url = sqs_create_queue(QueueName=queue_name)2231        assert queue_url.endswith(f"/{queue_name}")2232        response = sqs_http_client.get(queue_url)2233        assert "<UnknownOperationException" in response.text2234        assert response.status_code == 4042235    @pytest.mark.aws_validated2236    def test_get_create_queue_fails(self, sqs_create_queue, sqs_http_client):2237        queue1_url = sqs_create_queue()2238        queue2_name = f"test-queue-{short_uid()}"2239        response = sqs_http_client.get(2240            queue1_url,2241            params={2242                "Action": "CreateQueue",2243                "QueueName": queue2_name,2244            },2245        )2246        assert "<Code>InvalidAction</Code>" in response.text2247        assert "<Message>The action CreateQueue is not valid for this endpoint" in response.text2248        assert response.status_code == 4002249    @pytest.mark.aws_validated2250    def test_get_list_queues_fails(self, sqs_create_queue, sqs_http_client):2251        queue_url = sqs_create_queue()2252        response = sqs_http_client.get(2253            queue_url,2254            params={2255                "Action": "ListQueues",2256            },2257        )2258        assert "<Code>InvalidAction</Code>" in response.text2259        assert "<Message>The action ListQueues is not valid for this endpoint" in response.text2260        assert response.status_code == 4002261    @pytest.mark.aws_validated2262    def test_get_queue_url_works_for_same_queue(self, sqs_create_queue, sqs_http_client):2263        queue_url = sqs_create_queue()2264        response = sqs_http_client.get(2265            queue_url,2266            params={2267                "Action": "GetQueueUrl",2268                "QueueName": queue_url.split("/")[-1],2269            },2270        )2271        assert f"<QueueUrl>{queue_url}</QueueUrl>" in response.text2272        assert response.status_code == 2002273    @pytest.mark.aws_validated2274    def test_get_queue_url_work_for_different_queue(self, sqs_create_queue, sqs_http_client):2275        # for some reason this is allowed ð¤·2276        queue1_url = sqs_create_queue()2277        queue2_url = sqs_create_queue()2278        response = sqs_http_client.get(2279            queue1_url,2280            params={2281                "Action": "GetQueueUrl",2282                "QueueName": queue2_url.split("/")[-1],2283            },2284        )2285        assert f"<QueueUrl>{queue2_url}</QueueUrl>" in response.text2286        assert queue1_url not in response.text2287        assert response.status_code == 2002288    @pytest.mark.aws_validated2289    @pytest.mark.parametrize("strategy", ["domain", "path", "off"])2290    def test_endpoint_strategy_with_multi_region(2291        self,2292        strategy,2293        sqs_http_client,2294        create_boto_client,2295        aws_http_client_factory,2296        monkeypatch,2297        cleanups,2298    ):2299        monkeypatch.setattr(config, "SQS_ENDPOINT_STRATEGY", strategy)2300        queue_name = f"test-queue-{short_uid()}"2301        sqs_region1 = create_boto_client("sqs", "us-east-1")2302        sqs_region2 = create_boto_client("sqs", "eu-west-1")2303        queue_region1 = sqs_region1.create_queue(QueueName=queue_name)["QueueUrl"]2304        cleanups.append(lambda: sqs_region1.delete_queue(QueueUrl=queue_region1))2305        queue_region2 = sqs_region2.create_queue(QueueName=queue_name)["QueueUrl"]2306        cleanups.append(lambda: sqs_region2.delete_queue(QueueUrl=queue_region2))2307        if strategy == "off":2308            assert queue_region1 == queue_region22309        else:2310            assert queue_region1 != queue_region22311            assert "eu-west-1" in queue_region22312            # us-east-1 is the default region, so it's not necessarily part of the queue URL2313        client_region1 = aws_http_client_factory("sqs", "us-east-1")2314        client_region2 = aws_http_client_factory("sqs", "eu-west-1")2315        response = client_region1.get(2316            queue_region1, params={"Action": "SendMessage", "MessageBody": "foobar"}2317        )2318        assert response.ok2319        # shouldn't return anything2320        response = client_region2.get(2321            queue_region2, params={"Action": "ReceiveMessage", "VisibilityTimeout": "0"}2322        )2323        assert response.ok2324        assert "foobar" not in response.text2325        # should return the message2326        response = client_region1.get(2327            queue_region1, params={"Action": "ReceiveMessage", "VisibilityTimeout": "0"}2328        )...

test_validate.py

Source:test_validate.py

...5@pytest.mark.xfail(reason="there is no generalized way of server-side request validation yet")6class TestMissingParameter:7    @pytest.mark.aws_validated8    def test_opensearch(self, aws_http_client_factory):9        client = aws_http_client_factory("es", signer_factory=SigV4Auth)10        response = client.post(11            "/2021-01-01/opensearch/domain",12            data='{"foobar": "bazed"}',13        )14        assert (15            response.text16            == '{"message":"1 validation error detected: Value null at \'domainName\' failed to satisfy constraint: '17            'Member must not be null"}'18        )19    @pytest.mark.aws_validated20    def test_sns(self, aws_http_client_factory):21        client = aws_http_client_factory("sns", region="us-east-1")22        response = client.post(23            "/?Action=CreatePlatformApplication&Name=Foobar&Platform=Bar",24        )25        assert "<Code>ValidationError</Code>" in response.text26        assert (27            "<Message>1 validation error detected: Value null at 'attributes' failed to satisfy constraint: Member "28            "must not be null</Message>" in response.text29        )30    @pytest.mark.aws_validated31    def test_elasticache(self, aws_http_client_factory):32        client = aws_http_client_factory("elasticache")33        response = client.post(34            "/",35            params={36                "Action": "CreateCacheCluster",37            },38        )39        assert "<Code>InvalidParameterValue</Code>" in response.text40        assert (41            "<Message>The parameter CacheClusterIdentifier must be provided and must not be blank.</Message>"42            in response.text43        )44    @pytest.mark.aws_validated45    def test_sqs_create_queue(self, aws_http_client_factory):46        client = aws_http_client_factory("sqs")47        response = client.post(48            "/",49            params={50                "Action": "CreateQueue",51                "FooBar": "baz",52            },53        )54        assert "<Code>InvalidParameterValue</Code>" in response.text55        assert (56            "<Message>Value for parameter QueueName is invalid. Reason: Must specify a queue name.</Message>"57            in response.text58        )59    @pytest.mark.aws_validated60    def test_sqs_send_message(self, aws_http_client_factory, sqs_queue):61        client = aws_http_client_factory("sqs")62        response = client.post(63            "/",64            params={"Action": "SetQueueAttributes", "Version": "2012-11-05", "QueueUrl": sqs_queue},65        )66        assert "<Code>MissingParameter</Code>" in response.text67        assert (68            "<Message>The request must contain the parameter Attribute.Name.</Message>"69            in response.text...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.