TestMu 2025 - The Largest Virtual Software Testing Conference is Back | 50,000+ Attendees

How to use server_alive method in autotest

Best Python code snippet using autotest_python

redis_cache.py

...30    global check_deadline31    logging.warning("Redis server up")32    server_alive = True33    check_deadline = -134def check_server_alive() -> bool:35    global conn36    global server_alive37    global check_deadline38    if server_alive:39        return True40    if check_deadline < time.time():41        try:42            conn.ping()43            server_up()44            return True45        except:46            server_down()47            return False48def gen_hash(input_str):49    return str(hashlib.sha256(input_str.encode('utf-8')).hexdigest())50def generate_cache_key(input_dict):51    cache_key = key_prefix + ":" + gen_hash(json.dumps(input_dict))52    return cache_key53def cache_get(key):54    global conn55    try:56        cache_key = generate_cache_key(key)57        response = conn.get(cache_key)58        if response:59            decoded_response = json.loads(response)60            return decoded_response61    except redis.ConnectionError as e:62        server_down()63        logging.error("Redis server down: cache disabled in cache_get!")64    except json.JSONDecodeError:65        logging.error("Redis server corrupted response!")66    except Exception as e:67        logging.error("Redis server error: " + str(e))68def cache_get_protected(key):69    global server_alive70    if server_alive:71        return cache_get(key)72    else:73        return None74def cache_set_protected(key, value):75    global server_alive76    if server_alive:77        cache_set(key, value)78def cache_set(key, value):79    global conn80    global key_expire81    try:82        cache_key = generate_cache_key(key)83        value_json = u.to_json(value)84        conn.set(cache_key, value_json)85        conn.expire(cache_key, key_expire)86    except redis.ConnectionError as e:87        server_down()88        logging.error("Redis server down: cache disabled in cache_get!")89    except json.JSONDecodeError:90        logging.error("Redis key cannot be Json encoded! " + str(key))91    except Exception as e:92        server_down()93        logging.error("Redis cache disabled in cache_set! + Exception: " + str(e))94# decorator95def cache(**decorator_kwargs):96    global server_alive97    global key_prefix98    def decorator(func):99        @functools.wraps(func)100        def wrapper(*func_args, **func_kwargs):101            if not check_server_alive():102                return func(*func_args, **func_kwargs)103            cache_key = {'func': func.__globals__['__file__'] + '::' + func.__name__,104                         'func_args': str(func_args),105                         'func_kwargs': str(func_kwargs),106                         'decorator_kwargs': str(decorator_kwargs)107                         }108            response_from_cache = cache_get(cache_key)109            if response_from_cache:110                logging.debug(f"Redis returned object from cache for key ({cache_key}):" + str(response_from_cache))111                return response_from_cache112            else:113                response = func(*func_args, **func_kwargs)114                logging.debug(f"Redis returned object from wrapped function for key ({cache_key}): " + str(response))115                cache_set(cache_key, response)...

gtornado.py

Source: gtornado.py

