Unlock 30% off on Manual Testing Annual Plans this Holiday Season.

Copied to Clipboard!

How to use sqs_client method in localstack

Best Python code snippet using localstack_python

test_extended_messaging_client.py

Source:test_extended_messaging_client.py

1"""2The MIT License (MIT)3Copyright (c) 2021 Archetype Digital Inc.4Permission is hereby granted, free of charge, to any person obtaining a copy5of this software and associated documentation files (the "Software"), to deal6in the Software without restriction, including without limitation the rights7to use, copy, modify, merge, publish, distribute, sublicense, and/or sell8copies of the Software, and to permit persons to whom the Software is9furnished to do so, subject to the following conditions:10The above copyright notice and this permission notice shall be included in all11copies or substantial portions of the Software.12THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR13IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,14FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE15AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER16LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,17OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE18SOFTWARE.19"""20import hashlib21import json22import botocore23import pytest24from aws_sqs_ext_client.extended_messaging import SQSExtendedMessage25@pytest.fixture26def send_message_extended_client(sqs_extended_message, sqs_client):27    attributes = {'send_message': sqs_client.send_message}28    add_custom_method = sqs_extended_message.add_send_message_extended(None)29    add_custom_method(class_attributes=attributes)30    return attributes['send_message_extended']31@pytest.fixture32def receive_message_extended_client(sqs_extended_message, sqs_client):33    attributes = {'receive_message': sqs_client.receive_message}34    add_custom_method = sqs_extended_message.add_receive_message_extended(35        'creating-client-class.sqs')36    add_custom_method(class_attributes=attributes)37    return attributes['receive_message_extended']38@pytest.fixture39def delete_message_extended_client(sqs_extended_message, sqs_client):40    attributes = {'delete_message': sqs_client.delete_message}41    add_custom_method = sqs_extended_message.add_delete_message_extended(42        'creating-client-class.sqs')43    add_custom_method(class_attributes=attributes)44    return attributes['delete_message_extended']45@pytest.fixture46def send_message_batch_extended_client(sqs_extended_message, sqs_client):47    attributes = {'send_message_batch': sqs_client.send_message_batch}48    add_custom_method = sqs_extended_message.add_send_message_batch_extended(49        'creating-client-class.sqs')50    add_custom_method(class_attributes=attributes)51    return attributes['send_message_batch_extended']52@pytest.fixture53def delete_message_batch_extended_client(sqs_extended_message, sqs_client):54    attributes = {'delete_message_batch': sqs_client.delete_message_batch}55    add_custom_method = sqs_extended_message.add_delete_message_batch_extended(56        'creating-client-class.sqs')57    add_custom_method(class_attributes=attributes)58    return attributes['delete_message_batch_extended']59def test_add_custom_method(sqs_extended_message):60    def f(**kwargs):61        return True62    tests = [{63        'given': {64            'attributes': {'send_message': f},65            'callee': sqs_extended_message.add_send_message_extended,66            'args': ('creating-client-class.sqs'),67        },68        'wants': {69            'in_attributes': lambda x: 'send_message_extended' in x,70            'registration': (71                lambda x: x['send_message_extended'].__name__ ==72                'send_message_extended')73        }74    }, {75        'given': {76            'attributes': {'send_message': f},77            'callee': sqs_extended_message.add_send_message_extended,78            'args': ('creating-resource-class.sqs'),79        },80        'wants': {81            'in_attributes': lambda x: 'send_message_extended' in x,82            'registration': (83                lambda x: x['send_message_extended'].__name__ ==84                'send_message_extended')85        }86    }, {87        'given': {88            'attributes': {'receive_message': f},89            'callee': sqs_extended_message.add_receive_message_extended,90            'args': ('creating-client-class.sqs'),91        },92        'wants': {93            'in_attributes': lambda x: 'receive_message_extended' in x,94            'registration': (95                lambda x: x['receive_message_extended'].__name__ ==96                'receive_message_extended')97        }98    }, {99        'given': {100            'attributes': {'receive_messages': f},101            'callee': sqs_extended_message.add_receive_message_extended,102            'args': ('creating-resource-class.sqs.Queue'),103        },104        'wants': {105            'in_attributes': lambda x: 'receive_messages_extended' in x,106            'registration': (107                lambda x: x['receive_messages_extended'].__name__ ==108                'receive_message_extended')109        }110    }, {111        'given': {112            'attributes': {'delete_message': f},113            'callee': sqs_extended_message.add_delete_message_extended,114            'args': ('creating-client-class.sqs'),115        },116        'wants': {117            'in_attributes': lambda x: 'delete_message_extended' in x,118            'registration': (119                lambda x: x['delete_message_extended'].__name__ ==120                'delete_message_extended')121        }122    }, {123        'given': {124            'attributes': {'delete': f},125            'callee': sqs_extended_message.add_delete_message_extended,126            'args': ('creating-resource-class.sqs.Message'),127        },128        'wants': {129            'in_attributes': lambda x: 'delete_extended' in x,130            'registration': (131                lambda x: x['delete_extended'].__name__ ==132                'delete_message_extended')133        }134    }, {135        'given': {136            'attributes': {'send_message_batch': f},137            'callee': sqs_extended_message.add_send_message_batch_extended,138            'args': ('creating-client-class.sqs'),139        },140        'wants': {141            'in_attributes': lambda x: 'send_message_batch_extended' in x,142            'registration': (143                lambda x: x['send_message_batch_extended'].__name__ ==144                'send_message_batch_extended')145        }146    }, {147        'given': {148            'attributes': {'send_messages': f},149            'callee': sqs_extended_message.add_send_message_batch_extended,150            'args': ('creating-resource-class.sqs.Queue'),151        },152        'wants': {153            'in_attributes': lambda x: 'send_messages_extended' in x,154            'registration': (155                lambda x: x['send_messages_extended'].__name__ ==156                'send_message_batch_extended')157        }158    }, {159        'given': {160            'attributes': {'delete_message_batch': f},161            'callee': sqs_extended_message.add_delete_message_batch_extended,162            'args': ('creating-client-class.sqs'),163        },164        'wants': {165            'in_attributes': lambda x: 'delete_message_batch_extended' in x,166            'registration': (167                lambda x: x['delete_message_batch_extended'].__name__ ==168                'delete_message_batch_extended')169        }170    }, {171        'given': {172            'attributes': {'delete_messages': f},173            'callee': sqs_extended_message.add_delete_message_batch_extended,174            'args': ('creating-resource-class.sqs.Queue'),175        },176        'wants': {177            'in_attributes': lambda x: 'delete_messages_extended' in x,178            'registration': (179                lambda x: x['delete_messages_extended'].__name__ ==180                'delete_message_batch_extended')181        }182    }]183    for t in tests:184        attributes = t['given']['attributes']185        add_custom_method = t['given']['callee'](t['given']['args'])186        add_custom_method(class_attributes=attributes)187        assert t['wants']['in_attributes'](attributes)188        assert t['wants']['registration'](attributes)189##190# send_message_extended error handling test191##192def test_send_message_extended_wo_params(send_message_extended_client):193    with pytest.raises(ValueError) as excinfo:194        send_message_extended_client()195    assert "message body is required" in str(excinfo.value)196def test_send_message_extended_w_invalid_attribute(197        send_message_extended_client):198    with pytest.raises(ValueError) as excinfo:199        send_message_extended_client(200            MessageAttributes={'ExtendedPayloadSize': 'test'}201        )202    assert "ExtendedPayloadSize is reserved name" in str(excinfo.value)203def test_send_message_extended_wo_required_param_for_original(204        send_message_extended_client):205    with pytest.raises(206            botocore.exceptions.ParamValidationError) as excinfo:207        send_message_extended_client(MessageBody='{"message": "small text"}')208    assert 'Missing required parameter in input: "QueueUrl"' in str(209        excinfo.value)210##211# receive_message_extended error handling test212##213def test_receive_message_extended_w_invalid_attr(214        receive_message_extended_client):215    with pytest.raises(ValueError) as excinfo:216        receive_message_extended_client(AttributeNames={})217    assert "AttributeNames or MessageAttributeNames must be list" in str(218        excinfo.value)219def test_receive_message_extended_w_invalid_mattr(220        receive_message_extended_client):221    with pytest.raises(ValueError) as excinfo:222        receive_message_extended_client(MessageAttributeNames={})223    assert "AttributeNames or MessageAttributeNames must be list" in str(224        excinfo.value)225def test_receive_message_extended_wo_required_param_for_original(226        receive_message_extended_client):227    with pytest.raises(228            botocore.exceptions.ParamValidationError) as excinfo:229        receive_message_extended_client()230    assert 'Missing required parameter in input: "QueueUrl"' in str(231        excinfo.value)232def test_receive_message_extended_wo_response(233        receive_message_extended_client, sqs_client_queue):234    res = receive_message_extended_client(235        QueueUrl=sqs_client_queue['QueueUrl'],236        MessageAttributeNames=[])237    assert 'ResponseMetadata' in res238    assert 'Messages' not in res239##240# delete_message_extended error handling test241##242def test_delete_message_extended_wo_params(243        delete_message_extended_client):244    with pytest.raises(ValueError) as excinfo:245        delete_message_extended_client()246    assert "invalid call without ReceiptHandle" in str(excinfo.value)247##248# send_message_batch_extended error handling test249##250def test_send_message_batch_extended_wo_params(251        send_message_batch_extended_client):252    with pytest.raises(ValueError) as excinfo:253        send_message_batch_extended_client()254    assert "Entries (list) must be given" in str(excinfo.value)255    with pytest.raises(ValueError) as excinfo:256        send_message_batch_extended_client(Entries={'Messages': {}})257    assert "Entries (list) must be given" in str(excinfo.value)258    with pytest.raises(ValueError) as excinfo:259        send_message_batch_extended_client(Entries=[{}])260    assert (261        "message body is required, found in 0" in str(excinfo.value))262    with pytest.raises(ValueError) as excinfo:263        send_message_batch_extended_client(Entries=[264            {'MessageAttributes': {265                'ExtendedPayloadSize': {'key': 'value'}266            }}267        ])268    assert (269        "ExtendedPayloadSize is reserved name, found in 0"270        in str(excinfo.value))271##272# delete_message_batch_extended error handling test273##274def test_delete_message_batch_extended_wo_params(275        delete_message_batch_extended_client):276    with pytest.raises(ValueError) as excinfo:277        delete_message_batch_extended_client()278    assert "Entries (list) must be given" in str(excinfo.value)279    with pytest.raises(ValueError) as excinfo:280        delete_message_batch_extended_client(Entries={'Messages': {}})281    assert "Entries (list) must be given" in str(excinfo.value)282    with pytest.raises(ValueError) as excinfo:283        delete_message_batch_extended_client(Entries=[{}])284    assert (285        "missing ReceiptHandle, found 0" in str(excinfo.value))286# ##287# # simple send/receive/delete test288# ##289def test_extended_messaging_w_small_text(290        send_message_extended_client, receive_message_extended_client,291        delete_message_extended_client,292        sqs_client_queue, sqs_client):293    body = '{"message": "small text"}'294    attr = {295        'string_attr': {296            'StringValue': 'string_something',297            'DataType': 'String'298        },299        'binary_attr': {300            'BinaryValue': b'bytes_something',301            'DataType': 'Binary'302        },303    }304    # send305    res = send_message_extended_client(306        QueueUrl=sqs_client_queue['QueueUrl'], MessageBody=body,307        MessageAttributes=attr)308    assert 'MessageId' in res309    # receive310    res = sqs_client.receive_message(311        QueueUrl=sqs_client_queue['QueueUrl'], MessageAttributeNames=['All'],312        VisibilityTimeout=0, WaitTimeSeconds=0)313    res_extended = receive_message_extended_client(314        QueueUrl=sqs_client_queue['QueueUrl'], MessageAttributeNames=['All'])315    assert res.keys() == res_extended.keys()316    assert res_extended['Messages'][0]['MessageId'] == (317        res['Messages'][0]['MessageId'])318    assert res_extended['Messages'][0]['Body'] == body319    assert res_extended['Messages'][0]['MD5OfBody'] == (320        res['Messages'][0]['MD5OfBody'])321    assert res_extended['Messages'][0]['MD5OfBody'] == hashlib.md5(322        body.encode()).hexdigest()323    assert res_extended['Messages'][0]['MD5OfMessageAttributes'] == (324        res['Messages'][0]['MD5OfMessageAttributes'])325    # delete326    delete_message_extended_client(327        QueueUrl=sqs_client_queue['QueueUrl'],328        ReceiptHandle=res_extended['Messages'][0]['ReceiptHandle'])329    res = sqs_client.receive_message(330        QueueUrl=sqs_client_queue['QueueUrl'], WaitTimeSeconds=0)331    assert 'Messages' not in res332# ##333# # large send/receive/delete test334# ##335def test_extended_messaging_w_large_text(336        s3_bucket, sqs_client_queue, sqs_client, s3_client, big_message,337        bucket_name,338        send_message_extended_client, receive_message_extended_client,339        delete_message_extended_client):340    body = big_message341    # send342    res = send_message_extended_client(343        QueueUrl=sqs_client_queue['QueueUrl'],344        MessageBody=body)345    assert 'MessageId' in res346    # receive temporal message347    res = sqs_client.receive_message(348        QueueUrl=sqs_client_queue['QueueUrl'], MessageAttributeNames=['All'],349        VisibilityTimeout=0, WaitTimeSeconds=0)350    assert 'Messages' in res351    assert len(res['Messages']) == 1352    assert (353        's3BucketName' in res['Messages'][0]['Body'] and354        's3Key' in res['Messages'][0]['Body'])355    assert json.loads(356        res['Messages'][0]['Body'])['s3BucketName'] == bucket_name357    assert 'MD5OfBody' in res['Messages'][0]358    assert 'MD5OfMessageAttributes' in res['Messages'][0]359    # receive actual message360    res = receive_message_extended_client(361        QueueUrl=sqs_client_queue['QueueUrl'], MessageAttributeNames=['All'])362    assert 'Messages' in res363    assert len(res['Messages']) == 1364    assert res['Messages'][0]['Body'] == body365    assert res['Messages'][0]['MD5OfBody'] == hashlib.md5(366        body.encode()).hexdigest()367    assert 'MD5OfMessageAttributes' not in res['Messages'][0]368    # delete369    delete_message_extended_client(370        QueueUrl=sqs_client_queue['QueueUrl'],371        ReceiptHandle=res['Messages'][0]['ReceiptHandle'])372    res = sqs_client.receive_message(373        QueueUrl=sqs_client_queue['QueueUrl'], WaitTimeSeconds=1)374    assert 'Messages' not in res375    res = s3_client.list_objects_v2(Bucket=bucket_name)376    assert res['KeyCount'] == 0377def test_always_extended_messaging(378        s3_bucket, session, bucket_name, sqs_client, sqs_client_queue,379        delete_message_extended_client):380    sqs = SQSExtendedMessage(session, bucket_name, always_through_s3=True)381    attributes = {'send_message': sqs_client.send_message}382    add_custom_method = sqs.add_send_message_extended(None)383    add_custom_method(class_attributes=attributes)384    send_method = attributes['send_message_extended']385    body = '{"message": "small text"}'386    # send387    res = send_method(QueueUrl=sqs_client_queue['QueueUrl'], MessageBody=body)388    assert 'MessageId' in res389    # receive390    res = sqs_client.receive_message(391        QueueUrl=sqs_client_queue['QueueUrl'], MessageAttributeNames=['All'])392    assert 'Messages' in res393    assert len(res['Messages']) == 1394    assert (395        's3BucketName' in res['Messages'][0]['Body'] and396        's3Key' in res['Messages'][0]['Body'])397    assert json.loads(398        res['Messages'][0]['Body'])['s3BucketName'] == bucket_name399    # delete400    delete_message_extended_client(401        QueueUrl=sqs_client_queue['QueueUrl'],402        ReceiptHandle=res['Messages'][0]['ReceiptHandle'])403def test_extended_messaging_w_lower_threshold(404        s3_bucket, session, bucket_name, sqs_client, sqs_client_queue,405        delete_message_extended_client):406    sqs = SQSExtendedMessage(session, bucket_name, message_size_threshold=10)407    attributes = {'send_message': sqs_client.send_message}408    add_custom_method = sqs.add_send_message_extended(None)409    add_custom_method(class_attributes=attributes)410    send_method = attributes['send_message_extended']411    body = '{"message": "small text"}'412    # send413    res = send_method(QueueUrl=sqs_client_queue['QueueUrl'], MessageBody=body)414    assert 'MessageId' in res415    # receive416    res = sqs_client.receive_message(417        QueueUrl=sqs_client_queue['QueueUrl'], MessageAttributeNames=['All'])418    assert 'Messages' in res419    assert len(res['Messages']) == 1420    assert (421        's3BucketName' in res['Messages'][0]['Body'] and422        's3Key' in res['Messages'][0]['Body'])423    assert json.loads(424        res['Messages'][0]['Body'])['s3BucketName'] == bucket_name425    # delete426    delete_message_extended_client(427        QueueUrl=sqs_client_queue['QueueUrl'],428        ReceiptHandle=res['Messages'][0]['ReceiptHandle'])429# ##430# # large multiple send/receive/delete test431# ##432def test_multiple_extended_messaging_w_large_text(433        s3_bucket, sqs_client_queue, sqs_client, s3_client, big_message,434        bucket_name,435        send_message_batch_extended_client, receive_message_extended_client,436        delete_message_batch_extended_client):437    body1 = big_message438    body2 = json.loads(big_message)439    body2['id'] = 'TEST_BIG_MESSAGE2'440    body2 = json.dumps(body2)441    # send442    res = send_message_batch_extended_client(443        QueueUrl=sqs_client_queue['QueueUrl'],444        Entries=[445            {'Id': '1', 'MessageBody': body1},446            {'Id': '2', 'MessageBody': body2}])447    assert 'Successful' in res448    assert len(res['Successful']) == 2449    assert list(map(lambda x: x['Id'], res['Successful'])) == ['1', '2']450    res = receive_message_extended_client(451        QueueUrl=sqs_client_queue['QueueUrl'], MessageAttributeNames=['All'],452        MaxNumberOfMessages=10)453    rhs = list(map(lambda x: x['ReceiptHandle'], res['Messages']))454    # delete455    delete_message_batch_extended_client(456        QueueUrl=sqs_client_queue['QueueUrl'],457        Entries=[458            {'Id': '1', 'ReceiptHandle': rhs[0]},459            {'Id': '2', 'ReceiptHandle': rhs[1]},460        ])461    res = sqs_client.receive_message(462        QueueUrl=sqs_client_queue['QueueUrl'], WaitTimeSeconds=1)463    assert 'Messages' not in res464    res = s3_client.list_objects_v2(Bucket=bucket_name)...

