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

How to use _stop_thread method in pyatom

Best Python code snippet using pyatom_python

async_printing.py

Source:async_printing.py

1#!/usr/bin/env python2# coding=utf-83"""4A simple example demonstrating an application that asynchronously prints alerts, updates the prompt5and changes the window title6"""7import random8import threading9import time10from typing import List11import cmd212from cmd2 import ansi13ALERTS = ["Watch as this application prints alerts and updates the prompt",14          "This will only happen when the prompt is present",15          "Notice how it doesn't interfere with your typing or cursor location",16          "Go ahead and type some stuff and move the cursor throughout the line",17          "Keep typing...",18          "Move that cursor...",19          "Pretty seamless, eh?",20          "Feedback can also be given in the window title. Notice the alert count up there?",21          "You can stop and start the alerts by typing stop_alerts and start_alerts",22          "This demo will now continue to print alerts at random intervals"23          ]24class AlerterApp(cmd2.Cmd):25    """ An app that shows off async_alert() and async_update_prompt() """26    def __init__(self, *args, **kwargs) -> None:27        """ Initializer """28        super().__init__(*args, **kwargs)29        self.prompt = "(APR)> "30        # The thread that will asynchronously alert the user of events31        self._stop_thread = False32        self._alerter_thread = threading.Thread()33        self._alert_count = 034        self._next_alert_time = 035        # Create some hooks to handle the starting and stopping of our thread36        self.register_preloop_hook(self._preloop_hook)37        self.register_postloop_hook(self._postloop_hook)38    def _preloop_hook(self) -> None:39        """ Start the alerter thread """40        # This runs after cmdloop() acquires self.terminal_lock, which will be locked until the prompt appears.41        # Therefore this is the best place to start the alerter thread since there is no risk of it alerting42        # before the prompt is displayed. You can also start it via a command if its not something that should43        # be running during the entire application. See do_start_alerts().44        self._stop_thread = False45        self._alerter_thread = threading.Thread(name='alerter', target=self._alerter_thread_func)46        self._alerter_thread.start()47    def _postloop_hook(self) -> None:48        """ Stops the alerter thread """49        # After this function returns, cmdloop() releases self.terminal_lock which could make the alerter50        # thread think the prompt is on screen. Therefore this is the best place to stop the alerter thread.51        # You can also stop it via a command. See do_stop_alerts().52        self._stop_thread = True53        if self._alerter_thread.is_alive():54            self._alerter_thread.join()55    def do_start_alerts(self, _):56        """ Starts the alerter thread """57        if self._alerter_thread.is_alive():58            print("The alert thread is already started")59        else:60            self._stop_thread = False61            self._alerter_thread = threading.Thread(name='alerter', target=self._alerter_thread_func)62            self._alerter_thread.start()63    def do_stop_alerts(self, _):64        """ Stops the alerter thread """65        self._stop_thread = True66        if self._alerter_thread.is_alive():67            self._alerter_thread.join()68        else:69            print("The alert thread is already stopped")70    def _get_alerts(self) -> List[str]:71        """72        Reports alerts73        :return: the list of alerts74        """75        global ALERTS76        cur_time = time.monotonic()77        if cur_time < self._next_alert_time:78            return []79        alerts = []80        if self._alert_count < len(ALERTS):81            alerts.append(ALERTS[self._alert_count])82            self._alert_count += 183            self._next_alert_time = cur_time + 484        else:85            rand_num = random.randint(1, 20)86            if rand_num > 2:87                return []88            for i in range(0, rand_num):89                self._alert_count += 190                alerts.append("Alert {}".format(self._alert_count))91            self._next_alert_time = 092        return alerts93    def _generate_alert_str(self) -> str:94        """95        Combines alerts into one string that can be printed to the terminal96        :return: the alert string97        """98        global ALERTS99        alert_str = ''100        alerts = self._get_alerts()101        longest_alert = max(ALERTS, key=len)102        num_asterisks = len(longest_alert) + 8103        for i, cur_alert in enumerate(alerts):104            # Use padding to center the alert105            padding = ' ' * int((num_asterisks - len(cur_alert)) / 2)106            if i > 0:107                alert_str += '\n'108            alert_str += '*' * num_asterisks + '\n'109            alert_str += padding + cur_alert + padding + '\n'110            alert_str += '*' * num_asterisks + '\n'111        return alert_str112    def _generate_colored_prompt(self) -> str:113        """Randomly generates a colored prompt114        :return: the new prompt115        """116        fg_color = random.choice(list(ansi.FG_COLORS.keys()))117        bg_color = random.choice(list(ansi.BG_COLORS.keys()))118        return ansi.style(self.visible_prompt.rstrip(), fg=fg_color, bg=bg_color) + ' '119    def _alerter_thread_func(self) -> None:120        """ Prints alerts and updates the prompt any time the prompt is showing """121        self._alert_count = 0122        self._next_alert_time = 0123        while not self._stop_thread:124            # Always acquire terminal_lock before printing alerts or updating the prompt125            # To keep the app responsive, do not block on this call126            if self.terminal_lock.acquire(blocking=False):127                # Get any alerts that need to be printed128                alert_str = self._generate_alert_str()129                # Generate a new prompt130                new_prompt = self._generate_colored_prompt()131                # Check if we have alerts to print132                if alert_str:133                    # new_prompt is an optional parameter to async_alert()134                    self.async_alert(alert_str, new_prompt)135                    new_title = "Alerts Printed: {}".format(self._alert_count)136                    self.set_window_title(new_title)137                # No alerts needed to be printed, check if the prompt changed138                elif new_prompt != self.prompt:139                    self.async_update_prompt(new_prompt)140                # Don't forget to release the lock141                self.terminal_lock.release()142            time.sleep(0.5)143if __name__ == '__main__':144    import sys145    app = AlerterApp()146    app.set_window_title("Asynchronous Printer Test")...

