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

How to use schedule_task method in locust

Best Python code snippet using locust

test_temporal_scheduler.py

Source:test_temporal_scheduler.py

...31        def function1():32            """33            does nothing    def testCancelTaskSuccessfull(self):34            """35        t_id = self.ts.schedule_task(function1, delta_t=1)36        self.assertIsNotNone(t_id)37        self.assertTrue(self.ts.cancel_task(t_id))38    def testCancelTaskFail(self):39        """40        Tests unsuccessfull cancellation of a scheduled task.41        """42        def function1():43            """44            does nothing45            """46        t_id = self.ts.schedule_task(function1, delta_t=1)47        self.assertIsNotNone(t_id)48        self.assertFalse(self.ts.cancel_task(t_id+1))49    def testCallbackArgs(self):50        """51        test passing args to the task52        """53        self.task_args = {}54        def task(x, y, z):55            self.task_args = {"time": time.time(), "x": x, "y": y, "z": z}56        task_id = self.ts.schedule_task(task, absolute_time = time.time() + 1,57                                    task_args = (1,), task_kwargs={'y': 20, 'z': 300})58        self.assertIsNot(task_id, None)59        time.sleep(2)60        self.assertEqual(len(self.ts.timer.current_event_actions), 0)61        self.assertEqual(len(self.ts.timer.events), 0)62        self.assertEqual(self.task_args['x'], 1)63        self.assertEqual(self.task_args['y'], 20)64        self.assertEqual(self.task_args['z'], 300)65    def testCallbackTemporalAccuracy(self):66        """67        test temporal accuracy of callback executions68        """69        self.task_traces = [[] for i in range(4)]70        def task(task_nb):71            """72            Performs action of the tests task.73            Function records execution time on each call.74            Takes one argument, indicating ID if the task.75            """76            self.task_traces[task_nb].append(time.time())77        base_time = time.time()78        self.ts.schedule_task(task,79                            delta_t = 5,80                            task_args = (0,),81                            comment="task1, rel time 5")82        t2id = self.ts.schedule_task(task,83                                period = 3,84                                task_args = (1,),85                                comment = "test2, periodic, period = 3")86        self.ts.schedule_task(task,87                        absolute_time = time.time() + 5,88                        task_args = (2,))89        self.ts.schedule_task(task,90                        absolute_time = time.time() + 1,91                        task_args = (3,))92        # Sleep for 20 seconds, giving time to execute tasks.93        time.sleep(20)94        # Cancel periodic task95        self.assertTrue(self.ts.cancel_task(t2id))96        # Verify execution times.97        # Max deviation from the expected execution time98        max_delta = 0.0199        # Expected execution times (delays)100        task_expected = [ [5],101                          [3, 6, 9, 12, 15, 18],102                          [5],103                          [1],104                        ]105        # Convert absolute execution times to relative106        for task_times in self.task_traces:107            for exec_nb in range(len(task_times)):108                task_times[exec_nb] -= base_time109        # Do the checks110        # Test 1: check length of the record list111        self.assertEqual(len(task_expected), len(self.task_traces))112        for i in range(len(task_expected)):113            # Test 2: check length of the record list for a task114            self.assertEqual(len(task_expected[i]), len(self.task_traces[i]),115                              msg = "Diff len for task %s" % i)116            for j in range(len(self.task_traces[i])):117                # Test 3: check times118                self.assertAlmostEqual(self.task_traces[i][j],119                                        task_expected[i][j],120                                        delta = max_delta,121                                        msg = "Diff in task %s, exec %s: e:%s -> a:%s" %122                                                  (i, j, task_expected[i][j],123                                                   self.task_traces[i][j]))124    def testSchedulingCase1(self):125        """126        case 1: time is in the past.127        This is a dup of a test from another test128        For details see TimerThread.schedule_event129        """130        def task():131            """132            dummy task133            """134        # case 1: time is in the past. Reject135        # This is a dup of a test from another test136        with self.assertRaises(ValueError,137                                msg = "Past task can be scheduled"):138            self.ts.schedule_task(task, absolute_time = time.time()-1)139    def testSchedulingCase2(self):140        """141        case 2: scheduling to an empty scheduler.142        For details see TimerThread.schedule_event143        """144        def task(status):145            """146            Task increments "counter" field in status dict147            """148            status["counter"] += 1149        # case 2: scheduling to an empty scheduler.150        status = {"counter": 0}151        # Task is scheduled to execute after 1 second152        self.ts.schedule_task(task,153                          absolute_time = time.time() + 1,154                          task_args = (status,))155        # Wait for 2 seconds for the task to execute156        time.sleep(2)157        self.assertEqual(status["counter"], 1, msg="Case 2 task did not execute")158    def testSchedulingCase3(self):159        """160        case 3: Scheduled task is earlier than current head161        For details see TimerThread.schedule_event162        """163        def task(task_id, status):164            """165            Task saves execution time to status list166            """167            status[task_id - 1] = time.time()168        # case 3: Scheduled task is earlier than current head169        status = [0, 0]170        current_time = time.time()171        # Task 1 is scheduled to execute after 2 seconds172        self.ts.schedule_task(task,173                          absolute_time = current_time + 2,174                          task_args = (1, status,))175        # Task 2is scheduled to execute prior to task 1176        self.ts.schedule_task(task,177                          absolute_time = current_time + 1,178                          task_args = (2, status,))179        # Give both task time to complete180        time.sleep(3)181        self.assertAlmostEqual(status[0] - current_time, 2, delta = self.delta)182        self.assertAlmostEqual(status[1] - current_time, 1, delta = self.delta)183    def testSchedulingCase4(self):184        """185        case 4: New task, later than head, goes into the queue186        For details see TimerThread.schedule_event187        """188        def task():189            """190            dummy task191            """192        # case 4: New task, later than head, goes into the queue193        current_time = time.time()194        self.ts.schedule_task(task,195                          absolute_time = current_time + 1)196        # Task 2is scheduled to execute prior to task 1197        self.ts.schedule_task(task,198                          absolute_time = current_time + 2)199        # Reach into the bowels of the scheduler and check:200        # 1. length of the execution list201        self.assertEqual(len(self.ts.timer.current_event_actions), 1)202        # 2. length of the scheduler queue203        self.assertEqual(len(self.ts.timer.events), 1)204    def testSchedulingCase5(self):205        """206        case 5: New task at the same time as head, duplicate of a task on the list207        For details see TimerThread.schedule_event208        """209        def task():210            """211            dummy task212            """213        # case 5: New task at the same time as head,214        #           duplicate of a task on the list215        current_time = time.time()216        self.ts.schedule_task(task,217                          absolute_time = current_time + 1,218                          task_args=(1,))219        # check lengths of the execution list and the queue220        self.assertEqual(len(self.ts.timer.events), 0)221        self.assertEqual(len(self.ts.timer.current_event_actions), 1)222        #duplicate223        self.ts.schedule_task(task,224                          absolute_time = current_time + 1,225                          task_args=(1,))226        # check lengths of the execution list and the queue227        self.assertEqual(len(self.ts.timer.events), 0)228        self.assertEqual(len(self.ts.timer.current_event_actions), 1)229    def testSchedulingCase6(self):230        """231        case 6: New task at the same time as head, new task added to the list232        For details see TimerThread.schedule_event233        """234        def task():235            """236            dummy task237            """238        # case 6: New task at the same time as head,239        #            new task added to the list240        current_time = time.time()241        self.ts.schedule_task(task,242                          absolute_time = current_time + 1,243                          task_args=(1,))244        # check lengths of the execution list and the queue245        self.assertEqual(len(self.ts.timer.events), 0, msg = "Task added to events")246        self.assertEqual(len(self.ts.timer.current_event_actions), 1,247                                        msg = "task not added to current_event_actions")248        #duplicate249        self.ts.schedule_task(task,250                          absolute_time = current_time + 1,251                          task_args=(2,))252        # check lengths of the execution list and the queue253        self.assertEqual(len(self.ts.timer.events), 0,254                                        msg = "Task added to events")255        self.assertEqual(len(self.ts.timer.current_event_actions), 2,256                                        msg = "task not added to current-event_actions")257    def testSchedulingCase7(self):258        """259        case 7: New task for the event queue (not head), duplicate of a task in the queue260        For details see TimerThread.schedule_event261        """262        def task():263            """264            dummy task265            """266        # case 7: New task for the event queue (not head),267        #           duplicate of a task in the queue268        current_time = time.time()269        # fill current_event_actions first270        self.ts.schedule_task(task,271                          absolute_time = current_time + 1,272                          task_args=(1,))273        # sanity checks274        self.assertEqual(len(self.ts.timer.events), 0,275                                        msg = "Task added to events")276        self.assertEqual(len(self.ts.timer.current_event_actions), 1,277                                        msg = "task not added to current-event_actions")278        # task goes to events279        self.ts.schedule_task(task,280                          absolute_time = current_time + 2,281                          task_args=(1,))282        self.assertEqual(len(self.ts.timer.events), 1,283                                        msg = "Task added to events")284        self.assertEqual(len(self.ts.timer.current_event_actions), 1,285                                        msg = "task not added to current-event_actions")286        # dup!287        self.ts.schedule_task(task,288                          absolute_time = current_time + 2,289                          task_args=(1,))290        self.assertEqual(len(self.ts.timer.events), 1,291                                        msg = "Task added to events")292        self.assertEqual(len(self.ts.timer.current_event_actions), 1,293                                        msg = "task not added to current-event_actions")294    def testSchedulingCase8(self):295        """296         case 8: New task for the event queue (not head), new task added to the queue297        For details see TimerThread.schedule_event298        """299        def task():300            """301            dummy task302            """303        def task2():304            """305            dummy task306            """307        # case 8: New task for the event queue (not head),308        #           new task added to the queue309        current_time = time.time()310        # fill current_event_actions first311        self.ts.schedule_task(task,312                          absolute_time = current_time + 1,313                          task_args=(1,))314        # sanity checks315        self.assertEqual(len(self.ts.timer.events), 0,316                                        msg = "Task added to events")317        self.assertEqual(len(self.ts.timer.current_event_actions), 1,318                                        msg = "task not added to current-event_actions")319        # task goes to events320        self.ts.schedule_task(task,321                          absolute_time = current_time + 2,322                          task_args=(1,))323        self.assertEqual(len(self.ts.timer.events), 1,324                                        msg = "Task added to events")325        self.assertEqual(len(self.ts.timer.current_event_actions), 1,326                                        msg = "task not added to current-event_actions")327        # Other tasks328        # add kwargs329        self.ts.schedule_task(task,330                          absolute_time = current_time + 2,331                          task_args=(1,),332                          task_kwargs=(1,))333        # change task334        self.ts.schedule_task(task2,335                          absolute_time = current_time + 2,336                          task_args=(1,))337        #change args338        self.ts.schedule_task(task,339                          absolute_time = current_time + 2,340                          task_args=(2,))341        # Change time342        self.ts.schedule_task(task,343                          absolute_time = current_time + 3,344                          task_args=(1,))345        # There are 4 different tasks for t+2346        self.assertEqual(len(self.ts.timer.events[current_time + 2]), 4,347                                        msg = "Tasks not added to events tor t+2")348        # There is one task for t+3349        self.assertEqual(len(self.ts.timer.events[current_time + 3]), 1,350                                        msg = "Task not added to events for t+3")351        # There is one task waiting for execution352        self.assertEqual(len(self.ts.timer.current_event_actions), 1,353                                        msg = "task not added to current-event_actions")354    def testStatus(self):355        """356        Test retrieval of task status357        """358        self.task_traces = [[] for i in range(4)]359        def task(task_nb):360            """361            Performs action of the tests task.362            Function records execution time on each call.363            Takes one argument, indicating ID if the task.364            """365            self.task_traces[task_nb].append(time.time())366        scheduled_time = time.time() + 1367        t1_id = self.ts.schedule_task(task,368                          absolute_time = scheduled_time,369                          task_args = (1,))370        status = self.ts.status(t1_id)371        expected_status = {'id': 0,372                            'periodic': False,373                            'period': None,374                            'scheduled_execution': scheduled_time,375                            'previous_execution': None}376        self.assertEqual(status, expected_status)377    def testScheduleTaskFail(self):378        """379        Test situations when task scheduling is expected to fail380        """381        def task():382            """383            dummy task384            """385        with self.assertRaises(ValueError,386                                msg = "Task with no exec time can be scheduled"):387            self.ts.schedule_task(task)388        with self.assertRaises(ValueError,389                                msg = "Past task can be scheduled"):390            self.ts.schedule_task(task, absolute_time = time.time()-1)391        with self.assertRaises(ValueError,392                                msg = "Task with delta and absolute time can be scheduled"):393            self.ts.schedule_task(task, absolute_time = time.time()+1, delta_t=1)394    def testSchedulingEvents(self):395        """396        Test handling of thread synchronization using threading.Event397        """398        class eventAction(threading.Thread):399            """400            Implements a test thread waiting for an event.401            Class records time of each activation for further analysis402            """403            def __init__(self, sync):404                threading.Thread.__init__(self)405                self.sync = sync406                self.counter = 0407                # helper function to execute the threads408            def run(self):409                # wait for the event for max 5 seconds.410                self.sync.wait(5)411                if self.sync.is_set():412                    self.counter += 1413                self.sync.clear()414        sync = threading.Event()415        thread1 = eventAction(sync)416        thread1.start()417        self.ts.schedule_task(sync, delta_t = 1)418        self.assertEqual(thread1.counter, 0)419        # Start the threads and wait until they complete420        thread1.join()421        self.assertEqual(thread1.counter, 1)422    def testEventsTemporalAccuracy(self):423        """424        Test temporal accuracy of thread synchronization using threading.Event425        """426        class eventAction(threading.Thread):427            """428            Implements a test thread waiting for an event.429            Class records time of each activation for further analysis430            """431            def __init__(self, thread_id, sync, time_track, max_runs = 5):432                threading.Thread.__init__(self)433                self.thread_id = thread_id434                self.sync = sync435                self.track = time_track436                self.counter = 0437                self.max_runs = max_runs438            def run(self):439                while self.counter < self.max_runs:440                    # wait for the event for max 5 seconds.441                    # Record the time of the event, or -1 if timeout occurred442                    trigger_time = -1443                    if self.sync.wait(5):444                        trigger_time = time.time()445                    self.sync.clear()446                    self.track.append(trigger_time)447                    self.counter += 1448        sync = threading.Event()449        traces = [[], []]450        thread1_repeats = 50451        thread2_repeats = 10452        thread1 = eventAction(1, sync, traces[0], thread1_repeats)453        thread2 = eventAction(2, sync, traces[1], thread2_repeats)454        event_period = 0.03455        base_time = time.time()456        t_id = self.ts.schedule_task(sync, period = event_period)457        # Start the threads and wait until they complete458        thread1.start()459        thread2.start()460        thread1.join()461        thread2.join()462        # shut down synchronization task463        self.ts.cancel_task(t_id)464        # check for timeouts465        for task in range(2):466            for events in traces[task]:467                self.assertGreaterEqual(events, 0, msg = "Timeout in thread %s" % task)468        # Convert absolute execution times to relative469        for task_trace in traces:470            for i in range(len(task_trace)):...