test_ses_forwarder.py

Source:test_ses_forwarder.py

1import json2import uuid3from pprint import pprint4import boto35import pytest6import ses_forwarder.main as handler7def test_delete_sqs(sqs_client, sqs_queue):8    """9    Ensure that you can delete a message from the sqs queue10    """11    # add message to the queue12    sqs_client.send_message(QueueUrl=sqs_queue, MessageBody="example message")13    # retrieve the message from the queue14    queue_message = sqs_client.receive_message(15        QueueUrl=sqs_queue, MaxNumberOfMessages=116    )17    receipt_handle = queue_message["Messages"][0]["ReceiptHandle"]18    # get the queue_arn19    queue_attributes = sqs_client.get_queue_attributes(20        QueueUrl=sqs_queue, AttributeNames=["QueueArn"]21    )22    queue_arn = queue_attributes["Attributes"]["QueueArn"]23    response = handler.delete_sqs(queue_arn, receipt_handle)24    assert response is None25def test_handler(26    monkeypatch,27    s3_client,28    bucket,29    sqs_client,30    sqs_queue,31):32    """33    Ensure that the handler can take a received message and send an email34    """35    # setup environment variables36    monkeypatch.setenv("LAMBDA_TIMEOUT", "60")37    monkeypatch.setenv("MAIL_SENDER", "from@pieceofprivacy.com")38    # put example email into the s3 bucket39    key = f"pytest-{uuid.uuid4().hex}"40    with open("handlers/tests/events/test_email.txt", "rb") as f:41        s3_client.put_object(Body=f, Bucket=bucket, Key=key)42    # add message to the queue43    sqs_client.send_message(QueueUrl=sqs_queue, MessageBody="example message")44    # retrieve the message from the queue45    queue_message = sqs_client.receive_message(QueueUrl=sqs_queue)46    receipt_handle = queue_message["Messages"][0]["ReceiptHandle"]47    message_id = queue_message["Messages"][0]["MessageId"]48    # get the queue_arn49    queue_attributes = sqs_client.get_queue_attributes(50        QueueUrl=sqs_queue, AttributeNames=["QueueArn"]51    )52    queue_arn = queue_attributes["Attributes"]["QueueArn"]53    # construct message type for handler54    message = {"receipt": {"action": {"bucketName": bucket, "objectKey": key}}}55    body = {"Message": json.dumps(message)}56    event = {57        "Records": [58            {59                "messageId": message_id,60                "receiptHandle": receipt_handle,61                "eventSourceARN": queue_arn,62                "body": json.dumps(body),63            }64        ]65    }66    response = handler.handler(event, None)...