typewritethread.py

Source:typewritethread.py

1from threading import Thread, Timer2from pynput import keyboard, mouse3import time4'''5'''6# Listen to mouse event7# Access the clipboard or a memory area where login and password are stored8#9DOUBLE_CLICK = 0.3510class TypewriteThread(Thread):11    """Thread responsible for input a text into a field.12    The text is inserted when the user press double click on an input field.13    """14    only_one = []15    def __init__(self, text, timeout=30, *args, **kwargs):16        """Thread responsible for input a text into a field.17        The text is inserted when the user press double click on an input field.18        Args:19        text (str): input text20        timeout (int, optional): After timeout seconds the thread is stopped.21        Defaults to 30 secs.22        """23        super(TypewriteThread, self).__init__()24        self.text = text25        self.timeout = timeout26        self.start_time = 027        self._stop_thread = False28    def run(self):29        for twt in self.only_one:30            twt.interrupt()31        del self.only_one[:]32        self.only_one.append(self)33        # Save obj creation time34        self.writer = keyboard.Controller()35        self.mouseListener = mouse.Listener(on_click=self.on_click)36        self.mouseListener.name = f'{id(self)}-mouselistener'37        # Set a timeout38        t = Timer(self.timeout, self.interrupt)39        t.name = f'{id(self)}-timeout'40        t.start()41        # Mouselistener thread42        self.mouseListener.start()43        while True:44            time.sleep(0.2)45            if self._stop_thread:46                break47    def on_click(self, x, y, button, pressed):48        if not self.start_time and pressed:49            self.start_time = time.time()50        elif self.start_time and pressed:51            click = time.time()52            elapsed_time = click - self.start_time53            if elapsed_time < DOUBLE_CLICK:54                for c in self.text:55                    self.writer.press(c)56                    self.writer.release(c)57                # Interrupt58                self._stop_thread = True59                return False60            else:61                # Store the first mouse click62                self.start_time = click63    def interrupt(self):64        self._stop_thread = True65        if self.mouseListener.is_alive():66            self.mouseListener.stop()67        68if __name__ == '__main__':69    pass70    data = 'user\tpassword'71    daemon = TypewriteThread(text=data, timeout=10)72    daemon.name = 'typewrite'73    daemon.start()74    # The Main thread wait for daemon stop before process further75    print(daemon.getName())76    time.sleep(2)77    daemon.interrupt()78    print(daemon)79    daemon.join()...

backgroundmonitor.py

Source:backgroundmonitor.py

1import logging2import time3from pathlib import Path4from threading import Thread5from typing import Optional6from .monitor import PathScanner7logger = logging.getLogger(__name__)8class BackgroundMonitor(Thread):9    def __init__(self, root: Path, duty_cyle: float = 2):10        """11        Create the background monitor scanner.12        Args:13            root: The root path to scan14            duty_cycle: The time, in seconds, to spend on/off scanning15        """16        super().__init__()17        self.root = root18        self.duty_cycle = duty_cyle19        self._stop_thread = False20        # self.results: List[Dict] = []21        self._scanner = PathScanner(self.root)22    def run(self):23        while not self._stop_thread:24            self._scanner.scan(time_limit=self.duty_cycle)25            time.sleep(self.duty_cycle)26    def stop(self, timeout: Optional[float] = None):27        self._stop_thread = True28        self.join(timeout)29    def results(self):...

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.