How to use pop_runnable_tasks method in Lemoncheesecake

Best Python code snippet using lemoncheesecake

task.py

Source:task.py Github

copy

Full Screen

...39 if self._tasks_aborted:40 return "tests have been manually stopped"41 else:42 return None43def pop_runnable_tasks(remaining_tasks, completed_tasks, nb_tasks):44 runnable_tasks = list(45 itertools.islice(46 filter(lambda t: set(t.get_all_dependencies()).issubset(completed_tasks), remaining_tasks),47 0, nb_tasks48 )49 )50 for task in runnable_tasks:51 remaining_tasks.remove(task)52 _debug("pop runnable task %s" % task)53 yield task54def run_task(task, context, completed_task_queue):55 _debug("run task %s" % task)56 try:57 task.run(context)58 except TaskFailure as excp:59 task.result = TaskResultFailure(str(excp))60 except Exception:61 task.result = TaskResultException(serialize_current_exception())62 else:63 task.result = TaskResultSuccess()64 completed_task_queue.put(task)65def handle_task(task, context, completed_task_queue):66 _debug("handle task %s" % task)67 # skip task on dependency failure if any68 for dep_task in task.get_on_success_dependencies():69 if not isinstance(dep_task.result, TaskResultSuccess):70 if isinstance(dep_task.result, (TaskResultFailure, TaskResultSkipped)):71 reason = dep_task.result.reason72 else:73 reason = None74 skip_task(task, context, completed_task_queue, reason)75 return76 # skip task on external trigger if any77 skip_reason = context.is_task_to_be_skipped(task)78 if skip_reason:79 skip_task(task, context, completed_task_queue, reason=skip_reason)80 return81 # run task when all conditions are met82 run_task(task, context, completed_task_queue)83def schedule_tasks_to_be_run(tasks, context, pool, completed_tasks_queue):84 for task in tasks:85 pool.apply_async(handle_task, args=(task, context, completed_tasks_queue))86def skip_task(task, context, completed_task_queue, reason=""):87 _debug("skip task %s" % task)88 try:89 task.skip(context, reason)90 except Exception:91 task.result = TaskResultException(serialize_current_exception())92 else:93 task.result = TaskResultSkipped(reason)94 completed_task_queue.put(task)95def skip_all_tasks(tasks, remaining_tasks, completed_tasks, context, pool, completed_tasks_queue, reason):96 # schedule all tasks to be skipped...97 for task in remaining_tasks:98 pool.apply_async(skip_task, args=(task, context, completed_tasks_queue, reason))99 # ... and wait for their completion100 while len(completed_tasks) != len(tasks):101 completed_task = completed_tasks_queue.get()102 completed_tasks.add(completed_task)103def run_tasks(tasks, context, nb_threads=1):104 for task in tasks:105 check_task_dependencies(task)106 remaining_tasks = list(tasks)107 completed_tasks = set()108 pool = Pool(nb_threads)109 completed_tasks_queue = Queue()110 try:111 schedule_tasks_to_be_run(112 pop_runnable_tasks(remaining_tasks, completed_tasks, nb_threads),113 context, pool, completed_tasks_queue114 )115 while len(completed_tasks) != len(tasks):116 # wait for one task to complete117 completed_task = completed_tasks_queue.get()118 completed_tasks.add(completed_task)119 # schedule tasks to be run waiting for task success or simple completion120 tasks_to_be_run = pop_runnable_tasks(remaining_tasks, completed_tasks, nb_threads)121 schedule_tasks_to_be_run(tasks_to_be_run, context, pool, completed_tasks_queue)122 except KeyboardInterrupt:123 context.enable_task_abort()124 skip_all_tasks(125 tasks, remaining_tasks, completed_tasks, context, pool, completed_tasks_queue,126 _KEYBOARD_INTERRUPT_ERROR_MESSAGE127 )128 finally:129 pool.close()130 exceptions = [task.result.stacktrace for task in tasks if isinstance(task.result, TaskResultException)]131 if exceptions:132 raise LemoncheesecakeException(133 "Error(s) while running tasks, got exceptions:\n%s" % "\n".join(exceptions)134 )...

Full Screen

Full Screen

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.

LambdaTest Learning Hubs:

YouTube

You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.

Run Lemoncheesecake automation tests on LambdaTest cloud grid

Perform automation testing on 3000+ real desktop and mobile devices online.

Try LambdaTest Now !!

Get 100 minutes of automation test minutes FREE!!

Next-Gen App & Browser Testing Cloud

Was this article helpful?

Helpful

NotHelpful