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

How to use should_validate_request method in localstack

Best Python code snippet using localstack_python

decorators.py

Source:decorators.py

1import traceback2import logging3import time4import sys5import activitystreams as as_parser6import eventlet7from functools import wraps8from datetime import datetime9from uuid import uuid4 as uuid10from dino import validation11from dino import environ12from dino import utils13from dino.exceptions import NoSuchUserException14from dino.config import ConfigKeys15from dino.config import SessionKeys16from dino.config import ErrorCodes17logger = logging.getLogger()18def timeit(_logger, tag: str):19    def factory(view_func):20        @wraps(view_func)21        def decorator(*args, **kwargs):22            failed = False23            before = time.time()24            try:25                return view_func(*args, **kwargs)26            except Exception as e:27                failed = True28                _logger.exception(traceback.format_exc())29                _logger.error(tag + '... FAILED')30                environ.env.capture_exception(sys.exc_info())31                raise e32            finally:33                if not failed:34                    the_time = (time.time()-before)*100035                    if tag.startswith('on_') and environ.env.stats is not None:36                        environ.env.stats.timing('api.' + tag, the_time)37                    else:38                        _logger.debug(tag + '... %.2fms' % the_time)39        return decorator40    return factory41def locked_method(method):42    """Method decorator. Requires a lock object at self._lock"""43    def newmethod(self, *args, **kwargs):44        with self._lock:45            return method(self, *args, **kwargs)46    return newmethod47def _delayed_disconnect(sid: str):48    environ.env.disconnect_by_sid(sid)49def respond_with(gn_event_name=None, should_disconnect=False, emit_response=True):50    def factory(view_func):51        @wraps(view_func)52        def decorator(*args, **kwargs):53            tb = None54            try:55                status_code, data = view_func(*args, **kwargs)56            except Exception as e:57                environ.env.stats.incr(gn_event_name + '.exception')58                tb = traceback.format_exc()59                logger.error('%s: %s' % (gn_event_name, str(e)))60                environ.env.capture_exception(sys.exc_info())61                if should_disconnect and environ.env.config.get(ConfigKeys.DISCONNECT_ON_FAILED_LOGIN, False):62                    eventlet.spawn_after(seconds=1, func=_delayed_disconnect, sid=environ.env.request.sid)63                return 500, str(e)64            finally:65                if tb is not None:66                    logger.exception(tb)67            if status_code != 200:68                logger.warning('in decorator, status_code: %s, data: %s' % (status_code, str(data)))69                if should_disconnect and environ.env.config.get(ConfigKeys.DISCONNECT_ON_FAILED_LOGIN, False):70                    eventlet.spawn_after(seconds=1, func=_delayed_disconnect, sid=environ.env.request.sid)71            # in some cases the callback is enough72            if emit_response:73                response_message = environ.env.response_formatter(status_code, data)74                environ.env.emit(gn_event_name, response_message)75            return status_code, None76        return decorator77    return factory78def count_connections(connect_type=None):79    def factory(view_func):80        @wraps(view_func)81        def decorator(*args, **kwargs):82            try:83                if connect_type == 'connect':84                    environ.env.stats.incr('connections')85                elif connect_type == 'disconnect':86                    environ.env.stats.decr('connections')87                else:88                    logger.warning('unknown connect type "%s"' % connect_type)89            except Exception as e:90                logger.error('could not record statistics: %s' % str(e))91                environ.env.capture_exception(sys.exc_info())92            return view_func(*args, **kwargs)93        return decorator94    return factory95def pre_process(validation_name, should_validate_request=True):96    def factory(view_func):97        @wraps(view_func)98        def decorator(*a, **k):99            def _pre_process(*args, **kwargs):100                if not hasattr(validation.request, validation_name):101                    raise RuntimeError('no such attribute on validation.request: %s' % validation_name)102                try:103                    data = args[0]104                    if 'actor' not in data:105                        data['actor'] = dict()106                    # let the server determine the publishing time of the event, not the client107                    # use default time format, since activity streams only accept RFC3339 format108                    data['published'] = datetime.utcnow().strftime(ConfigKeys.DEFAULT_DATE_FORMAT)109                    data['id'] = str(uuid())110                    if should_validate_request:111                        data['actor']['id'] = str(environ.env.session.get(SessionKeys.user_id.value))112                        user_name = environ.env.session.get(SessionKeys.user_name.value)113                        if user_name is None or len(user_name.strip()) == 0:114                            try:115                                user_name = utils.get_user_name_for(data['actor']['id'])116                            except NoSuchUserException:117                                error_msg = '[%s] no user found for user_id "%s" in session' % \118                                            (validation_name, str(data['actor']['id']))119                                logger.error(error_msg)120                                return ErrorCodes.NO_USER_IN_SESSION, error_msg121                        data['actor']['displayName'] = utils.b64e(user_name)122                    activity = as_parser.parse(data)123                    # the login request will not have user id in session yet, which this would check124                    if should_validate_request:125                        is_valid, error_msg = validation.request.validate_request(activity)126                        if not is_valid:127                            logger.error('[%s] validation failed, error message: %s' % (validation_name, str(error_msg)))128                            return ErrorCodes.VALIDATION_ERROR, error_msg129                    is_valid, status_code, message = getattr(validation.request, validation_name)(activity)130                    if is_valid:131                        all_ok = True132                        if validation_name in environ.env.event_validator_map:133                            for validator in environ.env.event_validator_map[validation_name]:134                                all_ok, status_code, msg = validator(data, activity)135                                if not all_ok:136                                    logger.warning(137                                            '[%s] validator "%s" failed: %s' %138                                            (validation_name, str(validator), str(msg)))139                                    break140                        if all_ok:141                            args = (data, activity)142                            status_code, message = view_func(*args, **kwargs)143                except Exception as e:144                    logger.error('%s: %s' % (validation_name, str(e)))145                    logger.exception(traceback.format_exc())146                    environ.env.stats.incr('event.' + validation_name + '.exception')147                    environ.env.capture_exception(sys.exc_info())148                    return ErrorCodes.UNKNOWN_ERROR, str(e)149                if status_code == 200:150                    environ.env.stats.incr('event.' + validation_name + '.count')151                else:152                    environ.env.stats.incr('event.' + validation_name + '.error')153                    logger.warning('in decorator for %s, status_code: %s, message: %s' %154                                (validation_name, status_code, str(message)))155                return status_code, message156            start = time.time()157            exception_occurred = False158            try:159                environ.env.stats.incr('event.' + validation_name + '.count')160                return _pre_process(*a, **k)161            except:162                exception_occurred = True163                environ.env.stats.incr('event.' + validation_name + '.exception')164                raise165            finally:166                if not exception_occurred:167                    environ.env.stats.timing('event.' + validation_name, (time.time()-start)*1000)168        return decorator...

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.