db_util.py

Source:db_util.py

1# DB function for zwbot2# -*- coding: utf-8 -*-3import logging4from datetime import datetime, timedelta5from google.appengine.ext import db6# DB¼ÆÊýÆ÷²Ù×÷7class Counter(db.Model):8  count = db.IntegerProperty()9  is_title = db.IntegerProperty()10  current_title = db.StringProperty(multiline=False)11  current_word = db.StringProperty(multiline=False)12  rol_word1 = db.StringProperty(multiline=False)13  rol_word2 = db.StringProperty(multiline=False)14  rol_word3 = db.StringProperty(multiline=False)15  fatal_min = db.IntegerProperty()16class DB_Utility():17  def __init__(self):18    query = db.GqlQuery("select * from Counter")19    counter = query.get()20    21    if (not counter):22      counter = Counter()23      counter.count = 024      counter.is_title = 025      counter.current_title = ''26      counter.current_word = ''27      counter.rol_word1 = ''28      counter.rol_word2 = ''29      counter.rol_word3 = ''30      counter.fatal_min = -131      counter.put()32  33  # max ÊÇxlsÎÄ¼þµÄ×î´óÐÐÊý£¬´ïµ½ºó´ÓÍ·¿ªÊ¼34  def GetIncCounter(self, max):35    query = db.GqlQuery("select * from Counter")36    counter = query.get()37    38    result = counter.count                        # È¡µÃµ±Ç°µÄ¼ÆÊý(´Ó0¿ªÊ¼)39    counter.count += 140    counter.count = counter.count % max;          # ´Ó (0) - (max-1)41    42    counter.put()43    return(result)44  45  def GetCounter(self):46    q = db.GqlQuery("select * from Counter")47    counter = q.get()48    return(counter.count)49  50  def DecCounter(self):51    q = db.GqlQuery("select * from Counter")52    counter = q.get()53    if (counter.count > 0):54      counter.count -= 155      counter.put()56  57  58  def GetTitleFlag(self):59    q = db.GqlQuery("select * from Counter")60    counter = q.get()61    return(counter.is_title)62  63  def SetTitleFlag(self, title, title_str):64    q = db.GqlQuery("select * from Counter")65    counter = q.get()66    counter.is_title = title67    if (title == 1):68      counter.current_title = title_str69    counter.put()70  71  def GetTitleString(self):72    q = db.GqlQuery("select * from Counter")73    counter = q.get()74    return(counter.current_title.encode('utf-8'))75  76  77  def GetCurrentWord(self):78    q = db.GqlQuery("select * from Counter")79    counter = q.get()80    return(counter.current_word.encode('utf-8'))81  82  def SetCurrentWord(self, str_word):83    q = db.GqlQuery("select * from Counter")84    counter = q.get()85    counter.current_word = str_word86    counter.put()87  88  def GetRollingWords(self):89    q = db.GqlQuery("select * from Counter")90    counter = q.get()91    str_word = '%s  %s  %s' % (counter.rol_word3, counter.rol_word2, counter.rol_word1)92    return(str_word.encode('utf-8'))93  94  def SetRollingWords(self, str_word):95    q = db.GqlQuery("select * from Counter")96    counter = q.get()97    counter.rol_word3 = counter.rol_word298    counter.rol_word2 = counter.rol_word199    counter.rol_word1 = str_word100    counter.put()101  102  103  # ¶ÁÐ´ÍÆËÍÊ§°ÜµÄÖÜÆÚÊý104  def GetFatalMin(self):105    q = db.GqlQuery("select * from Counter")106    counter = q.get()107    return(counter.fatal_min)108  109  def SetFatalMin(self, iMin):110    q = db.GqlQuery("select * from Counter")111    counter = q.get()112    if (counter.fatal_min != iMin):113      counter.fatal_min = iMin114      counter.put()115#116# r17 ¼Æ»®ÈÎÎñ117#118class ScheduleTaskDb(db.Model):119  year = db.IntegerProperty()120  month = db.IntegerProperty()121  day = db.IntegerProperty()122  hour = db.IntegerProperty()123  minute = db.IntegerProperty()124  msg = db.StringProperty(multiline=True)125class Db_TaskHelper():126  def get_tasks(self, now):127    schedule_task = []128    query = db.GqlQuery("select * from ScheduleTaskDb where year=:1 and month=:2 and day=:3 and hour=:4", now.year, now.month, now.day, now.hour)129    for row in query:130      schedule_task.append(row)131    return (schedule_task)132  def list_tasks(slef):133    schedule_task = []134    query = db.GqlQuery("select * from ScheduleTaskDb")135    for row in query:136      schedule_task.append(row)137    return (schedule_task)138  def add_task(self, task_date, msg):139    schedule_task = ScheduleTaskDb()140    schedule_task.year = task_date.year141    schedule_task.month = task_date.month142    schedule_task.day = task_date.day143    schedule_task.hour = task_date.hour144    schedule_task.minute = task_date.minute145    schedule_task.msg = msg.decode("utf-8")146    147    logging.debug('add new task: [%s] %s' % (task_date.strftime("%Y-%m-%d %H:%M"), msg))148    schedule_task.put()149  def clean_task(self, task_date):150    query = db.GqlQuery("select * from ScheduleTaskDb where year=:1 and month=:2 and day=:3 and hour=:4 and minute=:5", task_date.year, task_date.month, task_date.day, task_date.hour, task_date.minute)151    schedule_task = query.get()152    153    logging.debug('delete task: [%s] %s' % (task_date.strftime("%Y-%m-%d %H:%M"), schedule_task.msg))...

