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

How to use wait_for_result method in localstack

Best Python code snippet using localstack_python

scheduling.py

Source:scheduling.py

1import logging2import numbers3import queue4import sys5import threading6import time7from .settings import process_jobs_max_batch_time8# This is the job queue. Add to it via jobs.put or call_on_main_thread.9# The main thread calls process_jobs every editor update (or every 2s in10# standalone mode) to process all outstanding jobs and then wait a bit longer11# for some fast replies.12_jobs = queue.Queue()13#####################14# This connection delays some dispatching until the main thread gets to it.15# That's necessary for accessing some Unity objects. It's also a source of deadlocks,16# so we try to avoid delays if possible.17def call_on_main_thread(f, wait_for_result = True):18    """19    Call a function on the main thread.20    21    If wait_for_result is True, will block until the main thread processes it 22    and return the value, or raise the exception it raised.23    If wait_for_result is False, then None is returned and exceptions will not 24    be raised25    """26    if wait_for_result and threading.current_thread() is threading.main_thread():27        # Only execute (and block) if we're on the main thread and want to get 28        # the result/raise exceptions. Otherwise we just queue the job29        return f()30    condition = threading.Condition()31    return_value = []32    exception = []33    def job():34        with condition:35            try:36                return_value.append(f())37            except:38                # Hand the exception to the other thread.39                exception.append(sys.exc_info()[1])40                if not wait_for_result:41                    # raise the exception on the main thread as no one is 42                    # waiting on it43                    raise44            condition.notify()45    with condition:46        _jobs.put(job)47        if wait_for_result:48            condition.wait()49            if len(exception):50                raise exception[0]51            else:52                return return_value[0]53        54        return None55def process_jobs(batch_time = process_jobs_max_batch_time):56    """57    Call this from the main loop on every editor update.58    If there are any jobs to process, process them all and keep processing, or 59    wait for jobs for `batch_time` seconds, giving threads time to send60    another request quickly if needed. The implication of this is at the last 61    milisecond, a job with a run time of 10 hours can be processed. The corollary62    is that a batch_time lower that the length of a job can be used to make sure 63    we process only one job.64    """65    if not isinstance(batch_time, numbers.Number):66        raise TypeError("'batch_time' argument must be numeric")67    68    # The main thread always holds the GIL. Explicitly yield with a time.sleep69    # so other threads can push jobs in the jobs queue.70    time.sleep(0.001)71    if _jobs.empty():72        return73    start = time.time()74    remaining = batch_time75    while remaining > 0:76        try:77            job = _jobs.get(timeout=remaining)78            try:79                job()80            except Exception as e:81                msg = f"An unexpected Exception occured while processing the job {job}: {e}"82                UnityEngine.Debug.LogException(msg)83                print(msg)84            _jobs.task_done()85        except queue.Empty:86            break87        elapsed = (time.time() - start)88        remaining = batch_time - elapsed89def process_all_jobs():90    while not _jobs.empty():91        process_jobs()92def make_exec_on_main_thread_decorator(wait_for_result):93    def decorator(f):94        """95        Decorator that will queue a job (function) for execution on the main96        thread.97        If wait_for_result is true then the function will return and throw98        exceptions normally.99        If it's false then the function will return None immediately after100        queueing the job, and exceptions will not propagate.101        """102        def func_wrapper(*args, **kwargs):103            call_on_main_thread(lambda: f(*args,**kwargs), wait_for_result=wait_for_result)104        return func_wrapper105    return decorator106exec_on_main_thread = make_exec_on_main_thread_decorator(wait_for_result = True)...

top_nav_goal.py

Source:top_nav_goal.py

...11    # send goal to move from Waypoint 3 to Waypoint 13, and back to the initial waypoint, WayPoint112    goal = GotoNodeGoal()13    goal.target = "WayPoint3"14    client.send_goal(goal)15    client.wait_for_result() 16    goal = GotoNodeGoal()17    goal.target = "WayPoint5"18    client.send_goal(goal)19    client.wait_for_result()20    goal = GotoNodeGoal()21    goal.target = "WayPoint7"22    client.send_goal(goal)23    client.wait_for_result()24    goal = GotoNodeGoal()25    goal.target = "WayPoint9"26    client.send_goal(goal)27    client.wait_for_result()28    goal = GotoNodeGoal()29    goal.target = "WayPoint10"30    client.send_goal(goal)31    client.wait_for_result()32    goal = GotoNodeGoal()33    goal.target = "WayPoint11"34    client.send_goal(goal)35    client.wait_for_result()36    goal = GotoNodeGoal()37    goal.target = "WayPoint12"38    client.send_goal(goal)39    client.wait_for_result()40    goal = GotoNodeGoal()41    goal.target = "WayPoint13"42    client.send_goal(goal)43    client.wait_for_result()44    goal = GotoNodeGoal()45    goal.target = "WayPoint10"46    client.send_goal(goal)47    client.wait_for_result()48    goal = GotoNodeGoal()49    goal.target = "WayPoint1"50    client.send_goal(goal)51    client.wait_for_result()...

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.