sqs.py

Source:sqs.py

1from pprint import pprint2import boto33from config import ENDPOINT_URL4def send_message(msg):5    pprint(msg)6    sqs_client = boto3.client('sqs', aws_access_key_id='secret_id', aws_secret_access_key='secret_key', endpoint_url=ENDPOINT_URL)7    response = sqs_client.send_message(8        QueueUrl=get_queue_url(sqs_client),9        MessageBody=(msg)10    )11    pprint(response)12    return response13def receive_messages():14    sqs_client = boto3.client('sqs', aws_access_key_id='secret_id', aws_secret_access_key='secret_key', endpoint_url=ENDPOINT_URL)15    messages = sqs_client.receive_message(16        QueueUrl=get_queue_url(sqs_client),17        MaxNumberOfMessages=10,18        WaitTimeSeconds=1019    )20    pprint(messages)21    return messages22def delete_message(receipt_handle):23    sqs_client = boto3.client('sqs', aws_access_key_id='secret_id', aws_secret_access_key='secret_key', endpoint_url=ENDPOINT_URL)24    response = sqs_client.delete_message(25        QueueUrl=get_queue_url(sqs_client),26        ReceiptHandle=receipt_handle27    )28    pprint(response)29def create_queue():30    sqs_client = boto3.client('sqs', aws_access_key_id='secret_id', aws_secret_access_key='secret_key', endpoint_url=ENDPOINT_URL)31    response = sqs_client.create_queue(32        QueueName='cars',33        Attributes={34            'DelaySeconds': '0',35            'VisibilityTimeout': '60',36        }37    )38    pprint(response)39def delete_queue():40    sqs_client = boto3.client('sqs', aws_access_key_id='secret_id', aws_secret_access_key='secret_key', endpoint_url=ENDPOINT_URL)41    response = sqs_client.delete_queue(42        QueueUrl=get_queue_url(sqs_client)43    )44def get_queue_url(sqs_client):45    response = sqs_client.get_queue_url(46        QueueName='cars'47    )48    return response['QueueUrl']49def purge_queue():50    sqs_client = boto3.client('sqs', aws_access_key_id='secret_id', aws_secret_access_key='secret_key', endpoint_url=ENDPOINT_URL)51    sqs_client.purge_queue(52        QueueUrl=get_queue_url(sqs_client)...

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.