process.py

Source:process.py

1from abc import ABC, abstractmethod2from datetime import datetime3from models import ScheduleTask4import threading5from init import db6from my_enum import MasterValveType, PumpType, TaskStatus7from pcf8575_manager import PCF8575Manager8from utils import get_master_valve, get_pump9import time10class ProcessIF(ABC):11    def __init__(self) -> None:12        self.is_running = False13        self.is_force_stop = False14    def start(self):15        self.thread = threading.Thread(target=self._task_operation)16        self.thread.start()17    18    def stop(self):19        if self.thread and self.is_running:20            self.is_force_stop = True21            self.thread.join()22            self.is_running = False23        pass24    25    def on_running(self):26        return self.is_running27    28    @abstractmethod29    def _task_body(self):30        pass31    32    def _task_operation(self):33        self.is_running = True34        self.is_force_stop = False35        36        self._task_body()37        38        self.is_running = False39        40        41class IrrigationProcess(ProcessIF):42    pass43    def __init__(self, schedule_task : ScheduleTask) -> None:44        super().__init__()45        self.schedule_task = schedule_task46        self.pump_type = PumpType.BORE_147        self.master_valve_type = MasterValveType.MASTER_148        self.pcf_manager = PCF8575Manager.get_instance()49        50    def _post_task(self):51        self.pump.turn_off()52        self.master_valve.turn_off()53        self.schedule_task.group.close_valves()54        55        self.schedule_task.finish_time = datetime.now()56        self.schedule_task.duration = time.time() - self.start_time57        self.schedule_task.status = TaskStatus.DONE58        db.session.commit() 59            60         61    def _task_body(self):62        try:63            self.start_time = time.time()64            self.schedule_task.start_time = datetime.now()65            self.schedule_task.status = TaskStatus.RUNNING66            db.session.commit()67            self.pump = get_pump(self.pump_type)  68            self.master_valve = get_master_valve(self.master_valve_type)     69            while self.is_force_stop == False and time.time() - self.start_time < self.schedule_task.plan_duration:70                self.master_valve.turn_on()71                self.schedule_task.group.open_valves()72                self.pump.turn_on()73            74            self._post_task()75        except Exception as e:76            self._post_task()77            78        79        80        return super()._task_body()81        82        83class ProcessManager:84    85    def __init__(self) -> None:...

