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

How to use task_stop method in autotest

Best Python code snippet using autotest_python

task_base.py

...15        """16        Initializer17        Args:18            stop_timeout_secs (int): Number of seconds to wait for process to exit19                upon calling task_stop(). If the process fails to stop before the20                specified timeout, it will attemp to kill the process via brute21                force. If you would like to wait indefinitely, pass in `None`.22        """23        self._stop_timeout_secs = stop_timeout_secs24        self._task_process = None25        self.task_stopping_event = multiprocessing.Event()26    def task_worker(self):27        raise NotImplementedError28    def task_run(self):29        if self.task_stopping_event.is_set():30            return31        self._task_process = multiprocessing.Process(target=self.task_worker)32        self._task_process.start()33    def task_stop(self):34        # Signal the process to stop35        self.task_stopping_event.set()36        # Wait for the process to exit37        self._task_process.join(self._stop_timeout_secs)38        # If the process didn't exit, attempt to kill it39        if self._task_process.is_alive():40            os.kill(self._task_process.pid, signal.SIGKILL)41        if self._task_process.is_alive():42            return False43        return True44#45# ThreadTaskBase =====================================================================46#47class ThreadTaskBase(object):48    """49    Base class for creating an object that gets spawned as a separate thread50    Child class needs to implement the task_worker method, which should be51    designed to return if task_stopping_event is set52    """53    def __init__(self, stop_timeout_secs=None):54        """55        Initializer56        Args:57            stop_timeout_secs (int): Number of seconds to wait for thread to exit58                upon calling task_stop(). If you would like to wait indefinitely,59                pass in None.60        """61        self._stop_timeout_secs = stop_timeout_secs62        self._task_thread = None63        self.task_stopping_event = threading.Event()64    def task_worker(self):65        raise NotImplementedError66    def task_run(self):67        if self.task_stopping_event.is_set():68            return69        self._task_thread = threading.Thread(target=self.task_worker)70        self._task_thread.start()71    def task_stop(self):72        # Signal the thread to stop73        self.task_stopping_event.set()74        # Wait for the thread to exit75        self._task_thread.join(self._stop_timeout_secs)76        if self._task_thread.is_alive():77            return False...

test_task_context.py

Source: test_task_context.py

...7    @hookimpl8    def task_start(self):9        self.events.append("task_start")10    @hookimpl11    def task_stop(self, failed):12        self.events.append(("task_stop", failed))13@pytest.fixture14def spy():15    instance = EventSpy()16    pm.register(instance)17    yield instance18    pm.unregister(instance)19def test_task_ok(spy):20    """task_context emits expected events during successful non-exit case"""21    # Initially no events22    assert not spy.events23    # Simulate a task24    with task_context():25        # task_start should have already been fired...

spider.py

Source: spider.py

1import threading2import time3from queue import Queue4from settings import MAX_PAGE, DL_PSD, FILE_DIR, MAX_QUEUE, MAX_THREAD5from httprequest import UrlHandler, Download6class Producer(threading.Thread):7    def __init__(self, queue, task,):8        threading.Thread.__init__(self)9        self.task_stop = False10        self.task = task11        self.queue = queue12    def run(self):13        tasks = self.task.get_tasks()14        while True:15            try:16                task = next(tasks)17            except StopIteration:18                break19            else:20                self.queue.put(task)21    def stop(self):22        self.task_stop = True23class Consumer(threading.Thread):24    def __init__(self, queue, donwload):25        threading.Thread.__init__(self)26        self.task_stop = False27        self.queue = queue28        self.download = donwload29    def run(self):30        while not self.task_stop:31            try:32                data_ = self.queue.get(timeout=4)33            except Exception as e:34                print(e.args)35                break36            else:37                self.download(data_, dl_dir=FILE_DIR, has_psd=DL_PSD, ).work()38                self.queue.task_done()39    def stop(self):40        self.task_stop = True41def wrapper(func, *args, **kwargs):42    def timer():43        start_time = time.time()44        func(*args, **kwargs)45        end_time = time.time()46        print('æ»ç¨æ¶:-->', int(end_time - start_time))47    return timer48@wrapper49def main():50    q = Queue(MAX_QUEUE)51    t1_thread = []52    t2_thread = []53    for url in ['http://www.uimaker.com/uimakerdown/list_36_{}.html'.format(i) for i in range(1, MAX_PAGE)]:54        t1 = Producer(q, UrlHandler(url))55        t1.start()56        t1_thread.append(t1)57    for i in range(MAX_THREAD):58        t2 = Consumer(q, Download)59        t2.start()60    for t1 in t1_thread:61        t1.join()62    for t2 in t2_thread:63        t2.join()64    q.join()65if __name__ == '__main__':...

Blogs

Check out the latest blogs from LambdaTest on this topic:

How to Position Your Team for Success in Estimation

Estimates are critical if you want to be successful with projects. If you begin with a bad estimating approach, the project will almost certainly fail. To produce a much more promising estimate, direct each estimation-process issue toward a repeatable standard process. A smart approach reduces the degree of uncertainty. When dealing with presales phases, having the most precise estimation findings can assist you to deal with the project plan. This also helps the process to function more successfully, especially when faced with tight schedules and the danger of deviation.

Why Selenium WebDriver Should Be Your First Choice for Automation Testing

Developed in 2004 by Thoughtworks for internal usage, Selenium is a widely used tool for automated testing of web applications. Initially, Selenium IDE(Integrated Development Environment) was being used by multiple organizations and testers worldwide, benefits of automation testing with Selenium saved a lot of time and effort. The major downside of automation testing with Selenium IDE was that it would only work with Firefox. To resolve the issue, Selenium RC(Remote Control) was used which enabled Selenium to support automated cross browser testing.

How to increase and maintain team motivation

The best agile teams are built from people who work together as one unit, where each team member has both the technical and the personal skills to allow the team to become self-organized, cross-functional, and self-motivated. These are all big words that I hear in almost every agile project. Still, the criteria to make a fantastic agile team are practically impossible to achieve without one major factor: motivation towards a common goal.

The Top 52 Selenium Open Source Projects On GitHub

Selenium, a project hosted by the Apache Software Foundation, is an umbrella open-source project comprising a variety of tools and libraries for test automation. Selenium automation framework enables QA engineers to perform automated web application testing using popular programming languages like Python, Java, JavaScript, C#, Ruby, and PHP.

Test Optimization for Continuous Integration

“Test frequently and early.” If you’ve been following my testing agenda, you’re probably sick of hearing me repeat that. However, it is making sense that if your tests detect an issue soon after it occurs, it will be easier to resolve. This is one of the guiding concepts that makes continuous integration such an effective method. I’ve encountered several teams who have a lot of automated tests but don’t use them as part of a continuous integration approach. There are frequently various reasons why the team believes these tests cannot be used with continuous integration. Perhaps the tests take too long to run, or they are not dependable enough to provide correct results on their own, necessitating human interpretation.

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.