1# -*- coding: utf-8 -2#3# This file is part of gunicorn released under the MIT license.4# See the NOTICE for more information.5import copy6import os7import sys8try:9    import tornado.web10except ImportError:11    raise RuntimeError("You need tornado installed to use this worker.")12import tornado.httpserver13from tornado.ioloop import IOLoop, PeriodicCallback14from tornado.wsgi import WSGIContainer15from gunicorn.workers.base import Worker16from gunicorn import __version__ as gversion17class TornadoWorker(Worker):18    @classmethod19    def setup(cls):20        web = sys.modules.pop("tornado.web")21        old_clear = web.RequestHandler.clear22        def clear(self):23            old_clear(self)24            if not "Gunicorn" in self._headers["Server"]:25                self._headers["Server"] += " (Gunicorn/%s)" % gversion26        web.RequestHandler.clear = clear27        sys.modules["tornado.web"] = web28    def handle_exit(self, sig, frame):29        if self.alive:30            super(TornadoWorker, self).handle_exit(sig, frame)31    def handle_request(self):32        self.nr += 133        if self.alive and self.nr >= self.max_requests:34            self.log.info("Autorestarting worker after current request.")35            self.alive = False36    def watchdog(self):37        if self.alive:38            self.notify()39        if self.ppid != os.getppid():40            self.log.info("Parent changed, shutting down: %s", self)41            self.alive = False42    def heartbeat(self):43        if not self.alive:44            if self.server_alive:45                if hasattr(self, 'server'):46                    try:47                        self.server.stop()48                    except Exception:49                        pass50                self.server_alive = False51            else:52                if not self.ioloop._callbacks:53                    self.ioloop.stop()54    def run(self):55        self.ioloop = IOLoop.instance()56        self.alive = True57        self.server_alive = False58        PeriodicCallback(self.watchdog, 1000, io_loop=self.ioloop).start()59        PeriodicCallback(self.heartbeat, 1000, io_loop=self.ioloop).start()60        # Assume the app is a WSGI callable if its not an61        # instance of tornado.web.Application or is an62        # instance of tornado.wsgi.WSGIApplication63        app = self.wsgi64        if not isinstance(app, tornado.web.Application) or \65           isinstance(app, tornado.wsgi.WSGIApplication):66            app = WSGIContainer(app)67        # Monkey-patching HTTPConnection.finish to count the68        # number of requests being handled by Tornado. This69        # will help gunicorn shutdown the worker if max_requests70        # is exceeded.71        httpserver = sys.modules["tornado.httpserver"]72        if hasattr(httpserver, 'HTTPConnection'):73            old_connection_finish = httpserver.HTTPConnection.finish74            def finish(other):75                self.handle_request()76                old_connection_finish(other)77            httpserver.HTTPConnection.finish = finish78            sys.modules["tornado.httpserver"] = httpserver79            server_class = tornado.httpserver.HTTPServer80        else:81            class _HTTPServer(tornado.httpserver.HTTPServer):82                def on_close(instance, server_conn):83                    self.handle_request()84                    super(_HTTPServer, instance).on_close(server_conn)85            server_class = _HTTPServer86        if self.cfg.is_ssl:87             _ssl_opt = copy.deepcopy(self.cfg.ssl_options)88             # tornado refuses initialization if ssl_options contains following89             # options90             del _ssl_opt["do_handshake_on_connect"]91             del _ssl_opt["suppress_ragged_eofs"]92             server = server_class(app, io_loop=self.ioloop,93                    ssl_options=_ssl_opt)94        else:95            server = server_class(app, io_loop=self.ioloop)96        self.server = server97        self.server_alive = True98        for s in self.sockets:99            s.setblocking(0)100            if hasattr(server, "add_socket"):  # tornado > 2.0101                server.add_socket(s)102            elif hasattr(server, "_sockets"):  # tornado 2.0103                server._sockets[s.fileno()] = s104        server.no_keep_alive = self.cfg.keepalive <= 0105        server.start(num_processes=1)...

Broadcaster.py

Source: Broadcaster.py

1import asyncio2import threading3import websockets4class Broadcaster(threading.Thread):5    clients = []6    board = []7    server_alive = True8    server_task = None9    stop_event = threading.Event()10    def run(self)-> None:11        asyncio.run(self.start_server())12    async def start_server(self):13        stop = asyncio.get_event_loop().run_in_executor(None, self.stop_event.wait)14        async with websockets.serve(self.handler, "localhost", 8765):15            await stop16    def setBoard(self, board):17        self.board = board18    async def handler(self, websocket):19        self.clients.append(websocket)20        while self.server_alive:21            message = await websocket.recv()22            if not self.server_alive:23                raise Exception("Server is now close")24            if message == 'retrieveGrid':25                await websocket.send("GRID:"+','.join(str(i) for i in self.board))26        # async for message in websocket:27        #     await websocket.send(message)28    def close(self):29        self.server_alive = False30        self.stop_event.set()31    async def broadcast(self, state):32        for ws in self.clients:...

Blogs

Check out the latest blogs from LambdaTest on this topic:

Keeping Quality Transparency Throughout the organization

In general, software testers have a challenging job. Software testing is frequently the final significant activity undertaken prior to actually delivering a product. Since the terms “software” and “late” are nearly synonymous, it is the testers that frequently catch the ire of the whole business as they try to test the software at the end. It is the testers who are under pressure to finish faster and deem the product “release candidate” before they have had enough opportunity to be comfortable. To make matters worse, if bugs are discovered in the product after it has been released, everyone looks to the testers and says, “Why didn’t you spot those bugs?” The testers did not cause the bugs, but they must bear some of the guilt for the bugs that were disclosed.

A Complete Guide To CSS Grid

Ever since the Internet was invented, web developers have searched for the most efficient ways to display content on web browsers.

Top 17 Resources To Learn Test Automation

Lack of training is something that creates a major roadblock for a tester. Often, testers working in an organization are all of a sudden forced to learn a new framework or an automation tool whenever a new project demands it. You may be overwhelmed on how to learn test automation, where to start from and how to master test automation for web applications, and mobile applications on a new technology so soon.

Testing Modern Applications With Playwright ????

Web applications continue to evolve at an unbelievable pace, and the architecture surrounding web apps get more complicated all of the time. With the growth in complexity of the web application and the development process, web application testing also needs to keep pace with the ever-changing demands.

A Complete Guide To CSS Houdini

As a developer, checking the cross browser compatibility of your CSS properties is of utmost importance when building your website. I have often found myself excited to use a CSS feature only to discover that it’s still not supported on all browsers. Even if it is supported, the feature might be experimental and not work consistently across all browsers. Ask any front-end developer about using a CSS feature whose support is still in the experimental phase in most prominent web browsers. ????

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.