KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use on_put method in localstack

Best Python code snippet using localstack_python

middleware.py

Source:middleware.py

1import falcon2import json3import jsonschema4import logging5def _get_request_schema(req, resource):6    if resource is None or req.method not in ['POST', 'PUT', 'PATCH']:7        return None8    # First try to get schema from method itself9    schema = getattr(10        getattr(resource, {'POST': 'on_post', 'PUT': 'on_put', 'PATCH': 'on_patch'}[req.method], None),11        '__request_schema__',12        None13    # Otherwise, fall back to schema defined directly in class14    ) or getattr(resource, '__request_schemas__', {}).get(15        {'POST': 'on_post', 'PUT': 'on_put', 'PATCH': 'on_patch'}[req.method]16    )17    return schema18def _get_response_schema(resource, req):19    try:20        method_name = {'POST': 'on_post', 'PUT': 'on_put', 'PATCH': 'on_patch', 'GET': 'on_get', 'DELETE': 'on_delete'}[req.method]21    except KeyError:22        return23    # First try to get schema from method itself24    return getattr(25        getattr(resource, method_name, None),26        '__response_schema__',27        None28    # Otherwise, fall back to schema defined directly in class29    ) or getattr(resource, '__response_schemas__', {}).get(method_name)30class _null_handler(logging.Handler):31    def emit(self, record):32        pass33class Middleware(object):34    def __init__(self, logger=None):35        if logger is None:36            # Default to no logging if no logger provided37            logger = logging.getLogger(__name__)38            logger.addHandler(_null_handler())39        self.logger = logger40    def process_resource(self, req, resp, resource, params):41        if _get_response_schema(resource, req) and not req.client_accepts_json:42            raise falcon.HTTPNotAcceptable('This API supports only JSON-encoded responses')43        if resource is None or req.method not in ['POST', 'PUT', 'PATCH']:44            return45        if _get_request_schema(req, resource) is not None:46            if req.content_type is None or 'application/json' not in req.content_type:47                raise falcon.HTTPUnsupportedMediaType('This API supports only JSON-encoded requests')48        if 'application/json' in req.content_type:49            body = req.stream.read()50            if not body:51                raise falcon.HTTPBadRequest(52                    'Empty request body',53                    'A valid JSON document is required'54                )55            schema = _get_request_schema(req, resource)56            try:57                req.context['doc'] = json.loads(body.decode('utf-8'))58            except (ValueError, UnicodeDecodeError) as error:59                if schema is not None:60                    raise falcon.HTTPBadRequest(61                        'Malformed JSON',62                        'Could not decode the request body.  The JSON was incorrect or not encoded as UTF-8'63                    )64                req.context['doc'] = body65            if schema is not None:66                try:67                    jsonschema.validate(req.context['doc'], schema)68                except jsonschema.exceptions.ValidationError as error:69                    raise falcon.HTTPBadRequest(70                        'Invalid request body',71                        json.dumps({'error': str(error)})72                    )73    def process_response(self, req, resp, resource):74        if 'result' not in req.context:75            return76        resp.body = json.dumps(req.context['result'])77        schema = _get_response_schema(resource, req)78        if schema is None:79            return80        try:81            jsonschema.validate(req.context['result'], schema)82        except jsonschema.exceptions.ValidationError as error:83            method_name = {'POST': 'on_post', 'PUT': 'on_put', 'PATCH': 'on_patch', 'GET': 'on_get', 'DELETE': 'on_delete'}[req.method]84            self.logger.error('Blocking proposed response from being sent from {0}.{1}.{2} to client as it does not match the defined schema: {3}'.format(resource.__module__, resource.__class__.__name__, method_name, str(error)))...

__init__.py

Source:__init__.py

...41					)42					return cv.create_json('success', cv.dictize(result))43				RESTfulCollectionEndpoint.on_get = on_get44			if create:45				def on_put(self, context):46					on_handle(context)47					on_create(context)48					instance = request_to_model(model_cls, context.request)49					context.session.save(instance).commit()50					return cv.create_json('success', {51						'created_id': model_cls.__table__.primary_key.value_on(instance)52					})53				RESTfulCollectionEndpoint.on_put = on_put54			cv.endpoint(region_route)(RESTfulCollectionEndpoint)55		if update or retrieve or delete:56			class RESTfulResourceEndpoint:57				def rest_get_model(self, context):58					return model_cls.rest_get(59						context.route.resource_id, context.session60					)61			62			if retrieve:63				def on_get(self, context):64					on_handle(context)65					on_retrieve(context)66					#	TODO: Variants.67					return cv.create_json('success', cv.dictize(68						self.rest_get_model(context)69					))70				RESTfulResourceEndpoint.on_get = on_get71			if update:72				def on_put(self, context):73					on_handle(context)74					on_update(context)75					resource = self.rest_get_model(context)76					request_to_update(resource, context.request)77					context.session.commit()78					return cv.create_json('success')79				RESTfulResourceEndpoint.on_put = on_put80			if delete:81				def on_delete(self, context):82					on_handle(context)83					on_delete_cb(context)84					context.session.delete(self.rest_get_model(context)).commit()85					return cv.create_json('success')86				RESTfulResourceEndpoint.on_delete = on_delete...

test_db_observer.py

Source:test_db_observer.py

1# -*- coding: utf-8 -*-2# Copyright 2018 ICON Foundation3#4# Licensed under the Apache License, Version 2.0 (the "License");5# you may not use this file except in compliance with the License.6# You may obtain a copy of the License at7#8# http://www.apache.org/licenses/LICENSE-2.09#10# Unless required by applicable law or agreed to in writing, software11# distributed under the License is distributed on an "AS IS" BASIS,12# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.13# See the License for the specific language governing permissions and14# limitations under the License.15from unittest.mock import Mock16import pytest17from iconservice import IconScoreDatabase18from iconservice.database.db import DatabaseObserver19from iconservice.icon_constant import IconScoreContextType20from iconservice.iconscore.context.context import ContextContainer21from iconservice.iconscore.icon_score_context import IconScoreContext22from tests import create_address23@pytest.fixture(scope="function")24def database_observer():25    return Mock(spec=DatabaseObserver)26@pytest.fixture(scope="function")27def score_db(context_db, database_observer):28    db = IconScoreDatabase(create_address(), context_db)29    db.set_observer(database_observer)30    return db31@pytest.fixture(scope="function")32def context(score_db):33    context = IconScoreContext(IconScoreContextType.DIRECT)34    context.current_address = score_db.address35    ContextContainer._push_context(context)36    yield context37    ContextContainer._clear_context()38def test_database_observer(context, score_db, database_observer):39    # PUT40    key: bytes = b"key1"41    value: bytes = b"value1"42    score_db.put(key, value)43    database_observer.on_put.assert_called()44    args, _ = database_observer.on_put.call_args45    assert key == args[1]46    assert None is args[2]47    assert value == args[3]48    last_value = value49    # UPDATE50    value: bytes = b"value2"51    score_db.put(key, value)52    database_observer.on_put.assert_called()53    args, _ = database_observer.on_put.call_args54    assert key == args[1]55    assert last_value is args[2]56    assert value == args[3]57    last_value = value58    # GET59    key: bytes = b"key1"60    value: bytes = score_db.get(key)61    database_observer.on_get.assert_called()62    args, _ = database_observer.on_get.call_args63    assert key == args[1]64    assert last_value is args[2]65    assert value == last_value66    # DELETE67    key: bytes = b"key1"68    score_db.delete(key)69    database_observer.on_delete.assert_called()70    args, _ = database_observer.on_delete.call_args71    assert key == args[1]...

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.