schedule_job_helper.py

Source:schedule_job_helper.py

1from graia.saya import Cube, Saya2from graia.scheduler.saya import SchedulerSchema3from graia.scheduler.timers import crontabify4from config import ScheduleTask5from schedule.JobNameDispatcher import JobNameDispatcher6class ScheduleJobHelper:7    def __init__(self, channel):8        self.channel = channel9        self.saya = Saya.current()10        self.cubeMap = {}11    def addScheduleJob(self, schedule_task: ScheduleTask, job_callback):12        cube = Cube(job_callback, self.__getSchemaByRules(schedule_task))13        cube.tag = schedule_task.name14        if schedule_task.name in self.cubeMap.keys():15            print(f"schedule job [{schedule_task.name}] already add,replace new one.")16            self.removeJob(schedule_task.name)17        self.cubeMap[schedule_task.name] = cube18        self.channel.content.append(cube)19        with self.saya.behaviour_interface.require_context(self.channel.module) as interface:20            interface.allocate_cube(cube)21    def removeScheduleJob(self, job_name):22        target_cube = self.cubeMap.pop(job_name)23        if target_cube:24            self.channel.cancel(target_cube.content)25            with self.saya.behaviour_interface.require_context(self.channel.module) as interface:26                interface.uninstall_cube(target_cube)27    def removeAllJob(self):28        for k in list(self.cubeMap.keys()):29            target_cube = self.cubeMap.pop(k)30            if target_cube:31                self.channel.cancel(target_cube.content)32                with self.saya.behaviour_interface.require_context(self.channel.module) as interface:33                    interface.uninstall_cube(target_cube)34    def removeJob(self, job_name):35        if job_name not in self.cubeMap.keys():36            return37        target_cube = self.cubeMap.pop(job_name)38        if target_cube:39            self.channel.cancel(target_cube.content)40            with self.saya.behaviour_interface.require_context(self.channel.module) as interface:41                interface.uninstall_cube(target_cube)42    def __getSchemaByRules(self, schedule_task: ScheduleTask):43        return SchedulerSchema(44            crontabify(schedule_task.rule),45            cancelable=True,46            dispatchers=[JobNameDispatcher(schedule_task)]...

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.