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...
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.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!