Join the 1-day Testing & QA Summit featuring 15+ Expert Speakers.Register for FREE! Join TestMu Conference

How to use test_dead_letter_queue method in localstack

Best Python code snippet using localstack_python

test_claims.py

Source:test_claims.py

...84        # This will implicitly verify that the claim is deleted.85        message_uri = urllib.parse.urlparse(claim_uri).path86        self.client.delete_messages(message_uri)87    @decorators.idempotent_id('c1975970-66e7-11e7-a771-fa163e40e1ff')88    def test_dead_letter_queue(self):89        # Post Messages90        QueueName = "QueueWithDLQ"91        DLQ_name = "DLQ"92        meta = {'ttl': 60, 'grace': 60}93        # Set dead letter queeu metadata94        op1 = {"op": "add",95               "path": "/metadata/_max_claim_count", "value": 2}96        op2 = {"op": "add",97               "path": "/metadata/_dead_letter_queue", "value": DLQ_name}98        op3 = {"op": "add",99               "path": "/metadata/_dead_letter_queue_messages_ttl",100               "value": 7799}101        metadata = [op1, op2, op3]102        self.client.create_queue(QueueName)...

test_sensor.py

Source:test_sensor.py

1import time2import uuid3from pathlib import Path4import pytest5from aiohttp import ClientSession, ClientConnectorError6import run_sensor7class MockResponse:8    def __init__(self, json, status):9        self._json = json10        self.status = status11    async def json(self):12        return self.json13    async def __aexit__(self, exc_type, exc, tb):14        pass15    async def __aenter__(self):16        return self17@pytest.fixture(autouse=True)18def override_queue_file_name(file_name_for_test):19    run_sensor.QUEUE_FILE_NAME = file_name_for_test20@pytest.fixture21def file_name_for_test():22    return "test_dead_letter_queue.pkl"23@pytest.fixture(autouse=True)24def cleanup_files(file_name_for_test):25    """Clean up temp files after each test."""26    yield27    run_sensor.delete_file(file_name_for_test)28@pytest.fixture29def mocked_endpoint(mocker):30    data = {}31    resp = MockResponse(data, 200)32    return mocker.patch("aiohttp.ClientSession.post", return_value=resp)33@pytest.fixture34def mocked_endpoint_no_internet(mocker):35    return mocker.patch(36        "aiohttp.ClientSession.post", side_effect=ClientConnectorError(None, OSError())37    )38@pytest.fixture39def queue_data():40    return [41        {42            "id": "2b1f5982-0382-49e2-9817-a54d77a47265",43            "event": {"type": "nominal", "readings": [50, 96, 25]},44            "timestamp": 1634720018,45        },46        {47            "id": "2b1f5982-0382-49e2-9817-a54d77a47265",48            "event": {"type": "nominal", "readings": [78, 83, 100]},49            "timestamp": 1634720023,50        },51    ]52@pytest.fixture53def populated_queue(queue_data):54    run_sensor.DEAD_LETTER_QUEUE.extend(queue_data)55@pytest.fixture56def blank_queue():57    run_sensor.DEAD_LETTER_QUEUE = []58@pytest.fixture59async def session():60    async with ClientSession(raise_for_status=True) as session:61        yield session62@pytest.fixture63def state():64    return {65        "id": uuid.uuid4(),66        "event": {67            "type": "nominal",68            "readings": [69                10,70                20,71                30,72            ],73        },74        "timestamp": int(time.time()),75    }76@pytest.fixture77def saved_state_file(queue_data):78    run_sensor.save_object_to_file(79        path=run_sensor.QUEUE_FILE_NAME, python_object=queue_data80    )81@pytest.mark.asyncio82async def test_send_state_with_queue(83    mocked_endpoint, session, state, populated_queue, queue_data84):85    """Test all items in the queue are sent to the API, as well as the current state."""86    # given ...87    # ... a client session88    # ... a populated queue89    # ... a mocked API endpoint90    # when ... we send the state to the API91    await run_sensor.send_state(session=session, state=state)92    # then ...93    # ... three separate calls are made to the API (two from the queue, one from current state)94    mock_calls = mocked_endpoint.call_args_list95    assert len(mock_calls) == 396    # ... the data sent to the API is as expected97    assert mock_calls[0].args[0] == run_sensor.REQUEST_ENDPOINT98    assert mock_calls[0][1]["json"] == queue_data[0]99    assert mock_calls[1].args[0] == run_sensor.REQUEST_ENDPOINT100    assert mock_calls[1][1]["json"] == queue_data[1]101    assert mock_calls[2].args[0] == run_sensor.REQUEST_ENDPOINT102    assert mock_calls[2][1]["json"] == state103    # ... the queue is cleared104    assert len(run_sensor.DEAD_LETTER_QUEUE) == 0105@pytest.mark.asyncio106async def test_send_state_no_internet(107    mocked_endpoint_no_internet,108    session,109    state,110    populated_queue,111    queue_data,112):113    """Simulate a scenario where no internet is available."""114    # given ...115    # ... a client session116    # ... a populated queue117    # ... a mocked API endpoint that throws a ClientConnectorError exception118    assert len(run_sensor.DEAD_LETTER_QUEUE) == 2119    # when ... the state is attempted to be sent over the internet120    await run_sensor.send_state(session=session, state=state)121    # then ...122    # ... three separate calls are made to the API (two from the queue, one from current state)123    mock_calls = mocked_endpoint_no_internet.call_args_list124    assert len(mock_calls) == 3125    # ... the data sent to the API is as expected126    assert mock_calls[0].args[0] == run_sensor.REQUEST_ENDPOINT127    assert mock_calls[0][1]["json"] == queue_data[0]128    assert mock_calls[1].args[0] == run_sensor.REQUEST_ENDPOINT129    assert mock_calls[1][1]["json"] == queue_data[1]130    assert mock_calls[2].args[0] == run_sensor.REQUEST_ENDPOINT131    assert mock_calls[2][1]["json"] == state132    # ... the queue size has increased by one133    assert len(run_sensor.DEAD_LETTER_QUEUE) == 3134    # ... the new item in the queue is the latest state135    assert run_sensor.DEAD_LETTER_QUEUE[2] == state136    # ... the save file was created137    saved_file = Path(run_sensor.QUEUE_FILE_NAME)138    assert saved_file.is_file()139@pytest.mark.asyncio140async def test_load_state(blank_queue, queue_data, saved_state_file):141    """Simulate a scenario where the state needs to be loaded from a file."""142    # given ...143    # ... a saved state file with two states saved144    # ... an empty dead letter queue in memory145    assert len(run_sensor.DEAD_LETTER_QUEUE) == 0146    # when ... we load the state from file147    run_sensor.load_state(run_sensor.QUEUE_FILE_NAME)148    # then ...149    # ...the in memory queue is populated with the expected states150    assert len(run_sensor.DEAD_LETTER_QUEUE) == 2...

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.