How to use response_operation method in localstack

Best Python code snippet using localstack_python

lambda_function.py

Source:lambda_function.py

1# !/usr/bin/python32# encoding: iso-8859-13"""4Lambda para realizar os processos no DynamoDB5"""6import json7import os8from S3Client import S3Client9def lambda_handler(event, context) -> str:10    """11    Recebe um Event no formato json.12    :param event: Contém informações da operação a ser realizada.13    :type event: STRING(JSON)14    :param context: Não utilizado15    :type context: None16    :return: Retorna um json17    :rtype: STRING(JSON)18    """19    operation = event['Operation']20    try:21        name_bucket = event['body']['name']22    except KeyError:23        name_bucket = os.environ['NAME_BUCKET']24    if operation == 'delete_bucket':25        s3 = S3Client()26        response_operation = s3.delete_bucket(name_bucket)27        if response_operation:28            status_code = 20029        else:30            status_code = 40031        response_lambda = {32            'status_code': status_code,33            'body': response_operation34        }35        response_lambda = json.dumps(response_lambda)36        response_lambda = json.loads(response_lambda)37        return response_lambda38    elif operation == 'create_bucket':39        s3 = S3Client()40        response_operation = s3.create_bucket(name_bucket)41        if response_operation:42            status_code = 20043        else:44            status_code = 40045        response_lambda = {46            'status_code': status_code,47            'body': response_operation48        }49        response_lambda = json.dumps(response_lambda)50        response_lambda = json.loads(response_lambda)51        return response_lambda52    elif operation == 'put_object':53        try:54            key = event['body']['key']55        except KeyError:56            key = os.environ['KEY_PUT_OBJECT']57        body = bytes(json.dumps(event['body']['body']), encoding='ISO-8859-1')58        s3 = S3Client()59        response_operation = s3.put_object(name_bucket, key, body)60        if response_operation:61            status_code = 20062        else:63            status_code = 40064        response_lambda = {65            'status_code': status_code,66            'body': response_operation67        }68        response_lambda = json.dumps(response_lambda)69        response_lambda = json.loads(response_lambda)70        return response_lambda71    elif operation == 'get_object':72        try:73            key = event['body']['key']74        except KeyError:75            key = os.environ['KEY_GET_OBJECT']76        s3 = S3Client()77        response_operation = s3.get_object(name_bucket, key)78        if response_operation:79            status_code = 20080        else:81            status_code = 40082        response_lambda = {83            'status_code': status_code,84            'body': response_operation85        }86        response_lambda = json.dumps(response_lambda)87        response_lambda = json.loads(response_lambda)88        return response_lambda89    elif operation == 'delete_object':90        key = event['body']['key']91        s3 = S3Client()92        response_operation = s3.delete_object(name_bucket, key)93        if response_operation:94            status_code = 20095        else:96            status_code = 40097        response_lambda = {98            'status_code': status_code,99            'body': response_operation100        }101        response_lambda = json.dumps(response_lambda)102        response_lambda = json.loads(response_lambda)103        return response_lambda104    elif operation == 'delete_objects':105        s3 = S3Client()106        list_files = s3.list_object(name_bucket)107        if list_files:108            for file in list_files:109                s3.delete_object(name_bucket, file)110            if len(list_files) > 0:111                status_code = 200112                response_operation = True113                response_lambda = {114                    'status_code': status_code,115                    'body': response_operation116                }117                response_lambda = json.dumps(response_lambda)118                response_lambda = json.loads(response_lambda)119                return response_lambda120            else:121                status_code = 400122                response_operation = False123                response_lambda = {124                    'status_code': status_code,125                    'body': response_operation126                }127                response_lambda = json.dumps(response_lambda)128                response_lambda = json.loads(response_lambda)129                return response_lambda130        else:131            status_code = 400132            response_operation = False133            response_lambda = {134                'status_code': status_code,135                'body': response_operation136            }137            response_lambda = json.dumps(response_lambda)138            response_lambda = json.loads(response_lambda)...

tests.py

Source:tests.py

1from bson import ObjectId2from starlette.testclient import TestClient3from src.main import app4from src.models import find_one5client = TestClient(app)6def check_db_operation(response):7    response_operation = response.json()['operation']8    object_id = ObjectId(response_operation['_id'])9    database_operation = find_one({'_id': object_id})10    assert database_operation['operator'] == response_operation['operator']11    assert database_operation['first_number'] == response_operation['first_number']12    assert database_operation['second_number'] == response_operation['second_number']13    assert database_operation['result'] == response_operation['result']14def test_root():15    response = client.get('/')16    assert response.status_code == 20017    assert response.json() == {'message': 'Calculator'}18    print('\tTest: test_root, success!')19def test_add():20    response = client.get('/add?first=345&second=678')21    assert response.status_code == 20022    assert response.json()['operation']['result'] == 102323    check_db_operation(response)24    print('\tTest: test_add, success!')25def test_subtract():26    response = client.get('/sub?first=344&second=23')27    assert response.status_code == 20028    assert response.json()['operation']['result'] == 32129    check_db_operation(response)30    print('\tTest: test_subtract, success!')31def test_multiply():32    response = client.get('/multi?first=600&second=65')33    assert response.status_code == 20034    assert response.json()['operation']['result'] == 3900035    check_db_operation(response)36    print('\tTest: test_multiply, success!')37def test_division():38    response = client.get('/div?first=877&second=47')39    assert response.status_code == 20040    assert round(response.json()['operation']['result'], 2) == 18.6641    check_db_operation(response)42    print('\tTest: test_division, success!')43def test_division_by_zero():44    response = client.get('/div?first=877&second=0')45    assert response.status_code == 20046    assert response.json() == {'error': 'Division by zero'}...

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.