How to use is_websocket_request method in localstack

Best Python code snippet using localstack_python

resource.py

Source:resource.py

1import re2import warnings3from .protocols.base import BaseProtocol4from .exceptions import WebSocketError5try:6    from collections import OrderedDict7except ImportError:8    class OrderedDict:9        pass10class WebSocketApplication(object):11    protocol_class = BaseProtocol12    def __init__(self, ws):13        self.protocol = self.protocol_class(self)14        self.ws = ws15    def handle(self):16        self.protocol.on_open()17        while True:18            try:19                message = self.ws.receive()20            except WebSocketError:21                self.protocol.on_close()22                break23            self.protocol.on_message(message)24    def on_open(self, *args, **kwargs):25        pass26    def on_close(self, *args, **kwargs):27        pass28    def on_message(self, message, *args, **kwargs):29        self.ws.send(message, **kwargs)30    @classmethod31    def protocol_name(cls):32        return cls.protocol_class.PROTOCOL_NAME33class Resource(object):34    def __init__(self, apps=None):35        self.apps = apps if apps else []36        if isinstance(apps, dict):37            if not isinstance(apps, OrderedDict):38                warnings.warn("Using an unordered dictionary for the "39                              "app list is discouraged and may lead to "40                              "undefined behavior.", UserWarning)41            self.apps = apps.items()42    # An app can either be a standard WSGI application (an object we call with43    # __call__(self, environ, start_response)) or a class we instantiate44    # (and which can handle websockets). This function tells them apart.45    # Override this if you have apps that can handle websockets but don't46    # fulfill these criteria.47    def _is_websocket_app(self, app):48        return isinstance(app, type) and issubclass(app, WebSocketApplication)49    def _app_by_path(self, environ_path, is_websocket_request):50        # Which app matched the current path?51        for path, app in self.apps:52            if re.match(path, environ_path):53                if is_websocket_request == self._is_websocket_app(app):54                    return app55        return None56    def app_protocol(self, path):57        # app_protocol will only be called for websocket apps58        app = self._app_by_path(path, True)59        if hasattr(app, 'protocol_name'):60            return app.protocol_name()61        else:62            return ''63    def __call__(self, environ, start_response):64        environ = environ65        is_websocket_call = 'wsgi.websocket' in environ66        current_app = self._app_by_path(environ['PATH_INFO'], is_websocket_call)67        if current_app is None:68            raise Exception("No apps defined")69        if is_websocket_call:70            ws = environ['wsgi.websocket']71            current_app = current_app(ws)72            current_app.ws = ws  # TODO: needed?73            current_app.handle()74            # Always return something, calling WSGI middleware may rely on it75            return []76        else:...

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.