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

How to use create_and_start method in localstack

Best Python code snippet using localstack_python

test_scheduler.py

Source:test_scheduler.py

...27    yield executor28    executor.shutdown()29class TestScheduler:30    @staticmethod31    def create_and_start(dispatcher) -> Tuple[Scheduler, threading.Thread]:32        scheduler = Scheduler(executor=dispatcher)33        thread = threading.Thread(target=scheduler.run)34        thread.start()35        return scheduler, thread36    def test_single_scheduled_run(self, dispatcher):37        scheduler, thread = self.create_and_start(dispatcher)38        task = DummyTask()39        invocation_time = time.time() + 0.240        scheduler.schedule(task, start=invocation_time)41        assert poll_condition(lambda: len(task.invocations) >= 1, timeout=5)42        scheduler.close()43        thread.join(5)44        assert len(task.invocations) == 145        assert task.invocations[0][0] == 146        assert task.invocations[0][1] == pytest.approx(invocation_time, 0.1)47    def test_period_run_nonfixed(self):48        task = DummyTask()49        scheduler, thread = self.create_and_start(None)50        scheduler.schedule(task, period=0.1, fixed_rate=False)51        scheduler.schedule(scheduler.close, start=time.time() + 0.5)52        thread.join(5)53        assert task.invocations[1][1] + 0.1 == pytest.approx(task.invocations[2][1], 0.05)54        assert task.invocations[2][1] + 0.1 == pytest.approx(task.invocations[3][1], 0.05)55        assert task.invocations[3][1] + 0.1 == pytest.approx(task.invocations[4][1], 0.05)56    def test_periodic_run_fixed_with_longer_task(self):57        task = DummyTask(fn=lambda: time.sleep(1))58        scheduler, thread = self.create_and_start(None)59        scheduler.schedule(task, period=0.5, fixed_rate=True)60        scheduler.schedule(scheduler.close, start=time.time() + 1.25)61        thread.join(5)62        assert len(task.invocations) == 363        first = task.invocations[0][1]64        assert first + 0.5 == pytest.approx(task.invocations[1][1], 0.1)65        assert first + 1 == pytest.approx(task.invocations[2][1], 0.1)66        assert poll_condition(lambda: len(task.completions) >= 3, timeout=5)67    def test_periodic_change_period(self, dispatcher):68        task = DummyTask()69        scheduler, thread = self.create_and_start(dispatcher)70        stask = scheduler.schedule(task, period=1, fixed_rate=True)71        def change_period(t: ScheduledTask, period: float):72            t.period = period73        scheduler.schedule(change_period, start=time.time() + 1.25, args=(stask, 0.5))74        scheduler.schedule(scheduler.close, start=time.time() + 3)75        thread.join(5)76        first = task.invocations[0][1]77        second = task.invocations[1][1]78        third = task.invocations[2][1]79        fourth = task.invocations[3][1]80        assert first + 1 == pytest.approx(second, 0.1)81        assert second + 1 == pytest.approx(third, 0.1)82        # changed to 0.583        assert third + 0.5 == pytest.approx(fourth, 0.1)84    def test_cancel_task(self, dispatcher):85        task1 = DummyTask()86        task2 = DummyTask()87        scheduler, thread = self.create_and_start(dispatcher)88        scheduler.schedule(task2.invoke, period=0.5)89        stask = scheduler.schedule(task1.invoke, period=0.5)90        scheduler.schedule(stask.cancel, start=time.time() + 0.75)91        scheduler.schedule(scheduler.close, start=time.time() + 1.5)92        thread.join(5)93        assert len(task1.invocations) == 294        assert len(task2.invocations) == 495    def test_error_handler(self):96        scheduler = Scheduler()97        event = threading.Event()98        def invoke():99            raise ValueError("unittest")100        def on_error(e):101            event.set()102        scheduler.schedule(invoke, on_error=on_error)103        scheduler.schedule(scheduler.close)104        scheduler.run()105        assert event.wait(5)106    def test_scheduling_reordering(self, dispatcher):107        task = DummyTask()108        scheduler, thread = self.create_and_start(dispatcher)109        t = time.time()110        scheduler.schedule(task, args=("task2",), start=t + 1)  # task two gets scheduled first111        time.sleep(0.25)112        scheduler.schedule(113            task, args=("task1",), start=t + 0.5114        )  # but task one has the shorter deadline115        scheduler.schedule(scheduler.close, start=t + 1.5)116        thread.join(5)117        assert len(task.invocations) == 2118        assert task.invocations[0][2][0] == "task1"119        assert task.invocations[1][2][0] == "task2"120    def test_close_interrupts_waiting_tasks(self, dispatcher):121        task = DummyTask()122        scheduler, thread = self.create_and_start(dispatcher)123        scheduler.schedule(task, start=time.time() + 1)124        time.sleep(0.25)125        scheduler.close()126        thread.join(5)...

Timer.py

Source:Timer.py

1##############################################################################2#  Copyright (c) 2016 Intel Corporation3# 4# Licensed under the Apache License, Version 2.0 (the "License");5#  you may not use this file except in compliance with the License.6#  You may obtain a copy of the License at7# 8#      http://www.apache.org/licenses/LICENSE-2.09# 10#  Unless required by applicable law or agreed to in writing, software11#  distributed under the License is distributed on an "AS IS" BASIS,12#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.13#  See the License for the specific language governing permissions and14#  limitations under the License.15##############################################################################16#    File Abstract: 17#    Implements a Timer.  You create a timer with an ID, start it, stop it, pause etc.18#    Call to Timer() takes 2 params - an ID and an action19#    Actions:20#       get - returns elapsed seconds21#       create - creates a timer, but does not start it22#       create_and_start - creates and starts timer23#       get_auto_create - creates, starts and gets a timer24#       stop - stops a timer25#       start - starts a timer, resumes ones that is paused, if stopped it resets counter26#       pause - pauses a timer, 'start' resumes27#28##############################################################################29import time30class TimerInfo:31    def __init__(self,strID):32        self.__ID = strID33        self.__Active = False34        self.__Paused = False35        self.__StartTime = 036        self.__PausedTime = 037    def __GetCurrentTime():38        return  int(round(time.time() )) 39    def getElapsedTime(self):40        if self.__Active:41            if not self.__Paused:42                return str(TimerInfo.__GetCurrentTime() - self.__StartTime)43            return str(self.__PausedTime - self.__StartTime)44        else:45            return "Timer [" + self.__ID +"] not started"46    def Start(self):47        if not self.__Active:48            self.__Active = True49        if self.__Paused:50            self.__Paused = False51            self.__PausedTime = 052        else:53            self.__StartTime = TimerInfo.__GetCurrentTime()54    def Pause(self):55        if self.__Active and not self.__Paused:56            self.__Paused = True57            self.__PausedTime = TimerInfo.__GetCurrentTime()58    def Stop(self):59        if self.__Active:60            self.__Active = False61            if self.__Paused:62                self.__Paused = False63                self.__PausedTime = 064# Actions include 'create','create_and_start','get_auto_create', 'get', 'start','stop','pause'65def Timer(ID,Action):66    if not hasattr(Timer, "timerMap"):67        Timer.timerMap = {}     68    Action = str(Action)69    _action = Action.lower()70    _ID = ID.lower()71    72    if _action == "create":73        if _ID in Timer.timerMap.keys():74            return "Timer with ID: " + ID + " already exists."75        timerState = TimerInfo(_ID)76        Timer.timerMap[_ID] = timerState77        return True78    if _action == "create_and_start":79        retVal = Timer(ID,"create")80        if retVal != True:81             return retVal82        timerState = Timer.timerMap[_ID]83        Timer(ID,"start")84        return Timer(ID,"get")85    if _action == "get_auto_create":86        if not _ID in Timer.timerMap.keys():87            Timer("create_and_start",ID)88        return Timer(ID,"get")89    if not _ID in Timer.timerMap.keys():90       return "Timer [" + ID +"] does not exist.  Can't perform action :" + Action91    92    timerState = Timer.timerMap[_ID]93    if _action == "get":94        return timerState.getElapsedTime()95    if _action == "start":96        return timerState.Start()97    if _action == "stop":98        return timerState.Stop()99    if _action == 'pause':100        return timerState.Pause()...

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.